lzf_0626
c564494f84
实测三连问：「季季盈90天起投多少」→「C1 客户能买它吗」→「那我能买它吗」，
第三问转人工。原因是 _topic_of 不认识适当性回答的格式——它那一行是
"南方季季盈90天为 R2（中低风险），与您当前的风险测评结果不匹配…"，
既没有冒号也不是标题行，于是被判成"取不出主语"，第三问就失去了指代对象。
补上这种格式：产品名在"为 R{1-5}"之前，且必须在冒号判断**之前**匹配（那一行没有冒号）。
以"为 R2"开头的行说明前面没有产品名，返回空串，而不是把等级本身当成主语。
修好后第三问给出与第二问一致的结论，这是对的：裁决是确定性的，同一个客户问同一只
产品，结论就该一样。新增两条单测钉住这两种情形。
513 lines
26 KiB
Python
513 lines
26 KiB
Python
"""客服 Agent：只回答能溯源到公司资料的问题，答不了就引导客户致电人工客服。
|
||
|
||
设计取向（金融场景，由业务方确定）：
|
||
|
||
- **确定性优先**：命中知识块后**直接返回原文**，不经模型改写。答案的字面内容来自公司
|
||
已发布的资料，模型不参与事实生成，因此不存在"编一个看起来合理的答案"的通路。
|
||
- **答不了就引导**：检索降级、未命中、置信度不足、意图未覆盖、模型异常——一律返回
|
||
引导客户拨打客服热线的固定话术，并置 `transfer_required=True` 留痕；
|
||
绝不用模型猜测答案。
|
||
- **本体保持薄**：只有意图分发与四条出口，不做多轮推理、不自主决策。这是业务方明确
|
||
要求的取向——"客服 Agent 本身不会很多内容，不会的就转人工"。
|
||
|
||
四条出口：
|
||
`faq` → 检索直返（不经模型）｜`product_inquiry` / `policy_explain` → 检索直返 + 来源引用
|
||
｜`chitchat` → 模型生成（提示词走发布配置）｜其余与异常 → 引导人工客服
|
||
"""
|
||
|
||
from typing import Any
|
||
|
||
from app.core.contracts import (
|
||
AgentDefinition,
|
||
AgentRequest,
|
||
CoreResult,
|
||
RequestContext,
|
||
SourceReference,
|
||
)
|
||
from app.core.errors import ForbiddenAgentError
|
||
from app.service.agent.base import BaseAgent
|
||
from app.service.model_gateway import DatabaseModelEndpointResolver
|
||
from app.service.runtime_config_service import load_active_prompt
|
||
|
||
AGENT_TYPE = "customer_service"
|
||
|
||
# 意图码必须三处对齐：AgentDefinition.supported_intents、agent_intent_config 的
|
||
# (agent_type, intent_code)、以及发布版 config_release 里 agent_tools 的
|
||
# `customer_service:<intent>` 白名单 key。缺任一处即失败关闭，这是底座的有意设计。
|
||
INTENT_FAQ = "faq"
|
||
INTENT_PRODUCT = "product_inquiry"
|
||
INTENT_POLICY = "policy_explain"
|
||
INTENT_SUITABILITY = "suitability_check"
|
||
INTENT_CHITCHAT = "chitchat"
|
||
INTENT_TRANSFER = "transfer_human"
|
||
BUSINESS_INTENTS = (INTENT_FAQ, INTENT_PRODUCT, INTENT_POLICY, INTENT_SUITABILITY)
|
||
|
||
TOOL_NAME = "search_knowledge"
|
||
# 适当性裁决工具。为什么必须走它而不是自己比大小：客户的风险等级只有底座能给出权威值
|
||
# （来自 fin_risk_assessment，且带测评有效期），Agent 自己判分等于绕开合规链路。
|
||
SUITABILITY_TOOL = "check_suitability"
|
||
|
||
# 风险等级名称：国标五级，稳定不变，只用于把裁决结果说成人话。
|
||
RISK_LEVEL_NAMES = {
|
||
1: "R1（低风险）", 2: "R2（中低风险）", 3: "R3（中风险）",
|
||
4: "R4（中高风险）", 5: "R5（高风险）",
|
||
}
|
||
|
||
# 三档置信阈值：方案 §2.3 要求的是「绝对阈值 AND（相对间隙 OR 分布优势）」混合判定，
|
||
# 只做绝对阈值会把口语化问法误判成"答不了"（这是实测踩到的坑）。
|
||
#
|
||
# 数值按本模型（qwen3.7-text-embedding-flash）**实测校准**，不是照搬经验值：
|
||
# · 库内问法：口语「我们公司叫什么名字」top1=0.592、标准「公司全称是什么」0.671、
|
||
# 带品牌名「南方科技的全称」0.855 —— 同样的正确答案，口语问法相似度天然更低；
|
||
# · 库外/越界：「你们公司什么时候上市」top1 最高 0.500、「今天天气怎么样」0.416，
|
||
# 且这些问题的 top1 与次优间隙都在 0.046 以内，而库内命中普遍在 0.09 以上。
|
||
# 于是：库内最低 0.579 / 库外最高 0.500，绝对分两侧都有余量；间隙 0.07 又能挡住
|
||
# 「存在并列候选」的不确定情形，避免因为一次高相似度的巧合就硬答。
|
||
HIGH_SCORE = 0.75 # ≥ 直接答（高置信不再要求间隙：分数已足够说明问题）
|
||
MID_SCORE = 0.55 # 需与 MIN_GAP 同时满足才答，答时附"信息可能不完整"提示
|
||
MIN_GAP = 0.07 # top1 领先次优的最小间隙；领先不足说明有并列候选，不硬答
|
||
|
||
TOP_K = 5
|
||
MAX_ANSWER_CHARS = 1200
|
||
REFERENCE_LIMIT = 3
|
||
|
||
COMPANY = "南方科技"
|
||
# 客服热线：正式号码确定后改这里（或改为读配置项，避免改代码）
|
||
HOTLINE = "400-XXX-XXXX"
|
||
SERVICE_HOURS = "每日 7:00-22:00"
|
||
|
||
FALLBACK_TEMPLATE = (
|
||
"抱歉，这个问题我暂时无法给出准确答复。为避免给您错误信息，"
|
||
f"建议您拨打客服热线 {HOTLINE}（{SERVICE_HOURS}）转人工客服咨询。"
|
||
)
|
||
# 客户侧只展示这一句固定话术（业务方确定）。原中置信的"信息可能不完整"提示已按此移除，
|
||
# 即中置信回答不再对客户标注不确定性——这是调整时知情的取舍，不是遗漏。
|
||
DISCLAIMER = "（以上内容由智能客服依据公司公开资料整理，不构成投资建议）"
|
||
|
||
CHITCHAT_PROMPT_CODE = "customer_service_chitchat"
|
||
CHITCHAT_TASK_TYPE = "chat"
|
||
DEFAULT_CHITCHAT_SYSTEM = (
|
||
f"你是{COMPANY}的智能客服助手。回应要简短、礼貌，并自然引导用户提出与基金、理财、"
|
||
"账户相关的问题。禁止承诺收益，禁止出现「保本」「稳赚」「无风险」「保证收益」"
|
||
"「预期收益率」「年化收益率」「安全」等表述。"
|
||
)
|
||
DEFAULT_CHITCHAT_TEMPLATE = "用户说：{message}\n请用不超过 40 字回应，并把话题引导到业务上。"
|
||
|
||
|
||
class CustomerServiceAgent(BaseAgent):
|
||
definition = AgentDefinition(
|
||
agent_type=AGENT_TYPE,
|
||
version="1.0.0",
|
||
allowed_roles=("customer",),
|
||
allowed_portals=("api",),
|
||
# 代码上限：实际可用范围由发布配置的意图白名单收窄（两者取交集）
|
||
allowed_tools=(TOOL_NAME, SUITABILITY_TOOL),
|
||
supported_intents=(
|
||
INTENT_FAQ, INTENT_PRODUCT, INTENT_POLICY, INTENT_SUITABILITY,
|
||
INTENT_CHITCHAT, INTENT_TRANSFER,
|
||
),
|
||
)
|
||
|
||
async def handle(self, request: AgentRequest, context: RequestContext) -> CoreResult:
|
||
intent = self._intent_code()
|
||
if intent == INTENT_CHITCHAT:
|
||
return await self._chitchat(request)
|
||
if intent == INTENT_TRANSFER:
|
||
return self._guide_to_human("客户主动要求转人工")
|
||
if intent not in BUSINESS_INTENTS:
|
||
# 分类失败或意图未覆盖：不猜，直接引导人工
|
||
return self._guide_to_human(f"意图未覆盖：{intent or '未识别'}")
|
||
if intent == INTENT_SUITABILITY:
|
||
# 适当性裁决是唯一会给出"能不能买"结论的出口，走独立实现：
|
||
# 它要组合「产品风险等级 + 客户档案等级」，不是知识检索能算出来的
|
||
return await self._answer_suitability(request, context)
|
||
return await self._answer_from_knowledge(request, context, intent)
|
||
|
||
# ---- 出口一：知识直返（faq / 产品 / 政策） ----
|
||
|
||
async def _answer_from_knowledge(
|
||
self, request: AgentRequest, context: RequestContext, intent: str
|
||
) -> CoreResult:
|
||
try:
|
||
output = await self.call_tool(
|
||
TOOL_NAME,
|
||
{"query": self._search_query(request), "top_k": TOP_K},
|
||
intent=intent,
|
||
context=context,
|
||
)
|
||
except ForbiddenAgentError:
|
||
# 白名单/权限类失败必须冒泡：那是配置错误，若被兜底话术吞掉，
|
||
# 运维会看到"客服一直引导人工"却查不出原因。
|
||
raise
|
||
except Exception:
|
||
return self._guide_to_human("知识检索调用失败")
|
||
|
||
if not isinstance(output, dict):
|
||
return self._guide_to_human("知识检索返回格式异常")
|
||
if output.get("degraded"):
|
||
reason = str(output.get("reason") or "未知")
|
||
return self._guide_to_human(f"知识检索降级：{reason}")
|
||
hits = output.get("hits")
|
||
if not isinstance(hits, list) or not hits:
|
||
return self._guide_to_human("知识库未命中")
|
||
|
||
best = hits[0]
|
||
if not isinstance(best, dict):
|
||
return self._guide_to_human("命中内容格式异常")
|
||
score = self._score(best.get("score"))
|
||
gap = score - self._second_score(hits)
|
||
# 混合判定：高置信直接答；中置信必须同时满足「分数够」与「领先次优够多」。
|
||
# 只满足其一的（分数够但两三个候选并驾齐驱）宁可引导人工——金融场景下
|
||
# "答不了"可接受，"答错"不可接受。
|
||
confident = score >= HIGH_SCORE
|
||
if not confident and not (score >= MID_SCORE and gap >= MIN_GAP):
|
||
return self._guide_to_human(f"置信度不足：score={score:.3f} gap={gap:.3f}")
|
||
|
||
# 命中的是行级子块时，先分辨客户问的是"某个字段"还是"整个产品"：
|
||
# 子块让「起投多少」拿到聚焦答案，但「介绍一下」会被某一行抢答。
|
||
best = self._prefer_section(request.message, best, hits)
|
||
content = str(best.get("content") or "").strip()
|
||
if not content:
|
||
return self._guide_to_human("命中内容为空")
|
||
|
||
# 正文只保留「答案 + 固定免责声明」：业务方要求客户侧只看到这一句固定话术，
|
||
# 因此不确定性提示与出处行都不再出现在正文里。
|
||
#
|
||
# 可追溯性不受影响：本次命中哪个知识块仍由审计（agent.tool_executed 的工具调用记录）
|
||
# 与消息表留痕，只是不面向客户展示。若将来要把出处给客户看，应当走
|
||
# source_references 的 knowledge 类型（需先让 ToolExecutor 登记本次可引用的 doc_id），
|
||
# 而不是继续往正文里拼字符串。
|
||
return CoreResult(
|
||
text=f"{content[:MAX_ANSWER_CHARS]}\n{DISCLAIMER}",
|
||
intent=self._classified_intent,
|
||
)
|
||
|
||
@staticmethod
|
||
def _prefer_section(message: str, best: dict[str, Any], hits: list[Any]) -> dict[str, Any]:
|
||
"""客户问整个产品时，用整节块替换掉抢答的那一行。
|
||
|
||
行级子块是为了让「起投多少」拿到聚焦答案，但「介绍一下」会被某一行抢答
|
||
（实测返回了"产品期限 90天封闭期"，而客户要的是整个产品）。
|
||
|
||
判据不用问句分类器，而是看问句与子块标签是否真的对得上。标签取自子块的
|
||
section 末段（"起投金额""风险等级"）：「起投多少」含"起投"、「风险高吗」含"风险"，
|
||
都算对得上；「介绍一下」与任何标签都不重合，说明客户要的是整节。
|
||
|
||
父块由检索层按子块分数的 0.9 折算后一并带回。万一没带回来就仍用子块——
|
||
宁可答得窄一点，也不要拿不相干的块去搪塞。
|
||
"""
|
||
doc_id = str(best.get("doc_id") or "")
|
||
# 末段恰为 2 位数字才是行级子块（PROD-007-04）；整节块自己的编号形如 PROD-901，
|
||
# 用"含连字符"判断会把整节块误判成子块。
|
||
parent_id, _, tail = doc_id.rpartition("-")
|
||
if not (parent_id and tail.isdigit() and len(tail) == 2):
|
||
return best # 命中的本来就是整节
|
||
# 标签取自子块 title 的末段（title 由 " · " 连接），如"起投金额""风险等级"
|
||
label = str(best.get("title") or "").split(" · ")[-1].strip()
|
||
if label and label[:2] in message:
|
||
return best # 客户问的正是这个字段
|
||
for hit in hits:
|
||
if isinstance(hit, dict) and str(hit.get("doc_id") or "") == parent_id:
|
||
return hit
|
||
return best
|
||
|
||
# ---- 出口一之二：适当性裁决（唯一给出"能不能买"结论的出口） ----
|
||
|
||
async def _answer_suitability(
|
||
self, request: AgentRequest, context: RequestContext
|
||
) -> CoreResult:
|
||
"""回答「以我的风险等级能不能买这只产品」。
|
||
|
||
为什么不能靠知识检索直接答：客户问「c1客户能买它吗」，答案是**两个事实的组合**
|
||
——该产品的风险等级（R2）与客户档案里的等级能不能匹配。检索只能给出"最像的那段
|
||
原文"，实测给的是 C1 的通用规则，答非所问。
|
||
|
||
三条硬约束，缺任何一条都转人工：
|
||
1. 产品风险等级**从知识库查出来**，不猜、也不采信问句里出现的"R2"字样；
|
||
2. 客户等级**由底座按档案解析**（check_suitability 内部读 fin_risk_assessment，
|
||
带测评有效期），**不采信客户自称**——这次实测里客户说"C1"，档案其实是 C2；
|
||
3. 产品名**只取上一轮回答里的主语**（`_topic_of`，它来自知识块字段，可信）。
|
||
客户第一句就直接问"XX 能买吗"时取不到，那就转人工：这个出口会给出"能不能买"
|
||
的结论，宁可答不了也不能答错。
|
||
"""
|
||
product = self._previous_topic(request)
|
||
if not product:
|
||
return self._guide_to_human("适当性问题里没识别出具体产品")
|
||
risk_level = await self._product_risk_level(product, context)
|
||
if risk_level is None:
|
||
return self._guide_to_human(f"未查到「{product}」的风险等级")
|
||
try:
|
||
decision = await self.call_tool(
|
||
SUITABILITY_TOOL,
|
||
{"customer_id": context.user_id, "product_risk_level": risk_level},
|
||
intent=INTENT_SUITABILITY,
|
||
context=context,
|
||
)
|
||
except ForbiddenAgentError:
|
||
# 与知识检索一致：白名单/权限类失败必须冒泡，那是配置错误，
|
||
# 被兜底话术吞掉的话运维只会看到"客服一直引导人工"却查不出原因
|
||
raise
|
||
except Exception:
|
||
return self._guide_to_human("适当性校验调用失败")
|
||
if not isinstance(decision, dict):
|
||
return self._guide_to_human("适当性校验返回格式异常")
|
||
return CoreResult(
|
||
text=f"{self._suitability_text(product, risk_level, decision)}\n{DISCLAIMER}",
|
||
intent=self._classified_intent,
|
||
)
|
||
|
||
@classmethod
|
||
def _previous_topic(cls, request: AgentRequest) -> str:
|
||
"""上一轮回答里的主语（产品名）；取不到返回空串。"""
|
||
for turn in reversed(request.history):
|
||
if turn.role == "assistant":
|
||
return cls._topic_of(turn.content)
|
||
return ""
|
||
|
||
async def _product_risk_level(self, product: str, context: RequestContext) -> int | None:
|
||
"""查产品的风险等级：问知识库要"风险等级"那一行，不从问句里猜。
|
||
|
||
产品手册里每个产品都有一行"风险等级 R2（中低风险）"，切分后是独立的行级子块，
|
||
所以按「{产品名} 风险等级」检索能直接命中。两重校验缺一不可：必须命中**风险等级
|
||
行**（否则可能匹配到"C1 可购买 R1、R2"那种列举，把 R1 当成产品等级），而且该行
|
||
必须属于**同一个产品**（否则会拿另一个产品的等级去做裁决）。
|
||
"""
|
||
try:
|
||
output = await self.call_tool(
|
||
TOOL_NAME,
|
||
{"query": f"{product} 风险等级", "top_k": 3},
|
||
intent=INTENT_SUITABILITY,
|
||
context=context,
|
||
)
|
||
except ForbiddenAgentError:
|
||
raise
|
||
except Exception:
|
||
return None
|
||
if not isinstance(output, dict) or output.get("degraded"):
|
||
return None
|
||
hits = output.get("hits")
|
||
if not isinstance(hits, list):
|
||
return None
|
||
for hit in hits:
|
||
if not isinstance(hit, dict):
|
||
continue
|
||
title = str(hit.get("title") or "")
|
||
content = str(hit.get("content") or "")
|
||
if "风险等级" not in title and "风险等级" not in content:
|
||
continue
|
||
if product not in title and product not in content:
|
||
continue
|
||
for level in range(1, 6):
|
||
if f"R{level}" in content:
|
||
return level
|
||
return None
|
||
|
||
@staticmethod
|
||
def _suitability_text(product: str, risk_level: int, decision: dict[str, Any]) -> str:
|
||
"""把裁决结果说成人话。
|
||
|
||
只说裁决本身与依据，不复述产品资料——客户问的是"我能不能买"，资料在前一问
|
||
已经给过了。拒绝对原因下断言：reason_code 可能是等级不匹配、测评过期或未测评，
|
||
统一说成"超出风险承受能力"是错的；需要签揭示书时也**不写具体持仓比例**，
|
||
那是豁免条款里的业务参数，让客户照着一个数字去操作容易出偏差，留给人工讲。
|
||
"""
|
||
level_name = RISK_LEVEL_NAMES.get(risk_level, f"R{risk_level}")
|
||
customer_level = decision.get("customer_risk_level")
|
||
if isinstance(customer_level, int):
|
||
lines = [f"您当前的风险测评等级为 C{customer_level}。"]
|
||
else:
|
||
# 档案里没有在有效期内的测评结果。这是合规上的"不能卖"，但话要说清楚是
|
||
# "还没测评/已过期"，不能写成"您的等级为无"这种客户看不懂的句子。
|
||
lines = ["您目前没有在有效期内的风险测评结果。"]
|
||
if decision.get("allowed"):
|
||
lines.insert(0, f"{product}为 {level_name}，在您的风险承受能力范围内，可以购买。")
|
||
if decision.get("required_disclosure"):
|
||
lines.append("购买前需签署产品风险揭示书，具体请咨询您的客户经理。")
|
||
if decision.get("requires_recording"):
|
||
lines.append("本次购买需进行双录（录音录像）。")
|
||
else:
|
||
lines.insert(
|
||
0,
|
||
f"{product}为 {level_name}，与您当前的风险测评结果不匹配，"
|
||
"按照投资者适当性管理规定暂时无法购买。",
|
||
)
|
||
lines.append("如需了解具体原因或申请重新测评，请联系您的客户经理或拨打客服热线。")
|
||
valid_until = str(decision.get("assessment_valid_until") or "")
|
||
if valid_until:
|
||
lines.append(f"风险测评有效期至 {valid_until[:10]}，过期需重新测评。")
|
||
return "\n".join(lines)
|
||
|
||
# ---- 出口二：闲聊（提示词走发布配置） ----
|
||
|
||
async def _chitchat(self, request: AgentRequest) -> CoreResult:
|
||
system, template = await self._chitchat_prompt()
|
||
message = request.message[:500]
|
||
try:
|
||
prompt = template.format(message=message, company=COMPANY)
|
||
except (KeyError, IndexError, ValueError):
|
||
# 发布配置里的占位符与代码不一致时回落默认模板：配置写错不应该让运行期崩
|
||
system = DEFAULT_CHITCHAT_SYSTEM
|
||
prompt = DEFAULT_CHITCHAT_TEMPLATE.format(message=message, company=COMPANY)
|
||
full_prompt = f"{system}\n\n{prompt}" if system else prompt
|
||
try:
|
||
endpoints = await DatabaseModelEndpointResolver().resolve(
|
||
agent_type=AGENT_TYPE, task_type="text_generation"
|
||
)
|
||
# 显式转成 list[object]：generate_with_model 形参是 list[object]，
|
||
# 而 list 不变型（invariant），直接传 list[ModelEndpointConfig] 过不了类型检查
|
||
endpoint_list: list[object] = list(endpoints)
|
||
execution = await self.generate_with_model(endpoint_list, full_prompt)
|
||
except Exception:
|
||
return self._guide_to_human("模型不可用")
|
||
text = (execution.text or "").strip()
|
||
if not text:
|
||
return self._guide_to_human("模型返回为空")
|
||
return CoreResult(
|
||
text=f"{text[:MAX_ANSWER_CHARS]}\n{DISCLAIMER}",
|
||
intent=self._classified_intent,
|
||
)
|
||
|
||
async def _chitchat_prompt(self) -> tuple[str, str]:
|
||
"""读取当前发布版本的闲聊提示词；未发布或读取失败时回落代码内置默认值。"""
|
||
try:
|
||
row = await load_active_prompt(CHITCHAT_PROMPT_CODE, CHITCHAT_TASK_TYPE, AGENT_TYPE)
|
||
except Exception:
|
||
row = None
|
||
if row is None:
|
||
return DEFAULT_CHITCHAT_SYSTEM, DEFAULT_CHITCHAT_TEMPLATE
|
||
return (
|
||
row.system_prompt or DEFAULT_CHITCHAT_SYSTEM,
|
||
row.user_prompt_template or DEFAULT_CHITCHAT_TEMPLATE,
|
||
)
|
||
|
||
# 客户这一句里出现这些词，说明主语要靠上文补全（"这个产品""该基金"）。
|
||
# 刻意不收"它/他"：单字指代在中文里极易误伤（"其他产品""其它"都含"他/它"），
|
||
# 而这类短句已经由 _MIN_STANDALONE_CHARS 覆盖，不需要靠它兜。
|
||
_REFERRING_WORDS = (
|
||
"这个", "那个", "这款", "这只", "该产品", "该基金", "上述", "前面提", "刚刚说",
|
||
)
|
||
# 短到这种程度的问题，自己通常不构成完整意图（"起投多少""风险高吗"）
|
||
_MIN_STANDALONE_CHARS = 8
|
||
|
||
@staticmethod
|
||
def _topic_of(answer: str) -> str:
|
||
"""从客服上一轮的回答里取出"这一轮在说哪个产品"，取不出来就返回空串。
|
||
|
||
回答是我们自己组装的，只有两种形状，都取自知识块的字段：
|
||
- 行级块：`南方季季盈90天：起投金额 1万元`（首个冒号前就是产品名）
|
||
- 整节块：`### 2.1 南方季季盈90天`（首个标题行去掉 # 号）
|
||
|
||
取不出来时宁可返回空、退化成不带主语，也不要拿兜底话术当主语：
|
||
「抱歉，这个问题我暂时无法给出准确答复」这种句子拿去检索只会把问题带偏。
|
||
"""
|
||
first = next((line.strip() for line in answer.splitlines() if line.strip()), "")
|
||
first = first.lstrip("#").strip()
|
||
# 适当性回答的首行是「{产品名}为 R2（中低风险），与您当前的风险测评结果不匹配…」，
|
||
# 产品名在"为 R"之前。这一条必须在冒号判断**之前**：那一行里没有冒号。
|
||
for level in range(1, 6):
|
||
index = first.find(f"为 R{level}")
|
||
if index >= 0:
|
||
candidate = first[:index].strip()
|
||
return candidate if 0 < len(candidate) <= 20 else ""
|
||
topic = first.split("：", 1)[0].strip() if "：" in first else first
|
||
# FAQ 型回答的首行是"问：C1 客户能买什么产品?"，冒号前只有一个"问"字；
|
||
# 拿它当主语只会把检索词污染成"问 那它风险高吗"。
|
||
if not topic or topic in {"问", "答"}:
|
||
return ""
|
||
if len(topic) > 20 or "，" in topic or "。" in topic:
|
||
return ""
|
||
return topic
|
||
|
||
@classmethod
|
||
def _search_query(cls, request: AgentRequest) -> str:
|
||
"""构造交给检索的查询串。
|
||
|
||
什么时候带上文（实测决定，两条都不能少）：
|
||
|
||
1. **客户这一句自己说不清楚时**才带。同一会话先问「季季盈90天的起投金额是多少」、
|
||
再问「基金赎回几天到账」，若无条件带上文，第二问会命中季季盈的产品块、
|
||
答出产品介绍——**答非所问**。在金融场景里这比"引导转人工"糟得多。
|
||
2. **带上文时只带"在说哪个产品"，不带上一轮的原话**。把上一轮整句拼进来会让
|
||
检索词语义变"宽"，反而只能命中粗粒度的整节块：实测「季季盈90天起投多少
|
||
那它风险高吗」命中的是整个产品小节，客户问的"风险"完全没有被聚焦；
|
||
换成「南方季季盈90天 那它风险高吗」才命中"风险等级"那一行。
|
||
|
||
换句话说：上文的作用是**补主语**，不是**补内容**。
|
||
|
||
只取客户的话、不取 Agent 自己的回答当内容：把 Agent 的措辞也拼进来会让检索偏向
|
||
自己上一轮的说法，而客户的真实意图可能已经在下一句里被修正过。这里从回答里取的
|
||
只有产品名这一个"主语"，不是它的论述。
|
||
"""
|
||
message = request.message.strip()
|
||
needs_context = (
|
||
len(message) < cls._MIN_STANDALONE_CHARS
|
||
or any(word in message for word in cls._REFERRING_WORDS)
|
||
)
|
||
if not needs_context:
|
||
return message[:500]
|
||
for turn in reversed(request.history):
|
||
if turn.role == "assistant":
|
||
topic = cls._topic_of(turn.content)
|
||
if topic:
|
||
return f"{topic} {message}"[:500]
|
||
break
|
||
return message[:500]
|
||
|
||
# ---- 出口三：引导人工客服（不做工单，只回话并留痕） ----
|
||
|
||
def _guide_to_human(self, reason: str) -> CoreResult:
|
||
return CoreResult(
|
||
text=FALLBACK_TEMPLATE,
|
||
intent=self._classified_intent,
|
||
transfer_required=True,
|
||
transfer_reason=reason[:200],
|
||
)
|
||
|
||
# ---- 辅助 ----
|
||
|
||
def _intent_code(self) -> str:
|
||
classified = self._classified_intent
|
||
return classified.intent if classified is not None else ""
|
||
|
||
@staticmethod
|
||
def _score(value: object) -> float:
|
||
"""把命中分数夹到 [0,1]：SourceReference.score 有 ge=0/le=1 约束。"""
|
||
try:
|
||
number = float(value) # type: ignore[arg-type]
|
||
except (TypeError, ValueError):
|
||
return 0.0
|
||
return min(1.0, max(0.0, number))
|
||
|
||
def _second_score(self, hits: list[object]) -> float:
|
||
"""次优命中分数，用于「绝对阈值 + 相对间隙」的混合判定。
|
||
|
||
只有一个命中时返回 0：此时间隙最大，是否回答交由绝对阈值把关，
|
||
而不是仅凭"没有竞争者"就认定可信。
|
||
"""
|
||
if len(hits) < 2 or not isinstance(hits[1], dict):
|
||
return 0.0
|
||
return self._score(hits[1].get("score"))
|
||
|
||
def _references(self, hits: list[object]) -> tuple[SourceReference, ...]:
|
||
"""保留待用：等基座支持 knowledge 类型引用后再启用。
|
||
|
||
目前 `governance.review_output` 只认可 memory / tool 两类来源，直接返回
|
||
knowledge 引用会被判为「引用未来自本次已授权召回结果」而让整个 run 失败。
|
||
启用前提是让 ToolExecutor 把工具返回的知识 doc_id 登记为本次可引用来源。
|
||
"""
|
||
references: list[SourceReference] = []
|
||
for hit in hits[:REFERENCE_LIMIT]:
|
||
if not isinstance(hit, dict):
|
||
continue
|
||
doc_id = str(hit.get("doc_id") or "").strip()
|
||
if not doc_id:
|
||
continue # 没有标识的命中无法回溯，丢弃而不是造一个假来源
|
||
title = str(hit.get("title") or "").strip()
|
||
references.append(SourceReference(
|
||
source_type="knowledge",
|
||
source_id=doc_id,
|
||
title=title[:200] or None,
|
||
score=self._score(hit.get("score")),
|
||
))
|
||
return tuple(references)
|