"""Anonymous customer-service orchestration without private customer access.""" from __future__ import annotations import json import logging import re from inspect import isawaitable from rag.intent import Intent, IntentResult logger = logging.getLogger(__name__) class QueryTooLongError(ValueError): pass class DataQueryRejected(ValueError): """NL2SQL 数据查询被拒绝（未开放、无权限或配额不足），message 可直接回复用户。""" async def _config(config_getter, key: str, default: str): value = config_getter(key, default) if isawaitable(value): value = await value return value or default async def _maybe_await(value): return await value if isawaitable(value) else value class AnonymousCustomerAgent: def __init__( self, *, context, rag_retrieve, intent_recognize, generate_answer, audit_writer, config_getter, data_query=None, ): self.context = context self.rag_retrieve = rag_retrieve self.intent_recognize = intent_recognize self.generate_answer = generate_answer self.audit_writer = audit_writer self.config_getter = config_getter # 可选 NL2SQL 数据查询依赖：签名 data_query(*, question, customer_id, # session_id, trace_id) -> dict；匿名 runtime 不装配（None），行为不变。 self.data_query = data_query async def handle( self, session_id: str, query: str, *, trace_id: str, customer_id: int | None = None, ) -> dict: if len(query) > 2000: raise QueryTooLongError("query长度不能超过2000字符") # 先取历史再写入当前问题，保证意图识别拿到的历史不含本轮输入；取不到历史不阻断请求 try: history = await self.context.get(session_id) except Exception: logger.exception("load conversation history failed: session_id=%s", session_id) history = [] await self.context.append(session_id, "user", query) if self._contains_sensitive_input(query): await _maybe_await(self.audit_writer( action="anon_sensitive_input", trace_id=trace_id, session_id=session_id, )) recognized = await _maybe_await(self.intent_recognize(query, history)) if isinstance(recognized, IntentResult): intent, search_query = recognized.intent, recognized.query else: intent, search_query = recognized, query sources = [] data_query_meta = None if intent == Intent.GUIDE_PURCHASE: answer = await _config( self.config_getter, "agent.customer.template.guide_purchase", "请前往开户页面办理。", ) elif intent == Intent.WANT_ADVISOR: answer = await _config( self.config_getter, "agent.customer.template.guide_advisor", "如需基金推荐，请联系投资顾问。", ) elif intent == Intent.OFF_TOPIC: answer = await _config( self.config_getter, "agent.customer.template.off_topic", "我是华夏科技的智能客服，只能解答基金与公司业务相关的问题，" "您可以问我基金知识、开户流程或公司信息～", ) elif intent == Intent.TRANSFER_HUMAN: answer = await _config( self.config_getter, "agent.customer.template.transfer_human", "好的，正在为您转接人工客服，请稍候。" "您也可以拨打官方客服热线 400-XXX-XXXX 直接联系我们。", ) elif intent == Intent.CHITCHAT: answer = await self._chitchat(session_id) elif intent == Intent.NL2SQL_REQUEST: if self.data_query is None or customer_id is None: # 匿名会话或未装配数据查询能力：引导登录，不触发任何数据库查询 answer = await _config( self.config_getter, "agent.customer.template.nl2sql_unavailable", "数据查询功能需要登录后使用，请先登录再来问我您的持仓和交易信息～", ) else: answer, sources, data_query_meta = await self._run_data_query( question=search_query, customer_id=customer_id, session_id=session_id, trace_id=trace_id, ) elif intent in (Intent.KNOWLEDGE_QA, Intent.COMPANY_INFO): try: # 用补全指代后的问题检索，省略主语的追问才能命中 sources = await _maybe_await(self.rag_retrieve(search_query, None)) except Exception: sources = [] if not sources: answer = await _config( self.config_getter, "agent.customer.template.fallback_human", "当前未找到匹配信息，请转人工客服。", ) else: messages = await self.context.get(session_id) prompt = messages + [ { "role": "system", "content": "仅根据提供的知识来源回答，不得编造基金推荐。", }, { "role": "system", "content": f"知识来源：{json.dumps(sources, ensure_ascii=False)}", }, ] try: answer = await _maybe_await(self.generate_answer(prompt)) except Exception: answer = await _config( self.config_getter, "agent.customer.template.fallback_human", "当前服务繁忙，请转人工客服。", ) else: answer = await _config( self.config_getter, "agent.customer.template.fallback_human", "当前未找到匹配信息，请转人工客服。", ) await self.context.append(session_id, "assistant", answer) result = { "answer": answer, "sources": sources, "intent": intent.value, "rewritten_query": search_query, "trace_id": trace_id, } if data_query_meta is not None: result["data_query"] = data_query_meta return result async def _run_data_query( self, *, question: str, customer_id: int, session_id: str, trace_id: str, ) -> tuple[str, list, dict]: """调用注入的 NL2SQL 数据查询能力，失败时统一降级为客服话术。""" try: payload = await _maybe_await( self.data_query( question=question, customer_id=customer_id, session_id=session_id, trace_id=trace_id, ) ) except DataQueryRejected as exc: return str(exc), [], None except Exception: logger.exception( "client data query failed: trace_id=%s session_id=%s customer_id=%s", trace_id, session_id, customer_id, ) answer = await _config( self.config_getter, "agent.customer.template.nl2sql_fallback", "暂时无法完成数据查询，请稍后再试或联系人工客服。", ) return answer, [], None if not isinstance(payload, dict) or not str(payload.get("answer") or "").strip(): return await _config( self.config_getter, "agent.customer.template.nl2sql_fallback", "暂时无法完成数据查询，请稍后再试或联系人工客服。", ), [], None meta = { key: payload[key] for key in ("query_id", "row_count", "truncated", "chart") if payload.get(key) is not None } return str(payload["answer"]), list(payload.get("sources") or []), meta or None async def _chitchat(self, session_id: str) -> str: """带对话历史调用 LLM 做受限闲聊，失败时退回固定话术。""" messages = await self.context.get(session_id) prompt = [ { "role": "system", "content": ( "你是华夏科技（基金代销金融机构）的智能客服助手。" "用户正在与你寒暄，请用一两句话简短、友好地回应，" "并自然地引导用户咨询基金知识、开户流程或公司信息。" "不得推荐任何基金产品，不得谈论具体收益，不得回答金融之外的实质性问题。" ), }, *messages, ] try: return await _maybe_await(self.generate_answer(prompt)) except Exception: return await _config( self.config_getter, "agent.customer.template.chitchat_fallback", "您好，我是华夏科技的智能客服，很高兴为您服务！" "您可以问我基金知识、开户流程或公司信息～", ) @staticmethod def _contains_sensitive_input(query: str) -> bool: return bool( re.search(r"(?