diff --git a/app/service/model_gateway.py b/app/service/model_gateway.py index 57a1837..79e0ee1 100644 --- a/app/service/model_gateway.py +++ b/app/service/model_gateway.py @@ -1,3 +1,4 @@ +import logging import os from collections.abc import Mapping from dataclasses import dataclass @@ -167,11 +168,22 @@ class DatabaseModelGateway: # 任何端点的能力名(deepseek 声明的是 text_generation/json_output/intent_classification), # 它需要的是「能生成结构化文本」的端点。若按同名筛选会得到空集,把记忆抽取打成 # 失败关闭——这是修复端点筛选时最容易引入的回归。 +logger = logging.getLogger(__name__) + TASK_CAPABILITY: dict[str, str] = { "embedding": "embedding", "intent_classification": "intent_classification", "memory_extraction": "text_generation", "text_generation": "text_generation", + # 风控的几处 task_type:它们要的都是"能生成文本"的端点,与 memory_extraction 同理。 + # 不登记就会落到下面"未映射 → 返回全部端点"的分支,而能否选到文本端点就取决于 + # `model_endpoint_config` 的**行顺序**——实测风控能跑通,仅仅因为 deepseek-flash(id=3) + # 恰好排在 qwen-embedding(id=5) 前面。这种"靠数据顺序才对"的隐式依赖必须消掉。 + "risk_agent_chat": "text_generation", + "risk_analysis": "text_generation", + "risk_script": "text_generation", + "risk_summary": "text_generation", + "daily_report_suggestion": "text_generation", } @@ -195,6 +207,14 @@ class DatabaseModelEndpointResolver: ))) capability = TASK_CAPABILITY.get(task_type) if capability is None: + # 未登记的 task_type 仍退回全部端点(保持原有保守策略:让故障表现为调用失败、 + # 而不是解析为空),但必须留下痕迹。静默退回会让"选端点靠表行顺序"这类问题 + # 在下游以"偶发调用失败"的形式冒出来,极难定位。 + logger.warning( + "模型端点筛选:task_type=%r 未登记能力映射,退回全部 active 端点;" + "请在 TASK_CAPABILITY 中补上它对应的能力", + task_type, + ) return endpoints matched = [ endpoint for endpoint in endpoints diff --git a/app/service/tool_executor.py b/app/service/tool_executor.py index 4bdfa79..3d2623f 100644 --- a/app/service/tool_executor.py +++ b/app/service/tool_executor.py @@ -65,16 +65,39 @@ class ToolExecutor: configured_tools: dict[str, tuple[str, ...]], context: RequestContext, ) -> ToolExecution: definition = self.registry.get(name) - allowed = configured_tools.get(intent, ()) - reason = None - if name not in allowed: + # 把三种"用不了"分开,并且**审计写详细、异常给通用**: + # + # 原先前两种情况共用一句"工具不在当前意图白名单",运维无法判断该去补发布配置、 + # 还是该改白名单内容——本项目已经因此踩坑两次(客服、风控的意图码都要求三处对齐, + # 而缺配置时是静默失败关闭)。权限与角色两处也只说"缺少工具权限",不说是哪一个。 + # + # 细节只进审计:异常 message 会随 API 响应返回给调用方, + # 白名单内容、权限码、角色集属于内部配置,不该出现在客户可见的响应里。 + reason = "" + detail = "" + if intent not in configured_tools: + reason = "该意图未配置工具白名单" + detail = ( + f"意图 {intent!r} 在发布版本里没有任何工具白名单" + f"(agent_tools / :{intent}),工具失败关闭" + ) + elif name not in configured_tools[intent]: reason = "工具不在当前意图白名单" + detail = ( + f"工具 {name!r} 不在意图 {intent!r} 的白名单 " + f"{list(configured_tools[intent])} 内" + ) elif definition.required_permission not in context.permissions: reason = "缺少工具权限" + detail = f"缺少权限 {definition.required_permission!r}" elif not set(definition.allowed_roles).intersection(context.roles): reason = "角色不能使用工具" + detail = ( + f"角色 {list(context.roles)} 与工具允许的角色 " + f"{list(definition.allowed_roles)} 无交集" + ) if reason: - await self._audit(name, intent, context, "denied", reason) + await self._audit(name, intent, context, "denied", detail) raise ForbiddenAgentError(reason) try: validated = definition.input_model.model_validate(arguments) diff --git a/tests/unit/service/test_tool_executor_denials.py b/tests/unit/service/test_tool_executor_denials.py new file mode 100644 index 0000000..00a54db --- /dev/null +++ b/tests/unit/service/test_tool_executor_denials.py @@ -0,0 +1,128 @@ +"""ToolExecutor 的拒绝路径:三种"用不了"必须能区分,且细节不得外泄。 + +**背景**:意图码要求三处对齐(`AgentDefinition.supported_intents` / `agent_intent_config` / +发布版 `agent_tools` 白名单),而缺配置时是**失败关闭**。原先"意图压根没发布白名单"与 +"白名单里没这个工具"共用一句「工具不在当前意图白名单」,运维无法判断该去补发布配置、 +还是该改白名单内容——本项目已因此踩坑两次(客服、风控)。 + +这个文件锁两件事: + +1. 三种拒绝的 message **互不相同**,各自指向不同的处置动作; +2. 白名单内容、权限码、角色集这些**内部配置只进审计**,不进异常 message —— + 后者会随 API 响应返回给调用方。 +""" + +from typing import Any + +import pytest +from pydantic import BaseModel + +from app.core.contracts import RequestContext +from app.core.errors import ForbiddenAgentError +from app.service.tool_executor import ToolDefinition, ToolExecutor, ToolRegistry + +PERMISSION = "knowledge:reference:read" + + +class _Args(BaseModel): + model_config = {"extra": "forbid"} + + query: str = "" + + +async def _handler(arguments: BaseModel, context: RequestContext) -> Any: + return {"ok": True} + + +def _executor() -> tuple[ToolExecutor, list[tuple[Any, ...]]]: + registry = ToolRegistry() + registry.register(ToolDefinition( + name="search_knowledge", + input_model=_Args, + handler=_handler, + required_permission=PERMISSION, + allowed_roles=("customer", "advisor"), + )) + executor = ToolExecutor(registry) + audits: list[tuple[Any, ...]] = [] + + async def fake_audit(name: str, intent: str, context: RequestContext, + status: str, reason: str) -> None: + audits.append((name, intent, status, reason)) + + executor._audit = fake_audit # type: ignore[method-assign] + return executor, audits + + +def _context(*, permissions: tuple[str, ...] = (), roles: tuple[str, ...] = ("customer",) + ) -> RequestContext: + return RequestContext(user_id="1", trace_id="t", permissions=permissions, roles=roles) + + +async def _reject( + executor: ToolExecutor, context: RequestContext, *, + intent: str, configured: dict[str, tuple[str, ...]], +) -> str: + with pytest.raises(ForbiddenAgentError) as excinfo: + await executor.execute( + name="search_knowledge", arguments={"query": "x"}, intent=intent, + configured_tools=configured, context=context, + ) + return str(excinfo.value) + + +@pytest.mark.asyncio +async def test_missing_intent_config_differs_from_missing_tool() -> None: + """「意图没发布白名单」与「白名单里没这个工具」必须是两句不同的话。""" + executor, _ = _executor() + context = _context(permissions=(PERMISSION,)) + + no_config = await _reject(executor, context, intent="faq", configured={}) + not_listed = await _reject( + executor, context, intent="faq", configured={"faq": ("other_tool",)} + ) + + assert no_config != not_listed + assert "未配置" in no_config + assert "白名单" in not_listed + + +@pytest.mark.asyncio +async def test_permission_and_role_failures_have_their_own_messages() -> None: + executor, _ = _executor() + configured = {"faq": ("search_knowledge",)} + + no_permission = await _reject(executor, _context(), intent="faq", configured=configured) + wrong_role = await _reject( + executor, _context(permissions=(PERMISSION,), roles=("risk_operator",)), + intent="faq", configured=configured, + ) + + assert "权限" in no_permission + assert "角色" in wrong_role + assert no_permission != wrong_role + + +@pytest.mark.asyncio +async def test_internal_detail_goes_to_audit_not_to_the_message() -> None: + """权限码这类内部配置只进审计,不进会返回给调用方的异常 message。""" + executor, audits = _executor() + + message = await _reject(executor, _context(), intent="faq", + configured={"faq": ("search_knowledge",)}) + + assert PERMISSION not in message + assert any(PERMISSION in str(entry[3]) for entry in audits), "审计里应当有具体缺哪个权限" + assert audits[0][2] == "denied" + + +@pytest.mark.asyncio +async def test_audit_names_what_to_fix_when_config_is_absent() -> None: + """缺配置时审计要指出该去补哪一类配置,而不是只说"不在白名单"。""" + executor, audits = _executor() + + await _reject(executor, _context(permissions=(PERMISSION,)), intent="faq", configured={}) + + detail = str(audits[0][3]) + assert "没有任何工具白名单" in detail + assert "faq" in detail