395 lines
16 KiB
Python
395 lines
16 KiB
Python
"""知识检索服务（Task 6）：意图 → 集合路由 + MySQL 二次校验 + 失败降级。
|
||
|
||
主链路（向量召回）::
|
||
|
||
KnowledgeQuery ──(白名单校验)──> embed ──> MilvusKnowledgeClient.search
|
||
└─> MySQL 二次校验（已发布 + 生效/失效日期）──> KnowledgeHit
|
||
|
||
失败降级：Milvus 不可用、embedding 维度不符之外的一切异常，**不冒泡**到问答主链路，
|
||
改为 MySQL `LIKE` 关键词检索，并在 `KnowledgeSearchResult.degraded=True` 上如实标注
|
||
（`degradation_reason` 给出可观测原因）。降级路径**同样**执行「已发布 + 有效期」过滤 ——
|
||
降级只是换召回通道，不放松可见性。
|
||
|
||
为什么生效日期过滤不在 Milvus 侧：三个集合的 schema 里没有 `effective_date` /
|
||
`expire_date` 字段（见 Task 6 计划 §schema）。无条件在向量库侧拼该过滤会让产品/政策集合
|
||
永远查不到数据；因此有效期一律以 `fin_knowledge_meta` 为准，在 MySQL 侧二次校验。
|
||
|
||
`intent` 是**稀疏标签**：写侧对普通知识省略该字段，契约里 `None` 表示"无显式标签、由集合名
|
||
推断"。因此读侧把「字段缺失」「空串」「非字符串」都当无标签处理，再按集合名回退
|
||
（`COLLECTION_INTENTS`），**不能只判 `is None`**。
|
||
|
||
集合名由本服务的路由表决定，**不接受调用方指定**；`KnowledgeQuery` 本身也没有集合名字段。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
from collections.abc import Callable, Mapping, Sequence
|
||
from datetime import UTC, date, datetime
|
||
from typing import Any
|
||
|
||
from sqlalchemy import bindparam, text
|
||
|
||
from app.core.errors import ForbiddenAgentError, RecoverableAgentError, ValidationAgentError
|
||
from app.core.knowledge_contracts import (
|
||
ALLOWED_COLLECTIONS,
|
||
VECTOR_DIM,
|
||
KnowledgeHit,
|
||
KnowledgeQuery,
|
||
KnowledgeSearchResult,
|
||
)
|
||
from app.infrastructure.db import SessionFactory
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
PUBLISHED_REVIEW_STATUS = "published"
|
||
ACTIVE_STATUS = "active"
|
||
|
||
#: 意图 → 检索集合。集合名不暴露给调用方，只由本表决定。
|
||
DEFAULT_ROUTES: dict[str, tuple[str, int]] = {
|
||
"faq": ("fin_faq_collection", 3),
|
||
"product_inquiry": ("fin_product_collection", 5),
|
||
"policy_explain": ("fin_policy_collection", 5),
|
||
}
|
||
|
||
#: 无显式 `intent` 标签时按集合名推断（`intent` 是稀疏标签，缺字段/空串都算无标签）。
|
||
COLLECTION_INTENTS: dict[str, str] = {
|
||
"fin_faq_collection": "faq",
|
||
"fin_product_collection": "product_inquiry",
|
||
"fin_policy_collection": "policy_explain",
|
||
}
|
||
|
||
#: 单次检索返回条数硬上限，防多意图叠加放大。
|
||
RESULT_LIMIT = 20
|
||
|
||
#: LIKE 关键词检索的关键词长度上限（防止构造超长模式）。
|
||
MAX_KEYWORD_LENGTH = 100
|
||
|
||
#: LIKE 通配符转义符。
|
||
LIKE_ESCAPE = "\\"
|
||
|
||
_HITS_SQL = text(
|
||
"""
|
||
SELECT id, milvus_collection, title, version, tags, content_text,
|
||
effective_date, expire_date, review_status, status
|
||
FROM fin_knowledge_meta
|
||
WHERE id IN :knowledge_ids
|
||
"""
|
||
).bindparams(bindparam("knowledge_ids", expanding=True))
|
||
|
||
_DEGRADED_SQL = text(
|
||
"""
|
||
SELECT id, milvus_collection, title, version, tags, content_text,
|
||
effective_date, expire_date, review_status, status
|
||
FROM fin_knowledge_meta
|
||
WHERE milvus_collection IN :collections
|
||
AND content_text LIKE :pattern ESCAPE :escape
|
||
ORDER BY updated_at DESC
|
||
LIMIT :limit
|
||
"""
|
||
).bindparams(bindparam("collections", expanding=True))
|
||
|
||
|
||
class KnowledgeRuntimeConfig:
|
||
"""检索运行配置：路由表 + 单次返回上限。默认即生产口径。"""
|
||
|
||
def __init__(
|
||
self,
|
||
*,
|
||
routes: Mapping[str, tuple[str, int]] | None = None,
|
||
result_limit: int = RESULT_LIMIT,
|
||
) -> None:
|
||
self.routes: dict[str, tuple[str, int]] = dict(routes or DEFAULT_ROUTES)
|
||
self.result_limit = max(1, int(result_limit))
|
||
|
||
def route(self, intent: str) -> tuple[str, int] | None:
|
||
return self.routes.get(intent)
|
||
|
||
|
||
class KnowledgeRetrievalService:
|
||
"""检索入口。`client` / `embedder` 用鸭子类型注入（不 import 具体实现，便于单测）。"""
|
||
|
||
def __init__(
|
||
self,
|
||
client: Any,
|
||
*,
|
||
embedder: Any = None,
|
||
session_factory: Callable[[], Any] | None = None,
|
||
config: KnowledgeRuntimeConfig | None = None,
|
||
vector_dim: int = VECTOR_DIM,
|
||
) -> None:
|
||
self.client = client
|
||
self.embedder = embedder
|
||
self._session_factory: Callable[[], Any] = session_factory or SessionFactory
|
||
self.config = config or KnowledgeRuntimeConfig()
|
||
self.vector_dim = int(vector_dim)
|
||
|
||
# --- 入口 -----------------------------------------------------------------
|
||
|
||
async def search(
|
||
self,
|
||
query: KnowledgeQuery,
|
||
*,
|
||
embedding_endpoints: Sequence[Any] | None = None,
|
||
embedder: Any = None,
|
||
) -> KnowledgeSearchResult:
|
||
"""执行检索。集合由意图映射，调用方无法指定集合名。"""
|
||
targets = self._assert_collections_allowed(query.intents)
|
||
top_k = min(int(query.top_k), self.config.result_limit)
|
||
searched = tuple(sorted({collection for collection, _ in targets}))
|
||
|
||
try:
|
||
vector = await self._embed(query.query, embedding_endpoints, embedder)
|
||
rows = await self._vector_search(targets, vector, top_k)
|
||
except (RecoverableAgentError, ForbiddenAgentError) as exc:
|
||
# 维度不符走失败关闭（ValidationAgentError），绝不会到这里。
|
||
return await self._degraded_search(query, searched, reason=str(exc))
|
||
|
||
verified = await self._verify(rows)
|
||
hits = self._to_hits(verified)
|
||
return KnowledgeSearchResult(hits=hits, degraded=False, searched_collections=searched)
|
||
|
||
# --- 路由与白名单 ---------------------------------------------------------
|
||
|
||
def _assert_collections_allowed(
|
||
self, intents: Sequence[str]
|
||
) -> list[tuple[str, int]]:
|
||
"""**任何网络调用之前**完成的集合白名单校验（含路由表本身的可信性检查）。"""
|
||
targets: list[tuple[str, int]] = []
|
||
unknown: list[str] = []
|
||
for intent in intents:
|
||
route = self.config.route(str(intent))
|
||
if route is None:
|
||
unknown.append(str(intent))
|
||
continue
|
||
collection, route_top_k = route
|
||
if collection not in ALLOWED_COLLECTIONS:
|
||
# 路由表被改坏时失败关闭，不把任意集合名送到 Milvus。
|
||
raise ForbiddenAgentError(f"知识集合不在白名单内：{collection}")
|
||
targets.append((collection, max(1, int(route_top_k))))
|
||
if not targets:
|
||
raise ForbiddenAgentError(f"无可检索的知识集合：未知意图 {sorted(unknown)}")
|
||
return targets
|
||
|
||
def assert_collection_allowed(self, collection: str) -> str:
|
||
"""供上层/适配器复用的集合白名单校验（非白名单抛 `ForbiddenAgentError`）。"""
|
||
if collection not in ALLOWED_COLLECTIONS:
|
||
raise ForbiddenAgentError(f"知识集合不在白名单内：{collection}")
|
||
return collection
|
||
|
||
# --- 向量召回 -------------------------------------------------------------
|
||
|
||
async def _embed(
|
||
self, keyword: str, embedding_endpoints: Sequence[Any] | None, embedder: Any
|
||
) -> list[float]:
|
||
active = embedder if embedder is not None else self.embedder
|
||
if active is None:
|
||
raise RecoverableAgentError("未配置向量化服务，无法执行向量检索")
|
||
if embedding_endpoints is None:
|
||
raise RecoverableAgentError("缺少已批准的向量端点，无法执行向量检索")
|
||
execution = await active.embed(list(embedding_endpoints), keyword, max_attempts=2)
|
||
vector = self._vector_of(execution)
|
||
self.assert_vector_dim(vector)
|
||
return vector
|
||
|
||
def assert_vector_dim(self, vector: Sequence[float]) -> None:
|
||
"""维度不符**失败关闭**：不静默返回空结果，也不降级掩盖配置错误。"""
|
||
if len(vector) != self.vector_dim:
|
||
raise ValidationAgentError(
|
||
f"检索向量维度不符：期望 {self.vector_dim}，实际 {len(vector)}"
|
||
)
|
||
|
||
@staticmethod
|
||
def _vector_of(execution: Any) -> list[float]:
|
||
raw = getattr(execution, "vector", None)
|
||
if raw is None and isinstance(execution, Mapping):
|
||
raw = execution.get("vector")
|
||
if raw is None:
|
||
raise RecoverableAgentError("向量化服务未返回向量")
|
||
return [float(item) for item in raw]
|
||
|
||
async def _vector_search(
|
||
self, targets: Sequence[tuple[str, int]], vector: list[float], top_k: int
|
||
) -> list[dict[str, Any]]:
|
||
rows: list[dict[str, Any]] = []
|
||
for collection, route_top_k in targets:
|
||
# 路由表里的 top_k 是**上限**（faq 只取 3 条），调用方只能收窄、不能放宽。
|
||
limit = max(1, min(top_k, route_top_k))
|
||
rows.extend(
|
||
await self.client.search(collection=collection, vector=vector, top_k=limit)
|
||
)
|
||
return rows
|
||
|
||
# --- MySQL 二次校验 -------------------------------------------------------
|
||
|
||
async def _verify(self, rows: Sequence[Mapping[str, Any]]) -> list[Mapping[str, Any]]:
|
||
"""按 `knowledge_id` 回表校验：已发布 + active + 在有效期内；并按分数降序保留。"""
|
||
if not rows:
|
||
return []
|
||
order = {str(row["knowledge_id"]): index for index, row in enumerate(rows)}
|
||
scores = {str(row["knowledge_id"]): self._score(row) for row in rows}
|
||
intents = {
|
||
str(row["knowledge_id"]): row.get("intent")
|
||
for row in rows
|
||
if isinstance(row.get("intent"), str) and str(row["intent"]).strip()
|
||
}
|
||
ids = [self._as_int(row["knowledge_id"]) for row in rows]
|
||
ids = [value for value in ids if value is not None]
|
||
if not ids:
|
||
return []
|
||
stored = await self._load_by_ids(ids)
|
||
verified = [
|
||
row
|
||
for row in stored
|
||
if self._is_visible(row) and str(row["id"]) in order
|
||
]
|
||
verified.sort(key=lambda row: scores.get(str(row["id"]), 0.0), reverse=True)
|
||
for row in verified:
|
||
row["score"] = scores.get(str(row["id"])) # type: ignore[index]
|
||
# 显式 intent 标签只存在于 Milvus 侧（MySQL 无该列）：随命中一并带过来，
|
||
# 否则会在回表后被"按集合名推断"覆盖掉，丢掉 chitchat / transfer_human。
|
||
row["intent"] = intents.get(str(row["id"])) # type: ignore[index]
|
||
return verified
|
||
|
||
async def _load_by_ids(self, ids: Sequence[int]) -> list[dict[str, Any]]:
|
||
async with self._session_factory() as session:
|
||
result = await session.execute(_HITS_SQL, {"knowledge_ids": tuple(ids)})
|
||
return [dict(row) for row in result.mappings().all()]
|
||
|
||
@staticmethod
|
||
def _is_visible(row: Mapping[str, Any]) -> bool:
|
||
"""「已发布 + 有效期」是唯一可见性口径，向量路径与降级路径共用。"""
|
||
if str(row.get("review_status")) != PUBLISHED_REVIEW_STATUS:
|
||
return False
|
||
if str(row.get("status")) != ACTIVE_STATUS:
|
||
return False
|
||
today = datetime.now(UTC).date()
|
||
effective = KnowledgeRetrievalService._as_date(row.get("effective_date"))
|
||
if effective is not None and effective > today:
|
||
return False
|
||
expire = KnowledgeRetrievalService._as_date(row.get("expire_date"))
|
||
return not (expire is not None and expire <= today)
|
||
|
||
@staticmethod
|
||
def _as_date(value: Any) -> date | None:
|
||
return value if isinstance(value, date) else None
|
||
|
||
# --- 降级路径 -------------------------------------------------------------
|
||
|
||
async def _degraded_search(
|
||
self, query: KnowledgeQuery, collections: Sequence[str], *, reason: str
|
||
) -> KnowledgeSearchResult:
|
||
"""Milvus 不可用时的兜底：MySQL `LIKE` 关键词检索，可见性过滤照旧。"""
|
||
logger.warning("知识向量检索降级为 MySQL LIKE：%s", reason)
|
||
pattern = f"%{self._escape_like(query.query.strip()[:MAX_KEYWORD_LENGTH])}%"
|
||
try:
|
||
async with self._session_factory() as session:
|
||
result = await session.execute(
|
||
_DEGRADED_SQL,
|
||
{
|
||
"collections": tuple(collections),
|
||
"pattern": pattern,
|
||
"escape": LIKE_ESCAPE,
|
||
"limit": min(int(query.top_k), self.config.result_limit),
|
||
},
|
||
)
|
||
rows = [dict(row) for row in result.mappings().all()]
|
||
except Exception as exc: # pragma: no cover - 连接层异常，按"无结果 + 已降级"返回
|
||
logger.warning("知识降级检索同样失败：%s", exc)
|
||
rows = []
|
||
verified = [row for row in rows if self._is_visible(row)]
|
||
return KnowledgeSearchResult(
|
||
hits=self._to_hits(verified),
|
||
degraded=True,
|
||
degradation_reason=reason or "milvus_unavailable",
|
||
searched_collections=tuple(sorted(collections)),
|
||
)
|
||
|
||
@staticmethod
|
||
def _escape_like(keyword: str) -> str:
|
||
escaped = keyword
|
||
for char in (LIKE_ESCAPE, "%", "_"):
|
||
escaped = escaped.replace(char, f"{LIKE_ESCAPE}{char}")
|
||
return escaped
|
||
|
||
# --- 结果装配 -------------------------------------------------------------
|
||
|
||
def _to_hits(self, rows: Sequence[Mapping[str, Any]]) -> tuple[KnowledgeHit, ...]:
|
||
hits: list[KnowledgeHit] = []
|
||
for row in rows:
|
||
hit = self._to_hit(row)
|
||
if hit is not None:
|
||
hits.append(hit)
|
||
return tuple(hits)
|
||
|
||
def _to_hit(self, row: Mapping[str, Any]) -> KnowledgeHit | None:
|
||
collection = row.get("milvus_collection")
|
||
if not isinstance(collection, str) or collection not in ALLOWED_COLLECTIONS:
|
||
# 契约要求 collection 必须在白名单内；不是我们的行直接丢弃而不是猜造。
|
||
return None
|
||
knowledge_id = self._as_int(row.get("id"))
|
||
if knowledge_id is None:
|
||
return None
|
||
return KnowledgeHit(
|
||
knowledge_id=str(knowledge_id),
|
||
collection=collection,
|
||
title=self._as_text(row.get("title")),
|
||
snippet=str(row.get("content_text") or ""),
|
||
score=self._score_or_none(row),
|
||
tags=self._as_tags(row.get("tags")),
|
||
version=self._as_text(row.get("version")),
|
||
intent=self._intent_of(row, collection),
|
||
)
|
||
|
||
def _intent_of(self, row: Mapping[str, Any], collection: str) -> str | None:
|
||
"""`intent` 是稀疏标签：缺字段 / 空串 / 非字符串都算无标签，按集合名推断。"""
|
||
raw = row.get("intent")
|
||
if isinstance(raw, str) and raw.strip():
|
||
return raw.strip()
|
||
return COLLECTION_INTENTS.get(collection)
|
||
|
||
@staticmethod
|
||
def _as_int(value: Any) -> int | None:
|
||
try:
|
||
return int(value)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
@staticmethod
|
||
def _as_text(value: Any) -> str | None:
|
||
if isinstance(value, str) and value.strip():
|
||
return value
|
||
return None
|
||
|
||
@staticmethod
|
||
def _as_tags(value: Any) -> tuple[str, ...]:
|
||
"""`tags` 在 Milvus 侧是逗号串，在 MySQL 侧是 JSON 数组；两种都吸收。"""
|
||
if isinstance(value, str):
|
||
text_value = value.strip()
|
||
if not text_value:
|
||
return ()
|
||
try:
|
||
parsed = json.loads(text_value)
|
||
except ValueError:
|
||
return tuple(item.strip() for item in text_value.split(",") if item.strip())
|
||
value = parsed
|
||
if isinstance(value, list | tuple):
|
||
return tuple(str(item) for item in value if str(item).strip())
|
||
return ()
|
||
|
||
@staticmethod
|
||
def _score(row: Mapping[str, Any]) -> float:
|
||
value = KnowledgeRetrievalService._score_or_none(row)
|
||
return value if value is not None else 0.0
|
||
|
||
@staticmethod
|
||
def _score_or_none(row: Mapping[str, Any]) -> float | None:
|
||
"""COSINE 距离即相似度（越大越相似），裁剪到契约要求的 0..1。"""
|
||
raw = row.get("score", row.get("distance"))
|
||
try:
|
||
score = float(raw) # type: ignore[arg-type]
|
||
except (TypeError, ValueError):
|
||
return None
|
||
return min(1.0, max(0.0, score))
|