lzf_0626
f09ea9e988
NL 线（含其并入的袁聪场外/推广域）。唯一冲突是 .gitignore —— 双方都往同一区域加了 .workdir/，取对方版本（他的更完整，含 .tmp/ 与说明），顺带修掉我之前用 Add-Content -Encoding utf8 造成的编码混合（read 工具当时报 invalid UTF-8）。 合并后修的问题 —— 都不是"改别人业务逻辑"，是让门禁能绿： 1. 缺运行依赖 python-docx。document_parser.py 解析 .docx 用它，但 requirements.txt 与 pyproject.toml 都没声明 —— 别人环境跑知识入库会直接 ModuleNotFoundError: No module named 'docx'。已补声明。 2. ruff 7 项：其中 tests/conftest.py 的 F821 Undefined name 'Path'（他的 tmp_path 修复 写了字符串注解 "Path" 却漏 import，运行时不求值所以没炸，但 mypy/ruff 会抓）、 tools/publish_customer_service_config.py 的 F841 inherited_keys 死变量（他改同 key 覆盖、换成 inherited_only 后忘删旧的）、3 处 E501，另 2 项 ruff --fix 自动修复。 3. 合规基线种子未跑：integration 的 test_compliance_seed_mysql 4 个用例要求 agent_negative_word 有 7 条 active 且已复核、agent_reply_template 覆盖 6 场景。 跑 tools/seed_compliance_baseline.py（11 条 active 规则 / 6 个场景模板）后 80 passed。 验证：ruff 干净 / mypy 180 文件 0 错 / unit+contract 1140 passed / integration 80 passed / 表数 68（alembic 已在 20260911_merge_risk_heads）。 唯一失败 tests/unit/repository/test_fund_readonly_contract.py 是双方一致的既有缺陷： 它断言 Base.metadata 里的 fin_* 表集合，而实测为空集 —— 即该测试依赖别的测试先导入模型的 副作用，单独跑必失败。NL 方也明确"不修不报"，此处照办，仅记录。
457 lines
15 KiB
Python
457 lines
15 KiB
Python
from typing import Any, cast
|
|
|
|
import pytest
|
|
|
|
from app.core.contracts import (
|
|
AgentDefinition,
|
|
AgentRequest,
|
|
IntentResult,
|
|
RequestContext,
|
|
ResolvedAgentConfig,
|
|
SourceReference,
|
|
ToolCallRecord,
|
|
)
|
|
from app.core.errors import ForbiddenAgentError
|
|
from app.service.agent.bootstrap import get_agent_factory
|
|
from app.service.agent.factory import AgentFactory
|
|
from app.service.agent.governance import review_output
|
|
from app.service.agent.implementations import risk_agent as public_risk_agent
|
|
from app.service.agent.implementations.risk_agent import (
|
|
EVIDENCE_TOOL,
|
|
INTENT_EVIDENCE,
|
|
INTENT_GENERAL,
|
|
INTENT_OVERVIEW,
|
|
INTENT_SEARCH,
|
|
OVERVIEW_TOOL,
|
|
SEARCH_TOOL,
|
|
RiskAgent,
|
|
)
|
|
from app.service.tool_executor import ToolExecution
|
|
|
|
|
|
class StubGovernance:
|
|
def __init__(self, tools_by_intent: dict[str, tuple[str, ...]] | None = None):
|
|
self.tools_by_intent = tools_by_intent or {}
|
|
|
|
async def resolve(self, definition: AgentDefinition, context: RequestContext):
|
|
del definition, context
|
|
return ResolvedAgentConfig(
|
|
config_version="contract",
|
|
prompt_version="contract",
|
|
model_endpoint="",
|
|
allowed_tools_by_intent=dict(self.tools_by_intent),
|
|
)
|
|
|
|
async def recall(self, context: RequestContext):
|
|
del context
|
|
return ()
|
|
|
|
async def review(self, result, context, config, memories, *, agent_type: str = ""):
|
|
return review_output(result, context, config, memories)
|
|
|
|
|
|
class StubToolExecutor:
|
|
def __init__(self, output: Any):
|
|
self.output = output
|
|
self.calls: list[tuple[str, str, dict[str, Any]]] = []
|
|
|
|
async def execute(
|
|
self,
|
|
*,
|
|
name: str,
|
|
arguments: dict[str, Any],
|
|
intent: str,
|
|
configured_tools: dict[str, tuple[str, ...]],
|
|
context: RequestContext,
|
|
) -> ToolExecution:
|
|
if name not in configured_tools.get(intent, ()):
|
|
raise ForbiddenAgentError("工具不在当前意图白名单")
|
|
self.calls.append((name, intent, arguments))
|
|
output = (
|
|
self.output.get(name, self.output)
|
|
if isinstance(self.output, dict)
|
|
else self.output
|
|
)
|
|
return ToolExecution(
|
|
output=output,
|
|
record=ToolCallRecord(tool_name=name, status="succeeded"),
|
|
references=(
|
|
SourceReference(
|
|
source_type="tool",
|
|
source_id=f"{context.trace_id}:{name}",
|
|
title=name,
|
|
),
|
|
),
|
|
)
|
|
|
|
|
|
class StubRiskModelClient:
|
|
def __init__(self, messages: list[dict[str, Any]]):
|
|
self.messages = list(messages)
|
|
self.calls: list[tuple[list[dict[str, Any]], list[dict[str, Any]]]] = []
|
|
|
|
async def chat(
|
|
self,
|
|
messages: list[dict[str, Any]],
|
|
*,
|
|
tools: list[dict[str, Any]],
|
|
) -> dict[str, Any]:
|
|
self.calls.append((messages, tools))
|
|
if not self.messages:
|
|
raise RuntimeError("测试模型消息已耗尽")
|
|
return self.messages.pop(0)
|
|
|
|
|
|
class FailingRiskModelClient:
|
|
async def chat(
|
|
self,
|
|
_messages: list[dict[str, Any]],
|
|
*,
|
|
tools: list[dict[str, Any]],
|
|
) -> dict[str, Any]:
|
|
del tools
|
|
raise RuntimeError("测试模型不可用")
|
|
|
|
|
|
def context() -> RequestContext:
|
|
return RequestContext(
|
|
user_id="990000002",
|
|
trace_id="risk-agent-trace",
|
|
roles=("risk_operator",),
|
|
permissions=("agent:run", "risk:alert:read"),
|
|
)
|
|
|
|
|
|
def request(message: str, intent: str) -> tuple[AgentRequest, IntentResult]:
|
|
return (
|
|
AgentRequest(
|
|
agent_type="risk",
|
|
message=message,
|
|
session_id="risk-session",
|
|
idempotency_key="risk-contract-key-0001",
|
|
),
|
|
IntentResult(intent=intent, confidence=0.95),
|
|
)
|
|
|
|
|
|
def build_factory(
|
|
tools_by_intent: dict[str, tuple[str, ...]],
|
|
output: Any,
|
|
*,
|
|
model_client: Any | None = None,
|
|
) -> tuple[AgentFactory, StubToolExecutor]:
|
|
executor = StubToolExecutor(output)
|
|
factory = AgentFactory(
|
|
cast(Any, StubGovernance(tools_by_intent)),
|
|
tool_executor=cast(Any, executor),
|
|
)
|
|
factory.register(
|
|
RiskAgent.definition,
|
|
lambda _context: RiskAgent(
|
|
RiskAgent.definition,
|
|
model_client=model_client or FailingRiskModelClient(),
|
|
),
|
|
)
|
|
return factory, executor
|
|
|
|
|
|
def test_risk_agent_and_tools_are_registered() -> None:
|
|
factory = get_agent_factory()
|
|
assert factory.definition("risk") == RiskAgent.definition
|
|
for tool_name in (OVERVIEW_TOOL, SEARCH_TOOL, EVIDENCE_TOOL):
|
|
tool = factory._tool_executor.registry.get(tool_name)
|
|
assert tool.read_only is True
|
|
assert tool.required_permission == "risk:alert:read"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_risk_overview_intent_calls_only_read_tool() -> None:
|
|
factory, executor = build_factory(
|
|
{INTENT_OVERVIEW: (OVERVIEW_TOOL,)},
|
|
{"total": 2, "levels": {"高": 1, "中": 1, "低": 0}, "pending": 2, "overdue": 0},
|
|
)
|
|
ctx = context()
|
|
agent = factory.create("risk", ctx)
|
|
agent._classified_intent = IntentResult(intent=INTENT_OVERVIEW, confidence=0.95)
|
|
req, _ = request("请查看风险概览", INTENT_OVERVIEW)
|
|
|
|
events = [event async for event in agent.execute(req, ctx, "run-risk-overview")]
|
|
|
|
assert executor.calls == [(OVERVIEW_TOOL, INTENT_OVERVIEW, {})]
|
|
assert "高风险 1 条" in events[-1].payload["result"]["result"]["text"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_risk_search_extracts_supported_filters() -> None:
|
|
factory, executor = build_factory(
|
|
{INTENT_SEARCH: (SEARCH_TOOL,)},
|
|
[
|
|
{
|
|
"alert_no": "ALERT-001",
|
|
"risk_level": "高",
|
|
"alert_type": "适当性错配",
|
|
"customer_no": "CUST-001",
|
|
"evidence_summary": "证据摘要",
|
|
}
|
|
],
|
|
)
|
|
ctx = context()
|
|
agent = factory.create("risk", ctx)
|
|
agent._classified_intent = IntentResult(intent=INTENT_SEARCH, confidence=0.95)
|
|
req, _ = request("查询高风险预警，规则 RW-007，客户编号 CUST-001", INTENT_SEARCH)
|
|
|
|
events = [event async for event in agent.execute(req, ctx, "run-risk-search")]
|
|
|
|
assert executor.calls[0][2] == {
|
|
"risk_level": "高",
|
|
"rule_code": "RW-007",
|
|
"customer_no": "CUST-001",
|
|
}
|
|
assert "ALERT-001" in events[-1].payload["result"]["result"]["text"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_risk_evidence_requires_alert_number() -> None:
|
|
factory, executor = build_factory(
|
|
{INTENT_EVIDENCE: (EVIDENCE_TOOL,)},
|
|
{
|
|
"alert": {
|
|
"alert_no": "ALERT-001",
|
|
"risk_level": "高",
|
|
"alert_type": "适当性错配",
|
|
"rule_codes": ["RW-007"],
|
|
"evidence_summary": "证据摘要",
|
|
},
|
|
"customer": {"customer_no": "CUST-001", "name": "张*"},
|
|
},
|
|
)
|
|
ctx = context()
|
|
agent = factory.create("risk", ctx)
|
|
agent._classified_intent = IntentResult(intent=INTENT_EVIDENCE, confidence=0.95)
|
|
req, _ = request("查询预警编号 ALERT-001 的证据", INTENT_EVIDENCE)
|
|
|
|
events = [event async for event in agent.execute(req, ctx, "run-risk-evidence")]
|
|
|
|
assert executor.calls[0][2] == {"alert_no": "ALERT-001"}
|
|
assert "预警编号：ALERT-001" in events[-1].payload["result"]["result"]["text"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_release_configuration_fails_closed() -> None:
|
|
factory, executor = build_factory({}, {})
|
|
ctx = context()
|
|
agent = factory.create("risk", ctx)
|
|
agent._classified_intent = IntentResult(intent=INTENT_OVERVIEW, confidence=0.95)
|
|
req, _ = request("查看概览", INTENT_OVERVIEW)
|
|
|
|
with pytest.raises(ForbiddenAgentError, match="白名单"):
|
|
async for _ in agent.execute(req, ctx, "run-risk-denied"):
|
|
pass
|
|
assert executor.calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_analysis_preset_routes_to_structured_analysis(monkeypatch) -> None:
|
|
async def fake_generate(_context, alert_no, output_type):
|
|
assert alert_no == "ALERT-001"
|
|
assert output_type == "工单摘要"
|
|
return {"type": output_type, "content": "工单标题：测试工单", "source": "模板降级输出"}
|
|
|
|
monkeypatch.setattr(
|
|
public_risk_agent.RiskAnalysisService,
|
|
"generate_for_context",
|
|
fake_generate,
|
|
)
|
|
factory, executor = build_factory({}, {})
|
|
ctx = context()
|
|
agent = factory.create("risk", ctx)
|
|
req, _ = request("当前预警编号：ALERT-001。请生成工单摘要", INTENT_GENERAL)
|
|
|
|
events = [event async for event in agent.execute(req, ctx, "run-risk-analysis")]
|
|
|
|
assert executor.calls == []
|
|
assert events[-1].payload["result"]["result"]["text"] == "工单标题：测试工单"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_autonomous_tool_loop_supports_multiple_rounds() -> None:
|
|
model_client = StubRiskModelClient([
|
|
{
|
|
"content": "",
|
|
"tool_calls": [{
|
|
"id": "call-search",
|
|
"type": "function",
|
|
"function": {
|
|
"name": SEARCH_TOOL,
|
|
"arguments": '{"risk_level":"高"}',
|
|
},
|
|
}],
|
|
},
|
|
{
|
|
"content": "",
|
|
"tool_calls": [{
|
|
"id": "call-evidence",
|
|
"type": "function",
|
|
"function": {
|
|
"name": EVIDENCE_TOOL,
|
|
"arguments": '{"alert_no":"ALERT-001"}',
|
|
},
|
|
}],
|
|
},
|
|
{
|
|
"content": "ALERT-001 需要人工复核，现有证据支持继续调查。",
|
|
},
|
|
])
|
|
factory, executor = build_factory(
|
|
{
|
|
INTENT_SEARCH: (SEARCH_TOOL,),
|
|
INTENT_EVIDENCE: (EVIDENCE_TOOL,),
|
|
},
|
|
{
|
|
SEARCH_TOOL: [{
|
|
"alert_no": "ALERT-001",
|
|
"risk_level": "高",
|
|
"alert_type": "适当性错配",
|
|
"customer_no": "CUST-001",
|
|
"evidence_summary": "证据摘要",
|
|
}],
|
|
EVIDENCE_TOOL: {
|
|
"alert": {
|
|
"id": 71,
|
|
"alert_no": "ALERT-001",
|
|
"risk_level": "高",
|
|
"alert_type": "适当性错配",
|
|
"rule_codes": ["RW-007"],
|
|
"evidence_summary": "证据摘要",
|
|
},
|
|
"customer": {"customer_no": "CUST-001", "name": "张*"},
|
|
},
|
|
},
|
|
model_client=model_client,
|
|
)
|
|
ctx = context()
|
|
agent = factory.create("risk", ctx)
|
|
agent._classified_intent = IntentResult(intent=INTENT_GENERAL, confidence=0.95)
|
|
req, _ = request("哪个预警更需要人工复核", INTENT_GENERAL)
|
|
|
|
events = [event async for event in agent.execute(req, ctx, "run-risk-autonomous")]
|
|
|
|
assert [call[0] for call in executor.calls] == [SEARCH_TOOL, EVIDENCE_TOOL]
|
|
assert executor.calls[1][2] == {"alert_no": "ALERT-001"}
|
|
assert events[-1].payload["result"]["result"]["text"].startswith("ALERT-001")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invalid_protocol_marker_falls_back_without_leaking() -> None:
|
|
model_client = StubRiskModelClient([
|
|
{"content": "<tool_calls><invoke name=\"search_risk_alerts\"></invoke></tool_calls>"},
|
|
])
|
|
factory, executor = build_factory(
|
|
{INTENT_SEARCH: (SEARCH_TOOL,)},
|
|
[],
|
|
model_client=model_client,
|
|
)
|
|
ctx = context()
|
|
agent = factory.create("risk", ctx)
|
|
agent._classified_intent = IntentResult(intent=INTENT_GENERAL, confidence=0.95)
|
|
req, _ = request("查看预警", INTENT_GENERAL)
|
|
|
|
events = [event async for event in agent.execute(req, ctx, "run-risk-protocol")]
|
|
|
|
text = events[-1].payload["result"]["result"]["text"]
|
|
assert "<tool_calls>" not in text
|
|
assert executor.calls == [(SEARCH_TOOL, INTENT_SEARCH, {})]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disposition_question_uses_read_only_fallback_without_model() -> None:
|
|
factory, executor = build_factory(
|
|
{INTENT_SEARCH: (SEARCH_TOOL,)},
|
|
[{
|
|
"alert_no": "ALERT-018",
|
|
"risk_level": "低",
|
|
"alert_type": "频繁交易初筛",
|
|
"customer_no": "CUST-018",
|
|
"rule_codes": ["RW-018"],
|
|
"evidence_summary": "交易来自有效定投工单",
|
|
}],
|
|
)
|
|
ctx = context()
|
|
agent = factory.create("risk", ctx)
|
|
agent._classified_intent = IntentResult(intent=INTENT_GENERAL, confidence=0.95)
|
|
req, _ = request("哪些预警可以按误报复核", INTENT_GENERAL)
|
|
|
|
events = [event async for event in agent.execute(req, ctx, "run-risk-disposition")]
|
|
|
|
text = events[-1].payload["result"]["result"]["text"]
|
|
assert executor.calls[0][0] == SEARCH_TOOL
|
|
assert "ALERT-018" in text
|
|
assert "不构成最终处置结论" in text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_general_list_question_uses_complete_search_summary() -> None:
|
|
factory, executor = build_factory(
|
|
{INTENT_SEARCH: (SEARCH_TOOL,)},
|
|
{
|
|
"total": 2,
|
|
"filters": {"risk_level": "低"},
|
|
"summary": {
|
|
"customer_groups": [
|
|
{
|
|
"customer_no": "CUST-001",
|
|
"customer_name": "张*",
|
|
"alert_count": 1,
|
|
"risk_levels": ["低"],
|
|
},
|
|
{
|
|
"customer_no": "CUST-002",
|
|
"customer_name": "李*",
|
|
"alert_count": 1,
|
|
"risk_levels": ["低"],
|
|
},
|
|
],
|
|
"product_groups": [
|
|
{
|
|
"product_code": "P-001",
|
|
"product_name": "稳健一号",
|
|
"alert_count": 2,
|
|
"risk_levels": ["低"],
|
|
},
|
|
],
|
|
"disposition_counts": {"可考虑放行": 2},
|
|
"complete": True,
|
|
},
|
|
"items": [],
|
|
},
|
|
)
|
|
ctx = context()
|
|
agent = factory.create("risk", ctx)
|
|
agent._classified_intent = IntentResult(intent=INTENT_GENERAL, confidence=0.95)
|
|
req, _ = request("当前低风险预警都是哪些客户的？他们买的什么产品？", INTENT_GENERAL)
|
|
|
|
events = [event async for event in agent.execute(req, ctx, "run-risk-complete-list")]
|
|
|
|
text = events[-1].payload["result"]["result"]["text"]
|
|
assert executor.calls[0][0] == SEARCH_TOOL
|
|
assert executor.calls[0][2]["risk_level"] == "低"
|
|
assert "CUST-001" in text
|
|
assert "CUST-002" in text
|
|
assert "稳健一号" in text
|
|
assert "全部命中记录" in text
|
|
|
|
|
|
def test_agent_prompt_requires_truncation_disclosure() -> None:
|
|
prompt = public_risk_agent._agent_system_prompt("查看当前预警")
|
|
|
|
assert "data_truncated=true" in prompt
|
|
assert "证据不完整" in prompt
|
|
|
|
|
|
def test_agent_prompt_requires_business_field_names() -> None:
|
|
prompt = public_risk_agent._agent_system_prompt("查看当前预警")
|
|
|
|
assert "field_meanings" in prompt
|
|
assert "customer_no" in prompt
|
|
assert "ack_status" in prompt
|