From 9aaacc242f74dd463565d07b372e1786fee079fb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BC=A0=E8=83=9C=E5=AE=87?= <17412268+zzzzz11122222@user.noreply.gitee.com> Date: Sat, 12 Sep 2026 11:15:24 +0800 Subject: [PATCH] =?UTF-8?q?chore:=20=E6=B8=85=E7=90=86=E8=BF=9D=E5=8F=8D?= =?UTF-8?q?=E5=BA=95=E5=BA=A7=E8=A7=84=E5=88=99=E7=9A=84=E6=AD=BB=E4=BB=A3?= =?UTF-8?q?=E7=A0=81=E5=B9=B6=E4=BF=AE=E6=AD=A3=E6=8E=A5=E5=8F=A3=E6=96=87?= =?UTF-8?q?=E6=A1=A3=E7=BC=96=E5=8F=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 删除生产死代码 app/service/knowledge_tool_service.py 与 app/infrastructure/milvus_knowledge_adapter.py：后者硬编码 Milvus 字段名， 违反 AGENTS.md §E，且仅被前者引用；生产检索链路实际走 knowledge_search_tool -> KnowledgeSearchService -> knowledge_schema 运行时探测。 - 删除上述两模块的单测，以及依赖 legacy 位置参数构造的 tests/unit/service/test_knowledge_retrieval.py。 - app/service/knowledge_retrieval_service.py 整文件回退底座版本， 移除 legacy 双构造与重复检索实现。 - docs/05-接口文档.md：客服画像候选改登记为 §8.5，恢复 §8.2 解析知识引用； 既有 §8.1-§8.4 编号全部保持，修复此前出现两个 8.3 的问题。 - app/model/profile.py：current_customer_id 改为普通可空列映射，与 alembic/baseline_generated.sql 及真实库一致；原 Computed 声明会让 ORM 把该列 从 INSERT 中排除，与「必须显式写入」的实际 schema 不符。 - 新增 docs/客服Agent接入底座扩展说明_v1.md，供集成分支评审逐项确认。 验证：pytest tests/unit tests/contract -> 1275 passed, 2 skipped, 0 failed； ruff check app tests tools alembic 通过；mypy app 通过（244 个源文件）。 --- .../milvus_knowledge_adapter.py | 74 --------- app/model/profile.py | 17 +- app/service/knowledge_retrieval_service.py | 65 +------- app/service/knowledge_tool_service.py | 55 ------- docs/05-接口文档.md | 42 ++--- docs/客服Agent接入底座扩展说明_v1.md | 154 ++++++++++++++++++ .../test_milvus_knowledge_adapter.py | 47 ------ .../unit/service/test_knowledge_retrieval.py | 87 ---------- .../service/test_knowledge_tool_service.py | 131 --------------- 9 files changed, 189 insertions(+), 483 deletions(-) delete mode 100644 app/infrastructure/milvus_knowledge_adapter.py delete mode 100644 app/service/knowledge_tool_service.py create mode 100644 docs/客服Agent接入底座扩展说明_v1.md delete mode 100644 tests/unit/infrastructure/test_milvus_knowledge_adapter.py delete mode 100644 tests/unit/service/test_knowledge_retrieval.py delete mode 100644 tests/unit/service/test_knowledge_tool_service.py diff --git a/app/infrastructure/milvus_knowledge_adapter.py b/app/infrastructure/milvus_knowledge_adapter.py deleted file mode 100644 index 9c05b8b..0000000 --- a/app/infrastructure/milvus_knowledge_adapter.py +++ /dev/null @@ -1,74 +0,0 @@ -from typing import Any - -from app.core.errors import ForbiddenAgentError, RecoverableAgentError -from app.core.knowledge_contracts import ALLOWED_KNOWLEDGE_COLLECTIONS - - -class MilvusKnowledgeClient: - def __init__(self, uri: str, token: str | None = None) -> None: - self._uri = uri - self._token = token - self._client: Any | None = None - - async def _ensure_client(self) -> Any: - if self._client is None: - from pymilvus import AsyncMilvusClient # type: ignore[import-untyped] - - self._client = AsyncMilvusClient(uri=self._uri, token=self._token) - return self._client - - async def search( - self, collection: str, vector: list[float], top_k: int - ) -> list[dict[str, Any]]: - if collection not in ALLOWED_KNOWLEDGE_COLLECTIONS: - raise ForbiddenAgentError("未授权的知识集合") - if len(vector) != 1024 or not 1 <= top_k <= 20: - raise RecoverableAgentError("知识检索参数无效") - try: - client = await self._ensure_client() - # Lite 重启后集合默认未加载；远程 Milvus 对重复加载保持幂等。 - load_collection = getattr(client, "load_collection", None) - if load_collection is not None: - await load_collection(collection_name=collection) - batches = await client.search( - collection_name=collection, - data=[vector], - limit=top_k, - output_fields=["knowledge_id", "title", "snippet", "tags", "version"], - search_params={"metric_type": "COSINE"}, - ) - except Exception as exc: - raise RecoverableAgentError("知识检索不可用") from exc - return [ - normalized - for batch in batches - for hit in batch - if (normalized := self._normalize_hit(hit)) is not None - ] - - @staticmethod - def _normalize_hit(hit: Any) -> dict[str, Any] | None: - """统一 Milvus SDK 的平铺与 entity 包装命中格式。""" - raw = dict(hit) - entity = raw.get("entity") - fields = entity if isinstance(entity, dict) else raw - knowledge_id = fields.get("knowledge_id") - snippet = fields.get("snippet") - score = raw.get("score", raw.get("distance", fields.get("score"))) - if ( - not isinstance(knowledge_id, str) - or not isinstance(snippet, str) - or not isinstance(score, (int, float)) - or isinstance(score, bool) - ): - return None - normalized: dict[str, Any] = { - "knowledge_id": knowledge_id, - "snippet": snippet, - "score": float(score), - } - for field in ("title", "tags", "version"): - value = fields.get(field) - if value is not None: - normalized[field] = value - return normalized diff --git a/app/model/profile.py b/app/model/profile.py index 22e6fe1..27f0bcc 100644 --- a/app/model/profile.py +++ b/app/model/profile.py @@ -4,15 +4,18 @@ 这里显式标注，避免后续有人按直觉写入而踩坑： 1. `user_facts.id` 在库里**没有 auto_increment**，插入时必须由应用显式提供主键； -2. `profile_snapshots.current_customer_id` 是**生成列**（`IF(is_current=1, customer_id, NULL)`）， - 与唯一键 `uk_profile_snapshot_current` 共同保证「每个客户最多一条当前快照」。生成列由数据库 - 维护，因此这里只映射为只读计算列，写入时不会提供该字段。 +2. `profile_snapshots.current_customer_id` **不是生成列**，而是普通可空列 + 唯一键 + `uk_profile_snapshot_current`：当前版本必须由写入方**显式写入**客户 ID（历史版本写 NULL）， + 才能保证「每个客户最多一条当前快照」。因此这里按普通可空列映射，**不能**声明 `Computed`—— + 声明成生成列会让 SQLAlchemy 把它从 INSERT 中排除，反而永远写不进去。 + （`docs/00` 第 783 行把它描述为「生成列」，与实际 DDL 及真实库不一致； + 以 `alembic/baseline_generated.sql`、`tools/seed_profile_demo.py` 和真实库为准。） """ from datetime import datetime from typing import Any -from sqlalchemy import CHAR, JSON, BigInteger, Boolean, Computed, DateTime, Float, String +from sqlalchemy import CHAR, JSON, BigInteger, Boolean, DateTime, Float, String from sqlalchemy.orm import Mapped, mapped_column from app.model.base import Base @@ -59,7 +62,5 @@ class ProfileSnapshot(Base): generated_at: Mapped[datetime | None] = mapped_column(DateTime) created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) updated_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) - # 生成列由数据库维护；映射为只读计算列，便于按当前快照查询，不参与 INSERT/UPDATE。 - current_customer_id: Mapped[int | None] = mapped_column( - BigInteger, Computed("IF(is_current = 1, customer_id, NULL)") - ) + # 普通可空列 + 唯一键，由写入方显式赋值（见模块 docstring 第 2 条），不是生成列。 + current_customer_id: Mapped[int | None] = mapped_column(BigInteger) diff --git a/app/service/knowledge_retrieval_service.py b/app/service/knowledge_retrieval_service.py index 012cc24..aab22b4 100644 --- a/app/service/knowledge_retrieval_service.py +++ b/app/service/knowledge_retrieval_service.py @@ -113,25 +113,16 @@ class KnowledgeRetrievalService: def __init__( self, client: Any, - *legacy_args: Any, + *, embedder: Any = None, session_factory: Callable[[], Any] | None = None, config: KnowledgeRuntimeConfig | None = None, vector_dim: int = VECTOR_DIM, ) -> None: - # 兼容早期客服工具的 positional 构造：embedder, vector_store, config, authority。 - self._legacy = len(legacy_args) == 3 - if self._legacy: - self.embedder = client - self.client = legacy_args[0] - self.config = legacy_args[1] - self._authority = legacy_args[2] - else: - self.client = client - self.embedder = embedder - self.config = config or KnowledgeRuntimeConfig() - self._authority = 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) # --- 入口 ----------------------------------------------------------------- @@ -139,14 +130,11 @@ class KnowledgeRetrievalService: async def search( self, query: KnowledgeQuery, - legacy_context: Any = None, *, embedding_endpoints: Sequence[Any] | None = None, embedder: Any = None, ) -> KnowledgeSearchResult: """执行检索。集合由意图映射，调用方无法指定集合名。""" - if self._legacy: - return await self._legacy_search(query) 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})) @@ -162,49 +150,6 @@ class KnowledgeRetrievalService: hits = self._to_hits(verified) return KnowledgeSearchResult(hits=hits, degraded=False, searched_collections=searched) - async def _legacy_search(self, query: KnowledgeQuery) -> KnowledgeSearchResult: - """兼容旧工具调用，仍复用同一意图路由和 MySQL 权威回查。""" - targets = self._assert_collections_allowed(query.intents) - collections = tuple(sorted({collection for collection, _ in targets})) - top_k = min(int(query.top_k), int(getattr(self.config, "result_limit", 20))) - try: - raw_vector = await self.embedder.embed(query.query) - vector = self._vector_of(raw_vector) - self.assert_vector_dim(vector) - rows: list[dict[str, Any]] = [] - for collection, route_top_k in targets: - rows.extend(await self.client.search( - collection, vector, min(top_k, route_top_k) - )) - if self._authority is not None: - verified = await self._authority.filter_published(tuple( - KnowledgeHit( - knowledge_id=str(row.get("knowledge_id")), - collection=collection, - title=row.get("title"), - snippet=str(row.get("snippet") or ""), - score=self._score(row), - ) - for row in rows - for collection, _ in targets - if str(row.get("collection") or collection) == collection - )) - return KnowledgeSearchResult( - hits=tuple(verified), searched_collections=collections - ) - return KnowledgeSearchResult(hits=self._to_hits(rows), searched_collections=collections) - except Exception: - if self._authority is None: - return KnowledgeSearchResult( - hits=(), degraded=True, degradation_reason="milvus_unavailable", - searched_collections=collections, - ) - fallback = await self._authority.search_keyword(query, collections, top_k) - return KnowledgeSearchResult( - hits=tuple(fallback), degraded=True, - degradation_reason="milvus_unavailable", searched_collections=collections, - ) - # --- 路由与白名单 --------------------------------------------------------- def _assert_collections_allowed( @@ -257,8 +202,6 @@ class KnowledgeRetrievalService: @staticmethod def _vector_of(execution: Any) -> list[float]: - if isinstance(execution, list | tuple): - return [float(item) for item in execution] raw = getattr(execution, "vector", None) if raw is None and isinstance(execution, Mapping): raw = execution.get("vector") diff --git a/app/service/knowledge_tool_service.py b/app/service/knowledge_tool_service.py deleted file mode 100644 index 2027d10..0000000 --- a/app/service/knowledge_tool_service.py +++ /dev/null @@ -1,55 +0,0 @@ -from typing import Protocol - -from app.core.config import get_settings -from app.core.contracts import RequestContext -from app.core.knowledge_contracts import KnowledgeQuery, KnowledgeSearchResult -from app.infrastructure.db import SessionFactory -from app.infrastructure.milvus_knowledge_adapter import MilvusKnowledgeClient -from app.service.knowledge_authority import KnowledgeMysqlAuthority -from app.service.knowledge_config import KnowledgeRuntimeConfig -from app.service.knowledge_retrieval_service import KnowledgeRetrievalService -from app.service.model_gateway import DatabaseModelGateway - - -class EmbeddingGateway(Protocol): - async def embed(self, *, endpoint_code: str, text: str, timeout_ms: int) -> list[float]: ... - - -class DatabaseEmbeddingAdapter: - def __init__( - self, endpoint_code: str, timeout_ms: int, *, gateway: EmbeddingGateway - ) -> None: - self._endpoint_code = endpoint_code - self._timeout_ms = timeout_ms - self._gateway = gateway - - async def embed(self, text: str) -> list[float]: - return await self._gateway.embed( - endpoint_code=self._endpoint_code, text=text, timeout_ms=self._timeout_ms - ) - - -async def query_knowledge_tool( - arguments: KnowledgeQuery, context: RequestContext -) -> KnowledgeSearchResult: - settings = get_settings() - if not settings.knowledge_embedding_endpoint_code: - return KnowledgeSearchResult( - degraded=True, degradation_reason="embedding_endpoint_unconfigured" - ) - # 知识向量端点与默认聊天端点隔离，避免回答模型被误用于检索。 - embedder = DatabaseEmbeddingAdapter( - settings.knowledge_embedding_endpoint_code, - settings.knowledge_embedding_timeout_ms, - gateway=DatabaseModelGateway(), - ) - # 兼容旧测试替身；真实 Settings 会优先提供本地/远程统一解析后的地址。 - milvus_uri = getattr(settings, "resolved_milvus_uri", settings.milvus_uri) - vector_store = MilvusKnowledgeClient(milvus_uri, token=settings.milvus_token or None) - # 权威元数据只读回查，确保对客答案始终来自已发布、有效的知识条目。 - async with SessionFactory() as session: - authority = KnowledgeMysqlAuthority(session) - service = KnowledgeRetrievalService( - embedder, vector_store, KnowledgeRuntimeConfig(), authority - ) - return await service.search(arguments, context) diff --git a/docs/05-接口文档.md b/docs/05-接口文档.md index ade4de4..859c450 100644 --- a/docs/05-接口文档.md +++ b/docs/05-接口文档.md @@ -616,26 +616,7 @@ Authorization: Bearer 记忆提取没有客户端写接口。`memory.extraction_requested` 由 `complete_run()` 与最终结果在同一事务写入 Outbox，再由 Worker 调用内部 `MemoryService`。更正、遗忘和监管删除属于独立隐私流程，本接口不临时复用 `memory_conflict`。 -### 8.2 客服画像候选（Phase 2） - -客服 Agent 不读取或直接修改正式画像。已登录用户明确陈述长期偏好、约束或目标时，系统 -异步生成 `memory_unit.status='candidate'` 候选；访客不会生成候选。候选不进入客服召回， -必须经过用户确认和管理员审核后才能晋升为 `active`。 - -```text -GET /api/v1/users/me/memory-candidates -POST /api/v1/users/me/memory-candidates/{candidate_id}/decisions -GET /api/v1/admin/customer-profile-candidates -POST /api/v1/admin/customer-profile-candidates/{candidate_id}/reviews -``` - -用户确认请求体为 `{ "decision": "confirmed" | "rejected" }`，需要 -`memory:candidate:confirm`；确认只将状态改为 `verified`。管理员审核请求体复用 -`ReviewPayload`，需要管理员角色和 `memory:candidate:review`；`approved` 会在事务内 -处理同键旧记忆冲突并将候选改为 `active`，`rejected` 将其改为 `rejected`。接口只返回 -结构化候选值，不返回对话证据摘录、密码、验证码或其他原始敏感内容。 - -### 8.3 解析知识引用 +### 8.2 解析知识引用 ```http GET /api/v1/knowledge-references/{reference_token} @@ -772,6 +753,27 @@ DELETE /api/v1/knowledge/{knowledge_id} **不返回 `real_name`、`birth_date`、`mobile_masked`、`trade_account` 等 PII**； `assessment_expired` 按**当前时间**重算，不采信快照里的历史布尔值。 +### 8.5 客服画像候选（Phase 2） + +> 编号说明：本节为客服二期新增，**不占用 §8.1–§8.4 既有号段**，以避免破坏 `AGENTS.md`、`docs/09`、`docs/14` 对「§8.3 知识库管理三端点」「§8.4 公共只读工具索引」的既有引用。 + +客服 Agent 不读取或直接修改正式画像。已登录用户明确陈述长期偏好、约束或目标时，系统 +异步生成 `memory_unit.status='candidate'` 候选；访客不会生成候选。候选不进入客服召回， +必须经过用户确认和管理员审核后才能晋升为 `active`。 + +```text +GET /api/v1/users/me/memory-candidates +POST /api/v1/users/me/memory-candidates/{candidate_id}/decisions +GET /api/v1/admin/customer-profile-candidates +POST /api/v1/admin/customer-profile-candidates/{candidate_id}/reviews +``` + +用户确认请求体为 `{ "decision": "confirmed" | "rejected" }`，需要 +`memory:candidate:confirm`；确认只将状态改为 `verified`。管理员审核请求体复用 +`ReviewPayload`，需要管理员角色和 `memory:candidate:review`；`approved` 会在事务内 +处理同键旧记忆冲突并将候选改为 `active`，`rejected` 将其改为 `rejected`。接口只返回 +结构化候选值，不返回对话证据摘录、密码、验证码或其他原始敏感内容。 + ## 9. 平台管理面接口 管理面只操作草稿、审核、激活、停用、回滚和归档流程，不提供绕过版本控制的通用 CRUD。所有更新和状态转换都需要 `If-Match`；创建、审核、激活、回滚和停用需要 `Idempotency-Key`。 diff --git a/docs/客服Agent接入底座扩展说明_v1.md b/docs/客服Agent接入底座扩展说明_v1.md new file mode 100644 index 0000000..1372016 --- /dev/null +++ b/docs/客服Agent接入底座扩展说明_v1.md @@ -0,0 +1,154 @@ +# 客服 Agent 接入底座扩展说明（v1） + +> 用途：提交 `qyqy_develop`（集成分支）评审时使用。 +> 本文档逐项列出「客服 Agent + RAG」能力为接入底座而**触碰的既有文件**、改动性质、合规依据， +> 以及**明确未改动**的底座边界，便于架构师逐条确认。 +> +> 依据：`AGENTS.md`、`docs/00`、`docs/01`、`docs/05`、`docs/09`、`docs/14`、`docs/20`。 + +--- + +## 一、总体口径 + +- 本次接入遵循 `AGENTS.md` 规则 7：客服 Agent 继承公共 `BaseAgent`、由 `AgentFactory` 创建， + **不绕过**公共鉴权、记忆、模型路由、工具、合规、审计与事件流程。 +- 数据库侧**未做任何基线改动**：没有新增/重命名/删除表，没有改字段类型、可空性或业务含义， + 没有新增迁移脚本。 +- 底座中所有「既有 Agent 默认行为」均通过**带默认值的声明位**扩展，默认值等于改动前行为， + 其他 Agent（`FundQueryDemoAgent`、`RiskAgent`、`PlatformProbeAgent`）路径不受影响。 + +--- + +## 二、A 类：纯新增，不影响既有代码 + +| 文件 | 说明 | +|---|---| +| `app/core/conversation_privacy.py` | 客服会话落库前的凭据最小化（密码/验证码/证件号/银行卡/手机号 → 占位符） | +| `app/api/controllers/visitor_tokens.py`、`app/api/schemas/visitor_tokens.py` | 访客短时令牌签发接口 | +| `app/service/agent/customer_service_routing.py` | 客服确定性意图路由（安全/账户/人工/合规/闲聊/公开知识） | +| `app/service/customer_service_session_memory_service.py` | 客服 Redis 短期会话记忆（30 分钟滑动 TTL、24 小时绝对上限、16 条/约 4096 Token 截断） | +| `app/service/customer_service_handover_context.py`、`customer_service_handover_admin_service.py` | 转人工上下文构造与管理员只读工单服务 | +| `app/service/customer_profile_candidate_service.py` | 画像候选（Phase 2，异步、需用户确认 + 管理员审核） | +| `app/service/knowledge_authority.py`、`knowledge_config.py`、`knowledge_publication_service.py` | 知识权威回查、运行期配置、受控发布工具所依赖的服务 | +| `app/infrastructure/milvus_profile_projection.py`、`neo4j_profile_projection.py` | 画像投影适配器（异步派生写入） | +| `app/worker/memory_sync_outbox_worker.py`、`customer_profile_candidate_worker.py` | 画像投影 Outbox 消费者、候选画像 Worker | +| `app/static/index.html` | 客服联调测试页（`/customer-service-test`） | +| `tools/publish_customer_service_knowledge.py`、`verify_customer_service_phase1.py`、`knowledge_import_preflight.py` | 受控发布、只读门禁、导入预检 | + +以上均为新增文件，不改变任何既有模块行为。 + +--- + +## 三、B 类：带默认值、默认行为不变的扩展（对齐底座扩展点，建议直接确认） + +| 文件 | 改动 | 默认值是否等于原行为 | +|---|---|---| +| `app/core/contracts.py` | `AgentDefinition` 新增 `requires_model_intent_classification=True`、`recalls_customer_memory=True`；`AgentRequestMetadata` 新增 `chitchat_streak`/`clarification_round`/`session_context`；`CoreResult` 新增 `clarification_required` | 是 | +| `app/service/agent/base.py` | `recall_memory()` 在声明关闭或访客时置空；`classify_intent()` 在声明关闭时返回 `None` | 是 | +| `app/service/agent/factory.py` | 仅在 `requires_model_intent_classification` 为真时绑定意图分类器 | 是 | +| `app/core/config.py` | 新增 `visitor_token_ttl_seconds`、`milvus_local_uri`、`knowledge_embedding_endpoint_code`、`knowledge_embedding_timeout_ms` 与 `resolved_milvus_uri` 属性 | 是（新增项均有默认值） | +| `app/core/knowledge_contracts.py` | 新增 `ALLOWED_KNOWLEDGE_COLLECTIONS` 别名（与 `ALLOWED_COLLECTIONS` 恒等）；`KnowledgeHit.answer` 可选字段；`KnowledgeSearchResult.hits` 补默认值 | 是（三个 `fin_*_collection` 白名单内容未变） | +| `app/model/knowledge.py` | 新增 `FinKnowledgeMeta = KnowledgeMeta` 别名 | 是 | +| `app/model/memory.py` | `__all__` 与 `ProfileSnapshot` 兼容转出 | 是 | +| `app/service/agent/bootstrap.py` | 新增注册 `query_knowledge` 工具，`handler` 复用 `knowledge_search_tool`（与 `search_knowledge` 同一实现），`required_permission="knowledge:query"` | 是（不新增能力面，实际范围仍由 `config_release` 白名单收口） | +| `app/service/memory_service.py` | `upsert_memory()` 新增 `status` 参数（`active`/`candidate`），`candidate` 不覆盖现有有效记忆 | 是（默认 `active`） | +| `app/service/public_platform_service.py` | 转人工工单 `reason_detail` 落库前做凭据脱敏 | 是（安全收紧） | +| `app/main.py` | 注册 `visitor_tokens_router`、挂载 `/customer-service-test` 静态页 | 是（纯新增装配） | +| `pyproject.toml`、`requirements.txt` | 新增 `milvus-lite>=3.2,<4` | 是（可选本地开发依赖） | +| `.gitignore` | 新增 `.worktrees/`、`data/milvus/` | 是 | + +--- + +## 四、C 类：需要架构师确认的底座语义扩展 + +### C1. 访客身份（`app/api/dependencies/auth.py` + `app/core/security.py`） + +- `security.py`：`JwtAuthenticator.authenticate()` 新增分支——当令牌含 `visitor: true` 声明时， + 返回 `roles=("visitor",)`、`permissions=("agent:run", "knowledge:query")`、`data_scope="public"`。 +- `auth.py`：`build_request_context()` 对 `visitor` 角色跳过 `IdentityService().resolve()` + （访客不在 `sys_user` 中，无法解析身份）。 + +**合规说明：** +- `visitor` 角色只可能出自用项目 RS256 私钥签名、且显式带 `visitor: true` 的令牌，普通用户令牌 + 无法携带该角色； +- 访客权限固定为 `agent:run` + `knowledge:query`，`data_scope="public"`， + **不包含任何账户、持仓、订单、银行卡、投诉进度或画像权限**； +- 该分支只影响访客令牌，非访客令牌仍走完整 `IdentityService` 身份解析。 + +**请求确认：** 是否接受在底座统一鉴权链路中增加这一受控访客身份；如需要，我们可补一份 +`docs/` 说明或在 `docs/05` 登记该令牌类型。 + +### C2. 客服 Agent 的角色与安全路由（`app/service/agent/implementations/customer_service.py`） + +- `allowed_roles` 由 `("customer",)` 扩为 `("visitor", "customer")`； +- `handle()` 首部插入确定性安全路由 `route_message()`（安全提示、合规拒答、账户入口、人工转接优先于检索）； +- 新增 `recalls_customer_memory=False`（客服不隐式召回长期/画像记忆）； +- 新增 `chitchat_streak == 4` 的一次性业务引导； +- 业务口径：`COMPANY` 由 `南方科技` 改为 `奶龙基金责任有限公司`。 + +**请求确认：** 角色扩容与行为契约变更（安全路由前置、闲聊计数）；品牌名属业务口径调整。 + +### C3. 画像快照 ORM 声明清理与 `current_customer_id` 映射更正 + +三处相关改动（**属于修既有缺陷，不是改设计**）： + +1. `app/model/risk_questionnaire.py`：原有内联 `ProfileSnapshot` 声明与 `app/model/profile.py` + 的同表声明**重复**，两个 declarative 类映射同一张表会让 SQLAlchemy 直接拒绝导入。 + 现改为从 `app.model.profile` 转出规范类，保留对既有调用方的兼容导出。 +2. `app/model/profile.py`：把 `current_customer_id` 映射为**普通可空列**。 +3. 依据：该列在**真实库中是普通可空列**（`EXTRA=''`、`GENERATION_EXPRESSION=''`，已实测）， + `alembic/baseline_generated.sql` 的建表语句也没有 `GENERATED` 子句， + `tools/seed_profile_demo.py` 明确写着「**不是生成列**，必须显式写入」。 + 因此**不能**声明 `Computed(...)`——那会让 SQLAlchemy 把它从 INSERT 中排除，反而永远写不进去。 +4. **附带发现（建议底座侧修正）：** `docs/00-新数据库基线设计.md` 第 783 行把该列描述为 + 「生成列」，与 `docs/02`/`alembic` 基线 DDL 及真实库不一致。本次未改动 `docs/00`， + 仅在此登记，请底座侧决定以哪一侧为准。 + +**请求确认：** 上述三点是否按「修缺陷」接受。 + +### C4. Milvus 双地址与部署约束（`app/core/config.py` + `app/service/health_service.py`） + +- 新增 `milvus_local_uri`（默认空串）与 `resolved_milvus_uri = milvus_local_uri or milvus_uri`； +- `health_service.py` 的健康检查改用 `resolved_milvus_uri`。 + +**约束（请一并确认并写入环境规范）：** +- 团队/生产环境的 `.env` **必须留空 `MILVUS_LOCAL_URI`**，否则健康检查与部分检索链路会指向 + 本地 Milvus Lite 文件，出现「健康检查正常、实际查的是另一个库」的隐性偏差； +- `.env` 中默认值为空串，**默认行为与改动前完全一致**。 + +--- + +## 五、D 类：本次已从 ZSY 侧移除、不进入集成分支的内容 + +为避免把违反底座规则的内容带进集成分支，以下内容已在本分支删除： + +| 内容 | 原因 | +|---|---| +| `app/service/knowledge_tool_service.py` | **生产死代码**（全仓唯一引用是它自己的单测）。生产路径是 `bootstrap.get_knowledge_search_service()` → `KnowledgeSearchService` | +| `app/infrastructure/milvus_knowledge_adapter.py` | **违反 `AGENTS.md` §E**：硬编码 `knowledge_id`/`snippet`/`tags` 等 Milvus 字段名，会打挂字段名不同的环境。且仅被上面那个死模块引用 | +| `tests/unit/service/test_knowledge_tool_service.py`、`tests/unit/infrastructure/test_milvus_knowledge_adapter.py` | 上述两模块的单测 | +| `tests/unit/service/test_knowledge_retrieval.py` | 依赖已移除的 legacy 位置参数构造 | +| `app/service/knowledge_retrieval_service.py` 的 legacy 双构造与 `_legacy_search` | 已整文件回退为底座版本，消除「同一服务两套构造语义 + 两套检索实现」的技术债 | + +检索仍统一走底座既有链路：`ToolExecutor` → `knowledge_search_tool` → `KnowledgeSearchService` +→ `app/core/knowledge_schema.py` 的 `detect_schema()` **运行时字段探测**，无任何硬编码字段名。 + +--- + +## 六、E 类：明确未改动的底座边界 + +- `docs/00`、`docs/02` 的表结构与字段语义：**未改动**； +- `alembic/` 迁移脚本：**未改动**； +- Milvus 字段名：**无硬编码**，全部走运行时探测； +- `docs/05-接口文档.md` 的既有章节编号：**§8.1–§8.4 全部保持原编号**，客服二期新增的 + 「客服画像候选」登记为 **§8.5**（不占用既有号段），另有 §9.7 追加；原有交叉引用 + 「§8.3 知识库管理三端点」「§8.4 公共只读工具索引」仍然成立； +- 其他业务 Agent、场外/推广/风控域：**未改动**行为契约。 + +--- + +## 七、F 类：需要业务方在环境侧确认的既有约束（非本次引入） + +1. `config_release` 是**环境数据、不随代码合并**：换环境需重新发布客服工具白名单。 +2. Milvus 集合 schema 因环境而异，检索层已运行时探测，**任何新增代码都不得硬编码字段名**。 +3. 本地 Milvus Lite 仅用于本地开发；团队环境应使用受管 Milvus 并留空 `MILVUS_LOCAL_URI`。 diff --git a/tests/unit/infrastructure/test_milvus_knowledge_adapter.py b/tests/unit/infrastructure/test_milvus_knowledge_adapter.py deleted file mode 100644 index f730288..0000000 --- a/tests/unit/infrastructure/test_milvus_knowledge_adapter.py +++ /dev/null @@ -1,47 +0,0 @@ -import pytest - -from app.core.errors import ForbiddenAgentError -from app.infrastructure.milvus_knowledge_adapter import MilvusKnowledgeClient - - -class FakeMilvus: - def __init__(self) -> None: - self.kwargs = None - - async def search(self, **kwargs): - self.kwargs = kwargs - return [[{ - "distance": 0.91, - "entity": { - "knowledge_id": "101", - "snippet": "开户说明", - "title": "基金开户", - "tags": ["开户"], - "version": "v1", - }, - }]] - - -@pytest.mark.asyncio -async def test_knowledge_adapter_uses_cosine_and_minimal_public_projection() -> None: - client = MilvusKnowledgeClient("http://unused") - fake = FakeMilvus() - client._client = fake - - hits = await client.search("fin_faq_collection", [0.1] * 1024, 3) - - assert hits[0]["knowledge_id"] == "101" - assert hits[0]["snippet"] == "开户说明" - assert hits[0]["score"] == 0.91 - assert fake.kwargs["collection_name"] == "fin_faq_collection" - assert fake.kwargs["limit"] == 3 - assert fake.kwargs["search_params"] == {"metric_type": "COSINE"} - assert fake.kwargs["output_fields"] == ["knowledge_id", "title", "snippet", "tags", "version"] - - -@pytest.mark.asyncio -async def test_knowledge_adapter_rejects_non_public_collection() -> None: - client = MilvusKnowledgeClient("http://unused") - - with pytest.raises(ForbiddenAgentError): - await client.search("customer_vectors", [0.1] * 1024, 3) diff --git a/tests/unit/service/test_knowledge_retrieval.py b/tests/unit/service/test_knowledge_retrieval.py deleted file mode 100644 index 8bc112b..0000000 --- a/tests/unit/service/test_knowledge_retrieval.py +++ /dev/null @@ -1,87 +0,0 @@ -import pytest - -from app.core.contracts import RequestContext -from app.core.errors import RecoverableAgentError -from app.core.knowledge_contracts import KnowledgeHit, KnowledgeQuery -from app.service.knowledge_config import KnowledgeRuntimeConfig -from app.service.knowledge_retrieval_service import KnowledgeRetrievalService - -# ruff: noqa: E501 - - -class FakeEmbedder: - async def embed(self, text: str) -> list[float]: - assert text == "开户" - return [0.1] * 1024 - - -class FakeVectorStore: - def __init__(self) -> None: - self.calls: list[tuple[str, int]] = [] - - async def search(self, collection: str, vector: list[float], top_k: int) -> list[dict[str, object]]: - assert len(vector) == 1024 - self.calls.append((collection, top_k)) - return [] - - -class FakeAuthority: - async def filter_published(self, hits: tuple[object, ...]) -> list[object]: - return [] - - async def search_keyword(self, query: object, collections: tuple[str, ...], top_k: int) -> list[object]: - return [] - - -class BrokenVectorStore: - async def search(self, collection: str, vector: list[float], top_k: int) -> list[dict[str, object]]: - raise RecoverableAgentError("知识检索不可用") - - -class FallbackAuthority: - def __init__(self) -> None: - self.calls: list[tuple[tuple[str, ...], int]] = [] - - async def filter_published(self, hits: tuple[object, ...]) -> list[object]: - return [] - - async def search_keyword(self, query: KnowledgeQuery, collections: tuple[str, ...], top_k: int) -> list[KnowledgeHit]: - self.calls.append((collections, top_k)) - return [KnowledgeHit( - knowledge_id="101", collection="fin_policy_collection", snippet="确认规则", - answer="工作日确认", score=1.0, - )] - - -@pytest.mark.asyncio -async def test_search_uses_faq_collection_for_faq_only() -> None: - vector_store = FakeVectorStore() - service = KnowledgeRetrievalService( - FakeEmbedder(), vector_store, KnowledgeRuntimeConfig(), FakeAuthority() - ) - - result = await service.search( - KnowledgeQuery(query="开户", intents=("faq",)), - RequestContext(user_id="visitor-1", trace_id="trace", roles=("visitor",), data_scope="public"), - ) - - assert vector_store.calls == [("fin_faq_collection", 3)] - assert result.searched_collections == ("fin_faq_collection",) - - -@pytest.mark.asyncio -async def test_milvus_failure_falls_back_to_published_active_unexpired_knowledge() -> None: - authority = FallbackAuthority() - service = KnowledgeRetrievalService( - FakeEmbedder(), BrokenVectorStore(), KnowledgeRuntimeConfig(), authority - ) - - result = await service.search( - KnowledgeQuery(query="开户", intents=("policy_explain",)), - RequestContext(user_id="visitor-1", trace_id="trace", roles=("visitor",), data_scope="public"), - ) - - assert authority.calls == [(("fin_policy_collection",), 5)] - assert result.degraded is True - assert result.degradation_reason == "milvus_unavailable" - assert result.hits[0].answer == "工作日确认" diff --git a/tests/unit/service/test_knowledge_tool_service.py b/tests/unit/service/test_knowledge_tool_service.py deleted file mode 100644 index 1310736..0000000 --- a/tests/unit/service/test_knowledge_tool_service.py +++ /dev/null @@ -1,131 +0,0 @@ -import pytest - -from app.core.contracts import RequestContext -from app.core.knowledge_contracts import KnowledgeHit, KnowledgeQuery, KnowledgeSearchResult -from app.service import knowledge_tool_service -from app.service.knowledge_tool_service import DatabaseEmbeddingAdapter, query_knowledge_tool - - -class FakeGateway: - def __init__(self) -> None: - self.calls: list[tuple[str, str, int]] = [] - - async def embed(self, *, endpoint_code: str, text: str, timeout_ms: int) -> list[float]: - self.calls.append((endpoint_code, text, timeout_ms)) - return [0.1] * 1024 - - -@pytest.mark.asyncio -async def test_embedding_adapter_uses_single_text_gateway_contract() -> None: - gateway = FakeGateway() - adapter = DatabaseEmbeddingAdapter("knowledge-embedding", 15000, gateway=gateway) - - vector = await adapter.embed("基金开户") - - assert len(vector) == 1024 - assert gateway.calls == [("knowledge-embedding", "基金开户", 15000)] - - -@pytest.mark.asyncio -async def test_query_tool_degrades_when_embedding_endpoint_is_unconfigured(monkeypatch) -> None: - class Settings: - knowledge_embedding_endpoint_code = "" - - monkeypatch.setattr("app.service.knowledge_tool_service.get_settings", lambda: Settings()) - - result = await query_knowledge_tool( - KnowledgeQuery(query="基金开户", intents=("faq",)), - RequestContext( - user_id="visitor-1", trace_id="trace", roles=("visitor",), data_scope="public" - ), - ) - - assert result.degraded is True - assert result.degradation_reason == "embedding_endpoint_unconfigured" - - -@pytest.mark.asyncio -async def test_query_tool_uses_configured_embedding_endpoint_and_read_only_dependencies( - monkeypatch, -) -> None: - class Settings: - knowledge_embedding_endpoint_code = "knowledge-embedding" - knowledge_embedding_timeout_ms = 15000 - milvus_uri = "http://milvus:19530" - milvus_token = "" - - class FakeGateway: - calls: list[tuple[str, str, int]] = [] - - async def embed( - self, *, endpoint_code: str, text: str, timeout_ms: int - ) -> list[float]: - self.calls.append((endpoint_code, text, timeout_ms)) - return [0.1] * 1024 - - class FakeMilvus: - def __init__(self, uri: str, token: str | None) -> None: - self.uri = uri - self.token = token - - class FakeSession: - async def __aenter__(self) -> object: - return object() - - async def __aexit__(self, exc_type, exc, traceback) -> None: - return None - - class FakeAuthority: - def __init__(self, session: object) -> None: - self.session = session - - class FakeRetrievalService: - def __init__(self, embedder, vector_store, config, authority) -> None: - self.embedder = embedder - self.vector_store = vector_store - self.config = config - self.authority = authority - - async def search( - self, query: KnowledgeQuery, context: RequestContext - ) -> KnowledgeSearchResult: - vector = await self.embedder.embed(query.query) - assert len(vector) == 1024 - assert isinstance(self.vector_store, FakeMilvus) - assert isinstance(self.authority, FakeAuthority) - assert self.config.routes["faq"] == ("fin_faq_collection", 3) - assert context.data_scope == "public" - return KnowledgeSearchResult( - hits=( - KnowledgeHit( - knowledge_id="1", - collection="fin_faq_collection", - snippet="snippet", - answer="answer", - ), - ), - searched_collections=("fin_faq_collection",), - ) - - gateway = FakeGateway() - monkeypatch.setattr(knowledge_tool_service, "get_settings", lambda: Settings()) - monkeypatch.setattr(knowledge_tool_service, "DatabaseModelGateway", lambda: gateway) - monkeypatch.setattr(knowledge_tool_service, "MilvusKnowledgeClient", FakeMilvus) - monkeypatch.setattr(knowledge_tool_service, "KnowledgeMysqlAuthority", FakeAuthority) - monkeypatch.setattr( - knowledge_tool_service, "KnowledgeRetrievalService", FakeRetrievalService - ) - monkeypatch.setattr(knowledge_tool_service, "SessionFactory", FakeSession) - - result = await query_knowledge_tool( - KnowledgeQuery(query="基金开户", intents=("faq",)), - RequestContext( - user_id="visitor-1", - trace_id="trace", - roles=("visitor",), - data_scope="public", - ), - ) - - assert result.hits[0].answer == "answer" - assert gateway.calls == [("knowledge-embedding", "基金开户", 15000)]