xuelong/bxh
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add customer service API deployment support

Browse Source
This commit is contained in:
3452078359-xuexue
2026-06-10 10:48:26 +08:00
parent 0594fc9f8c
commit e589073311
8 changed files with 777 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

350
app/api/travel_assistant.py
View File

@@ -9,7 +9,7 @@ import time
from typing import Any
from falkordb import FalkorDB
from fastapi import APIRouter, Depends, HTTPException
from fastapi import APIRouter, Depends, Header, HTTPException
from app.auth import CurrentUser
from app.config import settings
@@ -22,6 +22,7 @@ router = APIRouter()
ENTERPRISE_AGENT_TARGET_MS = 1200
ENTERPRISE_AGENT_WARN_MS = 2500
GRAPH_CACHE_TTL_SECONDS = 120
BAIXINGHUI_GRAPH_NAME = "baixinghui_travel_agency"
_GRAPH_DATA_CACHE: dict[tuple[str, str], tuple[float, Any]] = {}
ATTRACTION_ALIASES: dict[str, list[str]] = {
@@ -230,6 +231,54 @@ def _cached_fee_bindings(graph_name: str) -> list[dict[str, Any]]:
return _cached_graph_data(graph_name, "fee_bindings", _load_fee_bindings)
def _configured_api_keys() -> set[str]:
raw = str(getattr(settings, "ingest_api_keys", "") or "")
return {item.strip() for item in re.split(r"[,;\s]+", raw) if item.strip()}
def _bearer_token(value: str | None) -> str:
text = (value or "").strip()
if text.lower().startswith("bearer "):
return text[7:].strip()
return ""
async def _require_customer_service_api_key(
x_kg_api_key: str | None = Header(default=None, alias="X-KG-API-Key"),
x_api_key: str | None = Header(default=None, alias="X-API-Key"),
authorization: str | None = Header(default=None, alias="Authorization"),
) -> str:
keys = _configured_api_keys()
if not keys:
return "api-key-disabled"
candidate = (x_kg_api_key or x_api_key or _bearer_token(authorization)).strip()
if candidate and candidate in keys:
return candidate
raise HTTPException(status_code=401, detail="缺少或无效的客服接口 API Key")
def _normalize_customer_service_graph(value: Any) -> str:
text = str(value or "").strip()
if not text:
return BAIXINGHUI_GRAPH_NAME
lower = text.lower()
if lower in {"bxh", "baixinghui", "baixinghui_travel", "baixinghui_travel_agency"}:
return BAIXINGHUI_GRAPH_NAME
if "百姓惠" in text or "baixinghui" in lower:
return BAIXINGHUI_GRAPH_NAME
return text
def _external_bool(value: Any, default: bool = False) -> bool:
if value is None:
return default
if isinstance(value, bool):
return value
if isinstance(value, (int, float)):
return bool(value)
return str(value).strip().lower() in {"1", "true", "yes", "y", "on", "启用", "开启"}
def _extract_intent(question: str) -> dict[str, Any]:
q = question.strip()
party_size = None
@@ -552,7 +601,7 @@ def _normalize_intent(question: str, fallback: dict[str, Any], data: dict[str, A
return _complete_intent_defaults(question, merged)
async def _travel_intent_client() -> LlmClient | None:
async def _travel_llm_client(max_tokens: int = 1000) -> LlmClient | None:
cfg = await get_agent_settings()
global_cfg = cfg.get("global") or {}
if global_cfg.get("base_url") and global_cfg.get("api_key"):
@@ -561,7 +610,7 @@ async def _travel_intent_client() -> LlmClient | None:
global_cfg["api_key"],
global_cfg.get("model") or "deepseek-chat",
timeout=int(global_cfg.get("timeout") or 45),
max_tokens=1000,
max_tokens=max_tokens,
)
extract = cfg.get("extract") or {}
@@ -575,11 +624,15 @@ async def _travel_intent_client() -> LlmClient | None:
model_cfg["api_key"],
model_cfg.get("model") or "deepseek-chat",
timeout=int(extract.get("timeout") or 60),
max_tokens=1000,
max_tokens=max_tokens,
)
return None
async def _travel_intent_client() -> LlmClient | None:
return await _travel_llm_client(max_tokens=1000)
TRAVEL_INTENT_SYS = """你是旅行社客服行程推荐的自然语言需求解析器。只输出 JSON。
从客户原话中提取用于知识图谱检索的旅行需求，不要生成行程。
字段：
@@ -615,6 +668,204 @@ async def _llm_intent(question: str, fallback: dict[str, Any], enabled: bool) ->
return fallback, "llm_failed_rule_fallback"
CUSTOMER_SERVICE_FUSION_SYS = """你是百姓惠旅行社智能客服问答融合器。只输出 JSON。
你的任务是把知识图谱召回结果整理成可以直接回复客户的话术。
规则：
1. 只能使用 knowledge_base 中的线路、费用、酒店、餐饮、车辆、证据和 trace，不得编造未出现的价格、余位、房型、政策。
2. 如价格、余位、车辆、酒店房型、餐标、门票、小交通或天气政策不确定，必须说明需要按团期/供应商/景区政策二次核实。
3. 客户问“多少钱/报价/费用”时，优先说明图谱里的报价口径和计算边界；没有明确数字时不要补数字。
4. 客户问线路推荐时，优先给 1-3 个命中的线路或方案，并说明适合点。
5. 客户问资源替换或附近酒店餐饮时，说明是候选资源，不代表已锁定。
6. 语言要像真实客服，简洁、自然、可复制。
输出字段：
- answer: 给调用系统展示的完整回答
- customer_reply: 可直接发给客户的回复，和 answer 可相同但更口语
- confidence: 0 到 1 的数字
- next_questions: 2 到 4 个建议追问
- risk_notes: 0 到 4 条不能承诺或需核实的风险提示
"""
def _list_texts(value: Any, limit: int = 4, text_limit: int = 120) -> list[str]:
if not isinstance(value, list):
return []
return [_value(item, text_limit) for item in value[:limit] if _value(item, text_limit)]
def _summarize_named_items(value: Any, limit: int = 4) -> list[dict[str, str]]:
if not isinstance(value, list):
return []
items: list[dict[str, str]] = []
for item in value[:limit]:
if not isinstance(item, dict):
text = _value(item, 140)
if text:
items.append({"name": text, "summary": ""})
continue
name = _value(
item.get("name")
or item.get("hotel_name")
or item.get("restaurant_name")
or item.get("vehicle_name")
or item.get("title"),
80,
)
summary = _value(item.get("summary") or item.get("detail") or item.get("description") or item.get("price_text"), 180)
if name or summary:
items.append({"name": name, "summary": summary})
return items
def _summarize_plan_for_llm(plan: dict[str, Any]) -> dict[str, Any]:
days: list[dict[str, str]] = []
for day in (plan.get("daily_itinerary") or [])[:6]:
if not isinstance(day, dict):
continue
days.append({
"day": _value(day.get("day_index") or day.get("day") or "", 20),
"route": _value(day.get("from_to") or day.get("title") or "", 120),
"activity": _value(day.get("activity") or day.get("route_summary") or "", 180),
"hotel": _value(day.get("accommodation") or "", 100),
"meals": _value(day.get("meals") or day.get("meal_plan") or "", 140),
})
costs: list[dict[str, str]] = []
for item in (plan.get("cost_breakdown") or [])[:6]:
if isinstance(item, dict):
costs.append({
"category": _value(item.get("category"), 60),
"detail": _value(item.get("detail") or item.get("summary"), 180),
})
return {
"plan_name": _value(plan.get("plan_name") or plan.get("name"), 140),
"rank": plan.get("rank"),
"fit_score": plan.get("fit_score"),
"duration_days": plan.get("duration_days"),
"duration_nights": plan.get("duration_nights"),
"route_summary": _value(plan.get("route_summary"), 600),
"quote_summary": _value(plan.get("quote_summary") or plan.get("variant_summary"), 500),
"match_reasons": _list_texts(plan.get("match_reasons"), limit=5, text_limit=80),
"daily_itinerary": days,
"cost_breakdown": costs,
"hotels": _summarize_named_items(plan.get("hotels"), limit=4),
"restaurants": _summarize_named_items(plan.get("restaurants"), limit=4),
"vehicles": _summarize_named_items(plan.get("vehicles"), limit=4),
"policies": _summarize_named_items(plan.get("policies"), limit=4),
}
def _summarize_evidence_for_llm(item: dict[str, Any]) -> dict[str, str]:
return {
"type": _value(item.get("type"), 50),
"name": _value(item.get("name") or item.get("title"), 120),
"summary": _value(item.get("summary") or item.get("detail") or item.get("description"), 260),
"source": _value(item.get("source"), 120),
}
def _knowledge_pack_for_llm(agent_response: dict[str, Any]) -> dict[str, Any]:
trace = agent_response.get("trace") or {}
return {
"graph_name": agent_response.get("graph_name"),
"intent": agent_response.get("intent") or {},
"kg_answer_draft": _value(agent_response.get("copy_text"), 2400),
"plans": [
_summarize_plan_for_llm(plan)
for plan in (agent_response.get("plans") or [])[:5]
if isinstance(plan, dict)
],
"evidence": [
_summarize_evidence_for_llm(item)
for item in (agent_response.get("evidence") or [])[:12]
if isinstance(item, dict)
],
"sales_scripts": _summarize_named_items(agent_response.get("sales_scripts"), limit=4),
"follow_up_questions": _list_texts(agent_response.get("follow_up_questions"), limit=5, text_limit=120),
"trace": {
"response_mode": trace.get("response_mode"),
"response_mode_label": trace.get("response_mode_label"),
"method": trace.get("method"),
"quality_summary": trace.get("quality_summary") or {},
"retrieval_summary": trace.get("retrieval_summary") or {},
"data_gap": trace.get("data_gap"),
},
}
def _fallback_customer_service_answer(
agent_response: dict[str, Any],
source: str,
error: str | None = None,
) -> dict[str, Any]:
text = _value(agent_response.get("copy_text"), 3000)
if not text:
text = "当前知识图谱没有命中足够的线路或资源信息，建议先补充客户出行时间、人数、天数和必去景点后再核价。"
risk_notes = ["价格、余位、房型、车辆、门票和景区政策以团期及供应商二次核实为准。"]
trace = agent_response.get("trace") or {}
for check in (trace.get("quality_checks") or [])[:4]:
if isinstance(check, dict) and check.get("status") in {"warn", "fail"}:
note = _value(check.get("detail") or check.get("label"), 140)
if note and note not in risk_notes:
risk_notes.append(note)
return {
"answer": text,
"customer_reply": text,
"confidence": 0.68 if agent_response.get("plans") else 0.42,
"next_questions": _list_texts(agent_response.get("follow_up_questions"), limit=4, text_limit=120),
"risk_notes": risk_notes[:4],
"llm_used": False,
"fusion_source": source,
"llm_error": error,
}
def _confidence(value: Any, default: float = 0.72) -> float:
try:
return max(0.0, min(1.0, float(value)))
except Exception:
return default
async def _fuse_customer_service_answer(
question: str,
agent_response: dict[str, Any],
enabled: bool,
) -> dict[str, Any]:
if not enabled:
return _fallback_customer_service_answer(agent_response, "kg_template")
client = await _travel_llm_client(max_tokens=1600)
if not client:
return _fallback_customer_service_answer(agent_response, "kg_template_no_llm_configured")
payload = {
"question": question,
"knowledge_base": _knowledge_pack_for_llm(agent_response),
}
try:
data = await asyncio.to_thread(
client.chat_json,
CUSTOMER_SERVICE_FUSION_SYS,
json.dumps(payload, ensure_ascii=False),
)
except Exception as exc: # noqa: BLE001
return _fallback_customer_service_answer(agent_response, "kg_template_llm_failed", str(exc)[:220])
answer = _value(data.get("answer") or data.get("customer_reply"), 3000)
reply = _value(data.get("customer_reply") or answer, 2200)
if not answer:
return _fallback_customer_service_answer(agent_response, "kg_template_empty_llm_answer")
return {
"answer": answer,
"customer_reply": reply or answer,
"confidence": _confidence(data.get("confidence")),
"next_questions": _list_texts(data.get("next_questions"), limit=4, text_limit=120),
"risk_notes": _list_texts(data.get("risk_notes"), limit=4, text_limit=160),
"llm_used": True,
"fusion_source": "kg_llm_fusion",
"llm_error": None,
}
TASK_LABELS = {
"route_price": "线路报价",
"route_catalog": "线路清单",
@@ -2064,10 +2315,11 @@ def _enrich_agent_response(response: dict[str, Any], started_at: float) -> dict[
trace["agent_design"] = {
"current_test_goal": ["图查询命中质量", "客服回答可用性", "企业级响应速度", "报价/余位不可承诺边界"],
"future_api_contract": {
"endpoint": "POST /v1/admin/travel/assistant-query",
"endpoint": "POST /v1/admin/travel/customer-service-query",
"internal_test_endpoint": "POST /v1/admin/travel/assistant-query",
"request": ["question", "graph_name", "session_id/customer_context", "use_llm"],
"response": ["copy_text", "plans", "evidence", "follow_up_questions", "trace"],
"handoff": "客服系统消费结构化结果；trace 用于质检、审计和图查询优化。",
"response": ["answer", "customer_reply", "plans", "evidence", "follow_up_questions", "trace"],
"handoff": "外部客服系统消费融合回答和结构化证据；trace 用于质检、审计和图查询优化。",
},
}
return response
@@ -5023,3 +5275,87 @@ async def travel_assistant_query(
},
}
return _enrich_agent_response(response, started_at)
@router.post("/travel/customer-service-query")
async def travel_customer_service_query(
body: dict,
_api_key: str = Depends(_require_customer_service_api_key),
):
question = str(body.get("question") or body.get("text") or body.get("query") or "").strip()
if not question:
raise HTTPException(status_code=400, detail="请输入客户咨询内容")
graph_name = _normalize_customer_service_graph(
body.get("graph_name")
or body.get("knowledge_graph")
or body.get("kg")
or body.get("graph")
or BAIXINGHUI_GRAPH_NAME
)
context = ProjectContext(
tenant_id=str(body.get("tenant_id") or "travel_agency").strip(),
project_id=str(body.get("project_id") or "baixinghui_travel_agency").strip(),
graph_name=graph_name,
)
agent_body = dict(body)
agent_body["question"] = question
agent_body["graph_name"] = graph_name
agent_body["use_llm"] = _external_bool(body.get("use_llm"), True)
agent_response = await travel_assistant_query(
agent_body,
context=context,
_user={"username": "customer_service_api", "roles": ["api"]},
)
fusion = await _fuse_customer_service_answer(
question,
agent_response,
_external_bool(body.get("llm_fusion"), True),
)
trace = agent_response.get("trace") or {}
response: dict[str, Any] = {
"status": "ok",
"service": "baixinghui_customer_service",
"api_version": "2026-06-10",
"session_id": _value(body.get("session_id"), 120),
"question": question,
"graph_name": graph_name,
"answer": fusion["answer"],
"customer_reply": fusion["customer_reply"],
"confidence": fusion["confidence"],
"follow_up_questions": fusion.get("next_questions") or [],
"risk_notes": fusion.get("risk_notes") or [],
"knowledge": {
"plans": agent_response.get("plans") or [],
"evidence": agent_response.get("evidence") or [],
"sales_scripts": agent_response.get("sales_scripts") or [],
"quality_summary": trace.get("quality_summary") or {},
"retrieval_summary": trace.get("retrieval_summary") or {},
},
"routing": {
"intent": agent_response.get("intent") or {},
"response_mode": trace.get("response_mode"),
"response_mode_label": trace.get("response_mode_label"),
"planned_tasks": trace.get("planned_tasks") or [],
"method": trace.get("method"),
"intent_method": trace.get("intent_method"),
},
"llm": {
"intent_requested": bool((agent_response.get("intent") or {}).get("llm_requested")),
"intent_used": bool((agent_response.get("intent") or {}).get("llm_used")),
"fusion_used": bool(fusion.get("llm_used")),
"fusion_source": fusion.get("fusion_source"),
"error": fusion.get("llm_error"),
},
"trace": {
"latency_ms": trace.get("latency_ms"),
"speed": trace.get("speed") or {},
"quality_summary": trace.get("quality_summary") or {},
"retrieval_summary": trace.get("retrieval_summary") or {},
"graph_capabilities_used": trace.get("graph_capabilities_used") or [],
},
}
if _external_bool(body.get("return_raw_agent") or body.get("debug"), False):
response["raw_agent_response"] = agent_response
return response