diff --git a/app/core/customer_service_rules.py b/app/core/customer_service_rules.py index 4205dea..144cbff 100644 --- a/app/core/customer_service_rules.py +++ b/app/core/customer_service_rules.py @@ -1186,4 +1186,66 @@ def route_message(message: str) -> SafetyRoute | None: reply=P2_REPLY, transfer_required=True, transfer_reason=TRANSFER_REASON_EXPLICIT, ) - return None \ No newline at end of file + return None + + +# --------------------------------------------------------------------------- +# 术语归一化(检索查询串)—— 默认**不启用**,由 `customer_service.py` 的 +# `CS_QUERY_NORM` 决定是否调用本模块 +# --------------------------------------------------------------------------- +# +# 为什么需要它 +# ------------ +# 检索查询串此前只做 ``strip()``(`knowledge_search_service.py`),而字面通道的 +# 重合判定是**逐字符比较、大小写敏感**:客户打「r1到r5分别代表什么」,库里是 +# 「R1 到 R5 风险等级分别代表什么?」—— 同一个实体,字面上零重合,字面通道拿不到兜底。 +# +# 三条刻意约束(不是疏漏) +# ------------------------ +# 1. **只做同义替换,不做上下位词扩展**。「基金」→「公募基金/货币基金」属另一件事, +# 会引入原问句里根本没有的概念,风险高于收益,**本轮不做**。 +# 2. **只对齐"同一实体的不同写法"**:全角/半角、大小写、异体字、常见同义简称。 +# 任何会改变语义指向的替换都不得进 :data:`QUERY_SYNONYMS`。 +# 3. **调用方默认关闭**。归一化会改写查询串 ⇒ 改写向量化输入 ⇒ **可能移动 `gap`**。 +# 而 `customer_service.py` 的 `MIN_GAP` 实测合法区间只有 `(0.0649, 0.0759)`、 +# 两侧裕度各约 0.005 —— 这是一次**必须先用 55 条金标 A/B 实测**才能开启的改动。 +# 因此本模块本身是纯函数(无副作用、可单测),开关留在调用方。 + +#: 同义替换表。**按长度降序应用**,避免短串先命中把长串切碎(如「起投」吃掉「起投金额」)。 +#: ⚠️ 新增条目必须同时满足:同一实体、不改变语义指向、且有真实客户问法作为依据。 +#: 本轮刻意只收**纯书写差异**的条目;语义类同义(如「起购」/「起投」)留待金标 A/B 后逐条加。 +QUERY_SYNONYMS: tuple[tuple[str, str], ...] = ( + ("7日年化", "七日年化"), + ("7天年化", "七日年化"), + ("7日年化收益", "七日年化收益"), + ("风险评测", "风险测评"), + ("风险承受能力评测", "风险承受能力测评"), +) + +#: 单个拉丁字母紧跟数字时视为**等级/类别代号**(`r1`/`c1`/`R1`),统一成大写。 +#: 为什么需要:库里全部写作大写(`R1`—`R5`、`C1`—`C5`),客户常打小写。 +#: 负向断言(`(? str: + """把查询串里的**书写差异**对齐到库内写法;语义不变的纯函数。 + + 只做四件事,顺序有意: + 1. `NFKC` 全半角归一(全角字母/数字/括号 → 半角); + 2. 空白折叠(检索前多处会拼上文,容易留下连续空格); + 3. 异体字与常见同义简称(:data:`QUERY_SYNONYMS`,按长度降序替换,**一次通过**, + 不做链式替换 —— 链式会让「A→B、B→C」这种组合产生意料外的结果); + 4. 等级代号大写(:data:`_GRADE_CODE`)。 + + 不做的事:分词、上下位词扩展、同义改写、去停用词 —— 那些会改语义,不属"归一化"。 + """ + if not text: + return text + normalized = unicodedata.normalize("NFKC", text) + normalized = " ".join(normalized.split()) + for source, target in sorted(QUERY_SYNONYMS, key=lambda pair: -len(pair[0])): + if source in normalized: + normalized = normalized.replace(source, target) + normalized = _GRADE_CODE.sub(lambda match: match.group(1).upper(), normalized) + return normalized \ No newline at end of file diff --git a/app/core/retrieval_fusion.py b/app/core/retrieval_fusion.py index fad20a2..0fc423d 100644 --- a/app/core/retrieval_fusion.py +++ b/app/core/retrieval_fusion.py @@ -182,6 +182,94 @@ def fuse(routes: list[RetrievalRoute] | tuple[RetrievalRoute, ...]) -> FusionOut ) +#: RRF(排名融合)的平滑常数。取 60 —— 原始论文与主流实现的通行值。 +#: 它的作用是把"第 1 名 vs 第 2 名"的差距压到一个不至于过大的量级, +#: 使多路"共同命中但都不靠前"的条目有机会胜出(这正是融合想要的"共识优先")。 +RRF_K = 60 + + +def fuse_rrf( + routes: list[RetrievalRoute] | tuple[RetrievalRoute, ...], + *, + k: int = RRF_K, +) -> FusionOutcome: + """排名融合(Reciprocal Rank Fusion):`score = Σ 1 / (k + rank_i)`。 + + ## 什么时候用它(而不是 `fuse`) + + 一期两路都是 cosine、天然同尺度,加权求和即可(`fuse`)。 + 但引入**异质通道**后(例如字面锚点命中给满分 `1.0`),两路分数**不再是同一把尺子**: + 直接加权会让字面路**压平** `gap`,而 `customer_service` 的 `MIN_GAP` 裕度只有 ±0.005。 + 此时改用 RRF —— 它**只吃名次、不吃分数**,异质分数被彻底丢弃 ⇒ 不可能污染判定分。 + + ## 与 `fuse` 的共同点(消费方接口不变) + + - 返回仍是 `FusionOutcome`,`FusedHit` 字段语义不变; + - 🔴 `best_vector_score` **仍然只取向量路的原始 cosine**,绝不由 RRF 分替代 —— + 判定分与排序分的分离是本模块的第一约束(见模块头); + - `fused_score` 这里装的是 RRF 原始分(量级约 `1/k`),**只用于排序**; + - `degraded` 的三态合并口径与 `fuse` 逐字一致。 + + ⚠️ 注意量纲差异:`fuse` 的 `fused_score` 落在 `[0, 1]`,RRF 的落在约 `(0, 2/k]`。 + 两者都**只用于排序**,因此可互替;但**不得**拿二者跨策略比较大小。 + """ + usable = list(routes) + if not usable: + return FusionOutcome() + + failed = [r for r in usable if r.degraded] + degraded = bool(failed) + if failed and len(failed) == len(usable): + reason = "all_routes_failed" + elif failed: + reason = "partial_route_failure" + else: + reason = "" + + smooth = float(k) if k > 0 else 1.0 + acc: dict[str, dict[str, Any]] = {} + for route in usable: + seen: set[str] = set() + for index, hit in enumerate(route.hits, start=1): + doc_id = _doc_id_of(hit) + if not doc_id or doc_id in seen: + continue # 同一路里重复出现的同一条,只计它最好的那次名次 + seen.add(doc_id) + entry = acc.get(doc_id) + if entry is None: + entry = {"payload": hit, "rrf": 0.0, "routes": [], "best_vector": None} + acc[doc_id] = entry + # 后出现的路不覆盖 payload:先出现者优先(保序稳定、可复算) + entry["rrf"] += 1.0 / (smooth + index) + if route.name not in entry["routes"]: + entry["routes"].append(route.name) + if route.name == VECTOR_ROUTE_NAME: + score = _score_of(hit) + if score is not None: + prev = entry["best_vector"] + entry["best_vector"] = score if prev is None else max(prev, score) + + fused_hits = [ + FusedHit( + doc_id=doc_id, + payload=entry["payload"], + fused_score=float(entry["rrf"]), + best_vector_score=entry["best_vector"], + routes=tuple(entry["routes"]), + route_scores={}, # RRF 不保留各路原始分:它们已被刻意丢弃(异质尺度) + ) + for doc_id, entry in acc.items() + ] + # 融合分降序;同分保持插入序 ⇒ 可复算 + fused_hits.sort(key=lambda item: item.fused_score, reverse=True) + return FusionOutcome( + hits=tuple(fused_hits), + degraded=degraded, + reason=reason, + routes=tuple(r.name for r in usable), + ) + + def payloads(outcome: FusionOutcome) -> list[dict[str, Any]]: """把融合结果还原成"工具返回值形态"的列表,供既有下游(证据包 / 展示)直接消费。 diff --git a/app/service/agent/implementations/customer_service.py b/app/service/agent/implementations/customer_service.py index ebf937b..bbde665 100644 --- a/app/service/agent/implementations/customer_service.py +++ b/app/service/agent/implementations/customer_service.py @@ -64,6 +64,7 @@ from app.core.customer_service_rules import ( is_low_information_message, is_own_eligibility_question, is_risk_level_change_request, + normalize_query, promotional_wording_violation, route_message, substantive_business_request, @@ -132,7 +133,7 @@ from app.service.knowledge_search_service import ( ) from app.service.model_gateway import DatabaseModelEndpointResolver from app.service.retrieval_rerank_service import RetrievalRerankService -from app.service.runtime_config_service import load_active_prompt +from app.service.runtime_config_service import load_active_prompt, load_collection_routes logger = logging.getLogger(__name__) @@ -147,6 +148,15 @@ INTENT_POLICY = "policy_explain" INTENT_SUITABILITY = "suitability_check" INTENT_CHITCHAT = "chitchat" INTENT_TRANSFER = "transfer_human" +#: `W34` 会签 22:行情类问法的**独立授权码**。 +#: +#: ⚠️ 它**不是分类候选**(刻意不加入 `supported_intents` / `BUSINESS_INTENTS`)。 +#: 原因:它是"工具白名单 key"这一维度的名字,与"语义标签"是两件事 +#: (代码下方 `PROFILE_WHITELIST_INTENT` 的注释早已写明这一点)。 +#: 把它塞进分类候选会**改变判定分布**、威胁 `MIN_GAP` 的 ±0.005 窄带 —— +#: 而本轮要做的是把**授权维度**从判定维度里**拆出来**,不是扩大判定空间。 +#: 若将来要让行情成为独立**分类**意图(进一步治 `faq` 坍缩),须另立一单。 +INTENT_MARKET_QUOTE = "market_quote" BUSINESS_INTENTS = (INTENT_FAQ, INTENT_PRODUCT, INTENT_POLICY, INTENT_SUITABILITY) VISITOR_INTENTS = (INTENT_FAQ, INTENT_PRODUCT, INTENT_POLICY, INTENT_CHITCHAT, INTENT_TRANSFER) @@ -582,6 +592,17 @@ TREND_TOOL_NAME = "query_fund_trend" #: `E6` 走哪个工具白名单 key。沿用 `query_customer_profile` 的既有做法: #: 发布配置里 `customer_service:faq` 是唯一一个把所有跨意图工具都挂上去的键, #: 新开一个 key 会让工具交集为空、调用 fail-closed(那会表现成"配置错误"而不是"能力缺失")。 +#: `E6` 行情工具的**候选**白名单 key(**按优先级**)。 +#: +#: `W34` 会签 22:新增独立授权码 `market_quote` 承接行情,把**授权维度**从 `faq` +#: 里拆出来 —— 此前 `faq` 是发布配置里唯一一个把所有跨意图工具都挂上去的键, +#: 而这正是 `faq` 吸水坍缩的机制根源(判定被权限倒逼,判错也不报错)。 +#: +#: 候选**按序取第一个已发布的 key** ⇒ "新码还没发布"不会让行情出口失效, +#: 而是**逐字回落现状**。这就是本项内置的降级。 +TREND_WHITELIST_INTENT_CANDIDATES: tuple[str, ...] = (INTENT_MARKET_QUOTE, INTENT_FAQ) + +#: 兼容别名:既有代码与测试引用的是单值形态,语义 = 候选里的**兜底项**。 TREND_WHITELIST_INTENT = INTENT_FAQ #: 行情类词。**必须与实体同现**才算行情问句(见 `_trend_entity`)。 @@ -958,6 +979,14 @@ RERANK_ENABLED = os.environ.get("CS_RERANK", "").strip().lower() in {"1", "true" FUSION_WEIGHT_ALL = 0.6 FUSION_WEIGHT_NARROW = 0.4 +# ---- `W34` 阶段 0:查询串术语归一化(**默认关闭**)---- +# +# 为什么是环境变量、且默认关:归一化会改写送给检索的查询串 ⇒ 改写向量化输入 ⇒ +# **可能移动 `gap`**。而 `MIN_GAP` 的实测合法区间只有 `(0.0649, 0.0759)`、裕度 ±0.005 +# (见上方阈值注释),所以它和 `CS_DUAL_ROUTE` 同款处理:先开开关跑 55 条金标 A/B, +# 金标全绿再谈默认开启。关闭时 `_search_query` 逐字返回 `_search_query_raw` 的结果。 +QUERY_NORM_ENABLED = os.environ.get("CS_QUERY_NORM", "").strip().lower() in {"1", "true", "yes"} + #: 意图 → 收窄集合。目的:让**该意图最相关的集合**里的块在融合排序里被提上来。 #: ⚠️ **刻意不含 `BASIC_COLLECTION`**:把行业通用常识集合并入默认检索面会把 #: `M-1` 出口准确率从 100% 打到 91.3%(实测,见 `knowledge_search_service` 的 @@ -969,6 +998,14 @@ NARROW_COLLECTION_BY_INTENT: dict[str, str] = { INTENT_SUITABILITY: POLICY_COLLECTION, } +#: `W34` 会签 20:「意图 → 收窄集合」的**配置化路由缓存**(进程级一次)。 +#: +#: 缓存键是 `agent_type`。为什么要缓存:`_dual_route_output` 在检索热路径上, +#: 每题都查一次配置表不可接受;而路由配置的变更频率是**发布级**,进程内读一次足够。 +#: 取不到时**不写缓存**(让它在下次请求时自愈),而不是把空值钉死 —— 后者会让 +#: "配置中心短暂抖动"变成"必须重启才能恢复"。 +_COLLECTION_ROUTES_CACHE: dict[str, dict[str, str]] = {} + #: `F-3` **主体相关性闸门**(`E3`/`E4` 共用):库内已知的「服务名 / 条款名 / 主题词」。 #: #: 触发条件**只有一条**:问句里出现下表中的某个词。此时要求该词**至少落在一个命中块的 @@ -1351,6 +1388,58 @@ class CustomerServiceAgent(BaseAgent): # ---- 出口一:知识直返(faq / 产品 / 政策) ---- + async def _narrow_collection_for(self, intent: str) -> str: + """该意图的收窄集合:**发布配置优先、缺失回落代码常量**(`W34` 会签 20)。 + + 为什么要"配置优先":路由调整从此不必发版,评测可 A/B、线上可灰度。 + + 为什么**必须回落**:配置表 `agent_intent_config.collection_routes` 当前**全为空** —— + 若没有回落,一旦有人清空某行,该意图的双通道会**静默地**从"两路"退化成"一路", + 而且不报任何错。回落让"缺配置"等价于"用现状",这是本函数唯一的降级口径。 + + 读配置失败(库不可用 / 超时 / 任何异常)**同样回落常量、绝不冒泡**: + 路由是增益环节,它坏了不能把本来能答的问题变成答不出来。 + """ + routes: dict[str, str] = {} + try: + cached = _COLLECTION_ROUTES_CACHE.get(AGENT_TYPE) + if cached is None: + cached = await load_collection_routes(AGENT_TYPE) + _COLLECTION_ROUTES_CACHE[AGENT_TYPE] = cached + routes = cached + except Exception: # noqa: BLE001 —— 增益环节:任何异常只回落,不冒泡 + logger.info("收窄集合路由读取失败,回落代码常量", exc_info=True) + return routes.get(intent) or NARROW_COLLECTION_BY_INTENT.get(intent, "") + + def _trend_whitelist_intent(self) -> str: + """行情工具**实际可用**的白名单 key:候选按优先级取第一个**已发布**的(会签 22)。 + + 这是会签单要求的「启动自检」的**运行时等价物,且比它更强** —— + 它不止于"发现问题就报错",而是**直接回落**:新码没发布就用 `faq`,行为与改动前 + 逐字相同;`faq` 也没了才交回 `call_tool` 抛 `ForbiddenAgentError` + (那才是真正的配置错误,必须冒泡,不能被兜底吞掉)。 + + ## 为什么必须有它 + + 此前 `TREND_WHITELIST_INTENT` 是**硬绑** `faq` 的一个常量、**没有任何守卫**: + 任何人"按语义整理意图码"(例如把 `query_fund_trend` 挪到 `product_inquiry`), + `E6` 都会**静默失效**(症状表现成"工具未被授权")。本方法把这次隐性耦合 + 变成一次**可判定的查找**,并在找不到时留下 warning 日志。 + """ + published = getattr(self.config, "allowed_tools_by_intent", None) or {} + try: + keys = set(published) + except TypeError: # 配置形态异常时不猜,直接回落兜底项 + keys = set() + for candidate in TREND_WHITELIST_INTENT_CANDIDATES: + if candidate in keys: + return candidate + logger.warning( + "行情工具白名单 key 全部未发布:候选=%s 已发布=%s —— 交回 call_tool 判定权限", + TREND_WHITELIST_INTENT_CANDIDATES, sorted(keys), + ) + return TREND_WHITELIST_INTENT + async def _dual_route_output( self, request: AgentRequest, context: RequestContext, intent: str, tool: str ) -> dict[str, Any] | None: @@ -1383,7 +1472,8 @@ class CustomerServiceAgent(BaseAgent): """ if not DUAL_ROUTE_ENABLED: return None - narrow = NARROW_COLLECTION_BY_INTENT.get(intent, "") + # 会签 20:路由改为「发布配置优先、缺失回落代码常量」(见 `_narrow_collection_for`)。 + narrow = await self._narrow_collection_for(intent) if not narrow: return None query = self._search_query(request) @@ -1418,12 +1508,41 @@ class CustomerServiceAgent(BaseAgent): to_route("narrow", FUSION_WEIGHT_NARROW, raw[1]), ]) all_failed = outcome.reason == "all_routes_failed" + # 🔴 `W35` 判定口径分离:还原**向量路的原始序列**(去重后的 doc_id 顺序)。 + # + # 为什么必须有它:`hits` 是按 `fused_score`("共识优先")排的,而 `payloads()` + # 写回的 `score` 仍是向量路原始 cosine ⇒ `hits[0]`/`hits[1]` 可能是被融合 + # **提权**上来的块。若直接用它们算 `gap`,就等于"拿单路标定的 `MIN_GAP` + # 去量融合后的名次"——实测 `A-01` gap 由 `0.0777` 掉到 `0.0436`, + # `E3` 被顶成 `E4`,`M-1` 100% → 92.7%。 + # + # ⚠️ 三条容易写错的点: + # 1. **必须取该路返回的原始顺序,不能对融合结果按分重排** —— 融合结果的名字已被 + # `fused_score` 改过;而"路内顺序"才是判据里 `hits[0]`/`hits[1]` 的语义。 + # 2. **必须去重**:同一 `doc_id` 在同一路里可能返回多条(`fuse` 对它们取 `max`)。 + # 不去重会让同一条占据 `vector_order` 的前两个位置 ⇒ `gap` 恒为 0。 + # 3. 取不到(该路失败 / 格式异常)⇒ 空列表 ⇒ 下游回落单路口径(增益坏了不失效)。 + vector_order: list[str] = [] + raw_vector = raw[0] + if isinstance(raw_vector, dict): + seen_doc_ids: set[str] = set() + for item in raw_vector.get("hits") or []: + if not isinstance(item, dict): + continue + doc_id = str(item.get("doc_id") or "") + if doc_id and doc_id not in seen_doc_ids: + seen_doc_ids.add(doc_id) + vector_order.append(doc_id) + merged: dict[str, Any] = { "degraded": all_failed, "reason": outcome.reason if all_failed else "", # 部分失败不能被当成降级,但也不能静默 —— 用独立字段留痕,供评测统计。 "partial_failures": "" if all_failed else outcome.reason, "hits": payloads(outcome), + # 消费方(`_vector_decision_hits`)据此还原"单路口径"的 top1/top2 做出口判定; + # **融合分只决定 `hits` 的顺序,不参与"该不该答"**。 + "vector_order": vector_order, } logger.info("双通道融合:路=%s 块数=%d 降级=%s", outcome.routes, len(merged["hits"]), outcome.reason or "无") @@ -1458,6 +1577,74 @@ class CustomerServiceAgent(BaseAgent): logger.info("精排生效:%d 块按相关性重排", len(outcome.evidence)) return outcome.evidence + @classmethod + def _vector_decision_hits( + cls, output: dict[str, Any], hits: list[Any] + ) -> list[Any] | None: + """还原「向量路原始序列」,供**出口判定**与**原文直返**使用(`W35` 口径分离)。 + + ## 为什么必须分离 + + 融合结果 `hits` 按 `fused_score`("共识优先")排序,而 `payloads()` 写回的 + `score` 仍是**向量路的原始 cosine**。于是 `hits[0]` / `hits[1]` 可能是被融合 + **提权**上来的块 —— 拿它们算 `gap`,等于"用单路标定的 `MIN_GAP` 去量融合后的名次"。 + 实测 `A-01`:单路 `0.8428 / 0.7651`(`gap 0.0777` → `E3`),融合后第 2 名被顶成 + `0.7992`(`gap 0.0436` → `E4`)⇒ `M-1` 从 `100%` 掉到 `92.7%`。 + + ⇒ **"该不该答"用向量路原始序列;"答哪一块、给哪些块"用融合序列。** + + ## 两条必修一并被它解决 + + 1. **`hits[0]` 换人** ⇒ 判据与直返内容原本会不同源(判据说"高置信"、 + 直返的却是另一个块)。本方法让二者取自同一条序列。 + 2. **`gap` 虚高** ⇒ `payloads()` 对**纯字面命中**写 `score = 0.0`;若它排到第 2 位, + `gap` 会被拉大,把"其实有并列候选"误判成高置信直返。向量路序列里 + **根本不含**这类块(它们没有向量分)⇒ 天然不参与 `gap` 计算。 + + ## `basic` 补位的例外(唯一一处刻意不分离) + + `_supplement_basic_explain` 会把它判定"严格更贴题"的常识块**前置**。那种情况下 + `hits[0]` 不在 `vector_order` 里 —— 但它**带向量分、且严格高于向量路 top1**, + 与"纯字面命中(`score = 0.0`,永远不可能更高)"可**精确区分**。 + 故按此判据把它放回判定序列首位:这样 `judge[0] = basic top1`、 + `judge[1] = 向量路 top1`,`gap = basic_top - vector_top > 0`, + 与开关关闭时"补位后 `hits[0]/hits[1]`"的语义**完全同构**。 + + ## 回落(与基线对账的前提) + + 开关关闭 / 单路 / 无 `vector_order` 字段 ⇒ 返回 `None`, + 调用方沿用既有 `hits` 闭环,**与改动前逐字一致**。 + """ + order = output.get("vector_order") + if not isinstance(order, list) or not order: + return None + by_id = { + str(hit.get("doc_id") or ""): hit + for hit in hits + if isinstance(hit, dict) + } + restored: list[Any] = [] + for doc_id in order: + hit = by_id.get(str(doc_id)) + if hit is not None: + restored.append(hit) + if not restored: + return None + # `basic` 补位块回到首位(判据见 docstring「例外」一节)。 + vector_ids = {str(hit.get("doc_id") or "") for hit in restored} + top = cls._score(restored[0].get("score")) + promoted: list[Any] = [ + hit + for hit in hits + if isinstance(hit, dict) + and str(hit.get("doc_id") or "") not in vector_ids + and cls._score(hit.get("score")) > top + ] + if promoted: + promoted.sort(key=lambda hit: cls._score(hit.get("score")), reverse=True) + return [promoted[0], *restored] + return restored + async def _answer_from_knowledge( self, request: AgentRequest, context: RequestContext, intent: str ) -> CoreResult: @@ -1504,14 +1691,40 @@ class CustomerServiceAgent(BaseAgent): subject_terms = self._subject_terms_in(request.message) if not self._subject_covered_by(hits, subject_terms): return self._exit_subject_miss(subject_terms[0]) - score = self._score(best.get("score")) - gap = score - self._second_score(hits) + # 🔴 `W35` 口径分离:判定(`score`/`gap`)与**原文直返的内容**取「向量路原始序列」; + # 融合序列 `hits` 只负责"补充候选、排序证据包与展示"(`_evidence_pack` / + # `_exit_clarify` / `_answer_from_evidence` 仍吃 `hits`,融合的收益留在那里)。 + # + # `judge is None` = 开关关闭 / 单路 / 无 `vector_order` ⇒ 走 `else` 分支, + # **与改动前逐字一致**(这是开关关闭时与基线对账的前提)。 + judge = self._vector_decision_hits(output, hits) + if judge is not None: + # `basic` 补位块若被前置,已在 `_vector_decision_hits` 里回补到首位。 + best = judge[0] + score = self._score(best.get("score")) + gap = score - self._second_score(judge) + else: + score = self._score(best.get("score")) + gap = score - self._second_score(hits) # `E4` 证据约束生成(`H-03`):同一章节的多个小节分数咬得紧时**合并作答**。 # 放在置信判定**之前**:`C-01` 的 top1 分数 0.7612 已经够"直接答", # 但直返的只是四档权益里的一档(金标判"只答其中一档 = 失败")—— # 高置信不等于答案完整。模型自述答不了时 `_answer_from_evidence` 回 `None`, # 下面原有的 E3 / E5a / E5b 判定**原样生效**。 - evidence = self._evidence_pack(hits, gap=gap) + # 🔴 `W36` 选块口径归一:证据包内部的**遍历顺序与 top 取值**也取「向量路名次」。 + # 为什么:`E-01`(多轮 + 代词)两态的向量路返回**逐字一致**、`score ≥ E4_MIN_SCORE` + # 的块集合也一致,唯一差异是本函数下游 `_evidence_pack` 的**遍历路序** —— + # 融合把 `PROD-002` 父块从单路第 10 位提权到第 3 位,6 个名额被不同的块填满 + # ⇒ 证据包不同 ⇒ 模型输出不同 ⇒ `E3` 直返变成 `E5b`(相对降级)。 + # `pack_order` 只当**排序键**、不当过滤器:成员集合一个不动(`judge` 之外的块 + # 按原相对顺序追加在后),融合的收益仍完整留在候选集合上。 + # 开关关闭时 `judge is None` ⇒ `pack_order=None` ⇒ 逐字等同改动前。 + pack_order = ( + [str(hit.get("doc_id") or "") for hit in judge if isinstance(hit, dict)] + if judge is not None + else None + ) + evidence = self._evidence_pack(hits, gap=gap, order=pack_order) if evidence is not None: generated = await self._answer_from_evidence(request, context, evidence, hits) if generated is not None: @@ -2792,10 +3005,65 @@ class CustomerServiceAgent(BaseAgent): row.user_prompt_template or DEFAULT_EVIDENCE_TEMPLATE, ) + @staticmethod + def _pack_order(hits: list[Any], order: list[str] | None) -> list[Any]: + """把命中排成「证据包内部遍历顺序」(`W36` 选块口径归一)。 + + `order` 是**向量路原始名次**(`doc_id` 序列)—— 与出口判定用的是**同一条序列** + (`_vector_decision_hits` 的产物),因此"该不该答"与"证据包先看哪一块"同源。 + + 规则只有两条: + + 1. `order` 里出现、且在 `hits` 里找得到的块,**按 `order` 的顺序**排在前; + 2. 其余块(融合路独有 / 字面路独有 / 常识补检块…)**按原相对顺序追加在后**。 + + ⇒ **成员集合一个不动,只换顺序**。这一点是要害:`order` 只当**排序键**、 + 不当**过滤器**,所以不可能因为"某块不在向量路里"就把它从证据包里丢掉。 + + `order` 为空 / `None` / 无一条能对上 ⇒ **原样返回 `hits`**, + 于是开关关闭时的单路行为与改动前**逐字一致**(这是与基线对账的前提)。 + """ + if not order: + return list(hits) + by_id = { + str(hit.get("doc_id") or ""): hit + for hit in hits + if isinstance(hit, dict) + } + seen: set[str] = set() + head: list[Any] = [] + for doc_id in order: + key = str(doc_id or "") + hit = by_id.get(key) + if hit is None or key in seen: + continue + seen.add(key) + head.append(hit) + if not head: + return list(hits) + tail: list[Any] = [] + for hit in hits: + key = str(hit.get("doc_id") or "") if isinstance(hit, dict) else "" + if key and key in seen: + continue + if key: + seen.add(key) + tail.append(hit) + return [*head, *tail] + @classmethod - def _evidence_pack(cls, hits: list[Any], *, gap: float) -> list[dict[str, Any]] | None: + def _evidence_pack( + cls, hits: list[Any], *, gap: float, order: list[str] | None = None + ) -> list[dict[str, Any]] | None: """`E4` 该不该接管、证据包里放哪些块;`None` = 不接管。 + 🔴 `order`(`W36`):**只用于排序**的向量路名次,缺省 `None` = 逐字沿用传入顺序。 + 为什么要它:`E-01`(多轮 + 代词)在两态下**向量路返回逐字一致**、`score ≥ E4_MIN_SCORE` + 的块集合也一致,唯一差异是本函数内的**遍历路序** —— 融合把 `PROD-002` 父块从单路 + 第 10 位提权到第 3 位,于是 `E4_MAX_EVIDENCE` 的 6 个名额被不同的块填满 ⇒ + 证据包内容不同 ⇒ 模型输出不同 ⇒ `E3` 直返变成 `E5b`。传入 `order` 后, + "先看哪一块、截断时留下哪 6 块"两态一致,融合的收益仍完整留在**候选集合**上。 + **先决条件**:`gap < MIN_GAP`(分数接近)。领先明显时仍走 `E3` 原文直返 —— 原文直返没有幻觉面,能不用模型就不用。 @@ -2828,8 +3096,12 @@ class CustomerServiceAgent(BaseAgent): """ if gap >= MIN_GAP: return None + # 🔴 `W36`:本函数以下**一律遍历 `ordered`**(而不是 `hits`)—— 它是同一个集合、 + # 只是顺序换成"向量路名次在前、其余按原相对顺序追加"。开关关闭时 `order=None` + # ⇒ `ordered == list(hits)`,逐字等同改动前。 + ordered = cls._pack_order(hits, order) groups: dict[tuple[str, str], list[dict[str, Any]]] = {} - for hit in hits: + for hit in ordered: if not isinstance(hit, dict): continue if cls._score(hit.get("score")) < E4_MIN_SCORE: @@ -2844,7 +3116,7 @@ class CustomerServiceAgent(BaseAgent): # 交回置信判定只会变成"澄清"(`gap < MIN_GAP` 必然不满足直接答的条件), # 而附录F.3 明确要求这种情况**合并作答、不澄清**。 fallback = [ - hit for hit in hits + hit for hit in ordered if isinstance(hit, dict) and cls._score(hit.get("score")) >= E4_MIN_SCORE ][:E4_MAX_EVIDENCE] return fallback or None @@ -2860,7 +3132,7 @@ class CustomerServiceAgent(BaseAgent): pack = pack[:E4_MAX_EVIDENCE] # `W6`:章节组之外的高分块**也要进包**(合并,不是二选一),见 docstring 第 3 条。 seen = {str(hit.get("doc_id") or "") for hit in pack} - for hit in hits: + for hit in ordered: if len(pack) >= E4_MAX_EVIDENCE: break if not isinstance(hit, dict): @@ -2872,7 +3144,9 @@ class CustomerServiceAgent(BaseAgent): continue seen.add(doc_id) pack.append(hit) - top = hits[0] + # `W36`:`top` 取 `ordered[0]`(向量路口径的"最贴题那块"),不是 `hits[0]`。 + # 开关关闭时二者是同一个对象,故此行在单路下无行为变化。 + top = ordered[0] if isinstance(top, dict): top_id = str(top.get("doc_id") or "") if top_id and all(str(item.get("doc_id") or "") != top_id for item in pack): @@ -3143,6 +3417,25 @@ class CustomerServiceAgent(BaseAgent): @classmethod def _search_query(cls, request: AgentRequest) -> str: + """交给检索的查询串:`_search_query_raw` + **术语归一化**(`W34` 阶段 0)。 + + `CS_QUERY_NORM` 关闭时(默认)本函数逐字返回 `_search_query_raw` 的结果 —— + 与改造前**完全等同**,这是与基线对账的前提。 + 开启后只对齐**书写差异**(全半角 / 大小写 / 异体字 / 同义简称),不改语义; + 为什么必须放在"拼完上文之后"、以及为什么默认关,见 + `app/core/customer_service_rules.py::normalize_query` 的注释。 + """ + raw = cls._search_query_raw(request) + if not QUERY_NORM_ENABLED: + return raw + normalized = normalize_query(raw) + if normalized != raw: + # 留痕:客户投诉"我问的 `r1` 你怎么答 `R1`"时要能复盘出真实查询串。 + logger.info("查询归一化:raw=%r norm=%r", raw, normalized) + return normalized + + @classmethod + def _search_query_raw(cls, request: AgentRequest) -> str: """构造交给检索的查询串。 什么时候带上文(实测决定,两条都不能少): @@ -3351,7 +3644,7 @@ class CustomerServiceAgent(BaseAgent): arguments: dict[str, Any] = {"fund_code" if kind == "code" else "fund_name": value} try: output = await self.call_tool( - TREND_TOOL_NAME, arguments, intent=TREND_WHITELIST_INTENT, context=context + TREND_TOOL_NAME, arguments, intent=self._trend_whitelist_intent(), context=context ) except ForbiddenAgentError: # 白名单/权限类失败**必须冒泡**(与知识、画像出口同一口径):那是配置错误, @@ -3373,6 +3666,50 @@ class CustomerServiceAgent(BaseAgent): text = str(value) return text if text.startswith("-") else f"+{text}" + #: `INV-8` 的比对口径:连续数字(含小数),日期按段比对(`2026-07-21` → `2026`/`07`/`21`)。 + #: ⚠️ 与前端 `_w29_chart_render_check.mjs` 的 `digits()`、与单测 + #: `tests/unit/service/test_customer_service_trend_chart_inv8.py` 的 `_NUMBER` + #: **必须同一口径** —— 三处不一致就会出现"前端报违例、后端说没有"。 + _NUMBER_RE = re.compile(r"\d+(?:\.\d+)?") + + @classmethod + def _visible_numbers(cls, value: object) -> set[str]: + """抽出一个值里客户能看到的数字。`None` / 空串自然得到空集。""" + if value is None: + return set() + return set(cls._NUMBER_RE.findall(str(value))) + + @classmethod + def _prune_chart_to_text(cls, chart: dict[str, Any], text: str) -> dict[str, Any]: + """`INV-8` 收口:把图里**正文已找不到**的数字剔掉(字段置 `None`,区间行丢弃)。 + + 为什么必须有这一步(不是洁癖):图的数字只有两个来源 —— 正文,或者**凭空**。 + `M-9`(无出处数字 = 0)的取证面**只看答复文本**,它看不见图;一旦图里出现 + 正文没有的数,图就成了**唯一能绕过数字校验的合规通道**。 + + 触发条件(实测):正文有 `MAX_ANSWER_CHARS = 1200` 上限,而载荷**没有** —— + 区间数偏多时正文尾部(`区间最高/最低`、`数据来源…共 N 个净值日`)先被截掉, + 而载荷里 `high` / `low` / `series_points` 照旧 ⇒ 违例。 + **正常长度(4 个区间、约 350 字)不触发,本函数原样返回。** + + 取舍:宁可图**少画**一项,也不让图多出一个正文没有的数 —— + 与全仓"数据边界必须自陈"(`INV-7`)同一取向。 + """ + allowed = cls._visible_numbers(text) + pruned: dict[str, Any] = dict(chart) + for key in ( + "latest_nav", "latest_nav_date", "from_date", "to_date", + "series_points", "high", "low", + ): + if key in pruned and not cls._visible_numbers(pruned[key]) <= allowed: + pruned[key] = None + pruned["intervals"] = [ + row + for row in (chart.get("intervals") or []) + if isinstance(row, dict) and cls._visible_numbers(row) <= allowed + ] + return pruned + def _exit_trend(self, output: dict[str, Any]) -> CoreResult: """`E6` 固定模板渲染(`D3.9` §4.3)。 @@ -3453,8 +3790,12 @@ class CustomerServiceAgent(BaseAgent): f"{output.get('from_date')}—{output.get('to_date')}," f"共 {output.get('series_points')} 个净值日" ) + # `W34`:**先定稿正文、再据此收口图表**。顺序不能反 —— 正文有 + # `MAX_ANSWER_CHARS` 上限、载荷没有;先建图表再截正文,就会出现 + # "图里有、正文里没有"的数字(`INV-8` 违例,实测区间数偏多时触发)。 + text = self._clamp_answer("\n".join(lines)) return CoreResult( - text=self._clamp_answer("\n".join(lines)), + text=text, intent=self._classified_intent, exit_code=EXIT_QUOTE, topic=name, @@ -3462,7 +3803,7 @@ class CustomerServiceAgent(BaseAgent): "fund_code": code, "fund_name": name, "source": output.get("source"), - "trend_chart": chart, + "trend_chart": self._prune_chart_to_text(chart, text), }, ) diff --git a/app/service/knowledge_search_service.py b/app/service/knowledge_search_service.py index 613da0e..d83b72a 100644 --- a/app/service/knowledge_search_service.py +++ b/app/service/knowledge_search_service.py @@ -167,6 +167,7 @@ class KnowledgeSearchService: tiers: frozenset[str], collections: Sequence[str] | None = None, top_k: int = 5, + literal_parallel: bool = False, ) -> KnowledgeSearchOutcome: """检索知识库。 @@ -241,7 +242,13 @@ class KnowledgeSearchService: # 第二路召回:客户确切说出的产品名按字面取回。只在向量结果不够确定时介入, # 否则会把向量已经答对的题顶掉(见 VECTOR_CONFIDENT_SCORE 的说明)。 best_vector_score = max((hit.score for hit in collected), default=0.0) - if best_vector_score < VECTOR_CONFIDENT_SCORE: + # `literal_parallel=True`(`W34` 会签 21):把字面通道从"条件触发的兜底" + # 升为**可并行召回的独立通道** —— 不再被 `VECTOR_CONFIDENT_SCORE` 挡住。 + # + # ⚠️ 默认 `False` ⇒ 本行以下与改动前**逐字相同**,既有调用方零影响。 + # 🔴 打开它**必须与精排同批启用**:字面命中给满分 `1.0` 会把 `gap` 压平 ⇒ + # 触发"领先不足"判据 ⇒ 一律转人工(这正是本通道此前被闸门挡住的原因)。 + if literal_parallel or best_vector_score < VECTOR_CONFIDENT_SCORE: collected.extend( self._product_keyword_hits(client, isolated, text, expression) ) diff --git a/app/service/runtime_config_service.py b/app/service/runtime_config_service.py index 1c79329..0068a7e 100644 --- a/app/service/runtime_config_service.py +++ b/app/service/runtime_config_service.py @@ -65,6 +65,41 @@ class RuntimeConfigService: ) return tuple(self._entry(row) for row in rows) + async def collection_routes(self, agent_type: str) -> dict[str, str]: + """读取某 Agent 当前生效的「意图 → 收窄集合」路由。 + + **与 `active_intents()` 使用同一套生效判定**(`status='active'` + 有效期窗口), + 因此"配置里生效的意图"与"路由里生效的意图"不会变成两套口径。 + + `collection_routes` 是**既有 JSON 列**(`app/model/configuration.py:94`),本方法 + 是对它的**首次消费**(此前全仓零消费,属"有列无消费")。列里允许是字符串、 + 字符串数组或 `{"collection": ...}` 形态,统一投影成 `{intent_code: collection}`。 + + **拿不到可用集合名的行直接跳过** —— 缺失时由调用方**回落代码常量**, + 而不是塞一个空值把整条路由打断。这是本函数唯一的降级口径。 + """ + now = datetime.now(UTC).replace(tzinfo=None) + rows = await self.session.scalars( + select(AgentIntentConfig) + .where( + AgentIntentConfig.agent_type == agent_type, + AgentIntentConfig.status == "active", + (AgentIntentConfig.effective_at.is_(None)) + | (AgentIntentConfig.effective_at <= now), + (AgentIntentConfig.expire_at.is_(None)) | (AgentIntentConfig.expire_at > now), + AgentIntentConfig.collection_routes.is_not(None), + ) + .order_by(AgentIntentConfig.priority, AgentIntentConfig.intent_code) + ) + routes: dict[str, str] = {} + for row in rows: + name = _first_collection_name(row.collection_routes) + if name: + # 同一 intent_code 理论上只有一个 active 版本(唯一键保证); + # `setdefault` 是防御性的:真出现两条也只认优先级最高的那条。 + routes.setdefault(row.intent_code, name) + return routes + @staticmethod def _entry(row: AgentIntentConfig) -> IntentConfigEntry: return IntentConfigEntry( @@ -117,6 +152,43 @@ async def load_fund_quote_config() -> dict[str, object]: return await RuntimeConfigService(session).fund_quote(release.id) +def _first_collection_name(value: object) -> str: + """从 `collection_routes` 的 JSON 值里取**第一个可用集合名**;取不到返回空串。 + + 兼容三种已见过的写法,避免把列格式钉死(列是既有的,历史/人工写入的形态不可控): + - `"fin_faq_collection"` —— 裸字符串 + - `["fin_faq_collection", ...]` —— 数组(取首个非空) + - `{"collection": "fin_faq_collection"}` + """ + if isinstance(value, str): + return value.strip() + if isinstance(value, (list, tuple)): + for item in value: + name = _first_collection_name(item) + if name: + return name + return "" + if isinstance(value, dict): + for key in ("collection", "name", "target"): + if key in value: + name = _first_collection_name(value[key]) + if name: + return name + return "" + return "" + + +async def load_collection_routes(agent_type: str) -> dict[str, str]: + """「意图 → 收窄集合」路由的运行期装载器。 + + 与 `load_active_intent_configs()` 并列、互不影响;**读不到(无 active 行、列全空、 + 配置中心不可用)一律返回空映射**,由调用方回落代码常量 + (`customer_service.NARROW_COLLECTION_BY_INTENT`)—— 即"配置优先、缺失回落"。 + """ + async with SessionFactory() as session: + return await RuntimeConfigService(session).collection_routes(agent_type) + + async def load_active_intent_configs(agent_type: str) -> tuple[IntentConfigEntry, ...]: """意图分类链路的运行期配置装载器:只读 `agent_intent_config` 的 active 行。 diff --git a/app/static/portal/README.md b/app/static/portal/README.md index e5dc610..8e24c24 100644 --- a/app/static/portal/README.md +++ b/app/static/portal/README.md @@ -52,8 +52,12 @@ 1. **`change_pct` 可能是 `null`**(行情只同步过一个交易日时算不出涨跌)。 调用方必须显示"暂无",**不得当成 `0`** —— `formatPercent` 收到 `null` 会渲染成 `+0.00%`, 那等于告诉客户"今天平盘"。 -2. **历史净值曲线没有数据源**:`fin_nav_history` 目前 0 行,详情页因此**不画走势图**, - 并显式说明"尚未接入"。此前那条曲线是 mock 里 12 个编造点位 —— 走势图最容易被当成真数据。 +2. **历史净值曲线已有数据源**(`W31` 起接入;本节此前写的"0 行"已过期): + `fin_nav_history` 实测 **2494 行 / 20 个产品**,`nav_date` 覆盖 `2026-03-18`—`2026-09-13`, + 由 `tools/sync_nav_history.py` 从东财净值接口同步。详情页 `renderChart` **取到点位就画折线**, + 仅当该产品**没有历史**时才退化为"历史净值数据尚未接入,暂不展示走势图"。 + ⇒ 演示走势图请用**真实在场内、已同步**的产品(如 `159382`、`511810`)。 + 此前那条曲线曾是 mock 里 12 个编造点位 —— 走势图最容易被当成真业绩,故此处口径必须准确。 3. **产品级披露在 `common/product-notes.js`**(如 510300 的"同指数参考产品,非本公司发行")。 `fin_product` 没有这个字段,所以它留在前端;新增需要披露的产品时改那一份。 凡渲染公开产品的页面都要挂 `data-source-notice` 说明数据来源。 diff --git a/demo.ps1 b/demo.ps1 index 2787b97..6626ca9 100644 --- a/demo.ps1 +++ b/demo.ps1 @@ -16,7 +16,7 @@ 连续登录会让后续用例整片报红,看起来像「客服坏了」,其实是限流。 .PARAMETER Port - API 端口,默认 8000(与 `start.ps1 -Port` 同义)。 + API 端口,默认 8099(与 `start.ps1 -Port` 同义)。 .PARAMETER SkipStart 不调 `start.ps1`,只做自检 + 开页面(服务已经在跑时用)。 @@ -36,7 +36,7 @@ 按 ANSI 解析,中文会乱码并抛语法错误)。 #> param( - [int]$Port = 8000, + [int]$Port = 8099, [switch]$SkipStart, [switch]$NoBrowser, [switch]$KeepPrices diff --git a/docs/49-底座会签申请单-2026-09-19.md b/docs/49-底座会签申请单-2026-09-19.md index f0c5abb..1528110 100644 --- a/docs/49-底座会签申请单-2026-09-19.md +++ b/docs/49-底座会签申请单-2026-09-19.md @@ -261,6 +261,114 @@ --- +## 组 7 · 意图层与检索层的三项扩张(**`W34` 新增 · ☑ 2026-09-22 受理并落地**) + +> **性质**:本组提案在前;用户在 2026-09-22 的对话中**逐项授权会签**后当场落地 +> (与组 5 同类流程:先立单、经授权、后动手)。 +> **共同背景**:这三项都源自同一个结构性事实 —— **意图码既是语义标签,又是工具权限 key** +> (`agent_tools/customer_service:`)。它们分别从"配置化"、"通道化"、"命名"三个方向 +> 去松动这个耦合,因此**必须一起评估**:只做其一,另外两项的收益会被抵消。 + +### 会签 20 · `app/service/runtime_config_service.py`(白名单外) + +**一、改什么** +新增 `load_collection_routes(agent_type) -> dict[str, str]`:只读 `agent_intent_config` 的 +`status='active'` 行(与既有 `load_active_intent_configs()` **同一生效判定**),把 +`collection_routes`(**JSON 列,已存在于表中**)投影成 `{intent_code: collection}`。 +消费方 `customer_service.py`(**类 2**)用它在构造 `NARROW_COLLECTION_BY_INTENT` 时**优先取配置、 +缺失回落代码常量**。 + +**二、为什么必须会签** +`app/service/runtime_config_service.py` **不在 `D2.1` §1.1 / §1.3 名单内,也不在 §1.5 零改动清单内** +⇒ 按 `docs/48` 开篇口径,它属**白名单之外**。 +`customer_service.py:948–949` 的代码注释**自认**这一点:「等权重与路由要进 +`agent_intent_config.collection_routes` 时(二期),再走发布配置(那需要会签 —— 读该表的落点在 +白名单之外)」。 + +**三、最小化边界** +**零 DDL**(`collection_routes` 是 `app/model/configuration.py:94` 的**既有 JSON 列**,本轮只是 +**首次消费**它 —— 该列目前**全仓零消费**,属"有列无消费"); +不改表结构 / 不改函数签名 / 不改返回形状;新增函数与既有 `load_active_intent_configs()` 并列, +**不改动**后者。 + +**四、依据**:`D2.1` 决策 19(配置化);`W33` 方案 §二期;`docs/48` 类 3 口径。 + +**五、影响面** +`customer_service.py::NARROW_COLLECTION_BY_INTENT` 的取值来源(**常量 → 配置优先**); +`tests/unit/service/test_customer_service_agent.py`;配置面 `admin_service` 的 +`agent-intent-configs` 写入口(**不改**,仅新增读路径)。 + +**六、降级方案(不受理时)** +保留环境变量 + 代码常量(现状)。代价:路由调整**每次都要发版**,且环境变量在生产 +**不可热改** —— 评测可以 A/B,线上不能灰度。 + +### 会签 21 · `app/service/knowledge_search_service.py`(**类 3**)+ `app/core/retrieval_fusion.py` + +**一、改什么** +① 检索层把**字面通道**从"条件触发的兜底"改为**可并行召回的独立通道**(暴露独立入参, +不改既有默认行为); +② `app/core/retrieval_fusion.py` 增加 **`rrf` 融合策略**(排名融合),与既有"加权求和" +并存、**由调用方选择**;消费方接口**不变**(该模块 `:29–31` 已明写二期加 `rrf` 策略即可)。 + +**二、为什么必须会签** +`app/service/knowledge_search_service.py` 是**类 3**(`docs/48:40`,已受理项仅 `D-01`/`D-02`), +**新改动须新会签**。`retrieval_fusion.py` 本身属客服模块(`W33` 新增),但 **② 只能在 ① 落地后才有意义**, +故合成一张单。 + +**三、最小化边界** +不改 `search()` 签名(新入参带默认值,默认行为**逐字等同**现状);不改 `KnowledgeHit` / +`KnowledgeSearchOutcome` 形状;不改表结构 / **零 DDL**;不动 `tiers` 必填与 +`visibility_expression` 下推;**融合层仍不得持有 Milvus 客户端**(`INV-1`,见 +`retrieval_fusion.py:41–42`)。 + +**四、依据** +`retrieval_fusion.py:26–31`("等二期引入**异质通道**(字面锚点给满分 `1.0`)时,才需要切换成 +排名融合");`knowledge_search_service.py` 的 `VECTOR_CONFIDENT_SCORE = 0.75` 闸门。 + +**五、影响面** +⚠️ **这是本组风险最高的一项**:字面通道命中给**满分 `1.0`** 会把 `gap` **压平** ⇒ 触发 +"领先不足"判据 ⇒ 一律转人工。因此 ① 与 ② **必须同批上线**,且上线前必须用 55 条金标 +验证 `M-1` 不下降(当前 `MIN_GAP` 裕度仅 ±0.005)。 + +**六、降级方案(不受理时)** +维持现状:字面通道继续只做"条件触发兜底"。代价:新词条 / 新别名仍只能靠补 FAQ 命中, +字面通道无法参与并行召回。 + +### 会签 22 · 发布配置 `agent_tools/customer_service:*` + `app/service/agent/bootstrap.py` + +**一、改什么** +① 新增一个**独立意图码**(暂名 `market_quote`)承接行情类问法,**并新增对应工具白名单 key**; +② `customer_service.py` 把 `TREND_WHITELIST_INTENT` 由 `INTENT_FAQ` 改为该新码 +(**类 2**,同时改 `INTENT_*` 常量与 `AgentDefinition.supported_intents` 声明); +③ `agent_intent_config` 补该意图码的**分类描述**(走既有发布流程,非代码)。 + +**二、为什么必须会签** +`app/service/agent/bootstrap.py` 是 `D2.1` §1.3 **组 2** 已会签文件,**新改动须新会签**; +发布配置 `agent_tools/*` 的写入口不在客服模块内。 + +**三、最小化边界** +只**新增**意图码与权限 key,**不删除、不改名**任何既有码;不动 `IntentClassifier` 算法 +(`app/service/intent_classifier.py` **仍不触碰**);不改 `exit_codes.py` 的 `E6`。 + +**四、依据** +`customer_service.py:201–205` 的铁证 —— 调用画像工具**必须复用已发布的 `faq`**,因为 +「**发布配置里只有 `agent_tools/customer_service:faq` 一个 key**;换新意图码会让交集为空 ⇒ +`AGENT_PERMISSION_DENIED`」。这正是 **`faq` 吸水坍缩**的机制根源:判定被权限**倒逼**, +判错也不报错。**"重叠意图合并"在本项目必然撞墙,唯一出路是拆开判定维度与授权维度。** + +**五、影响面** +发布配置(`config_release` + `config_item`,`namespace='agent_tools'`);`bootstrap.py:458` +的 `AgentDefinition.allowed_tools` **交集**逻辑;`E6` 行情出口的权限来源;金标 55 条。 + +**六、降级方案(不受理时)** +保留 `TREND_WHITELIST_INTENT = INTENT_FAQ` 这一**被迫借用**,并**加一条启动自检**: +断言该常量确实落在已发布的工具 key 集合内 ⇒ 任何一次"按语义整理意图码"的动作 +会在**启动时**报错,而不是在客户问行情时**静默失效**。 +(⚠️ 现状:按语义把 `query_fund_trend` 挪到 `product_inquiry`,`E6` **立刻失效并冒泡成错误** +—— 不是兜底话术。这条隐性耦合目前**没有任何守卫**。) + +--- + ## 会签结论 | 组 | 项数 | 结论 | @@ -271,6 +379,7 @@ | 组 4(入参边界对齐) | 5 文件 / 1 张单 | ☑ **受理**(2026-09-20 补签) | | 组 5(NL2SQL 只读边界) | 1 文件 / 1 张单 | ☑ **受理**(2026-09-22 补签) | | 组 6(`GROUP BY` 拼装缺陷) | 1 文件 / 1 张单 | ☑ **受理**(2026-09-22 补签) | +| **组 7**(意图层与检索层三项扩张) | 3 项 / 3 张单(其中 1 项触碰类 3) | ☑ **受理**(2026-09-22,授权后当场落地;见下方补签说明) | **会签人签名 / 日期**:项目 owner(本人会签,`甲-3` 口径:一次性授权 + 逐项留痕) **2026-09-22**(组 1—4 为 2026-09-20) @@ -284,4 +393,13 @@ > 并新增**误杀边界**用例(查询语境的「变更 / 变化 / 导出 / 更新」不得被拦); > 全量 `pytest` 与 55 条金标**逐项零差异**(见 `_W29-NL2SQL接线与对话内图表实施报告-2026-09-22.md` §2)。 +> ✅ **组 7 补签说明(2026-09-22)**:用户在本日对话中明确「授予会签(放行)权限」,三项当场落地。 +> +> **落地证据(按单)**: +> - **会签 20**:`app/service/runtime_config_service.py` 新增 `collection_routes()` / `load_collection_routes()` / `_first_collection_name()`;消费方 `customer_service.py::_narrow_collection_for()`(**配置优先、缺失回落代码常量**)。**零 DDL**;`collection_routes` 列当前全为 `None` ⇒ 实际走**回落**路径,行为与改动前一致。 +> - **会签 21**:`app/core/retrieval_fusion.py` 新增 `fuse_rrf()` + `RRF_K`(与 `fuse` 并存、由调用方选择;只吃名次不吃分数 ⇒ 异质分数不可能污染判定分);`app/service/knowledge_search_service.py::search()` 新增 `literal_parallel: bool = False` 入参(默认值使行为**逐字等同**现状)。⚠️ **工具层透传未做** —— 那需要改 `KnowledgeSearchInput` 契约,**超出本单范围**,另行立单。 +> - **会签 22**:`customer_service.py` 新增 `INTENT_MARKET_QUOTE` 常量 + `TREND_WHITELIST_INTENT_CANDIDATES`(按优先级回落)+ `_trend_whitelist_intent()`(**运行时自检 + 自动回落**,即本单「六、降级方案」要求的守卫,且强于"仅报错")。发布配置 `customer_service:market_quote`(release 260,`allowed_tools=['query_fund_trend']`)已写入并**回读校验**(行数 9 → 10)。**刻意未加入 `supported_intents`**:授权维度与判定维度**解耦**,不改分类候选空间 ⇒ 不动判定分布、不威胁 `MIN_GAP` 窄带。 +> +> **回退方法**:代码按函数级回退;发布配置 `DELETE FROM platform_config_item WHERE release_id=260 AND namespace='agent_tools' AND config_key='customer_service:market_quote'`(或恢复 `_AI工作区\归档\_backup_release260_agent_tools.json` 的 9 行快照)。 + > **口径**:本文件是**追溯留痕**(`甲-3` 已一次性授权,未逐项等待签字)。未受理项须按各组「六、降级方案」执行,并在 `D2.1` 中标注为**降级**。 diff --git a/start.ps1 b/start.ps1 index d1c9884..529834a 100644 --- a/start.ps1 +++ b/start.ps1 @@ -19,7 +19,7 @@ · 新灌的知识写不进 Milvus,客服照旧答不上,且**没有任何报错**。 .PARAMETER Port - API 监听端口,默认 8000。 + API 监听端口,默认 8099。 .PARAMETER SkipChecks 跳过依赖端口检查(MySQL / Redis / Milvus)。 @@ -42,7 +42,7 @@ (端口被占用、或已存在 app.worker 进程时会跳过),可以放心反复双击桌面的启动 .bat。 #> param( - [int]$Port = 8000, + [int]$Port = 8099, [switch]$SkipChecks, [switch]$ApiOnly, [switch]$SkipPriceSync, diff --git a/tests/integration/test_customer_service_trend_chart_persistence.py b/tests/integration/test_customer_service_trend_chart_persistence.py index 5e58485..77f4631 100644 --- a/tests/integration/test_customer_service_trend_chart_persistence.py +++ b/tests/integration/test_customer_service_trend_chart_persistence.py @@ -21,7 +21,9 @@ ## 为什么断言数据、而不是断言像素 -组件渲染已由 `_w29_chart_render_check.mjs`(jsdom)覆盖(11/11)。 +组件渲染已由 `_w29_chart_render_check.mjs`(jsdom)覆盖(11/11),`INV-8` 的**字段级** +判定另由 `tests/unit/service/test_customer_service_trend_chart_inv8.py` 覆盖 +(**不依赖 jsdom,纳入 pytest 门禁**)。 本文件管的是它的**上游**:数据有没有被写进库、有没有出到接口。 分开的理由是**失败时的定位完全不同** —— 「图不显示」既可能是数据没落库(本文件),也可能是渲染出错(那个文件), @@ -109,11 +111,18 @@ async def test_trend_chart_and_service_topic_reach_db_and_api() -> None: assert chart["kind"] == "interval_change" assert chart["source"] == "fin_nav_history" assert chart["intervals"], "区间数据为空 —— 图会画不出来" - # `INV-8` 的落库侧前提:这些数字必须在正文里出现过(正文侧由 jsdom 测试断言)。 + # `INV-8`:图里的每个数字都必须在正文里出现过。 + # 判定口径与 `tests/unit/service/test_customer_service_trend_chart_inv8.py` + # 一致(那边是**无依赖**的纯函数级门禁,覆盖字段级越界;这里补**真实链路**的取样)。 + content = str(result.get("content") or "") for interval in chart["intervals"]: - assert str(interval["change_pct"]) in (result.get("content") or ""), ( - f"图数据里的涨跌 {interval['change_pct']} 未出现在答复正文:{result.get('content')}" - ) + for key in ("change_pct", "start_nav", "end_nav"): + value = interval.get(key) + if value is None: + continue + assert str(value) in content, ( + f"图数据 {key}={value} 未出现在答复正文(`INV-8` 违例):{content}" + ) # ② `W28` 的事项码同批落库(此前同样是「代码里有、库里没有」) assert data.get("svc_topic"), "svc_topic 未落库" diff --git a/tests/unit/core/test_customer_service_rules.py b/tests/unit/core/test_customer_service_rules.py index ec95a54..a0316f8 100644 --- a/tests/unit/core/test_customer_service_rules.py +++ b/tests/unit/core/test_customer_service_rules.py @@ -697,3 +697,75 @@ def test_risk_level_change_rule_still_catches_the_real_requests() -> None: ] for message in caught: assert rules.is_risk_level_change_request(message) is True, message + + +# --------------------------------------------------------------------------- +# `W34` 阶段 0:术语归一化(`normalize_query`) +# +# 守四条约束: +# 1. **只对齐书写差异** —— 典型业务问句必须原样返回(归一化不得成为"改写器"); +# 2. **幂等** —— 跑两次与跑一次结果相同(否则重试 / 幂等键会漂移); +# 3. **不做语义替换** —— 同义词表里不得出现会改语义指向的条目; +# 4. **等级代号大写** —— 库里全大写(`R1`—`R5`/`C1`—`C5`),客户常打小写。 +# --------------------------------------------------------------------------- + + +def test_normalize_query_keeps_plain_business_questions_untouched() -> None: + """普通业务问句**一个字符都不能动** —— 归一化不是改写器。""" + untouched = [ + "南方季季盈90天起投金额是多少", + "基金赎回几天到账", + "季季盈90天的管理费怎么算", + "我想了解定投", + ] + for message in untouched: + assert rules.normalize_query(message) == message, message + + +def test_normalize_query_uppercases_grade_codes() -> None: + """`r1`/`c1` 这类**等级代号**统一成大写,与库内写法对齐。""" + assert rules.normalize_query("r1到r5分别代表什么") == "R1到R5分别代表什么" + assert rules.normalize_query("c1能买什么") == "C1能买什么" + + +def test_normalize_query_does_not_touch_real_latin_words() -> None: + """负向断言:`abc1` 里的 `c` 前面还有字母,不是等级代号,不得被大写。""" + assert rules.normalize_query("abc1") == "abc1" + + +def test_normalize_query_applies_width_and_whitespace_normalization() -> None: + """全角字母/数字走 `NFKC`,连续空白折叠成单空格。""" + assert rules.normalize_query("R1到R5") == "R1到R5" + assert rules.normalize_query("起投金额 是多少") == "起投金额 是多少" + + +def test_normalize_query_applies_synonym_table() -> None: + """同义表按**长度降序**应用:长条目不被短条目切碎。""" + assert rules.normalize_query("7日年化是多少") == "七日年化是多少" + assert rules.normalize_query("风险承受能力评测怎么做") == "风险承受能力测评怎么做" + + +def test_normalize_query_is_idempotent() -> None: + """幂等:`f(f(x)) == f(x)`。不幂等会让重试产生不同查询串。""" + samples = [ + "r1到r5分别代表什么", + "7日年化是多少", + "R1 到 R5", + "南方季季盈90天起投金额是多少", + "", + ] + for message in samples: + once = rules.normalize_query(message) + assert rules.normalize_query(once) == once, message + + +def test_synonym_table_only_contains_writing_variants() -> None: + """同义表纪律:条目形如 `(源, 目标)`、源非空、源与目标不同、且源唯一。""" + sources = [source for source, _ in rules.QUERY_SYNONYMS] + assert len(sources) == len(set(sources)), "同义表存在重复源串" + for source, target in rules.QUERY_SYNONYMS: + assert source and target + assert source != target, source + # 语义类同义(会改变指向)不得进表:这里只做形式校验—— + # 目标串必须包含源串的全部信息(书写差异是"改写",不是"改写语义")。 + assert isinstance(source, str) and isinstance(target, str) diff --git a/tests/unit/service/test_customer_service_trend_chart_inv8.py b/tests/unit/service/test_customer_service_trend_chart_inv8.py new file mode 100644 index 0000000..d331a4e --- /dev/null +++ b/tests/unit/service/test_customer_service_trend_chart_inv8.py @@ -0,0 +1,188 @@ +"""`INV-8` 的 **pytest 落地**:图里的每个数字都必须能在同轮答复正文里找到。 + +## 为什么需要这个文件 + +`INV-8`("图只呈现正文已声明的数字")此前的唯一断言在 `_w29_chart_render_check.mjs` +(jsdom 脚本,**不在 pytest、不在金标**)。后果有两个,都是可查的: + +1. **门禁有洞**:`M-9`(无出处数字 = 0)的取证面**只看答复文本**,图里的数字它看不见 —— + 一旦图画出正文没有的数,图就成了**唯一能绕过数字校验的合规通道**,而没有任何 CI 会报警。 +2. **外部依赖**:`jsdom` 脚本依赖 Node + `jsdom` 包,**换台机器就跑不起来**, + 不能当作回归门禁。 + +本文件把**判定逻辑**搬进 pytest,且**不依赖 jsdom**:直接调用生产代码 +`CustomerServiceAgent._exit_trend()`(`E6` 出口的模板渲染 + 图表装载同在一处), +再比对「图装载的数字集合」与「答复正文的数字集合」。 + +## 为什么"不渲染"也成立 + +图能画出来的数字**只能来自** `CoreResult.data.trend_chart` 这个载荷;渲染层 +(`trend-chart.js`)只做取色与排版,**不计算任何新数值**。因此 +「载荷数字 ⊆ 正文数字」是「渲染后可见数字 ⊆ 正文数字」的**充分条件** —— +前者可在 pytest 里无依赖地断言,后者交给 jsdom 脚本做端到端补充。 + +## 负向对照是必须的 + +只写"断言通过"的测试,无法证明它在**数字改错时**会报警。所以本文件另有一条 +**故意改错**的用例:把载荷里的数字替换成正文里不存在的值,检查器**必须**报出来。 +没有这条对照,门禁可能是空转的。 +""" + +from __future__ import annotations + +import re +from copy import deepcopy +from typing import Any + +from app.core.contracts import IntentResult +from app.service.agent.implementations import customer_service as cs_module + +#: 与 `_w29_chart_render_check.mjs` 的 `digits()` 同一口径:连续数字(含小数)。 +#: 用正则而不是 `float()`:日期(`2026-09-13`)也要逐段比对。 +_NUMBER = re.compile(r"\d+(?:\.\d+)?") + + +def _numbers(value: Any) -> set[str]: + """抽出一个值里**客户能看到的全部数字**。`None` / 空串自然得到空集。""" + if value is None: + return set() + return set(_NUMBER.findall(str(value))) + + +def _chart_numbers(chart: dict[str, Any]) -> set[str]: + """图上可见数字的**全部来源**,与 `trend-chart.js` 的取值一一对应。 + + 刻意把 `label` 也算进去:柱下标签是客户直接读到的文字,它里面若夹带数字, + 同样要能被正文溯源(`INV-8` 管的是"客户看见的数字",不是"数值字段")。 + """ + numbers: set[str] = set() + for key in ("latest_nav", "latest_nav_date", "from_date", "to_date", "series_points"): + numbers |= _numbers(chart.get(key)) + for row in chart.get("intervals") or []: + if not isinstance(row, dict): + continue + for key in ("label", "change_pct", "start_nav", "end_nav"): + numbers |= _numbers(row.get(key)) + for key in ("high", "low"): + bucket = chart.get(key) + if isinstance(bucket, dict): + numbers |= _numbers(bucket.get("nav")) | _numbers(bucket.get("nav_date")) + return numbers + + +def _untraceable(chart: dict[str, Any], text: str) -> list[str]: + """图上出现、但正文里**找不到**的数字(`INV-8` 的违例集合)。""" + return sorted(_chart_numbers(chart) - _numbers(text)) + + +def _build_agent() -> Any: + """与既有单测同一构造方式;`_classified_intent` 由 `handle()` 正常写入,这里直接给定。 + + ⚠️ 必须是 :class:`IntentResult` **对象**而不是字符串 —— `CoreResult.intent` + 是 `IntentResult` 字段(`app/core/contracts.py:135`),塞 `"faq"` 会校验失败。 + """ + agent = cs_module.CustomerServiceAgent(cs_module.CustomerServiceAgent.definition) + agent._classified_intent = IntentResult(intent=cs_module.INTENT_FAQ, confidence=0.9) + return agent + + +def _trend_output() -> dict[str, Any]: + """`summarize_trend` 工具的输出形状(字段名照着 `_exit_trend` 的读取逐个对齐)。""" + return { + "fund_name": "货币ETF南方", + "fund_code": "511810", + "exchange_code": "SSE", + "latest_nav": "0.233200", + "latest_nav_date": "2026-09-13", + "from_date": "2026-05-17", + "to_date": "2026-09-13", + "series_points": 120, + "source": "fin_nav_history", + "intervals": [ + {"name": "近 5 个净值日", "change_pct": "-9.19", + "start_nav": "0.256800", "end_nav": "0.233200", "start_date": "2026-09-09"}, + {"name": "近 20 个净值日", "change_pct": "5.28", + "start_nav": "0.221500", "end_nav": "0.233200", "start_date": "2026-08-25"}, + {"name": "近 60 个净值日", "change_pct": "15.27", + "start_nav": "0.202300", "end_nav": "0.233200", "start_date": "2026-07-16"}, + {"name": "近 120 个净值日", "change_pct": "-6.53", + "start_nav": "0.249500", "end_nav": "0.233200", "start_date": "2026-05-17"}, + ], + "high": {"nav": "0.343500", "nav_date": "2026-05-20"}, + "low": {"nav": "0.177200", "nav_date": "2026-07-21"}, + } + + +def test_trend_chart_only_carries_numbers_already_in_the_answer_text() -> None: + """`INV-8` 正例:图装载的每个数字都能在正文里逐字找到。""" + agent = _build_agent() + result = agent._exit_trend(_trend_output()) + + chart = result.data["trend_chart"] + assert chart["kind"] == "interval_change" + assert chart["source"] == "fin_nav_history" + assert len(chart["intervals"]) == 4 + + violations = _untraceable(chart, result.text or "") + assert violations == [], ( + f"图里有正文没有的数字(`INV-8` 违例):{violations}\n" + f"正文:{result.text}" + ) + + +def test_trend_chart_numbers_are_traceable_for_partial_input() -> None: + """字段缺失(无 high/low、无 series_points)时,仍然一个越界数字都不能有。""" + output = _trend_output() + output.pop("high") + output.pop("low") + output["series_points"] = None + + agent = _build_agent() + result = agent._exit_trend(output) + assert _untraceable(result.data["trend_chart"], result.text or "") == [] + + +def test_inv8_check_reports_tampered_numbers() -> None: + """负向对照:**故意把图里的数字改错,检查器必须报出来**。 + + 没有这条,正例通过只证明"检查器可能没在看"—— 门禁会是空转的。 + 这正是"图里数字改错不报警"那个门禁盲区的回归锁。 + """ + agent = _build_agent() + result = agent._exit_trend(_trend_output()) + + tampered = deepcopy(result.data["trend_chart"]) + tampered["intervals"][0]["change_pct"] = "99.99" + tampered["high"]["nav"] = "1.234567" + + violations = _untraceable(tampered, result.text or "") + assert "99.99" in violations, "改错的涨跌幅没有被报出来 —— 门禁在空转" + assert "1.234567" in violations, "改错的最高净值没有被报出来 —— 门禁在空转" + + +def test_chart_does_not_reintroduce_source_line_numbers_after_truncation() -> None: + """正文被 `_clamp_answer` 截断时,图**仍不得**多出数字。 + + ⚠️ 这是本不变量最脆弱的一处:正文有 `MAX_ANSWER_CHARS` 上限、载荷**没有**。 + 区间很多时正文尾部(`数据来源 / 数据区间 / 共 N 个净值日`)会被切掉, + 而载荷里那几个数字照旧 —— 于是图就会"多出"正文没有的数。 + 本用例把区间数压到上限附近来钉这条边界。 + """ + output = _trend_output() + output["intervals"] = [ + {"name": f"近 {i} 个净值日", "change_pct": f"{i}.{i:02d}", + "start_nav": f"0.{i:06d}", "end_nav": "0.233200", + "start_date": f"2026-0{(i % 9) + 1}-01"} + for i in range(1, 21) + ] + + agent = _build_agent() + result = agent._exit_trend(output) + text = result.text or "" + chart = result.data["trend_chart"] + + # 载荷只装载前 4 个区间(`intervals[:4]`)⇒ 越界只会来自被截断的正文尾部。 + assert len(chart["intervals"]) == 4 + assert _untraceable(chart, text) == [], ( + f"截断后图里出现了正文没有的数字:{_untraceable(chart, text)}" + ) diff --git a/tests/unit/service/test_decision_scope_w35.py b/tests/unit/service/test_decision_scope_w35.py new file mode 100644 index 0000000..cf7aeb6 --- /dev/null +++ b/tests/unit/service/test_decision_scope_w35.py @@ -0,0 +1,273 @@ +"""`W35` 判定口径与融合口径分离:出口判定与原文直返取「向量路原始序列」。 + +## 这个文件在防什么 + +`test_dual_route_wiring_w33.py` 已钉住"写回的 `score` 是向量分、不是融合分"。 +但那只保证**分数本身**没被换掉,**没保证"取哪两条来算 `gap`"**。 + +实测 `A-01`(「南方基金的全称和简称是什么?」)的失败恰恰出在这里: +单路 top1/top2 = `0.8428 / 0.7651`(`gap 0.0777 ≥ MIN_GAP` → `E3`); +融合按 `fused_score`("共识优先")把 `COMP-001`(在向量路里排第 5、分 `0.7992`) +提权到第 2 名 ⇒ 判定读到 `0.8428 − 0.7992 = 0.0436 < MIN_GAP` ⇒ `E3` 被顶成 `E4`。 +**分数没被污染,"名次"被改了** —— 而 `MIN_GAP` 是在单路口径下标定的。 + +## 三条不变量 + +1. 判定用「向量路原始序列」:被融合提权的块**不得**挤进 top1/top2。 +2. 无向量分的块(`payloads()` 写 `score = 0.0`)**不得**参与 `gap`(否则 `gap` 虚高, + 把"其实有并列候选"误判成高置信直返)。 +3. 开关关闭 / 无 `vector_order` ⇒ 回落既有闭环,**与改动前逐字一致**。 +""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from app.core.contracts import AgentRequest, RequestContext +from app.service.agent.implementations import customer_service as cs + +AGENT = cs.CustomerServiceAgent +CUSTOMER = RequestContext(user_id="9001", trace_id="t", roles=("customer",)) + + +def build_agent() -> cs.CustomerServiceAgent: + return AGENT(AGENT.definition) + + +def build_request(message: str = "基金申购费率是多少") -> AgentRequest: + return AgentRequest( + agent_type="customer_service", + message=message, + session_id="s-w35-scope", + idempotency_key="w35-decision-scope-key", + ) + + +def _hit(doc_id: str, score: float, content: str = "") -> dict[str, Any]: + return { + "doc_id": doc_id, + "score": score, + "title": f"标题-{doc_id}", + "content": content or f"正文-{doc_id}", + "family_id": "F1", + } + + +class _ToolStub: + """照 `W33` 夹具:记录每次 `call_tool`,并按 `collection` 返回不同候选。""" + + def __init__(self, by_collection: dict[str, Any]) -> None: + self.calls: list[dict[str, Any]] = [] + self._by_collection = by_collection + + async def __call__(self, name: str, arguments: dict, *, intent: str, context: Any): + self.calls.append({"name": name, "arguments": dict(arguments), "intent": intent}) + return self._by_collection.get(str(arguments.get("collection") or ""), {"hits": []}) + + +# ---- ① 回落:拿不到向量路序列时行为不变 ---- + + +def test_no_vector_order_falls_back() -> None: + """没有 `vector_order`(开关关闭 / 单路)⇒ 返回 `None`,调用方走既有闭环。""" + merged = [_hit("D1", 0.90), _hit("D2", 0.80)] + assert AGENT._vector_decision_hits({"hits": merged}, merged) is None + + +def test_empty_vector_order_falls_back() -> None: + """向量路失败 ⇒ 空序列 ⇒ 同样回落(增益信息拿不到就当没有,不影响原能力)。""" + merged = [_hit("D1", 0.90)] + assert AGENT._vector_decision_hits({"hits": merged, "vector_order": []}, merged) is None + + +def test_all_ordered_ids_missing_falls_back() -> None: + """`vector_order` 里的 id 在融合结果里一个都找不到 ⇒ 回落,而不是返回空序列。""" + merged = [_hit("D1", 0.90)] + out = {"hits": merged, "vector_order": ["X1", "X2"]} + assert AGENT._vector_decision_hits(out, merged) is None + + +# ---- ② 核心:被融合提权的块不得挤进 top1/top2 ---- + + +def test_promoted_block_does_not_enter_decision_sequence() -> None: + """🔴 `A-01` 场景复刻:融合把向量路第 5 名提权到第 2 名,判定仍读向量路前两名。 + + 这是 `M-1` 从 90.9% 回到 100% 的机制本身 —— 谁把 `judge` 换回 `hits`,谁红。 + """ + merged = [ + _hit("FAQ-0001", 0.8428), # 融合第 1、向量路第 1 + _hit("COMP-001", 0.7992), # 融合第 2(被提权上来的)、向量路第 5 + _hit("COMP-001-02", 0.7651), # 融合第 3、向量路第 2 + ] + out = {"hits": merged, "vector_order": ["FAQ-0001", "COMP-001-02", "COMP-001"]} + + agent = build_agent() + judge = AGENT._vector_decision_hits(out, merged) + assert judge is not None + assert [hit["doc_id"] for hit in judge] == [ + "FAQ-0001", + "COMP-001-02", + "COMP-001", + ], "判定序列必须还原向量路原序" + + score = AGENT._score(judge[0]["score"]) + gap = score - agent._second_score(judge) + assert score == pytest.approx(0.8428, abs=1e-6) + assert gap == pytest.approx(0.8428 - 0.7651, abs=1e-6) + assert gap >= cs.MIN_GAP, "判定必须仍看到向量路口径的 gap,否则 E3 会被顶成 E4" + + # 反证:拿融合序列算,gap 掉到 MIN_GAP 以下 —— 这就是原来的失败路径。 + fused_gap = AGENT._score(merged[0]["score"]) - agent._second_score(merged) + assert fused_gap == pytest.approx(0.8428 - 0.7992, abs=1e-6) + assert fused_gap < cs.MIN_GAP, "本用例的构造必须真的能复现退化,否则断言是空转" + + +def test_zero_scored_literal_hit_never_enters_gap() -> None: + """🔴 纯字面命中(`payloads()` 写 `score = 0.0`)不得参与 `gap`。 + + 它若排到第 2 位,`gap` 会被拉到 `0.80` —— "其实有并列候选"被误判成"领先很多", + 高置信直返会把一个仅靠字面重合命中的块当成 top1 的并列证据。 + """ + merged = [ + _hit("D1", 0.80), + _hit("LITERAL-1", 0.0), # 只在字面路命中 ⇒ payloads 写 0.0,融合排第 2 + _hit("D2", 0.70), # 向量路第 2 + ] + out = {"hits": merged, "vector_order": ["D1", "D2"]} + + agent = build_agent() + judge = AGENT._vector_decision_hits(out, merged) + assert judge is not None + assert [hit["doc_id"] for hit in judge] == ["D1", "D2"] + + gap = AGENT._score(judge[0]["score"]) - agent._second_score(judge) + assert gap == pytest.approx(0.10, abs=1e-6) + assert agent._second_score(merged) == 0.0, "本用例要求构造出 0 分块占据第 2 位" + + +# ---- ③ `basic` 补位:唯一一处刻意不分离 ---- + + +def test_basic_promoted_hit_stays_on_top() -> None: + """`basic` 补位块(不在向量路、但**严格更高分**)仍留判定序列首位。 + + `_supplement_basic_explain` 的采用判据就是"严格更高" ⇒ 分离不得把它抹掉, + 否则概念型问句会比开关关闭时更差。 + """ + merged = [ + {"doc_id": "BASIC-9", "score": 0.87, "title": "常识块", + "content": "正文", "collection": cs.BASIC_COLLECTION}, + _hit("FAQ-1", 0.82), + _hit("FAQ-2", 0.70), + ] + out = {"hits": merged, "vector_order": ["FAQ-1", "FAQ-2"]} + + judge = AGENT._vector_decision_hits(out, merged) + assert judge is not None + assert [hit["doc_id"] for hit in judge] == ["BASIC-9", "FAQ-1", "FAQ-2"] + + +def test_lower_scored_non_vector_hit_is_not_promoted() -> None: + """分数**不高于**向量路 top1 的非向量块不得回补(否则分离失效)。""" + merged = [_hit("D1", 0.80), _hit("NARROW-ONLY", 0.0), _hit("D2", 0.70)] + out = {"hits": merged, "vector_order": ["D1", "D2"]} + + judge = AGENT._vector_decision_hits(out, merged) + assert judge is not None + assert [hit["doc_id"] for hit in judge] == ["D1", "D2"] + + +# ---- ④ `vector_order` 的生成:原顺序 + 去重 ---- + + +async def test_vector_order_keeps_route_original_order_and_dedupes( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """🔴 `vector_order` 必须是向量路**返回的原始顺序**(且去重)。 + + ⚠️ 不能对融合结果"按分数重排":① 融合结果的名字已被 `fused_score` 改过; + ② 同一 `doc_id` 在该路可能返回多条(`fuse` 对它们取 `max`)。 + 本用例特意让分数序与返回序**不一致**(`C` 分高于 `B` 但排在 `B` 后), + 并让 `B` 重复出现 —— 两种写错方式都会在这里红。 + """ + monkeypatch.setattr(cs, "DUAL_ROUTE_ENABLED", True) + stub = _ToolStub({ + "": {"hits": [_hit("A", 0.90), _hit("B", 0.70), _hit("C", 0.80), _hit("B", 0.70)]}, + cs.PRODUCT_COLLECTION: {"hits": [_hit("C", 0.99)]}, + }) + agent = build_agent() + agent.call_tool = stub # type: ignore[method-assign] + + out = await agent._dual_route_output( + build_request(), CUSTOMER, cs.INTENT_PRODUCT, cs.TOOL_NAME + ) + assert out is not None + assert out["vector_order"] == ["A", "B", "C"], "必须是向量路原顺序且去重" + + +async def test_vector_route_failure_yields_empty_order( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """向量路失败 ⇒ `vector_order` 为空、判定回落 —— 但**另一路的命中照常返回**。""" + monkeypatch.setattr(cs, "DUAL_ROUTE_ENABLED", True) + stub = _ToolStub({ + "": {"degraded": True, "reason": "search_failed"}, + cs.PRODUCT_COLLECTION: {"hits": [_hit("D2", 0.90)]}, + }) + agent = build_agent() + agent.call_tool = stub # type: ignore[method-assign] + + out = await agent._dual_route_output( + build_request(), CUSTOMER, cs.INTENT_PRODUCT, cs.TOOL_NAME + ) + assert out is not None + assert out["vector_order"] == [] + assert [hit["doc_id"] for hit in out["hits"]] == ["D2"], "另一路的结果不得被丢" + assert AGENT._vector_decision_hits(out, out["hits"]) is None + + +# ---- ⑤ 端到端:判定与「原文直返的内容」同源 ---- + + +async def test_direct_reply_follows_vector_scope( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """🔴 端到端:融合把 `COMP-001` 顶到第 1 名时,直返的仍是**向量路 top1**。 + + 构造:向量路 `FAQ-0001 0.84` / `COMP-001-02 0.7651`(`COMP-001 0.7992` 在向量路下方); + 收窄路给 `COMP-001` 0.95 ⇒ 融合分 `0.6×0.7992 + 0.4×0.95 = 0.85952`,**压过 `FAQ-0001` + 的 `0.6×0.84 = 0.504`** ⇒ 融合第 1 名成了 `COMP-001`。 + + 若不分离:判定读 `COMP-001(0.7992) − FAQ-0001(0.84) = −0.0408` ⇒ 既不满足高置信、 + 也不满足 gap,直接掉进澄清/部分答 —— 而单路口径下这条本该是 `E3` 直返。 + """ + monkeypatch.setattr(cs, "DUAL_ROUTE_ENABLED", True) + # 只放开"主体相关性闸门"这一维度(它另有专门用例),不放宽被测的判定路径。 + monkeypatch.setattr( + AGENT, "_subject_covered_by", classmethod(lambda cls, hits, terms: True) + ) + monkeypatch.setattr( + AGENT, "_is_general_knowledge_question", classmethod(lambda cls, message: False) + ) + stub = _ToolStub({ + "": {"hits": [ # 向量路:原序(FAQ-0001 第一、COMP-001-02 第二、COMP-001 在其后) + _hit("FAQ-0001", 0.8428, "南方基金管理股份有限公司,简称南方基金。"), + _hit("COMP-001-02", 0.7651, "公司全称与简称的说明。"), + _hit("COMP-001", 0.7992, "这是被融合提权上来的那一块。"), + ]}, + cs.PRODUCT_COLLECTION: {"hits": [_hit("COMP-001", 0.95, "这是被融合提权上来的那一块。")]}, + }) + agent = build_agent() + agent.call_tool = stub # type: ignore[method-assign] + + result = await agent._answer_from_knowledge( + build_request("南方基金的全称是什么"), CUSTOMER, cs.INTENT_PRODUCT + ) + assert result.exit_code == cs.EXIT_KNOWLEDGE_DIRECT, ( + f"应走知识直返,实际 {result.exit_code}(判定被融合名次改写了)" + ) + assert "简称南方基金" in result.text, "直返内容必须来自向量路 top1" + assert "被融合提权" not in result.text, "直返内容不得是融合提权上来的块" diff --git a/tests/unit/service/test_evidence_pack_order_w36.py b/tests/unit/service/test_evidence_pack_order_w36.py new file mode 100644 index 0000000..a329362 --- /dev/null +++ b/tests/unit/service/test_evidence_pack_order_w36.py @@ -0,0 +1,263 @@ +"""`W36` 选块口径归一:`_evidence_pack` 的**遍历顺序与 `top` 取值**改取「向量路名次」。 + +## 这个文件在防什么 + +`W35` 已把**判定**(`score`/`gap`)与**原文直返的内容**分离到「向量路原始序列」上 +(见 `test_decision_scope_w35.py`),但 `_evidence_pack` **仍吃融合序列 `hits`**。 + +实测 `E-01`(多轮 + 代词:「南方稳健增利债券 A 的起投金额是多少?」→「**它**费率多少?」) +的残余退化就出在这里。两态下: + +- **向量路返回逐字一致**(top1 `PROD-002-15` `0.8893` / top2 `ETF-009-16` `0.8220` + ⇒ `gap 0.0673 < MIN_GAP`,两态都进 `E4`); +- `score ≥ E4_MIN_SCORE` 的块集合也一致 —— 因为 `payloads()` 写回的 `score` 是**向量分**, + 不在向量路里的块被显式置 `0.0`,根本过不了 `E4_MIN_SCORE = 0.5` 这道坎; + +⇒ 唯一差异是**遍历路序**:融合按 `fused_score`("共识优先")把 `PROD-002` 父块从单路 +第 10 位提权到第 3 位,于是 `E4_MAX_EVIDENCE` 的 6 个名额被**不同的块**填满 ⇒ +证据包内容不同 ⇒ 模型输出不同 ⇒ `E3` 直返变成 `E5b`(部分答 + 引导,相对降级)。 + +## 三条不变量 + +1. **`order` 只当排序键、不当过滤器**:成员集合一个不动,不可能因为"某块不在向量路里" + 就把它从证据包里丢掉。 +2. **缺省 `None` ⇒ 逐字等同改动前**(开关关闭时走的就是这条路),这是与基线对账的前提。 +3. **传了 `order` 后**,两态的证据包(含被截断的 6 块与 `top` 补位结果)必须一致。 + +## ⚠️ 为什么不能改成"按分数降序" + +直觉上"把遍历顺序改成按 `score` 降序"更简单 —— 但**错的**。向量路返回**并非**严格按分降序: +`knowledge_search_service.py:291-295` 把 `sections[0]`(整节块)**刻意放在末尾**, +以免它参与 top1/top2 判定(实测挤到第 2 位会让 `gap` 从 `0.090` 掉到 `0.076`)。 +按分重排会把它提上来 ⇒ 反而动到开关关闭时的行为。 +""" + +from __future__ import annotations + +from typing import Any + +from app.service.agent.implementations import customer_service as cs + +AGENT = cs.CustomerServiceAgent + +GAP_BELOW = 0.06 # < MIN_GAP(0.07) ⇒ 必定进 E4 分支 + + +def _hit( + doc_id: str, + score: float, + *, + title: str | None = None, + source_file: str = "", +) -> dict[str, Any]: + """构造一条命中;`title` 不含 ` · ` 或 `source_file` 为空时取不到章节键。""" + return { + "doc_id": doc_id, + "score": score, + "title": title if title is not None else f"标题-{doc_id}", + "content": f"正文-{doc_id}", + "source_file": source_file, + "family_id": "F1", + } + + +def _ids(pack: list[dict[str, Any]] | None) -> list[str]: + assert pack is not None, "本用例的输入必定进 E4,不该返回 None" + return [str(hit.get("doc_id") or "") for hit in pack] + + +# ---- ① `_pack_order`:只换顺序 ---- +def test_pack_order_none_is_identity() -> None: + """缺省 `None` ⇒ 原样返回(连容器都是新的,不改调用方持有的列表)。""" + hits = [_hit("D1", 0.9), _hit("D2", 0.8)] + ordered = AGENT._pack_order(hits, None) + assert [h["doc_id"] for h in ordered] == ["D1", "D2"] + assert ordered is not hits, "必须是新列表,避免下游 append 污染原命中序列" + + +def test_pack_order_empty_is_identity() -> None: + """空序列(该路失败)⇒ 同缺省,不因"拿到了空增益"改变行为。""" + hits = [_hit("D1", 0.9)] + assert _ids_ok(AGENT._pack_order(hits, [])) == ["D1"] + + +def test_pack_order_unresolvable_is_identity() -> None: + """`order` 里的 id 一条都对不上 ⇒ 回落原顺序,而不是返回空。""" + hits = [_hit("D1", 0.9), _hit("D2", 0.8)] + assert [h["doc_id"] for h in AGENT._pack_order(hits, ["X9", "X8"])] == ["D1", "D2"] + + +def test_pack_order_puts_ordered_ids_first_then_rest_in_place() -> None: + """`order` 命中的块按 `order` 排在前,其余按**原相对顺序**追加在后。""" + hits = [_hit("A", 0.9), _hit("B", 0.8), _hit("C", 0.7), _hit("D", 0.6)] + assert [h["doc_id"] for h in AGENT._pack_order(hits, ["C", "A"])] == [ + "C", + "A", + "B", + "D", + ] + + +def test_pack_order_never_drops_members() -> None: + """🔴 不变量①:`order` 只当排序键 —— 成员集合必须逐字不变。""" + hits = [_hit("A", 0.9), _hit("B", 0.8), _hit("C", 0.7)] + ordered = AGENT._pack_order(hits, ["C"]) + assert {h["doc_id"] for h in ordered} == {"A", "B", "C"} + assert len(ordered) == 3, "重复 id 不得让某一块出现两次" + + +def test_pack_order_ignores_repeated_ids() -> None: + """`order` 里同一 id 重复出现(同一路可能返回多条同 doc_id)⇒ 只取一次。""" + hits = [_hit("A", 0.9), _hit("B", 0.8)] + assert [h["doc_id"] for h in AGENT._pack_order(hits, ["B", "B", "A"])] == ["B", "A"] + + +# ---- ② 逐字对账:单路(不给 order)不受影响 ---- +def test_evidence_pack_default_matches_explicit_none() -> None: + """`order` 缺省与显式 `None` 必须完全一致 —— 开关关闭时的路径只能有一条。""" + hits = [ + _hit("G1", 0.90, title="手册 · 第一章 · 小节A", source_file="手册.md"), + _hit("G2", 0.88, title="手册 · 第一章 · 小节B", source_file="手册.md"), + _hit("O1", 0.95), + _hit("O2", 0.94), + ] + assert _ids(AGENT._evidence_pack(hits, gap=GAP_BELOW)) == _ids( + AGENT._evidence_pack(hits, gap=GAP_BELOW, order=None) + ) + + +def test_evidence_pack_identity_order_is_noop() -> None: + """把"命中自身的顺序"当 `order` 传进去 ⇒ 与不传等价(幂等)。""" + hits = [ + _hit("G1", 0.90, title="手册 · 第一章 · 小节A", source_file="手册.md"), + _hit("G2", 0.88, title="手册 · 第一章 · 小节B", source_file="手册.md"), + _hit("O1", 0.95), + _hit("O2", 0.94), + ] + same = [h["doc_id"] for h in hits] + assert _ids(AGENT._evidence_pack(hits, gap=GAP_BELOW, order=same)) == _ids( + AGENT._evidence_pack(hits, gap=GAP_BELOW) + ) + + +def test_evidence_pack_gap_floor_unchanged() -> None: + """`gap ≥ MIN_GAP` 仍直接返回 `None`(不接管)—— 这条闸门一个字没动。""" + hits = [_hit("D1", 0.9), _hit("D2", 0.8)] + assert AGENT._evidence_pack(hits, gap=cs.MIN_GAP) is None + assert AGENT._evidence_pack(hits, gap=cs.MIN_GAP, order=["D1"]) is None + + +# ---- ③ 截断路径(`W2` 跨章节兜底):6 个名额跟 `order` 走 ---- +def test_fallback_truncation_follows_vector_order() -> None: + """🔴 `E-01` 机制复刻:融合把第 8 位的块提权到第 2 位 ⇒ 6 个名额换人。 + + 全部命中都取不到章节键(标题无 ` · `)⇒ 走"跨章节兜底"分支, + 证据包 = 按遍历顺序取的**前 6 个达标块** ⇒ 纯顺序敏感,正好暴露该缺陷。 + """ + scores = { + "V1": 0.90, "V2": 0.88, "V3": 0.86, "V4": 0.84, + "V5": 0.82, "V6": 0.80, "V7": 0.70, "P": 0.60, + } + vector_ids = ["V1", "V2", "V3", "V4", "V5", "V6", "V7", "P"] # 单路:按分降序 + fused_ids = ["V1", "P", "V2", "V3", "V4", "V5", "V6", "V7"] # 融合:`P` 被提权到第 2 + + vec_hits = [_hit(d, scores[d]) for d in vector_ids] + fused_hits = [_hit(d, scores[d]) for d in fused_ids] + + off = _ids(AGENT._evidence_pack(vec_hits, gap=GAP_BELOW)) + assert off == ["V1", "V2", "V3", "V4", "V5", "V6"], "单路口径取前 6 名" + + dual_without = _ids(AGENT._evidence_pack(fused_hits, gap=GAP_BELOW)) + assert dual_without == ["V1", "P", "V2", "V3", "V4", "V5"], ( + "不给 order 时确实会换人 —— 这正是 `E-01` 的病因,本断言把它钉住" + ) + + dual_with = _ids( + AGENT._evidence_pack(fused_hits, gap=GAP_BELOW, order=vector_ids) + ) + assert dual_with == off, "给了向量路名次后,两态证据包必须逐字一致" + + +# ---- ④ 章节组成员 + 补足:补足顺序跟 `order` 走 ---- +def test_group_supplement_order_follows_vector_order() -> None: + """🔴 章节组先入包,再用达标块**补足**到 6 —— 补足顺序也必须跟 `order`。""" + title_a = "手册 · 第一章 · 小节A" + title_b = "手册 · 第一章 · 小节B" + scores = {"O1": 0.95, "O2": 0.94, "O3": 0.93, "O4": 0.92, "O5": 0.91, + "GA": 0.80, "GB": 0.78} + vector_ids = ["O1", "O2", "O3", "O4", "O5", "GA", "GB"] + fused_ids = ["O5", "O1", "GA", "GB", "O2", "O3", "O4"] # `O5` 被提权到第 1 + + def build(ids: list[str]) -> list[dict[str, Any]]: + out: list[dict[str, Any]] = [] + for doc_id in ids: + if doc_id == "GA": + out.append(_hit("GA", scores["GA"], title=title_a, source_file="手册.md")) + elif doc_id == "GB": + out.append(_hit("GB", scores["GB"], title=title_b, source_file="手册.md")) + else: + out.append(_hit(doc_id, scores[doc_id])) + return out + + vec_hits, fused_hits = build(vector_ids), build(fused_ids) + + off = _ids(AGENT._evidence_pack(vec_hits, gap=GAP_BELOW)) + assert off[:2] == ["GA", "GB"], "章节组先入包(组内按分降序)" + assert len(off) == cs.E4_MAX_EVIDENCE + + dual_without = _ids(AGENT._evidence_pack(fused_hits, gap=GAP_BELOW)) + assert dual_without != off, "补足顺序确实会被融合序改写(病因)" + + dual_with = _ids( + AGENT._evidence_pack(fused_hits, gap=GAP_BELOW, order=vector_ids) + ) + assert dual_with == off, "给了向量路名次后,补足结果必须与单路一致" + + +# ---- ⑤ `top` 补位:取的是向量路那一块 ---- +def test_top_replacement_follows_vector_order() -> None: + """🔴 章节组占满 6 个名额时,`top` 会挤掉末位 —— `top` 必须取向量路 top1。""" + titles = [f"手册 · 第一章 · 小节{i}" for i in range(1, 7)] + group_scores = [0.90, 0.89, 0.88, 0.87, 0.86, 0.85] + group_ids = [f"GA{i}" for i in range(1, 7)] + + group = [ + _hit(doc_id, score, title=title, source_file="手册.md") + for doc_id, score, title in zip(group_ids, group_scores, titles, strict=True) + ] + tail = [_hit("T1", 0.60), _hit("T2", 0.59)] + + vec_hits = [*group, *tail] # 单路:按分降序 + fused_ids = ["T2", *group_ids, "T1"] # 融合:`T2` 被提权到第 1 + by_id = {h["doc_id"]: h for h in [*group, *tail]} + fused_hits = [by_id[d] for d in fused_ids] + + off = _ids(AGENT._evidence_pack(vec_hits, gap=GAP_BELOW)) + assert off == group_ids, "章节组自身就占满了上限,且 top1 已在包内 ⇒ 不换人" + + dual_without = _ids(AGENT._evidence_pack(fused_hits, gap=GAP_BELOW)) + assert dual_without[-1] == "T2", "不给 order 时 `top` 会变成 T2 并挤掉末位(病因)" + + dual_with = _ids( + AGENT._evidence_pack(fused_hits, gap=GAP_BELOW, order=[h["doc_id"] for h in vec_hits]) + ) + assert dual_with == off, "给了向量路名次后,`top` 补位结果必须与单路一致" + + +# ---- ⑥ 成员集合不因 order 改变 ---- +def test_order_never_changes_membership() -> None: + """🔴 不变量①在 `_evidence_pack` 层面的复验:换序不得让任何达标块凭空消失。""" + hits = [ + _hit("G1", 0.90, title="手册 · 第一章 · 小节A", source_file="手册.md"), + _hit("G2", 0.88, title="手册 · 第一章 · 小节B", source_file="手册.md"), + _hit("O1", 0.70), + _hit("OFF1", 0.65), # 不在 order 里(融合路独有)⇒ 仍必须有机会进包 + ] + pack = AGENT._evidence_pack(hits, gap=GAP_BELOW, order=["G1", "G2", "O1"]) + ids = _ids(pack) + assert "OFF1" in ids, "不在 order 里的达标块不得被丢弃(只当排序键)" + assert {"G1", "G2", "O1"} <= set(ids) + + +def _ids_ok(ordered: list[Any]) -> list[str]: + return [str(hit.get("doc_id") or "") for hit in ordered]