diff --git a/app/service/agent/implementations/customer_service.py b/app/service/agent/implementations/customer_service.py index 3d5ca89..7b22636 100644 --- a/app/service/agent/implementations/customer_service.py +++ b/app/service/agent/implementations/customer_service.py @@ -159,9 +159,16 @@ FALLBACK_TEMPLATE = ( "抱歉，这个问题我暂时无法给出准确答复。为避免给您错误信息，" f"建议您拨打客服热线 {HOTLINE}（{SERVICE_HOURS}）转人工客服咨询。" ) -# 客户侧只展示这一句固定话术（业务方确定）。原中置信的"信息可能不完整"提示已按此移除， -# 即中置信回答不再对客户标注不确定性——这是调整时知情的取舍，不是遗漏。 -DISCLAIMER = "（以上内容由智能客服依据公司公开资料整理，不构成投资建议）" +# 免责声明**只由治理层注入**（`PlatformGovernance.review` → `review_output`，话术取自 +# `agent_reply_template` 的 `TPL_DISCLAIMER`，取不到时退回 `FALLBACK_DISCLAIMER`）。 +# +# 这里曾自行拼一句 `DISCLAIMER`，合并后与治理层的话术**同时出现**，客户会看到两条 +# 意思重复的声明的（实测复现）。更重要的是分工问题：话术是合规文案，属于**发布配置**， +# 改文案不该改代码；Agent 自己拼等于把可配置的合规文案硬编码进业务逻辑， +# 而且治理层无法判断"业务是不是已经加过了"（它只认自己追加过的那个形状）。 +# 因此本文件不再定义、也不再引用任何免责声明常量。 + +# 客服热线：正式号码确定后改这里（或改为读配置项，避免改代码） CHITCHAT_PROMPT_CODE = "customer_service_chitchat" CHITCHAT_TASK_TYPE = "chat" @@ -291,15 +298,15 @@ class CustomerServiceAgent(BaseAgent): if not content: return self._guide_to_human("命中内容为空") - # 正文只保留「答案 + 固定免责声明」：业务方要求客户侧只看到这一句固定话术， - # 因此不确定性提示与出处行都不再出现在正文里。 + # 正文只保留答案本身：固定免责声明由**治理层**统一追加（见文件头 `DISCLAIMER` 说明）， + # 业务代码不再拼字符串——否则会出现两条重复声明，且合规文案变成不可配置的硬编码。 # # 可追溯性不受影响：本次命中哪个知识块仍由审计（agent.tool_executed 的工具调用记录） # 与消息表留痕，只是不面向客户展示。若将来要把出处给客户看，应当走 # source_references 的 knowledge 类型（需先让 ToolExecutor 登记本次可引用的 doc_id）， # 而不是继续往正文里拼字符串。 return CoreResult( - text=f"{content[:MAX_ANSWER_CHARS]}\n{DISCLAIMER}", + text=content[:MAX_ANSWER_CHARS], intent=self._classified_intent, ) @@ -383,7 +390,7 @@ class CustomerServiceAgent(BaseAgent): f"风险测评结果为准，不以本次自述为准。\n{text}" ) return CoreResult( - text=f"{text}\n{DISCLAIMER}", + text=text, intent=self._classified_intent, ) @@ -505,7 +512,7 @@ class CustomerServiceAgent(BaseAgent): if not text: return self._guide_to_human("模型返回为空") return CoreResult( - text=f"{text[:MAX_ANSWER_CHARS]}\n{DISCLAIMER}", + text=text[:MAX_ANSWER_CHARS], intent=self._classified_intent, ) diff --git a/app/service/knowledge_search_service.py b/app/service/knowledge_search_service.py index 36e00f2..1889751 100644 --- a/app/service/knowledge_search_service.py +++ b/app/service/knowledge_search_service.py @@ -19,11 +19,45 @@ PRODUCT_COLLECTION = "fin_product_collection" POLICY_COLLECTION = "fin_policy_collection" DEFAULT_COLLECTIONS: tuple[str, ...] = (FAQ_COLLECTION, PRODUCT_COLLECTION, POLICY_COLLECTION) -# 检索输出字段：与灌库脚本写入的 schema 对齐 -OUTPUT_FIELDS = ( - "doc_id", "title", "content", "chapter", "section", - "tags", "doc_no", "version", "source_file", "visibility", -) +# 检索输出字段：**按现库集合的真实 schema**（2026-09-11 实测 describe_collection）。 +# +# ⚠️ 这里与灌库脚本 `tools/load_knowledge_milvus.py` 的设计字段**不一致**，是刻意的： +# 那个脚本自述"临时脚本，跑完即删"，所需的 `knowledge/_chunks.jsonl` 不在仓库里， +# 它那套 schema（`doc_id` 主键 + `chapter`/`section`/`doc_no`/`visibility`）在本环境 +# **从未建起来过**。现库三个集合的真实字段是： +# knowledge_id / title / snippet / tags / version / intent / embedding +# 字段名不匹配的代价是**静默零召回**：Milvus 会对不存在的字段直接报错， +# 三个集合全失败 → `degraded=True` → 客服一律"引导人工"，知识问答整体失效（实测复现）。 +# +# 对齐策略：**只改读取侧**，不改检索逻辑与 `KnowledgeHit` 的对外形状—— +# · `doc_id` ← `knowledge_id`（`KnowledgeHit.doc_id` 的名字保持不动， +# 下游 Agent 与 `test_knowledge_keyword_recall` 都依赖它） +# · `content` ← `snippet`（现库正文存在 snippet 字段） +# · `title` / `version` 同名直取 +# · `chapter`/`section`/`doc_no`/`source_file`/`visibility` 现库没有 → 留空/默认， +# 依赖它们的增强逻辑（父子块选择、来源编号）自然退化为"不启用"而不会报错。 +# +# 若将来按灌库脚本的 schema 重建集合并回填，把下面的 `_FIELD_ALIASES` 改回 +# `{"doc_id": "doc_id", "content": "content", ...}` 并把 `_HAS_VISIBILITY` 置 True 即可， +# 检索逻辑一行都不用动。 +_FIELD_ALIASES: dict[str, str] = { + "doc_id": "knowledge_id", + "title": "title", + "content": "snippet", + "tags": "tags", + "version": "version", + "intent": "intent", + # 现库未落这些字段 → 显式不请求，读取处按缺省值处理 +} + +#: 现库集合是否有 `visibility` 字段。没有时**不能**再拼 `visibility == "public"`： +#: 该表达式会让 Milvus 报 "field visibility not exist"，整次检索失败。 +_HAS_VISIBILITY = False + +#: 现库集合是否有 `source_file` 字段（没有时来源标题只能靠 `doc_no`，通常为空）。 +_HAS_SOURCE_FILE = False + +OUTPUT_FIELDS = tuple(_FIELD_ALIASES.values()) # 关键字精确召回：客户问到产品名这类**专有名词**时，字面匹配比相似度更确定。 # @@ -149,7 +183,13 @@ class KnowledgeSearchService: return KnowledgeSearchOutcome(degraded=True, reason="embedding_empty") targets = tuple(collections or self._collections) - expression = None if include_internal else 'visibility == "public"' + # 可见性过滤只在集合真有该字段时拼：现库没有 `visibility`，拼上去会让 Milvus + # 报 "field visibility not exist"，三个集合全失败 → 一律走兜底（实测复现）。 + # 代价必须说清楚：`_HAS_VISIBILITY=False` 时**没有**检索层的内部资料硬隔离， + # 只能靠入库侧只放对外知识来保证（现库 356 行均为对外知识，已核对）。 + expression = ( + None if (include_internal or not _HAS_VISIBILITY) else 'visibility == "public"' + ) collected: list[KnowledgeHit] = [] failures = 0 for collection in targets: @@ -263,14 +303,14 @@ class KnowledgeSearchService: rows = lookup( collection_name=PRODUCT_COLLECTION, filter=expression, - output_fields=["doc_id", "title"], + output_fields=[_FIELD_ALIASES["doc_id"], _FIELD_ALIASES["title"]], limit=KEYWORD_SCAN_LIMIT, ) except Exception: return [] matched_ids = [ - str(row.get("doc_id") or "") + str(row.get(_FIELD_ALIASES["doc_id"]) or "") for row in (rows if isinstance(rows, list) else []) if isinstance(row, dict) and self._overlap_length(query, str(row.get("title") or "")) >= MIN_KEYWORD_OVERLAP @@ -283,7 +323,7 @@ class KnowledgeSearchService: try: details = lookup( collection_name=PRODUCT_COLLECTION, - filter=f"doc_id in [{quoted}]", + filter=f"{_FIELD_ALIASES['doc_id']} in [{quoted}]", output_fields=list(OUTPUT_FIELDS), limit=len(matched_ids), ) @@ -294,20 +334,10 @@ class KnowledgeSearchService: for row in details if isinstance(details, list) else []: if not isinstance(row, dict): continue - content = str(row.get("content") or "") + content = str(row.get(_FIELD_ALIASES["content"]) or "") if not content: continue - hits.append(KnowledgeHit( - doc_id=str(row.get("doc_id") or ""), - title=str(row.get("title") or ""), - content=content, - score=KEYWORD_MATCH_SCORE, - source_file=str(row.get("source_file") or ""), - visibility=str(row.get("visibility") or "public"), - doc_no=str(row.get("doc_no") or ""), - version=str(row.get("version") or ""), - chapter=str(row.get("chapter") or ""), - )) + hits.append(self._hit_from_row(row, score=KEYWORD_MATCH_SCORE)) # 一个产品名可能命中多个块（产品概览、费率表各一块）：全都保留， # 是不是"只有一个明确候选"交给上层的 gap 判定，这里不替它做选择。 return hits @@ -347,7 +377,8 @@ class KnowledgeSearchService: try: rows = lookup( collection_name=collection, - filter=self._anded(expression, f"doc_id in [{quoted}]"), + filter=self._anded(expression, + f"{_FIELD_ALIASES['doc_id']} in [{quoted}]"), output_fields=list(OUTPUT_FIELDS), limit=KEYWORD_SCAN_LIMIT, ) @@ -356,21 +387,11 @@ class KnowledgeSearchService: for row in rows if isinstance(rows, list) else []: if not isinstance(row, dict): continue - doc_id = str(row.get("doc_id") or "") - content = str(row.get("content") or "") + doc_id = str(row.get(_FIELD_ALIASES["doc_id"]) or "") + content = str(row.get(_FIELD_ALIASES["content"]) or "") if not content or doc_id not in parent_scores: continue - found.append(KnowledgeHit( - doc_id=doc_id, - title=str(row.get("title") or ""), - content=content, - score=parent_scores[doc_id], - source_file=str(row.get("source_file") or ""), - visibility=str(row.get("visibility") or "public"), - doc_no=str(row.get("doc_no") or ""), - version=str(row.get("version") or ""), - chapter=str(row.get("chapter") or ""), - )) + found.append(self._hit_from_row(row, score=parent_scores[doc_id])) return found @staticmethod @@ -390,6 +411,26 @@ class KnowledgeSearchService: """把可见性过滤与 doc_id 过滤合成一个 Milvus 表达式。""" return f"({expression}) and ({extra})" if expression else extra + @staticmethod + def _hit_from_row(row: Any, *, score: float) -> KnowledgeHit: + """把一行 Milvus 标量查询结果折成 `KnowledgeHit`。 + + 字段名经 `_FIELD_ALIASES` 映射（现库是 `knowledge_id`/`snippet`）；现库没有的字段 + （`source_file`/`doc_no`/`chapter`/`visibility`）按缺省值处理——留空字符串而不是 + 编造内容，`reference_title` 会因此退回纯标题，来源引用退化但不失真。 + """ + return KnowledgeHit( + doc_id=str(row.get(_FIELD_ALIASES["doc_id"]) or ""), + title=str(row.get(_FIELD_ALIASES["title"]) or ""), + content=str(row.get(_FIELD_ALIASES["content"]) or ""), + score=score, + source_file=str(row.get("source_file") or "") if _HAS_SOURCE_FILE else "", + visibility="public", + doc_no="", + version=str(row.get(_FIELD_ALIASES["version"]) or ""), + chapter="", + ) + @staticmethod def _parse(raw: Any, collection: str) -> list[KnowledgeHit]: """把 pymilvus 的 `[[{id, distance, entity}]]` 折叠成命中列表（纯函数，不抛异常）。""" @@ -401,18 +442,9 @@ class KnowledgeSearchService: entity = row.get("entity") if isinstance(row, dict) else None if not isinstance(entity, dict): continue - content = str(entity.get("content") or "") + content = str(entity.get(_FIELD_ALIASES["content"]) or "") if not content: continue # 没有正文的命中无法作为答案来源，直接丢弃而不是猜造 - hits.append(KnowledgeHit( - doc_id=str(entity.get("doc_id") or ""), - title=str(entity.get("title") or ""), - content=content, - score=float(row.get("distance") or 0.0), - source_file=str(entity.get("source_file") or ""), - visibility=str(entity.get("visibility") or "public"), - doc_no=str(entity.get("doc_no") or ""), - version=str(entity.get("version") or ""), - chapter=str(entity.get("chapter") or ""), - )) + hits.append(KnowledgeSearchService._hit_from_row( + entity, score=float(row.get("distance") or 0.0))) return hits diff --git a/tests/unit/service/test_customer_service_topic_matrix.py b/tests/unit/service/test_customer_service_topic_matrix.py index fe1c1a7..f9401ab 100644 --- a/tests/unit/service/test_customer_service_topic_matrix.py +++ b/tests/unit/service/test_customer_service_topic_matrix.py @@ -13,14 +13,19 @@ from typing import Any +from app.service.agent.governance import FALLBACK_DISCLAIMER from app.service.agent.implementations.customer_service import ( - DISCLAIMER, FALLBACK_TEMPLATE, CustomerServiceAgent, ) PRODUCT = "南方季季盈90天" +#: 固定免责声明由**治理层**注入（`PlatformGovernance.review` → `review_output`）， +#: 业务 Agent 不再自己拼——否则客户会看到两条重复声明（合并时实测复现）。 +#: 这里取治理层的代码兜底文案，只用于还原"治理后"的正文形状。 +DISCLAIMER = FALLBACK_DISCLAIMER + def _decision(**overrides: Any) -> dict[str, Any]: base: dict[str, Any] = { @@ -32,8 +37,8 @@ def _decision(**overrides: Any) -> dict[str, Any]: def _replied(body: str) -> str: - """出口交给客户的正文形状：`{正文}\\n{DISCLAIMER}`（四个出口都这样组装）。""" - return f"{body}\n{DISCLAIMER}" + """出口交给客户的正文形状：`{正文}\\n\\n{免责声明}`（治理层统一追加）。""" + return f"{body}\n\n{DISCLAIMER}" # ---- 出口一：知识直返（正文 = 知识块 content + DISCLAIMER） ---- diff --git a/tests/unit/service/test_knowledge_keyword_recall.py b/tests/unit/service/test_knowledge_keyword_recall.py index a164d67..2328f4f 100644 --- a/tests/unit/service/test_knowledge_keyword_recall.py +++ b/tests/unit/service/test_knowledge_keyword_recall.py @@ -29,13 +29,17 @@ async def _embed(text: str) -> list[float]: def _row(doc_id: str, title: str, score: float, content: str = "正文") -> dict[str, Any]: - """构造 pymilvus 的 `{distance, entity}` 行（`search()` 的返回形状）。""" + """构造 pymilvus 的 `{distance, entity}` 行（`search()` 的返回形状）。 + + 字段名按**现库集合的真实 schema**（`knowledge_id` / `snippet`），不是灌库脚本里那套 + （`doc_id` / `content`）——后者在本环境的集合从未建起来过，按它写替身会让测试假绿： + 业务代码读取 `snippet` 拿到空串，命中被静默丢弃，而断言又恰好只查 `doc_id`。 + """ return { "distance": score, "entity": { - "doc_id": doc_id, "title": title, "content": content, - "source_file": "x.md", "visibility": "public", "doc_no": "", - "version": "", "chapter": "", + "knowledge_id": doc_id, "title": title, "snippet": content, + "tags": "", "version": "", "intent": "", }, } @@ -48,9 +52,8 @@ def _flat(doc_id: str, title: str, content: str = "正文") -> dict[str, Any]: 测试红了一次——是测试写错，不是业务代码有问题（业务代码"没有正文就不作答"是对的）。 """ return { - "doc_id": doc_id, "title": title, "content": content, - "source_file": "x.md", "visibility": "public", "doc_no": "", - "version": "", "chapter": "", + "knowledge_id": doc_id, "title": title, "snippet": content, + "tags": "", "version": "", "intent": "", } @@ -73,8 +76,11 @@ class FakeClient: def query(self, **kwargs: Any) -> Any: self.query_calls += 1 - if "content" not in kwargs.get("output_fields", []): - return [{"doc_id": d, "title": t} for d, t in self._product_titles] + # 业务代码请求的字段名取自 `_FIELD_ALIASES`（现库是 `knowledge_id`/`snippet`）： + # 这里按"有没有要正文"区分两轮查询，而不是写死某一套字段名。 + output_fields = kwargs.get("output_fields", []) + if "snippet" not in output_fields and "content" not in output_fields: + return [{"knowledge_id": d, "title": t} for d, t in self._product_titles] pattern = str(kwargs.get("filter") or "") return [row for doc_id, row in self._product_rows.items() if doc_id in pattern] diff --git a/tools/publish_customer_service_config.py b/tools/publish_customer_service_config.py index e636eab..5a95f23 100644 --- a/tools/publish_customer_service_config.py +++ b/tools/publish_customer_service_config.py @@ -34,10 +34,15 @@ ADMIN = "9003" AGENT_TYPE = "customer_service" TOOL_NAME = "search_knowledge" SUITABILITY_TOOL = "check_suitability" +# 画像只读工具：客服的"出口零"（本人风险等级/投资偏好/测评是否过期）走它取权威字段。 +# 那个出口复用的是 `faq` 意图 key（见 `customer_service.PROFILE_WHITELIST_INTENT`）， +# 所以**必须**把它加进 `faq` 的白名单里；漏了的表现是"问画像一律转人工"， +# 而画像出口本身是对的——这是纯配置缺口（实测复现过）。 +PROFILE_TOOL = "query_customer_profile" # 只有会调用工具的意图才需要白名单；chitchat（模型生成）与 transfer_human（引导人工） # 都不查知识库。给它们配空白名单反而会掩盖"配置漏配"，因此不发布这两条。 INTENT_TOOLS: dict[str, tuple[str, ...]] = { - "faq": (TOOL_NAME,), + "faq": (TOOL_NAME, PROFILE_TOOL), "product_inquiry": (TOOL_NAME,), "policy_explain": (TOOL_NAME,), # 适当性裁决要两步：先从知识库拿到产品的风险等级，再由底座按档案里的客户等级裁决 @@ -208,9 +213,25 @@ async def main() -> int: for intent, tools in INTENT_TOOLS.items() ] inherited_keys = {(str(i["namespace"]), str(i["item_key"])) for i in inherited} + # **同 key 的继承项必须被本次新定义覆盖**，不能原样搬过去。 + # 血泪教训（实测）：上一版发布的是 `faq = ["query_knowledge", ...]`，而那个工具 + # 已随"客服检索改为 search_knowledge"从代码上限移除；原样继承会让 admin 服务的 + # 子集校验（白名单 ⊆ 代码限定的 allowed_tools）直接 422 拒绝整次发布， + # 报错是"配置超出 Agent 工具上限"，看不出是继承造成的。 + inherited_only = [ + item for item in inherited + if (str(item["namespace"]), str(item["item_key"])) not in { + ("agent_tools", f"{AGENT_TYPE}:{intent}") for intent in INTENT_TOOLS + } + ] + dropped = [item for item in inherited if item not in inherited_only] + for item in dropped: + print(f" [覆盖] {item['namespace']}/{item['item_key']} 将由本次定义替换" + f"（原值 {item['value_json']}）") pending = [ item for item in new_items - if (str(item["namespace"]), str(item["item_key"])) not in inherited_keys + if (str(item["namespace"]), str(item["item_key"])) not in + {(str(i["namespace"]), str(i["item_key"])) for i in inherited_only} ] if not pending: print("客服白名单已存在于当前生效版本，无需发布") @@ -228,9 +249,9 @@ async def main() -> int: print(f"\n发布版本 id={release_id}") base = f"/api/v1/admin/config-releases/{release_id}/platform-config-items" - for item in [*inherited, *pending]: + for item in [*inherited_only, *pending]: response = await post(client, base, auth=auth, payload=item) - mark = "继承" if item in inherited else "新增" + mark = "继承" if item in inherited_only else "新增" print(f" [{mark}] {item['namespace']}/{item['item_key']} → {response.status_code}") if response.status_code != 201: print(f" 失败：{response.text[:200]}")