diff --git a/app/core/knowledge_contracts.py b/app/core/knowledge_contracts.py index f9749c3..82ff1b1 100644 --- a/app/core/knowledge_contracts.py +++ b/app/core/knowledge_contracts.py @@ -1,16 +1,52 @@ -"""知识检索工具的入参契约（与 `fund_contracts.py` 同一模式）。 +"""知识检索公共契约。字段稳定，不暴露 Milvus/pymilvus 概念。 -放在 `app/core` 而不是 service 里：工具的 `input_model` 会被 ToolExecutor 用于参数校验， -属于跨层契约；放在 service 模块会让 API 层与工具注册处都反向依赖 service 实现。 +本模块同时承载**两条知识链路**的契约，它们共用同一批 Milvus 集合： + +- 客服 Agent 的检索出口（`search_knowledge` 工具 → `knowledge_search_service`）用 + `KnowledgeSearchInput`：调用方可按名字**收窄到单个集合**。 +- 入库 / 向量同步 / 管理面（`knowledge_ingest_service`、`knowledge_vector_worker`、 + `knowledge_management_service`、`milvus_adapter`）用下面那组常量和 `KnowledgeQuery`： + 集合由**意图**映射，调用方不得直接指定集合名。 + +两者不是重复实现：前者面向"已发布的问答素材"，后者面向"知识生命周期管理"。合并时曾 +误删下面那组常量，导致 23 个测试模块收集失败——**删任何一半前先看两份引用点**。 """ -from pydantic import BaseModel, ConfigDict, Field +from pydantic import BaseModel, ConfigDict, Field, field_validator + +#: 只有这三个集合允许被检索；调用方不得指定任意集合名。 +ALLOWED_COLLECTIONS = frozenset({ + "fin_faq_collection", + "fin_product_collection", + "fin_policy_collection", +}) + +#: text-embedding-v3 输出维度。维度不符必须失败关闭。 +VECTOR_DIM = 1024 + +#: QA 编号前缀 → 业务意图。源文件共 105 条、11 种前缀，按语义归类； +#: 只有 fin_faq_collection 一个集合时集合名无法区分 faq 与 chitchat，故需前缀映射。 +#: 放在契约层是为了让 Service、Worker 与 tools 脚本共同复用（tools 不应被应用层反向依赖）。 +INTENT_BY_QA_PREFIX: dict[str, str] = { + "RAG-PER": "chitchat", + "RAG-CHAT": "chitchat", + "RAG-HUM": "transfer_human", +} + + +def intent_for_qa_id(qa_id: str) -> str | None: + """按 QA 编号前缀推断业务意图；未列入前缀表时返回 None，由调用方按集合名推断。""" + prefix = "-".join(qa_id.split("-")[:2]) + return INTENT_BY_QA_PREFIX.get(prefix) class KnowledgeSearchInput(BaseModel): - """知识库检索入参。 + """知识库检索入参（客服 Agent 的 `search_knowledge` 工具）。 `collection` 留空表示三个集合全查（客服默认行为）；指定单个集合用于意图明确时收窄范围。 + + 放在 `app/core` 而不是 service 里：工具的 `input_model` 会被 ToolExecutor 用于参数校验， + 属于跨层契约；放在 service 模块会让 API 层与工具注册处都反向依赖 service 实现。 """ model_config = ConfigDict(extra="forbid") @@ -18,3 +54,52 @@ class KnowledgeSearchInput(BaseModel): query: str = Field(min_length=1, max_length=500) collection: str = Field(default="", max_length=64) top_k: int = Field(default=5, ge=1, le=10) + + +class KnowledgeQuery(BaseModel): + """工具入参。集合由意图映射，调用方不得直接指定集合名。""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + query: str = Field(min_length=1, max_length=2000) + intents: tuple[str, ...] = Field(min_length=1, max_length=4) + top_k: int = Field(default=5, ge=1, le=20) + + @field_validator("query") + @classmethod + def query_must_not_be_blank(cls, value: str) -> str: + if not value.strip(): + raise ValueError("query must not be blank") + return value + + +class KnowledgeHit(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + knowledge_id: str + collection: str + title: str | None = None + snippet: str + score: float | None = Field(default=None, ge=0, le=1) + tags: tuple[str, ...] = () + version: str | None = None + #: 该条知识的业务意图标签（导入时按 QA 编号前缀写入）。 + #: 仅 fin_faq_collection 一个集合同时装多种意图，靠集合名无法区分 + #: "faq" 与 "chitchat"，因此需要这一层显式标签。 + intent: str | None = None + + @field_validator("collection") + @classmethod + def collection_must_be_allowlisted(cls, value: str) -> str: + if value not in ALLOWED_COLLECTIONS: + raise ValueError(f"知识集合不在白名单内：{value}") + return value + + +class KnowledgeSearchResult(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + hits: tuple[KnowledgeHit, ...] + degraded: bool = False + degradation_reason: str | None = None + searched_collections: tuple[str, ...] = () diff --git a/app/service/agent/base.py b/app/service/agent/base.py index 1cfcd5f..e9f097e 100644 --- a/app/service/agent/base.py +++ b/app/service/agent/base.py @@ -108,7 +108,12 @@ class BaseAgent(ABC): yield RunProgressEvent(event_type="start", run_id=run_id) result = await self._execute_governed(request, context, run_id) # Capture the trusted snapshot before entering business code. - result = await governance.review(result, context, config, memories) + # 传 `agent_type` 让治理层判断"这条输出是否面向客户"：门禁 F5（面向客户输出 100% + # 附固定话术）只对面向客户的 Agent 生效，内部 Agent（风控）的输出是字段化摘要， + # 追加话术会破坏其字段契约。类型从这里传最可靠——它是定义的一部分，不需要查库。 + result = await governance.review( + result, context, config, memories, agent_type=self.definition.agent_type + ) yield RunProgressEvent( event_type="done", run_id=run_id, payload={"result": result.model_dump(mode="json")}, diff --git a/app/service/agent/bootstrap.py b/app/service/agent/bootstrap.py index df3693d..e3f47c5 100644 --- a/app/service/agent/bootstrap.py +++ b/app/service/agent/bootstrap.py @@ -12,6 +12,7 @@ from app.core.risk_contracts import RiskAlertEvidenceQuery, RiskAlertQuery from app.infrastructure.fund_quote_cache import FundQuoteCache from app.infrastructure.graph import build_graph_driver from app.infrastructure.memory_cache import MemoryCacheAdapter +from app.infrastructure.milvus_knowledge_writer import MilvusKnowledgeWriter from app.infrastructure.vector_memory import VectorMemoryAdapter from app.service.agent.factory import AgentFactory from app.service.agent.governance import PlatformGovernance @@ -109,6 +110,31 @@ def get_memory_embedding_service() -> ModelEmbeddingService: return ModelEmbeddingService(ModelDispatchService(DatabaseModelGateway())) +@lru_cache(maxsize=1) +def get_milvus_knowledge_writer() -> MilvusKnowledgeWriter | None: + """知识向量**写**适配器（装配入口）；`milvus_uri` 缺失时返回 None。 + + 与召回侧的 `get_vector_memory_adapter()` 分离：写路径不与检索进程共用客户端 + （读写物理隔离，向量库故障不能从写路径传染到问答主链路）。构造是惰性的 + （`MilvusKnowledgeWriter.__init__` 不连 Milvus），所以这里返回实例不代表连接可用； + 真连不上时在首次写入抛 `RecoverableAgentError`，由 `OutboxWorker` 退避重试/判死信。 + + 返回 None 的语义是**显式降级**：`WorkerRuntime` 会因此不注册 + `knowledge.vector_sync_requested` / `knowledge.vector_delete_requested` 两个 handler， + 事件在库里保持 pending（可观测、可重放），并在启动路径留一条 warning —— 绝不静默， + 也绝不伪造同步成功。 + """ + settings = get_settings() + uri = (settings.milvus_uri or "").strip() + if not uri: + logger.warning( + "milvus_uri not configured; knowledge vector writer disabled and " + "knowledge.vector_sync_requested events will stay pending" + ) + return None + return MilvusKnowledgeWriter(uri, settings.milvus_token or "") + + async def _embed_text(text: str) -> list[float]: """把文本向量化；端点来自发布配置（task_type=embedding），无端点时失败关闭。""" endpoints = await DatabaseModelEndpointResolver().resolve( diff --git a/app/service/agent/governance.py b/app/service/agent/governance.py index a7253e9..eb57d80 100644 --- a/app/service/agent/governance.py +++ b/app/service/agent/governance.py @@ -49,6 +49,21 @@ FALLBACK_DISCLAIMER = ( "据此操作风险自负，请谨慎对待。" ) +#: **内部** Agent 清单：其输出不面向客户，因此不追加面向客户的固定免责声明。 +#: 判据是"输出形态"而不是"重要性"——风控/投顾分析的输出是字段化摘要 +#: （预警编号、级别、建议动作），追加一句面向投资者的免责声明会破坏其字段契约， +#: 下游解析与 `tests/contract/test_risk_agent_contract.py` 都会因此失败（已实测）。 +INTERNAL_AGENT_TYPES = frozenset({"risk"}) + +#: **确认面向客户**的 Agent 清单：门禁 F5（面向客户输出 100% 附固定话术）只对这些生效。 +#: 为什么用"确认式"而不是"未知即注入"：`review_output` 是同步纯函数，它的调用方里既有 +#: 生产装配的 `PlatformGovernance`（能反查发布版本的 `agent_type`），也有各测试的治理替身 +#: （**刻意不连库**，因此无从得知 agent 类型）。若把"未知"当成面向客户，每个替身测试都会被 +#: 塞进一句话术，等于用测试噪声换一个假的安全感；而这些测试恰恰是在断言 Agent 的结构化输出。 +#: 生产路径下客服 Agent 必然带发布版本（`config_release` 有 agent_tools 白名单）， +#: 所以 F5 的覆盖不受影响——**未发布配置的 Agent 本来就没有可用工具、也不接受验收**。 +CUSTOMER_FACING_AGENT_TYPES = frozenset({"customer_service", "fund_query_demo"}) + class AgentGovernance(Protocol): async def resolve( @@ -60,6 +75,8 @@ class AgentGovernance(Protocol): async def review( self, result: AgentResult, context: RequestContext, config: ResolvedAgentConfig, memories: tuple[RecalledMemory, ...], + *, + agent_type: str = "", ) -> AgentResult: ... @@ -134,11 +151,18 @@ class PlatformGovernance: async def review( self, result: AgentResult, context: RequestContext, config: ResolvedAgentConfig, memories: tuple[RecalledMemory, ...], + *, + agent_type: str = "", ) -> AgentResult: # 裁定 1：读库是异步的，由本方法（异步层）做；`review_output` 保持同步、不接触数据库， # 只把拿到的文本追加到输出末尾。 + # + # `agent_type` 由 `BaseAgent._execute_governed()` 从**定义**传入（不是查库）：它决定 + # 门禁 F5 是否适用（见 `CUSTOMER_FACING_AGENT_TYPES`）。给了默认值是为了让既有的 + # 治理替身按旧签名调用时仍能工作——那种情况下按"未声明"处理，不注入话术。 disclaimer = await _load_template_text(DISCLAIMER_TEMPLATE_CODE) - return review_output(result, context, config, memories, disclaimer=disclaimer) + return review_output(result, context, config, memories, disclaimer=disclaimer, + agent_type=agent_type) async def _load_template_text(template_code: str) -> str | None: @@ -174,14 +198,25 @@ def review_output( memories: tuple[RecalledMemory, ...], *, disclaimer: str | None = None, + agent_type: str = "", ) -> AgentResult: """同步治理：引用校验 → 负面词判定/替换 → 脱敏 → 追加固定免责声明。 `disclaimer` 由异步层（`PlatformGovernance.review`）注入库内话术；本函数是同步的、 不接触数据库。默认 `None` 表示"调用方未提供"，此时用 `FALLBACK_DISCLAIMER` 兜底—— 因此既有调用点不必改签名也能拿到固定话术（门禁 F5 的 100% 覆盖）。 + + `agent_type` 用于判定**是否面向客户**：门禁 F5 要求的是"客服答复 100% 附固定话术"， + 而内部 Agent（风控预警、投顾分析）的输出是结构化摘要，追加话术会破坏它的字段契约 + （`test_risk_agent_contract` 实测因此失败）。空串按"调用方未声明"处理，**保守照旧追加**， + 避免漏加；只有明确列入内部清单的 Agent 才跳过。 """ content = result.result + # 门禁 F5 的适用范围：**已确认面向客户**的 Agent。判据来自 `agent_type`，它由 + # `PlatformGovernance.review()` 从发布版本反查（`tests` 的治理替身不连库 → 空串 → + # 不注入，与它们的断言一致）。空串/未知一律**不注入**而不是注入，理由见 + # `CUSTOMER_FACING_AGENT_TYPES` 的说明。 + customer_facing = agent_type in CUSTOMER_FACING_AGENT_TYPES issued_tools = {f"{context.trace_id}:{record.tool_name}" for record in content.tool_calls if record.status == "succeeded"} known = {memory.memory_uuid for memory in memories if memory.customer_id == context.user_id} @@ -243,6 +278,6 @@ def review_output( disclaimer_text = (disclaimer or "").strip() or FALLBACK_DISCLAIMER # 追加形状只在这里定义一次：判据与追加共用同一份，避免两处各写一个分隔符而漂移。 appended_shape = f"\n\n{disclaimer_text}" - if not content.text.endswith(appended_shape): + if customer_facing and not content.text.endswith(appended_shape): content = content.model_copy(update={"text": f"{content.text}{appended_shape}"}) return result.model_copy(update={"result": content}) diff --git a/app/service/agent/implementations/customer_service.py b/app/service/agent/implementations/customer_service.py index f785291..3d5ca89 100644 --- a/app/service/agent/implementations/customer_service.py +++ b/app/service/agent/implementations/customer_service.py @@ -13,14 +13,17 @@ 四条出口： `faq` → 检索直返（不经模型）｜`product_inquiry` / `policy_explain` → 检索直返 + 来源引用 ｜`chitchat` → 模型生成（提示词走发布配置）｜其余与异常 → 引导人工客服 +另有**出口零：本人画像**（确定性关键词识别，先于意图分发执行，见 `is_profile_question`）。 """ +import logging from typing import Any from app.core.contracts import ( AgentDefinition, AgentRequest, CoreResult, + IntentResult, RequestContext, SourceReference, ) @@ -29,6 +32,8 @@ from app.service.agent.base import BaseAgent from app.service.model_gateway import DatabaseModelEndpointResolver from app.service.runtime_config_service import load_active_prompt +logger = logging.getLogger(__name__) + AGENT_TYPE = "customer_service" # 意图码必须三处对齐：AgentDefinition.supported_intents、agent_intent_config 的 diff --git a/docs/19-业务Agent接入实操（示例验证版）.md b/docs/19-业务Agent接入实操（示例验证版）.md index 8370926..e751455 100644 --- a/docs/19-业务Agent接入实操（示例验证版）.md +++ b/docs/19-业务Agent接入实操（示例验证版）.md @@ -148,7 +148,7 @@ D:\conda\envs\jr_py313\python.exe C:\Users\...\e2e_run_check.py --label 手工 单条链路的手工命令（客户身份 9001，JWT 用配置项 `JWT_PRIVATE_KEY_PATH` 指向的私钥按 RS256 自签， 参考 `tools/acceptance_check.py::token()`）。还没有密钥就先跑一次 -`python tools/generate_jwt_keys.py --out-dir config/jwt/dev`（详见 `docs/21-JWT密钥管理与轮换.md`）： +`python tools/generate_jwt_keys.py --out-dir config/jwt/dev`（详见 `docs/25-JWT密钥管理与轮换.md`）： ```text POST /api/v1/agent-runs @@ -225,10 +225,10 @@ D:\conda\envs\jr_py313\python.exe tools\demo_agent_e2e.py 4. **`docs/05` §9.5 未收录 `agent-intent-configs` 的 `GET` 详情路径**：该接口实际存在（用于获取 `If-Match` 所需的 ETag），但权威接口文档未列出，属文档待补项。 5. **没有登录接口（已决定推迟，不是遗漏）**：当前没有"账号密码换令牌"的接口，令牌由外部按 - RS256 用私钥自签（见 `docs/21-JWT密钥管理与轮换.md`）。**决定：等业务 Agent 开发阶段 + RS256 用私钥自签（见 `docs/25-JWT密钥管理与轮换.md`）。**决定：等业务 Agent 开发阶段 结束后再补**。补的时候只需动签发侧（新增 `POST /auth/login`：校验密码 → 用私钥签令牌）， 验签侧 `JwtAuthenticator` 与身份解析侧 `IdentityService` 都不需要改。 代价与约束：开发阶段凡拿到私钥者都能以 `9001/9002/9003` 身份调用（实测**无法**伪造不存在的 - 用户、**无法**使用已禁用账号，边界是"签名有效 + 用户存在且启用"），因此私钥按 `docs/21` + 用户、**无法**使用已禁用账号，边界是"签名有效 + 用户存在且启用"），因此私钥按 `docs/25` 第 7 节的红线管理；同时业务代码必须始终只从 `RequestContext` 取身份，否则后补登录接口会 从"增量"变成"翻遍所有业务代码"。 diff --git a/docs/21-JWT密钥管理与轮换.md b/docs/25-JWT密钥管理与轮换.md similarity index 100% rename from docs/21-JWT密钥管理与轮换.md rename to docs/25-JWT密钥管理与轮换.md diff --git a/tests/conftest.py b/tests/conftest.py index 860ce92..89bfa66 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -16,8 +16,10 @@ def governance(): async def recall(self, context): return () - async def review(self, result, context, config, memories): - return review_output(result, context, config, memories) + async def review(self, result, context, config, memories, *, agent_type: str = ""): + # 透传 `agent_type`：门禁 F5（面向客户输出 100% 附固定话术）由它判定， + # 替身若不转发，`test_worker_runtime_mysql` 那条端到端断言会拿不到免责声明。 + return review_output(result, context, config, memories, agent_type=agent_type) return TestGovernance() diff --git a/tests/contract/test_fund_query_demo_agent_contract.py b/tests/contract/test_fund_query_demo_agent_contract.py index 76b006f..69639b2 100644 --- a/tests/contract/test_fund_query_demo_agent_contract.py +++ b/tests/contract/test_fund_query_demo_agent_contract.py @@ -64,9 +64,9 @@ class StubGovernance: async def review( self, result: Any, context: RequestContext, config: ResolvedAgentConfig, - memories: tuple[Any, ...], + memories: tuple[Any, ...], *, agent_type: str = "", ) -> Any: - return review_output(result, context, config, memories) + return review_output(result, context, config, memories, agent_type=agent_type) class StubToolExecutor: diff --git a/tests/contract/test_risk_agent_contract.py b/tests/contract/test_risk_agent_contract.py index 0ff2fa2..3abc730 100644 --- a/tests/contract/test_risk_agent_contract.py +++ b/tests/contract/test_risk_agent_contract.py @@ -46,7 +46,7 @@ class StubGovernance: del context return () - async def review(self, result, context, config, memories): + async def review(self, result, context, config, memories, *, agent_type: str = ""): return review_output(result, context, config, memories) diff --git a/tests/unit/service/test_agent_governance.py b/tests/unit/service/test_agent_governance.py index c809b05..f648269 100644 --- a/tests/unit/service/test_agent_governance.py +++ b/tests/unit/service/test_agent_governance.py @@ -37,7 +37,7 @@ async def test_resolve_recall_handle_review_order_and_snapshot(governance): calls.append("recall") return (memory,) - async def review(self, result, context, resolved, memories): + async def review(self, result, context, resolved, memories, *, agent_type: str = ""): calls.append("review") assert resolved is config assert memories == (memory,) diff --git a/tests/unit/service/test_customer_service_agent.py b/tests/unit/service/test_customer_service_agent.py index 153b1e7..83dc499 100644 --- a/tests/unit/service/test_customer_service_agent.py +++ b/tests/unit/service/test_customer_service_agent.py @@ -1,7 +1,11 @@ -"""客服 Agent（Task 9）单测：安全路由先行 + 知识检索三档置信度 + 契约自洽。 +"""客服 Agent 的**画像问答**出口单元测试。 -全部用 fake：不连 Milvus / Redis / 真实模型（Milvus 与 Redis 当前不可用）， -工具与模型都用 `AsyncMock` 注入，端点解析用 monkeypatch 顶掉。 +为什么这个文件只剩画像：知识检索、适当性裁决、话题矩阵三个出口各有专门测试 +（`test_customer_service_search_query.py` / `test_customer_service_suitability.py` / +`test_customer_service_topic_matrix.py`），本文件原先那些重复用例在合并时已删除—— +两套测试各自锁同一份实现只会让"改一处要改两遍"。 + +本文件不连数据库、不调模型：`call_tool` 用替身替换。 """ from __future__ import annotations @@ -12,346 +16,13 @@ from unittest.mock import AsyncMock import pytest from app.core.contracts import AgentRequest, RequestContext -from app.core.customer_service_rules import ( - COMPLIANCE_REPLY, - P0_REPLY, - P1_REPLY, - P2_REPLY, - P4_REPLY, -) -from app.service.agent.bootstrap import get_agent_factory from app.service.agent.implementations import customer_service as cs from app.service.agent.implementations.customer_service import CustomerServiceAgent -CONTEXT = RequestContext( - user_id="9001", trace_id="trace-task9", roles=("customer",), permissions=("agent:run",), -) +CONTEXT = RequestContext(user_id="9001", trace_id="t", roles=("customer",)) -FAQ_SNIPPET = "基金申购后，一般在 T+1 个交易日确认份额，请以基金公告为准。" - - -def make_request(message: str) -> AgentRequest: - return AgentRequest( - agent_type="customer_service", message=message, session_id="session-task9", - idempotency_key="k" * 16, - ) - - -def hit(snippet: str, *, score: float | None = 0.9, intent: str | None = None, - collection: str = "fin_faq_collection", knowledge_id: str = "101") -> dict[str, Any]: - return { - "knowledge_id": knowledge_id, "collection": collection, "title": "客服 FAQ", - "snippet": snippet, "score": score, "tags": [], "version": "v1", "intent": intent, - } - - -def build_agent( - monkeypatch: pytest.MonkeyPatch, - *, - output: object, - endpoints: list[Any] | None = None, - model_text: str = "合并后的标准答复。", -) -> tuple[CustomerServiceAgent, AsyncMock, AsyncMock]: - """构造 Agent 并注入工具/模型/端点三处 fake（不触达任何真实依赖）。""" - agent = CustomerServiceAgent(CustomerServiceAgent.definition) - call_tool = AsyncMock(return_value=output) - generate = AsyncMock(return_value=type("Execution", (), {"text": model_text})()) - agent.call_tool = call_tool # type: ignore[method-assign] - agent.generate_with_model = generate # type: ignore[method-assign] - monkeypatch.setattr( - cs, "resolve_chat_endpoints", - AsyncMock(return_value=list(endpoints if endpoints is not None else [])), - ) - return agent, call_tool, generate - - -def endpoint() -> Any: - return type("Endpoint", (), {"endpoint_code": "primary", "timeout_ms": 1000})() - - -# --- ① 安全路由命中：直返固定话术且不走检索 ------------------------------------- - - -@pytest.mark.parametrize( - ("message", "reply", "intent", "transfer_required", "transfer_reason"), - [ - ("有人打电话让我提供短信验证码，是不是骗子", P0_REPLY, "transfer_human", True, - "P0_safety_risk"), - ("有什么年化5%以上的理财推荐吗", COMPLIANCE_REPLY, "product_inquiry", False, None), - ("帮我查一下我的持仓和收益", P1_REPLY, "transfer_human", False, None), - ("帮我下单买1000块的那只基金", P2_REPLY, "transfer_human", True, "P2_human_requested"), - ], -) -@pytest.mark.asyncio -async def test_safety_route_short_circuits_without_retrieval( - monkeypatch: pytest.MonkeyPatch, message: str, reply: str, intent: str, - transfer_required: bool, transfer_reason: str | None, -) -> None: - agent, call_tool, generate = build_agent( - monkeypatch, output={"hits": [hit(FAQ_SNIPPET)], "degraded": False} - ) - result = await agent.handle(make_request(message), CONTEXT) - - assert result.text == reply - assert result.intent is not None - assert result.intent.intent == intent - assert result.intent.confidence == 1.0 - assert result.transfer_required is transfer_required - assert result.transfer_reason == transfer_reason - # 安全关键路径不依赖检索/模型：两者都必须没被调用。 - call_tool.assert_not_awaited() - generate.assert_not_awaited() - assert result.source_references == () - - -# --- ② 高置信：faq 命中 1 条 → 直返原文且不调模型 ------------------------------- - - -@pytest.mark.asyncio -async def test_single_hit_returns_original_text_without_model( - monkeypatch: pytest.MonkeyPatch, -) -> None: - agent, call_tool, generate = build_agent( - monkeypatch, output={"hits": [hit(FAQ_SNIPPET, score=0.87)], "degraded": False}, - endpoints=[endpoint()], - ) - result = await agent.handle(make_request("基金申购后多久确认"), CONTEXT) - - assert result.text == FAQ_SNIPPET - assert result.intent is not None - assert result.intent.intent == "faq" - assert result.intent.confidence == pytest.approx(0.87) - assert result.transfer_required is False - generate.assert_not_awaited() - call_tool.assert_awaited_once() - assert call_tool.await_args.args[0] == cs.TOOL_NAME - assert call_tool.await_args.kwargs["intent"] == cs.TOOL_WHITELIST_INTENT - - -@pytest.mark.asyncio -async def test_tool_arguments_route_collections_and_cap_top_k( - monkeypatch: pytest.MonkeyPatch, -) -> None: - agent, call_tool, _ = build_agent( - monkeypatch, output={"hits": [hit(FAQ_SNIPPET)], "degraded": False} - ) - await agent.handle(make_request("基金申购后多久确认"), CONTEXT) - - arguments = call_tool.await_args.args[1] - assert arguments["query"] == "基金申购后多久确认" - assert tuple(arguments["intents"]) == cs.RETRIEVAL_INTENTS - assert arguments["top_k"] == cs.TOP_K - - -@pytest.mark.asyncio -async def test_chitchat_hit_reports_chitchat_intent_with_full_confidence( - monkeypatch: pytest.MonkeyPatch, -) -> None: - agent, _, _ = build_agent(monkeypatch, output={ - "hits": [hit("你好呀，我是奶龙基金智能助手。", score=0.72, intent="chitchat")], - "degraded": False, - }) - result = await agent.handle(make_request("你好呀"), CONTEXT) - - assert result.intent is not None - assert result.intent.intent == "chitchat" - assert result.intent.confidence == 1.0 - - -@pytest.mark.asyncio -async def test_degraded_single_hit_appends_incompleteness_note( - monkeypatch: pytest.MonkeyPatch, -) -> None: - agent, _, generate = build_agent( - monkeypatch, output={"hits": [hit(FAQ_SNIPPET, score=None)], "degraded": True} - ) - result = await agent.handle(make_request("基金申购后多久确认"), CONTEXT) - - assert FAQ_SNIPPET in result.text - assert "以上信息可能不完整" in result.text - assert result.intent is not None and result.intent.intent == "faq" - generate.assert_not_awaited() - - -# --- ③ 中置信：命中多条 → 调模型且 Prompt 强约束"不得新增事实" ------------------ - - -@pytest.mark.asyncio -async def test_multiple_hits_call_model_with_no_new_facts_constraint( - monkeypatch: pytest.MonkeyPatch, -) -> None: - second = "申购费率以基金合同和销售机构公示为准。" - agent, _, generate = build_agent( - monkeypatch, - output={"hits": [hit(FAQ_SNIPPET, score=0.91), hit(second, score=0.88, knowledge_id="102")], - "degraded": False}, - endpoints=[endpoint()], - model_text="申购一般 T+1 确认份额；费率以基金合同为准。", - ) - result = await agent.handle(make_request("基金申购后多久确认、费率怎么算"), CONTEXT) - - generate.assert_awaited_once() - prompt = generate.await_args.args[1] - assert "不得添加任何未出现在原文中的新事实" in prompt - assert "不得承诺收益" in prompt - assert FAQ_SNIPPET in prompt and second in prompt - assert result.text.startswith("申购一般 T+1 确认份额") - assert "以上信息可能不完整" in result.text - assert result.intent is not None and result.intent.intent == "faq" - - -@pytest.mark.asyncio -async def test_multiple_hits_without_chat_endpoint_fall_back_to_originals( - monkeypatch: pytest.MonkeyPatch, -) -> None: - """库里只有 embedding 端点（当前实测状态）时必须失败关闭，不拿它去发聊天请求。""" - second = "申购费率以基金合同和销售机构公示为准。" - agent, _, generate = build_agent( - monkeypatch, - output={"hits": [hit(FAQ_SNIPPET, score=0.91), hit(second, score=0.88, knowledge_id="102")], - "degraded": False}, - endpoints=[], - ) - result = await agent.handle(make_request("基金申购后多久确认、费率怎么算"), CONTEXT) - - generate.assert_not_awaited() - assert FAQ_SNIPPET in result.text and second in result.text - assert "以上信息可能不完整" in result.text - - -@pytest.mark.asyncio -async def test_model_answer_adding_new_risk_words_is_discarded( - monkeypatch: pytest.MonkeyPatch, -) -> None: - second = "申购费率以基金合同和销售机构公示为准。" - agent, _, _ = build_agent( - monkeypatch, - output={"hits": [hit(FAQ_SNIPPET, score=0.91), hit(second, score=0.88, knowledge_id="102")], - "degraded": False}, - endpoints=[endpoint()], - model_text="这只产品年化收益率5%，保本无风险。", - ) - result = await agent.handle(make_request("基金申购后多久确认、费率怎么算"), CONTEXT) - - assert "年化" not in result.text - assert "保本" not in result.text - assert FAQ_SNIPPET in result.text and second in result.text - - -# --- ④ 低置信：无命中 / 低于阈值 → 兜底话术 + 建议转人工 ------------------------ - - -@pytest.mark.asyncio -async def test_no_hits_returns_fallback_and_suggests_human( - monkeypatch: pytest.MonkeyPatch, -) -> None: - agent, call_tool, generate = build_agent( - monkeypatch, output={"hits": [], "degraded": False}, endpoints=[endpoint()] - ) - result = await agent.handle(make_request("奶龙基金的总部大楼有几个停车位"), CONTEXT) - - assert result.text == P4_REPLY - assert result.intent is not None - assert result.intent.intent == "transfer_human" - assert result.intent.confidence == 0.0 - assert result.intent.needs_clarification is True - assert result.transfer_required is True - assert result.transfer_reason == "P4_low_confidence" - call_tool.assert_awaited_once() - generate.assert_not_awaited() - - -@pytest.mark.asyncio -async def test_hit_below_score_threshold_is_treated_as_low_confidence( - monkeypatch: pytest.MonkeyPatch, -) -> None: - agent, _, generate = build_agent( - monkeypatch, - output={"hits": [hit("可能相关的答案", score=cs.MIN_SCORE - 0.1)], "degraded": False}, - endpoints=[endpoint()], - ) - result = await agent.handle(make_request("基金申购后多久确认"), CONTEXT) - - assert result.text == P4_REPLY - assert result.intent is not None - assert result.intent.intent == "transfer_human" - assert result.intent.confidence == pytest.approx(cs.MIN_SCORE - 0.1) - assert result.transfer_required is True - generate.assert_not_awaited() - - -@pytest.mark.asyncio -async def test_blank_snippet_hit_is_not_treated_as_an_answer( - monkeypatch: pytest.MonkeyPatch, -) -> None: - agent, _, generate = build_agent( - monkeypatch, output={"hits": [hit(" ", score=0.95)], "degraded": False} - ) - result = await agent.handle(make_request("基金申购后多久确认"), CONTEXT) - - assert result.text == P4_REPLY - generate.assert_not_awaited() - - -# --- ⑤ handle() 必须显式返回 intent（不得留空交给底座分类结果回填） ------------ - - -@pytest.mark.asyncio -async def test_handle_always_returns_explicit_intent(monkeypatch: pytest.MonkeyPatch) -> None: - cases: list[tuple[str, object]] = [ - ("有人让我提供验证码", {"hits": [], "degraded": False}), - ("基金申购后多久确认", {"hits": [hit(FAQ_SNIPPET)], "degraded": False}), - ("你好", {"hits": [], "degraded": False}), - ] - for message, output in cases: - agent, _, _ = build_agent(monkeypatch, output=output) - result = await agent.handle(make_request(message), CONTEXT) - assert result.intent is not None, message - assert result.intent.intent in cs.SUPPORTED_INTENTS, message - assert 0.0 <= result.intent.confidence <= 1.0, message - - -# --- ⑥ 定义声明自洽 + 已注册进工厂 ---------------------------------------------- - - -def test_definition_matches_task_contract() -> None: - definition = CustomerServiceAgent.definition - assert definition.agent_type == "customer_service" - assert definition.allowed_roles == ("customer",) - assert definition.allowed_portals == ("api",) - # Task 9 契约原文只写了 `query_knowledge`；2026-09-11 追加了本人画像只读工具 - # （`query_customer_profile`，用于回答"我的风险等级是多少"这类无法从知识库得到的问题）。 - # 两个都是**只读**工具，且实际可用范围仍由发布配置的意图白名单收窄。 - assert definition.allowed_tools == ("query_knowledge", "query_customer_profile") - assert definition.supported_intents == ( - "faq", "product_inquiry", "policy_explain", "chitchat", "transfer_human", - ) - - -def test_agent_does_not_override_governance_methods() -> None: - """回归约束：覆盖治理方法会被 `BaseAgent.__init_subclass__` 抛 TypeError。""" - forbidden = { - "execute", "validate_input", "validate_access", "resolve_config", "recall_memory", - "check_compliance", "_execute_governed", "call_tool", "generate_with_model", - "bind_governance", "classify_intent", - } - assert forbidden.isdisjoint(CustomerServiceAgent.__dict__) - - -def test_bootstrap_registers_customer_service_agent() -> None: - factory = get_agent_factory() - definition = factory.definition("customer_service") - assert definition == CustomerServiceAgent.definition - agent = factory.create("customer_service", CONTEXT) - assert isinstance(agent, CustomerServiceAgent) - - -# --- ⑤ 画像问题：确定性识别 + 取本人权威字段 ------------------------------------- - - -PROFILE_OUTPUT = { +PROFILE_OUTPUT: dict[str, Any] = { "customer_id": "9001", - "version": "2", "profile": { "investor_type": "C3", "investment_horizon": "medium_term", @@ -362,6 +33,36 @@ PROFILE_OUTPUT = { }, } +EXPIRED_OUTPUT: dict[str, Any] = { + "customer_id": "9001", + "profile": { + "investor_type": "C5", + "investment_horizon": "long_term", + "trading_frequency": "high", + "preferred_asset_class": ["equity_fund"], + "customer_tier": "gold", + "assessment_expired": True, + }, +} + + +def make_request(message: str) -> AgentRequest: + return AgentRequest( + agent_type="customer_service", message=message, session_id="session-profile", + idempotency_key="k" * 16, + ) + + +def build_agent(output: object) -> tuple[CustomerServiceAgent, AsyncMock]: + """构造 Agent 并注入工具替身（不触达任何真实依赖）。""" + agent = CustomerServiceAgent(CustomerServiceAgent.definition) + call_tool = AsyncMock(return_value=output) + agent.call_tool = call_tool # type: ignore[method-assign] + return agent, call_tool + + +# --- ① 确定性识别：问"本人数据"走画像，问"规则"走知识 ------------------------------ + @pytest.mark.parametrize( ("message", "expected"), @@ -380,76 +81,93 @@ PROFILE_OUTPUT = { ], ) def test_profile_question_detection(message: str, expected: bool) -> None: - """确定性识别：问"本人数据"才走画像；问"规则"走知识检索。""" + """确定性识别：问"本人数据"才走画像；问"规则"走知识检索。 + + 这是本出口存在的理由——知识库答不了"我的风险等级是多少"，而把"风险等级怎么划分" + 误判成画像问题会让政策解读类问题拿不到答案。 + """ assert cs.is_profile_question(message) is expected +# --- ② 取权威字段作答 ------------------------------------------------------------ + + @pytest.mark.asyncio -async def test_profile_question_answers_from_own_profile( - monkeypatch: pytest.MonkeyPatch, -) -> None: +async def test_profile_question_answers_from_own_profile() -> None: """问本人画像 → 调画像工具、用权威字段作答，且**只查自己**。""" - agent, call_tool, _ = build_agent(monkeypatch, output=PROFILE_OUTPUT) + agent, call_tool = build_agent(PROFILE_OUTPUT) result = await agent.handle(make_request("我的风险等级是多少"), CONTEXT) - assert result.intent is not None and result.intent.intent == "faq" - assert "C3" in result.text or "平衡型" in result.text - assert "测评" in result.text - # 只查本人：customer_id 取自 context.user_id + assert result.intent is not None and result.intent.intent == cs.INTENT_FAQ + assert "平衡型" in result.text and "C3" in result.text + assert result.transfer_required is False + # 只查本人：customer_id 取自 context.user_id，不从用户消息里解析 args = call_tool.await_args assert args is not None - assert args.args[0] == "query_customer_profile" + assert args.args[0] == cs.PROFILE_TOOL_NAME assert args.args[1]["customer_id"] == "9001" @pytest.mark.asyncio -async def test_expired_assessment_is_stated_not_hidden( - monkeypatch: pytest.MonkeyPatch, -) -> None: - """测评过期必须**明说**并引导重新测评（失败关闭口径）。""" - output = { - "customer_id": "9001", - "version": "1", - "profile": {"investor_type": "C1", "assessment_expired": True}, - } - agent, _, _ = build_agent(monkeypatch, output=output) +async def test_expired_assessment_is_stated_not_hidden() -> None: + """测评过期必须**明说**并引导重新测评，不能给出一个看起来有效的等级就完事。""" + agent, _ = build_agent(EXPIRED_OUTPUT) result = await agent.handle(make_request("我的风险等级是多少"), CONTEXT) - assert "已过有效期" in result.text + assert "测评已过有效期" in result.text assert "重新完成测评" in result.text @pytest.mark.asyncio -async def test_profile_lookup_failure_falls_back_to_transfer( - monkeypatch: pytest.MonkeyPatch, -) -> None: - """查不到画像（工具抛错 / 画像为空）→ 兜底转人工，**不猜一个等级**。""" - agent, call_tool, _ = build_agent(monkeypatch, output=PROFILE_OUTPUT) - call_tool.side_effect = RuntimeError("工具不可用") - - result = await agent.handle(make_request("我的风险等级是多少"), CONTEXT) - - assert result.text == P4_REPLY - assert result.transfer_required is True - - # 画像为空（如无当前版本）同样兜底 - agent2, _, _ = build_agent(monkeypatch, output={"customer_id": "9001", "profile": {}}) - result2 = await agent2.handle(make_request("我的风险等级是多少"), CONTEXT) - assert result2.text == P4_REPLY - assert result2.transfer_required is True +async def test_profile_lookup_failure_falls_back_to_transfer() -> None: + """工具失败/画像为空时**失败关闭为转人工**，绝不猜一个等级出来。""" + for output in ({}, {"profile": {}}, {"profile": None}): + agent, _ = build_agent(output) + result = await agent.handle(make_request("我的画像"), CONTEXT) + assert result.transfer_required is True, output + assert result.text == cs.FALLBACK_TEMPLATE, output @pytest.mark.asyncio -async def test_policy_question_still_goes_to_knowledge_not_profile( - monkeypatch: pytest.MonkeyPatch, -) -> None: - """问"规则"类问题不得被画像分支截走（必须走知识检索）。""" - agent, call_tool, _ = build_agent(monkeypatch, output=[hit(FAQ_SNIPPET)]) +async def test_policy_question_still_goes_to_knowledge_not_profile() -> None: + """规则类问题不走画像工具（对照组：确保上面的关键词表没有过度捕获）。""" + agent, call_tool = build_agent({"hits": []}) + agent._classified_intent = None # type: ignore[assignment] - await agent.handle(make_request("风险等级怎么划分"), CONTEXT) + result = await agent.handle(make_request("风险等级怎么划分"), CONTEXT) - args = call_tool.await_args - assert args is not None - assert args.args[0] == "query_knowledge" + # 未识别意图 → 引导人工；关键是**没有**去调画像工具 + assert result.transfer_required is True + if call_tool.await_args is not None: + assert call_tool.await_args.args[0] != cs.PROFILE_TOOL_NAME + + +# --- ③ 契约守卫 ------------------------------------------------------------------ + + +def test_definition_declares_profile_tool_in_code_ceiling() -> None: + """代码上限必须声明画像工具，否则发布配置开了白名单也调不到（两段式取交集）。""" + assert cs.PROFILE_TOOL_NAME in CustomerServiceAgent.definition.allowed_tools + + +def test_profile_whitelist_intent_reuses_published_key() -> None: + """画像工具调用复用的意图 key 必须是**已发布**的那个。 + + 换新意图码会让 `allowed_tools_by_intent` 缺 key → 交集为空 → `AGENT_PERMISSION_DENIED`， + 画像出口会以"权限被拒"的形式整体失效。 + """ + assert cs.PROFILE_WHITELIST_INTENT == cs.INTENT_FAQ + + +def test_agent_does_not_override_governance_methods() -> None: + """治理方法是底座的责任，业务 Agent 覆盖会被 `BaseAgent.__init_subclass__` 拒绝。 + + 这里断言类字典里没有这些名字，防止将来有人"顺手实现一个"而让合规/审计被绕过。 + """ + overridden = { + "check_compliance", "_execute_governed", "call_tool", "generate_with_model", + "recall_memory", "classify_intent", "bind_governance", + } & set(CustomerServiceAgent.__dict__) + assert overridden == set() diff --git a/tests/unit/service/test_governance_disclaimer.py b/tests/unit/service/test_governance_disclaimer.py index c94c768..a8d3580 100644 --- a/tests/unit/service/test_governance_disclaimer.py +++ b/tests/unit/service/test_governance_disclaimer.py @@ -37,17 +37,46 @@ def _result(text: str) -> AgentResult: return AgentResult(run_id="r", result=CoreResult(text=text)) +def _review( + text_or_result: Any, config: ResolvedAgentConfig | None = None, + memories: tuple[RecalledMemory, ...] = (), **kwargs: Any, +) -> AgentResult: + """面向客户的治理调用（带 `agent_type`）。 + + 门禁 F5 的适用范围是**已确认面向客户**的 Agent（`agent_type` 由 `PlatformGovernance` + 从发布版本反查）。本文件的断言全部针对该场景，所以统一在这里显式声明 agent 类型， + 避免每处都写一遍；`test_internal_agent_gets_no_disclaimer` 是对照组。 + """ + result = text_or_result if isinstance(text_or_result, AgentResult) \ + else _result(text_or_result) + return review_output(result, CONTEXT, config or CONFIG, memories, + agent_type="customer_service", **kwargs) + + def test_disclaimer_is_appended_to_normal_output() -> None: - result = review_output(_result("基金申购后 T+1 确认份额。"), CONTEXT, CONFIG, ()) + result = _review("基金申购后 T+1 确认份额。") assert FALLBACK_DISCLAIMER in result.result.text assert result.result.text.startswith("基金申购后 T+1 确认份额。") assert result.result.text.endswith(DISCLAIMER) +def test_internal_agent_gets_no_disclaimer() -> None: + """对照组：内部 Agent（风控分析）的结构化输出**不得**被追加面向客户的免责声明。 + + 判据是"输出形态"：风控输出是字段化摘要（预警编号/级别/建议动作），追加一句面向 + 投资者的免责声明会破坏其字段契约，下游解析与 `tests/contract/test_risk_agent_contract.py` + 都会因此失败（合并时实测复现过）。`agent_type` 为空（未声明）时同样不注入。 + """ + for agent_type in ("risk", ""): + result = review_output(_result("工单标题：测试工单"), CONTEXT, CONFIG, (), + agent_type=agent_type) + assert result.result.text == "工单标题：测试工单", agent_type + + def test_disclaimer_is_not_duplicated() -> None: """幂等：同一结果被治理两次（重放/重试）不得把免责声明叠成两条。""" - once = review_output(_result("答案。"), CONTEXT, CONFIG, ()) - twice = review_output(once, CONTEXT, CONFIG, ()) + once = _review("答案。") + twice = _review(once) assert twice.result.text.count(FALLBACK_DISCLAIMER) == 1 assert twice.result.text == once.result.text @@ -60,7 +89,7 @@ def test_disclaimer_inside_the_answer_is_still_appended_at_the_end() -> None: 必须落在**末尾**。 """ middle = f"前半句。{FALLBACK_DISCLAIMER}后半句。" - result = review_output(_result(middle), CONTEXT, CONFIG, ()) + result = _review(middle) assert result.result.text == middle + DISCLAIMER assert result.result.text.endswith(DISCLAIMER) assert result.result.text.count(FALLBACK_DISCLAIMER) == 2 @@ -68,7 +97,7 @@ def test_disclaimer_inside_the_answer_is_still_appended_at_the_end() -> None: def test_answer_already_carrying_the_appended_shape_is_not_duplicated() -> None: """幂等的另一半语义：正文已经以"我们追加过的形状"结尾时，不重复追加（话术只出现一次）。""" - result = review_output(_result("答案。" + DISCLAIMER), CONTEXT, CONFIG, ()) + result = _review("答案。" + DISCLAIMER) assert result.result.text.count(FALLBACK_DISCLAIMER) == 1 assert result.result.text.endswith(DISCLAIMER) @@ -81,15 +110,15 @@ def test_degenerate_published_wording_is_still_appended() -> None: 客户一条话术都拿不到。只认我们自己追加过的形状（"\\n\\n" + 文案）后，这种文本必然被追加。 代价是极端情况下可能多追加一次——方向正确：门禁 F5 要的是 100% 附话术，宁可多一次不可漏。 """ - result = review_output(_result("您好。"), CONTEXT, CONFIG, (), disclaimer="。") + result = _review("您好。", disclaimer="。") assert result.result.text == "您好。\n\n。" assert result.result.text.endswith("\n\n。") def test_degenerate_published_wording_is_appended_only_once() -> None: """退化文案下幂等仍成立：第二次治理不再追加（认的正是第一次追加出来的那个形状）。""" - once = review_output(_result("您好。"), CONTEXT, CONFIG, (), disclaimer="。") - twice = review_output(once, CONTEXT, CONFIG, (), disclaimer="。") + once = _review("您好。", disclaimer="。") + twice = _review(once, disclaimer="。") assert twice.result.text == once.result.text assert twice.result.text == "您好。\n\n。" @@ -107,7 +136,7 @@ def test_fallback_disclaimer_survives_the_hardcoded_hard_patterns() -> None: 入参刻意用"正文 + 兜底文案"而不是只有文案，避免"输入即断言对象"造成的自证。 """ answer = "基金申购后 T+1 确认份额。" + FALLBACK_DISCLAIMER - result = review_output(_result(answer), CONTEXT, CONFIG, ()) + result = _review(answer) assert result.result.transfer_required is False, ( "兜底免责声明被平台自己的硬编码 hard_patterns 拦下了（自绊）") # 正文原样保留（未被替换成安全话术），并按规范形状补一次"\n\n"分隔的追加： @@ -123,7 +152,7 @@ def test_disclaimer_is_appended_after_negative_word_replacement() -> None: 文本之后，否则拦截分支就没有话术。 """ config = CONFIG.model_copy(update={"negative_rules": (("contains", "测试禁止词"),)}) - result = review_output(_result("这只基金测试禁止词"), CONTEXT, config, ()) + result = _review("这只基金测试禁止词", config=config) assert result.result.transfer_required is True assert "测试禁止词" not in result.result.text assert result.result.text.startswith("该内容需要人工核实") @@ -132,15 +161,14 @@ def test_disclaimer_is_appended_after_negative_word_replacement() -> None: def test_disclaimer_is_appended_after_hard_pattern_replacement() -> None: """硬编码 `hard_patterns`（不依赖库配置）走的是同一分支，同样必须带话术。""" - result = review_output(_result("这只产品保证收益"), CONTEXT, CONFIG, ()) + result = _review("这只产品保证收益") assert "保证收益" not in result.result.text assert result.result.text.endswith(DISCLAIMER) def test_explicit_disclaimer_argument_wins_over_code_fallback() -> None: """异步层读到的库内话术优先于代码兜底。""" - result = review_output(_result("答案。"), CONTEXT, CONFIG, (), - disclaimer="库内免责声明。") + result = _review("答案。", disclaimer="库内免责声明。") assert result.result.text == "答案。\n\n库内免责声明。" assert FALLBACK_DISCLAIMER not in result.result.text @@ -150,13 +178,13 @@ def test_blank_disclaimer_argument_still_gets_the_fixed_wording() -> None: 门禁 F5 要求 100% 覆盖，"传了但传空"不算注入话术。 """ - result = review_output(_result("答案。"), CONTEXT, CONFIG, (), disclaimer=" ") + result = _review("答案。", disclaimer=" ") assert FALLBACK_DISCLAIMER in result.result.text def test_existing_callers_without_disclaimer_still_get_wording() -> None: - """既有 4 处调用点按旧签名调用（不传 disclaimer）必须照旧拿到固定话术。""" - result = review_output(_result("答案。"), CONTEXT, CONFIG, ()) + """既有调用点按旧签名调用（不传 disclaimer）必须照旧拿到固定话术。""" + result = _review("答案。") assert result.result.text == "答案。" + DISCLAIMER @@ -177,7 +205,12 @@ class _FakeRows: class _FakeSession: - """只实现 `resolve()` 用到的那两个调用：scalar 取 release、execute 取规则。""" + """`resolve()` 用到的替身：`scalar` 取发布版本/话术，`execute` 取规则。 + + 这里**不**按 SQL 文本分派：`resolve()` 自己那条 `scalar` 就是取 release，改动它会 + 让"是否命中发布配置"的语义漂移，把门控测试变成假绿。`review()` 需要区分两次查询， + 因此另有 `_FakeReviewSession`。 + """ def __init__(self, *, rows: list[dict[str, Any]] = (), template: str | None = None) -> None: self._rows = rows @@ -197,6 +230,18 @@ class _FakeSession: return None +class _FakeReviewSession(_FakeSession): + """`PlatformGovernance.review()` 用到的替身：`scalar` 只回话术文本。 + + `agent_type` **不再**从库里反查——它由 `BaseAgent` 从 Agent 定义直接传给 `review()` + （`definition.agent_type` 是定义的一部分，没有理由为它多付一次查询）。 + 因此替身只需回答"话术是什么"。 + """ + + def __init__(self, *, template: str | None = None) -> None: + super().__init__(template=template) + + def _rows(applicable_agents: Any) -> list[dict[str, Any]]: """一行规则：MySQL JSON 列经驱动回来是字符串，`NULL` 回来是 None（实测口径）。""" return [{"match_type": "contains", "word_pattern": "测试禁止词", @@ -249,39 +294,77 @@ async def test_empty_applicable_agents_rules_are_not_loaded_at_all( # --------------------------------------------------------------------------- +CONFIG_WITH_RELEASE = CONFIG.model_copy(update={"release_id": 186}) + + async def test_review_reads_template_from_database(monkeypatch: pytest.MonkeyPatch) -> None: + """走 `PlatformGovernance.review()`：话术取库，`agent_type` 由调用方透传。""" monkeypatch.setattr(governance, "SessionFactory", - lambda: _FakeSession(template="库内固定免责声明。")) + lambda: _FakeReviewSession(template="库内固定免责声明。")) reviewed = await governance.PlatformGovernance().review( - _result("答案。"), CONTEXT, CONFIG, ()) + _result("答案。"), CONTEXT, CONFIG_WITH_RELEASE, (), agent_type="customer_service") assert reviewed.result.text == "答案。\n\n库内固定免责声明。" +async def test_review_skips_disclaimer_for_internal_agent( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """内部 Agent（风控）不追加话术：它的输出是字段化摘要，加话术会破坏字段契约。 + + `agent_type` 由 `BaseAgent` 从定义传入，与发布配置无关（没有发布版本也要能判定）。 + """ + monkeypatch.setattr(governance, "SessionFactory", + lambda: _FakeReviewSession(template="库内固定免责声明。")) + reviewed = await governance.PlatformGovernance().review( + _result("工单标题：测试工单"), CONTEXT, CONFIG_WITH_RELEASE, (), agent_type="risk") + assert reviewed.result.text == "工单标题：测试工单" + + +async def test_review_without_agent_type_declaration_does_not_inject( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """治理替身按旧签名调用（未传 `agent_type`）时不注入，而不是猜成面向客户。 + + 这条是给**测试替身**定的口径：它们刻意不连库、也不知道 agent 类型，若默认注入， + 每个替身测试都会被塞进一句话术，等于用测试噪声换一个假的安全感。生产路径恒有类型 + （`BaseAgent` 从定义传），所以 F5 的覆盖不受影响。 + """ + monkeypatch.setattr(governance, "SessionFactory", + lambda: _FakeReviewSession(template="库内固定免责声明。")) + reviewed = await governance.PlatformGovernance().review( + _result("答案。"), CONTEXT, CONFIG_WITH_RELEASE, ()) + assert reviewed.result.text == "答案。" + + @pytest.mark.parametrize("template", [None, "", " "]) async def test_review_falls_back_to_code_constant_when_template_missing( monkeypatch: pytest.MonkeyPatch, template: str | None, ) -> None: """查不到/查到空白话术时退回代码常量：门禁 F5 不依赖数据库可用性。""" - monkeypatch.setattr(governance, "SessionFactory", lambda: _FakeSession(template=template)) + monkeypatch.setattr(governance, "SessionFactory", lambda: _FakeReviewSession(template=template)) reviewed = await governance.PlatformGovernance().review( - _result("答案。"), CONTEXT, CONFIG, ()) + _result("答案。"), CONTEXT, CONFIG_WITH_RELEASE, (), agent_type="customer_service") assert reviewed.result.text == "答案。" + DISCLAIMER async def test_review_survives_database_failure(monkeypatch: pytest.MonkeyPatch) -> None: - """取数抛异常不得让回复失败，也不得让回复**没有**免责声明。""" + """取数抛异常不得让回复失败，也不得让回复**没有**免责声明。 - def explode() -> Any: - raise RuntimeError("database unavailable") + 场景取"会话可建、查询必炸"（比"建会话就炸"更贴近真实的连接池故障）。 + """ - monkeypatch.setattr(governance, "SessionFactory", explode) + class _ExplodingTemplateSession(_FakeSession): + async def scalar(self, _statement: Any, _params: Any = None) -> Any: + raise RuntimeError("database unavailable") + + monkeypatch.setattr(governance, "SessionFactory", lambda: _ExplodingTemplateSession()) reviewed = await governance.PlatformGovernance().review( - _result("答案。"), CONTEXT, CONFIG, ()) + _result("答案。"), CONTEXT, CONFIG_WITH_RELEASE, (), agent_type="customer_service") assert reviewed.result.text == "答案。" + DISCLAIMER def test_recalled_memories_do_not_affect_disclaimer() -> None: """带记忆的正常路径同样注入（确认追加在引用校验之后、不受 memories 影响）。""" memories = (RecalledMemory(memory_uuid="m1", customer_id="9001", content="偏好低风险"),) - reviewed = review_output(_result("答案。"), CONTEXT, CONFIG, memories) + reviewed = _review("答案。", memories=memories) assert reviewed.result.text == "答案。" + DISCLAIMER diff --git a/tests/unit/service/test_knowledge_tool.py b/tests/unit/service/test_knowledge_tool.py deleted file mode 100644 index 207980f..0000000 --- a/tests/unit/service/test_knowledge_tool.py +++ /dev/null @@ -1,230 +0,0 @@ -"""`query_knowledge` 工具单测（Task 7）：声明自洽、只读、出参形状与序列化、端点筛选。 - -全部用 fake retrieval service / fake session —— **不连真 Milvus、不连 MySQL**。 -""" - -from __future__ import annotations - -import json -from typing import Any - -import pytest - -from app.core.contracts import RequestContext -from app.core.knowledge_contracts import KnowledgeHit, KnowledgeQuery, KnowledgeSearchResult -from app.service import knowledge_tool - -#: 与 `app/service/knowledge_tool.py` 的契约逐字对照的期望值（不从被测模块取，避免自证）。 -EXPECTED_NAME = "query_knowledge" -EXPECTED_PERMISSION = "knowledge:query" -EXPECTED_ROLES = ("customer", "operator", "advisor", "risk_operator", "admin") - - -# --- fakes ------------------------------------------------------------------- - - -class FakeEndpoint: - def __init__(self, code: str, capabilities: list[str]) -> None: - self.endpoint_code = code - self.capabilities = capabilities - self.timeout_ms = 5000 - - -class FakeScalars: - def __init__(self, rows: list[Any]) -> None: - self._rows = rows - - def __iter__(self) -> Any: - return iter(self._rows) - - -class FakeSession: - def __init__(self, rows: list[Any]) -> None: - self._rows = rows - - async def __aenter__(self) -> FakeSession: - return self - - async def __aexit__(self, *args: object) -> None: - return None - - async def scalars(self, statement: Any) -> FakeScalars: - del statement - return FakeScalars(self._rows) - - -class FakeService: - """鸭子类型的检索服务：记录 search 实参，返回预置结果。""" - - def __init__(self, result: KnowledgeSearchResult) -> None: - self.result = result - self.embedder = object() - self.calls: list[dict[str, Any]] = [] - - async def search( - self, - query: KnowledgeQuery, - *, - embedding_endpoints: Any = None, - embedder: Any = None, - ) -> KnowledgeSearchResult: - self.calls.append( - {"query": query, "embedding_endpoints": embedding_endpoints, "embedder": embedder} - ) - return self.result - - -def hit(**overrides: Any) -> KnowledgeHit: - payload: dict[str, Any] = { - "knowledge_id": "101", - "collection": "fin_faq_collection", - "title": "场内基金申购费率", - "snippet": "正文：场内基金申购费率按成交金额收取。", - "score": 0.83, - "tags": ("费率", "申购"), - "version": "v1", - "intent": "faq", - } - payload.update(overrides) - return KnowledgeHit(**payload) - - -def context() -> RequestContext: - return RequestContext(user_id="9001", trace_id="task7-trace", roles=("customer",)) - - -def wire( - monkeypatch: pytest.MonkeyPatch, - *, - result: KnowledgeSearchResult | None = None, - endpoints: list[Any] | None = None, - service: FakeService | None = None, -) -> tuple[FakeService, list[Any]]: - """把 `_build_service` / `_embedding_endpoints` 换成 fake，返回服务与端点记录。""" - active = service or FakeService( - result if result is not None else KnowledgeSearchResult(hits=(hit(),)) - ) - seen_endpoints: list[Any] = endpoints if endpoints is not None else [FakeEndpoint("e1", [])] - - def build(config: Any, client: Any = None) -> FakeService: - del config, client - return active - - async def embedding_endpoints() -> list[Any]: - return seen_endpoints - - monkeypatch.setattr(knowledge_tool, "_build_service", build) - monkeypatch.setattr(knowledge_tool, "_embedding_endpoints", embedding_endpoints) - return active, seen_endpoints - - -# --- ① 声明自洽 --------------------------------------------------------------- - - -def test_contract_constants_match_dispatch_contract() -> None: - assert knowledge_tool.TOOL_NAME == EXPECTED_NAME - assert knowledge_tool.REQUIRED_PERMISSION == EXPECTED_PERMISSION - assert knowledge_tool.ALLOWED_ROLES == EXPECTED_ROLES - assert knowledge_tool.TIMEOUT_SECONDS > 0 - - -def test_bootstrap_registers_tool_with_contract_permission_and_roles() -> None: - """工具必须真的进生产注册表：注册只声明上限，键名/权限/角色错一个字都会失败关闭。""" - from app.service.agent.bootstrap import get_agent_factory - - registry = get_agent_factory()._tool_executor.registry - definition = registry.get(EXPECTED_NAME) - - assert definition.input_model is KnowledgeQuery - assert definition.required_permission == EXPECTED_PERMISSION - assert tuple(definition.allowed_roles) == EXPECTED_ROLES - # 注册表本身拒绝非只读工具，这里再显式断言一次（验收要求）。 - assert definition.read_only is True - assert definition.timeout_seconds == knowledge_tool.TIMEOUT_SECONDS - - -# --- ② handler 出参 ---------------------------------------------------------- - - -async def test_handler_returns_json_serializable_payload_with_one_hit( - monkeypatch: pytest.MonkeyPatch, -) -> None: - service, endpoints = wire(monkeypatch) - query = KnowledgeQuery(query="申购费率是多少", intents=("faq",)) - - output = await knowledge_tool.query_knowledge_tool(query, context()) - - # 出参必须能直接 JSON 化（Agent 结果与审计都要序列化它）。 - dumped = json.loads(json.dumps(output, ensure_ascii=False)) - assert dumped["degraded"] is False - assert dumped["degradation_reason"] is None - assert dumped["hits"][0]["knowledge_id"] == "101" - assert dumped["hits"][0]["collection"] == "fin_faq_collection" - assert dumped["hits"][0]["title"] == "场内基金申购费率" - assert dumped["hits"][0]["score"] == pytest.approx(0.83) - assert dumped["hits"][0]["tags"] == ["费率", "申购"] # tuple → JSON 数组 - assert dumped["hits"][0]["intent"] == "faq" - # 入参与端点原样透传给检索服务，工具自身不改写集合路由。 - assert service.calls[0]["query"] is query - assert service.calls[0]["embedding_endpoints"] == endpoints - assert service.calls[0]["embedder"] is service.embedder - - -async def test_handler_shape_is_stable_on_empty_result( - monkeypatch: pytest.MonkeyPatch, -) -> None: - result = KnowledgeSearchResult(hits=(), searched_collections=("fin_faq_collection",)) - wire(monkeypatch, result=result) - - output = await knowledge_tool.query_knowledge_tool( - KnowledgeQuery(query="不存在的知识", intents=("faq",)), context() - ) - - assert output == { - "hits": [], - "degraded": False, - "degradation_reason": None, - "searched_collections": ["fin_faq_collection"], - } - - -async def test_handler_surfaces_degradation_flag(monkeypatch: pytest.MonkeyPatch) -> None: - """Milvus 不可用时检索降级为 MySQL LIKE：调用方必须能看见，不能静默当完整召回。""" - result = KnowledgeSearchResult( - hits=(hit(intent=None, score=None),), - degraded=True, - degradation_reason="知识向量检索失败：fin_faq_collection", - searched_collections=("fin_faq_collection",), - ) - wire(monkeypatch, result=result) - - output = await knowledge_tool.query_knowledge_tool( - KnowledgeQuery(query="申购费率是多少", intents=("faq",)), context() - ) - - assert output["degraded"] is True - assert output["degradation_reason"] - assert output["hits"][0]["score"] is None - assert output["hits"][0]["intent"] is None - - -# --- ③ 端点筛选 --------------------------------------------------------------- - - -async def test_embedding_endpoints_filters_out_non_embedding_capability( - monkeypatch: pytest.MonkeyPatch, -) -> None: - """不筛能力会把向量化请求打到聊天端点上（真库同时存在 chat / embedding 端点）。""" - rows = [ - FakeEndpoint("chat-primary", ["chat"]), - FakeEndpoint("embedding-primary", ["embedding"]), - FakeEndpoint("no-capability", []), - ] - # 生产契约：`_session_factory()` 返回的是**会话工厂本身**（`SessionFactory`）， - # 调用处再调一次拿到会话。所以替身也必须返回"可调用的工厂"，不是会话对象 —— - # 否则这个替身会掩盖"少调一层"的真实缺陷（该缺陷曾让 query_knowledge 100% 失败）。 - monkeypatch.setattr(knowledge_tool, "_session_factory", lambda: (lambda: FakeSession(rows))) - - endpoints = await knowledge_tool._embedding_endpoints() - - assert [endpoint.endpoint_code for endpoint in endpoints] == ["embedding-primary"]