张胜宇
2c3a5188fb
- app/core/retrieval_fusion.py（新增）: 两路候选加权融合，三层分数分离 - app/service/retrieval_rerank_service.py（新增）: LLM 精排，只出 id / 越界剔除 / 失败静默回落 - customer_service.py: 接线，开关 CS_DUAL_ROUTE / CS_RERANK，未设置时一行都不执行 - 测试 61 条（融合 22 / 精排 22 / 接线 17） - agent_intent_config 补 5 条 active（1 -> 6） 铁律: vector_score 只用于出口判定，绝不改写；融合与精排只改「看哪些块、什么顺序」 门禁: 2608 passed / 3 skipped / 0 failed；金标 55 条逐项零差异，M-1 恒 100%
193 lines
8.5 KiB
Python
193 lines
8.5 KiB
Python
"""LLM 精排：只对**已成形证据包内的块**做排序（`W33` 一期，**纯新增，免会签**）。
|
||
|
||
## 它在链路里的位置（只此一处）
|
||
|
||
```
|
||
检索 → 融合 → 出口判定（E3/E4/E5）→ E4 组装证据包 → 【本模块排序】→ E4 生成答复
|
||
↑ 就是这里，之前没有任何东西
|
||
```
|
||
|
||
## 五条硬边界（每条都有理由，不是风格偏好）
|
||
|
||
1. **只在 `E4` 内生效**。`E4` 的先决条件是 `gap < MIN_GAP`（分数接近、需要合并作答）。
|
||
在此**之外**做精排会把 `E3` 的"原文直返、零生成调用"整批拖进模型 —— 那正是最该省的部分。
|
||
2. **不碰判定分数**。本模块**不改任何 `score`**，只改**列表顺序**。理由见
|
||
`app/core/retrieval_fusion.py` 头部：`MIN_GAP` 的实测可选取值区间是 `(0.0649, 0.0759)`，
|
||
裕度约 `0.005`；改分数就会翻出口。
|
||
3. **默认只排序、不删块**（`mode="order"`）。证据包大小由既有的 `E4_MAX_EVIDENCE` 控制，
|
||
本模块不参与取舍 —— 少给一块证据就可能把 `M-4 事实正确率` 打下来，而"排序"已经能拿到
|
||
大部分收益（生成模型对靠前的证据注意力更高）。`mode="select"` 仅作为可度量的备选模式存在，
|
||
**默认不启用**。
|
||
4. **只认 id、不认别的**。输出契约固定为 `{"ordered_ids": [...]}`，**不接受理由字段** ——
|
||
理由会引入新事实与新数字，触碰 `M-9 无出处数字 = 0` 这条零容忍指标。
|
||
5. **失败一律静默回落原顺序**（超时 / 非 JSON / 异常 / id 非法），并把原因写进
|
||
`RerankOutcome.reason` 供留痕。这是全仓一致的范式：**增益调用坏了，不能让本来能答的
|
||
问题变成答不出来**（同一取向见 `customer_service.py` 的 `_supplement_basic_explain`）。
|
||
|
||
## 档位安全
|
||
|
||
本模块**不持有 Milvus 客户端、不做任何检索**；输入必须是**已经过 `visibility` 过滤**的证据包。
|
||
`select` 模式也只能从给定候选里挑，**不得新增 id** —— 越界 id 直接忽略。
|
||
少这一条约束，访客内容就可能经"精排"这一侧被读出来（`INV-1 档位不可越`）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
import re
|
||
from collections.abc import Awaitable, Callable, Sequence
|
||
from dataclasses import dataclass
|
||
from typing import Any
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
#: 端点解析器：返回可用的模型端点列表（由调用方注入，便于单测与解耦）。
|
||
ResolveEndpoints = Callable[[], Awaitable[list[object]]]
|
||
#: 生成函数：`(endpoints, prompt) -> 带 .text 的对象`（生产上绑 `Agent.generate_with_model`）。
|
||
GenerateFn = Callable[[list[object], str], Awaitable[Any]]
|
||
|
||
#: 精排提示词：**极简、只出 JSON、不给理由**。
|
||
RERANK_SYSTEM = (
|
||
"你是检索结果排序器。你只能依据给出的候选来排序，"
|
||
"不得引入候选之外的任何信息、名称或数字，也不得解释理由。"
|
||
)
|
||
|
||
_TEMPLATE = """按「与用户问题的相关性」给下面的候选块排序。
|
||
|
||
规则：
|
||
- 只能使用给出的 id，不得新增、不得改写 id。
|
||
- 只输出一个 JSON 对象，字段为 ordered_ids，值为 id 的数组。
|
||
- 不要输出理由、不要输出 Markdown、不要输出其它任何字段。
|
||
|
||
输出格式：{{"ordered_ids": ["id1", "id2"]}}
|
||
|
||
用户问题：{question}
|
||
|
||
候选：
|
||
{candidates}"""
|
||
|
||
#: 单块送模型的正文上限（够判断相关性即可，省 token 也少一次越界机会）。
|
||
_CONTENT_PREVIEW = 240
|
||
_MAX_QUESTION = 500
|
||
|
||
_JSON_OBJECT = re.compile(r"\{.*\}", re.DOTALL)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class RerankOutcome:
|
||
"""精排结果。`applied=False` 时 `evidence` 与输入**同序**，`reason` 说明回落原因。"""
|
||
|
||
evidence: list[dict[str, Any]]
|
||
applied: bool = False
|
||
reason: str = ""
|
||
#: 模型给出的合法顺序（仅用于留痕与排查；越界 id 已在解析时剔除）。
|
||
ordered_ids: tuple[str, ...] = ()
|
||
|
||
|
||
class RetrievalRerankService:
|
||
"""证据包精排。构造后**无状态**，可复用。"""
|
||
|
||
def __init__(
|
||
self,
|
||
resolve_endpoints: ResolveEndpoints,
|
||
generate: GenerateFn,
|
||
*,
|
||
timeout_seconds: float = 6.0,
|
||
mode: str = "order",
|
||
) -> None:
|
||
if mode not in ("order", "select"):
|
||
raise ValueError("mode 只能是 'order' 或 'select'")
|
||
if timeout_seconds <= 0:
|
||
raise ValueError("timeout_seconds 必须为正")
|
||
self._resolve = resolve_endpoints
|
||
self._generate = generate
|
||
self._timeout = timeout_seconds
|
||
self._mode = mode
|
||
|
||
async def rerank(
|
||
self, question: str, evidence: Sequence[dict[str, Any]], *, limit: int | None = None
|
||
) -> RerankOutcome:
|
||
items = [item for item in evidence if isinstance(item, dict)]
|
||
# 0 或 1 块没有可排的余地 —— **不要为它花一次模型调用**
|
||
if len(items) <= 1:
|
||
return RerankOutcome(evidence=list(items), reason="nothing_to_rank")
|
||
|
||
ids = [str(item.get("doc_id") or "") for item in items]
|
||
if any(not doc_id for doc_id in ids):
|
||
# 缺 doc_id 的块无法被引用，排序意义不完整 ⇒ 不排（而不是猜）
|
||
return RerankOutcome(evidence=list(items), reason="missing_doc_id")
|
||
|
||
prompt = self._build_prompt(str(question or "")[:_MAX_QUESTION], items)
|
||
try:
|
||
endpoints = await self._resolve()
|
||
execution = await asyncio.wait_for(
|
||
self._generate(list(endpoints), prompt), timeout=self._timeout
|
||
)
|
||
except TimeoutError:
|
||
return RerankOutcome(evidence=list(items), reason="timeout")
|
||
except Exception: # noqa: BLE001 —— 增益调用，任何异常都只回落
|
||
logger.info("精排调用失败，回落原顺序", exc_info=True)
|
||
return RerankOutcome(evidence=list(items), reason="call_failed")
|
||
|
||
ordered_ids = self._parse(getattr(execution, "text", "") or "", set(ids))
|
||
if not ordered_ids:
|
||
return RerankOutcome(evidence=list(items), reason="unparsable_output")
|
||
|
||
order_index = {doc_id: index for index, doc_id in enumerate(ordered_ids)}
|
||
by_id = {str(item.get("doc_id")): item for item in items}
|
||
# 模型漏掉的块追加到尾部，保持它们在原顺序里的相对位置 —— **不丢块**
|
||
tail = [item for item in items if str(item.get("doc_id")) not in order_index]
|
||
ordered = [by_id[doc_id] for doc_id in ordered_ids] + tail
|
||
|
||
if self._mode == "select" and limit is not None and 0 < limit < len(ordered):
|
||
ordered = ordered[:limit]
|
||
|
||
return RerankOutcome(
|
||
evidence=ordered, applied=True, reason="", ordered_ids=tuple(ordered_ids)
|
||
)
|
||
|
||
@staticmethod
|
||
def _build_prompt(question: str, items: Sequence[dict[str, Any]]) -> str:
|
||
lines: list[str] = []
|
||
for item in items:
|
||
doc_id = str(item.get("doc_id") or "")
|
||
title = str(item.get("title") or "") or "未命名"
|
||
content = str(item.get("content") or "").strip()[:_CONTENT_PREVIEW]
|
||
lines.append(f"id={doc_id} 标题={title}\n{content}")
|
||
return f"{RERANK_SYSTEM}\n\n" + _TEMPLATE.format(
|
||
question=question, candidates="\n---\n".join(lines)
|
||
)
|
||
|
||
@staticmethod
|
||
def _parse(text: str, allowed: set[str]) -> list[str]:
|
||
"""取 `ordered_ids`；**只保留候选集合内的 id**（越界 id 一律剔除，不报错）。
|
||
|
||
容错三件事与 `E4` 的 `_parse_evidence_output` 同取向：去代码块围栏、允许 JSON 前后
|
||
有解释性文字、只认第一个对象。
|
||
"""
|
||
raw = text.strip()
|
||
if raw.startswith("```"):
|
||
raw = raw.removeprefix("```")
|
||
raw = raw.removeprefix("json").removesuffix("```").strip()
|
||
match = _JSON_OBJECT.search(raw)
|
||
if match is None:
|
||
return []
|
||
try:
|
||
payload = json.loads(match.group(0))
|
||
except (json.JSONDecodeError, TypeError):
|
||
return []
|
||
if not isinstance(payload, dict):
|
||
return []
|
||
values = payload.get("ordered_ids")
|
||
if not isinstance(values, (list, tuple)):
|
||
return []
|
||
out: list[str] = []
|
||
for value in values:
|
||
doc_id = str(value)
|
||
# 越界 / 重复：剔除。**这是 `INV-1` 的一道防线** ——模型不得借精排引入新的块。
|
||
if doc_id in allowed and doc_id not in out:
|
||
out.append(doc_id)
|
||
return out
|