feat: T21-5 知识库 Tool 接入对话——kb_tools(search_knowledge)挂统一注册表第三层; ToolSpec 新增 skip_access_check(公开知识无归属语义, advisor 无绑定客户可查, 意图层限定 customer/advisor); 意图关键词 customer/advisor 组尾挂 kb 词(Core 词优先, risk 不开放); tool_node 按 spec 白名单注入 query; summarize 溯源口径; 单测 13 例, 425 绿

2026-09-07 12:05:36 +08:00
parent e2a444453e
commit 1c51149912
4 changed files with 347 additions and 9 deletions
@@ -0,0 +1,229 @@
+"""T21-5 知识库 Tool 单测：注册表合并 / skip_access_check / 意图扩展 / 带参注入。
+
+覆盖：
+1. kb_tools.search_knowledge Tool 函数（rag_service mock，返回结构）；
+2. get_registered_tool 三层合并（core → risk → kb）；
+3. run_tool 对 skip_access_check 的处理（advisor 无绑定客户 → success 而非
+   AUTH_403 blocked；未知角色也放行——公开知识无归属语义，意图层已限范围）；
+4. match_intent：customer/advisor 命中 kb 词、Core 词优先、risk 不命中；
+5. tool_node 按 spec 白名单注入 query（agent_service 集成，FakeLLM 模式同既有测试）。
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from app.service import rag_service, tool_service
+from app.service.agent_service import tool_node
+from app.tool import kb_tools
+
+
+@pytest.fixture()
+def fake_rag(monkeypatch):
+    """mock rag_service.search_knowledge（不连 Ollama/Milvus）。"""
+    calls: dict = {}
+
+    def fake_search(query, *, product_id=None, doc_type=None, top_k=3):
+        calls.update({"query": query, "top_k": top_k})
+        return {
+            "query": query,
+            "results": [
+                {
+                    "id": "PROD-110022_1",
+                    "score": 0.82,
+                    "product_id": "PROD-110022",
+                    "product_name": "稳健债基 A",
+                    "doc_type": "rule",
+                    "risk_level": "R1",
+                    "source_doc_id": "KB-PROD-110022",
+                    "source_version": "2026.09",
+                    "effective_date": "2026-09-01",
+                    "chunk_text": "申购 T 日 15:00 前提交按当日净值确认，T+1 确认份额。",
+                    "chunk_no": 1,
+                }
+            ],
+            "source_refs": [
+                {
+                    "source_doc_id": "KB-PROD-110022",
+                    "source_version": "2026.09",
+                    "product_id": "PROD-110022",
+                    "product_name": "稳健债基 A",
+                }
+            ],
+        }
+
+    monkeypatch.setattr(rag_service, "search_knowledge", fake_search)
+    return calls
+
+
+CUSTOMER_ACTOR = {"actor_id": "CUST-1", "roles": ["customer"], "token_type": "dev"}
+ADVISOR_ACTOR = {"actor_id": "ADV-1", "roles": ["advisor"], "token_type": "dev"}
+
+
+class TestToolFunction:
+    def test_returns_hits_and_refs(self, fake_rag):
+        out = kb_tools.search_knowledge(query="申购规则")
+        assert out["hit_count"] == 1
+        assert out["results"][0]["source_doc_id"] == "KB-PROD-110022"
+        assert fake_rag["query"] == "申购规则"
+
+    def test_registry_shape(self):
+        spec = kb_tools.KB_TOOL_REGISTRY["search_knowledge"]
+        assert spec["requires_customer"] is False
+        assert spec["skip_access_check"] is True
+        assert "query" in spec["param_whitelist"]
+
+
+class TestRegistryMerge:
+    def test_three_layer_merge(self):
+        # core / risk / kb 各取一个代表
+        assert tool_service.get_registered_tool("query_holdings") is not None
+        assert tool_service.get_registered_tool("alert_query") is not None
+        assert tool_service.get_registered_tool("search_knowledge") is not None
+        assert tool_service.get_registered_tool("no_such_tool") is None
+
+
+class TestRunToolSkipAccess:
+    def test_advisor_without_customer_succeeds(self, fake_rag, monkeypatch):
+        # 核心：advisor 无绑定客户（customer_id=""）查知识库必须 success
+        # （无 skip_access_check 时会被 assert_tool_access 拒 AUTH_403_NOT_ASSIGNED）
+        monkeypatch.setattr(tool_service, "_core_ro", lambda: object())
+        monkeypatch.setattr(tool_service, "_risk_repo", lambda: object())
+        monkeypatch.setattr(
+            tool_service, "_session_repo", lambda: type("R", (), {"insert_tool_call": lambda *a, **k: None})()
+        )
+        record = tool_service.run_tool(
+            tool_name="search_knowledge",
+            agent_type="advisor",
+            actor=ADVISOR_ACTOR,
+            customer_id="",
+            tool_input={"query": "基金申购费率"},
+            session_id="S1",
+        )
+        assert record["status"] == "success"
+        assert record["data"]["hit_count"] == 1
+
+    def test_unknown_role_also_passes(self, fake_rag, monkeypatch):
+        # 公开知识：非四角色（如 platform 误调）也放行——意图层已限定开放范围
+        monkeypatch.setattr(tool_service, "_core_ro", lambda: object())
+        monkeypatch.setattr(tool_service, "_risk_repo", lambda: object())
+        monkeypatch.setattr(
+            tool_service, "_session_repo", lambda: type("R", (), {"insert_tool_call": lambda *a, **k: None})()
+        )
+        record = tool_service.run_tool(
+            tool_name="search_knowledge",
+            agent_type="customer",
+            actor={"actor_id": "X", "roles": ["unknown_role"], "token_type": "dev"},
+            customer_id="",
+            tool_input={"query": "费率"},
+            session_id="S1",
+        )
+        assert record["status"] == "success"
+
+    def test_bad_param_still_blocked(self, fake_rag, monkeypatch):
+        # 白名单外参数仍拒（skip_access_check 只豁免归属，不豁免入参校验）
+        monkeypatch.setattr(tool_service, "_core_ro", lambda: object())
+        monkeypatch.setattr(tool_service, "_risk_repo", lambda: object())
+        monkeypatch.setattr(
+            tool_service, "_session_repo", lambda: type("R", (), {"insert_tool_call": lambda *a, **k: None})()
+        )
+        record = tool_service.run_tool(
+            tool_name="search_knowledge",
+            agent_type="customer",
+            actor=CUSTOMER_ACTOR,
+            customer_id="CUST-1",
+            tool_input={"customer_id": "HACK"},
+            session_id="S1",
+        )
+        assert record["status"] == "blocked"
+        assert record["error_code"] == "TOOL_BAD_PARAM"
+
+
+class TestIntent:
+    def test_customer_kb_keywords(self):
+        assert tool_service.match_intent("customer", "稳健债基的申购费率是多少") == "search_knowledge"
+        assert tool_service.match_intent("customer", "赎回几天到账") == "search_knowledge"
+        assert tool_service.match_intent("advisor", "这只基金的定投起点") == "search_knowledge"
+
+    def test_core_keywords_take_priority(self):
+        # Core 词更特异（组序在前）：持仓类问句不被 kb 词抢走
+        assert tool_service.match_intent("customer", "看一下我的持仓") == "query_holdings"
+        assert tool_service.match_intent("customer", "我的风险测评结果") == "query_customer_profile"
+
+    def test_risk_branch_no_kb(self):
+        # 拍板：风控不开放知识检索
+        assert tool_service.match_intent("risk", "基金申购费率") is None
+        assert tool_service.match_intent("analyst", "基金申购费率") is None
+
+
+class TestToolNodeInjection:
+    """tool_node 按 spec 白名单注入 query（state→tool_input）。"""
+
+    def _state(self, agent_type="customer", msg="基金申购费率"):
+        return {
+            "agent_type": agent_type,
+            "history": [],
+            "user_message": msg,
+            "messages": [],
+            "reply": "",
+            "has_disclaimer": False,
+            "session_id": "S1",
+            "trace_id": "T1",
+            "actor": CUSTOMER_ACTOR,
+            "customer_id": "CUST-1",
+            "tool_results": [],
+        }
+
+    def test_query_injected_from_user_message(self, fake_rag, monkeypatch):
+        captured: dict = {}
+
+        def fake_run_tool(**kwargs):
+            captured.update(kwargs)
+            return {"tool_name": kwargs["tool_name"], "status": "success", "error_code": None,
+                    "data": {"hit_count": 0, "results": [], "source_refs": []}, "latency_ms": 1}
+
+        monkeypatch.setattr(tool_service, "run_tool", fake_run_tool)
+        tool_node(self._state())
+        assert captured["tool_input"] == {"query": "基金申购费率"}
+
+    def test_no_input_for_core_tools(self, fake_rag, monkeypatch):
+        # Core Tool（白名单无 query）不注入 tool_input，维持 T-04 口径
+        captured: dict = {}
+
+        def fake_run_tool(**kwargs):
+            captured.update(kwargs)
+            return {"tool_name": kwargs["tool_name"], "status": "success", "error_code": None,
+                    "data": {}, "latency_ms": 1}
+
+        monkeypatch.setattr(tool_service, "run_tool", fake_run_tool)
+        tool_node(self._state(msg="看一下我的持仓"))
+        assert captured["tool_input"] is None
+
+
+class TestSummarize:
+    def test_kb_summary_with_refs(self):
+        record = {
+            "tool_name": "search_knowledge",
+            "status": "success",
+            "error_code": None,
+            "data": {
+                "hit_count": 1,
+                "results": [{
+                    "product_name": "稳健债基 A", "doc_type": "rule", "score": 0.82,
+                    "chunk_text": "申购 T 日 15:00 前提交按当日净值确认，T+1 确认份额。本金部分。",
+                }],
+                "source_refs": [{"source_doc_id": "KB-PROD-110022", "source_version": "2026.09"}],
+            },
+        }
+        text = tool_service.summarize(record)
+        assert "命中 1 条" in text
+        assert "稳健债基 A" in text
+        assert "KB-PROD-110022@2026.09" in text
+
+    def test_kb_summary_empty(self):
+        record = {
+            "tool_name": "search_knowledge", "status": "success", "error_code": None,
+            "data": {"hit_count": 0, "results": [], "source_refs": []},
+        }
+        text = tool_service.summarize(record)
+        assert "未命中" in text and "不要编造" in text