feat: 迁移奶龙风控业务模块与演示文档

2026-09-10 21:03:44 +08:00
parent 6516ccb385
commit a94d5c754d
62 changed files with 10815 additions and 0 deletions
@@ -0,0 +1,184 @@
+"""奶龙风控智能助手业务对话端到端验收。"""
+
+from __future__ import annotations
+
+import asyncio
+import datetime as dt
+import json
+import uuid
+from pathlib import Path
+from typing import Any
+
+import httpx
+import jwt
+from sqlalchemy import text
+
+from app.core.config import get_settings
+from app.infrastructure.db import SessionFactory
+from app.main import create_app
+from app.worker.runtime import WorkerRuntime
+
+PRIVATE_KEY = Path("config/jwt/jwt-private.pem").read_text(encoding="utf-8")
+RISK_USER = "9002"
+AGENT_TYPE = "risk"
+
+CASES = (
+    {
+        "name": "风险概览",
+        "message": "查看当前风险概览",
+        "expected_tools": {"get_risk_overview"},
+    },
+    {
+        "name": "完整筛选回答",
+        "message": "当前低风险预警都是哪些客户的？他们买的都是什么产品？",
+        "expected_tools": {"search_risk_alerts"},
+    },
+    {
+        "name": "误报研判",
+        "message": "哪些预警可以按误报复核？",
+        "expected_tools": {"search_risk_alerts", "get_alert_evidence"},
+    },
+)
+
+
+def token(subject: str) -> str:
+    now = dt.datetime.now(dt.UTC)
+    return jwt.encode(
+        {
+            "sub": subject,
+            "iss": get_settings().jwt_issuer,
+            "aud": get_settings().jwt_audience,
+            "exp": now + dt.timedelta(minutes=30),
+            "nbf": now - dt.timedelta(seconds=5),
+            "jti": str(uuid.uuid4()),
+        },
+        PRIVATE_KEY,
+        algorithm="RS256",
+    )
+
+
+async def audit_rows(trace_id: str, run_id: str) -> list[dict[str, object]]:
+    async with SessionFactory() as session:
+        rows = (
+            await session.execute(
+                text(
+                    "SELECT action_type, detail FROM interaction_audit "
+                    "WHERE JSON_UNQUOTE(JSON_EXTRACT(detail,'$.trace_id'))=:trace "
+                    "OR JSON_UNQUOTE(JSON_EXTRACT(detail,'$.run_id'))=:run "
+                    "ORDER BY id"
+                ),
+                {"trace": trace_id, "run": run_id},
+            )
+        ).mappings().all()
+    result = []
+    for row in rows:
+        detail = row["detail"]
+        if isinstance(detail, str):
+            detail = json.loads(detail)
+        result.append({"action_type": row["action_type"], "detail": detail})
+    return result
+
+
+async def run_case(
+    client: httpx.AsyncClient,
+    auth: dict[str, str],
+    case: dict[str, Any],
+) -> dict[str, Any]:
+    accepted = await client.post(
+        "/api/v1/agent-runs",
+        headers=auth,
+        json={
+            "agent_type": AGENT_TYPE,
+            "message": case["message"],
+            "session_id": f"risk-business-e2e-{uuid.uuid4()}",
+            "idempotency_key": uuid.uuid4().hex,
+        },
+    )
+    if accepted.status_code != 202:
+        raise SystemExit(
+            f"{case['name']} 受理失败：{accepted.status_code} {accepted.text}"
+        )
+    accepted_data = accepted.json()["data"]
+    run_id = str(accepted_data["run_id"])
+    trace_id = str(accepted_data["trace_id"])
+
+    await WorkerRuntime().execute(run_id)
+    detail_response = await client.get(f"/api/v1/agent-runs/{run_id}", headers=auth)
+    if detail_response.status_code != 200:
+        raise SystemExit(
+            f"{case['name']} 结果查询失败："
+            f"{detail_response.status_code} {detail_response.text}"
+        )
+    detail = detail_response.json()["data"]
+    events = await client.get(
+        f"/api/v1/agent-runs/{run_id}/events",
+        headers={**auth, "Accept": "text/event-stream"},
+    )
+    audits = await audit_rows(trace_id, run_id)
+    calls = ((detail.get("result") or {}).get("tool_calls") or {}).get("calls") or []
+    return {
+        "name": case["name"],
+        "run_id": run_id,
+        "trace_id": trace_id,
+        "status": detail.get("status"),
+        "error_code": detail.get("error_code"),
+        "content": (detail.get("result") or {}).get("content") or "",
+        "tool_calls": calls,
+        "events": events,
+        "audits": audits,
+        "expected_tools": case["expected_tools"],
+    }
+
+
+def validate_result(result: dict[str, Any]) -> None:
+    name = str(result["name"])
+    if result["status"] != "succeeded":
+        raise SystemExit(
+            f"{name} 执行失败：status={result['status']} "
+            f"error_code={result['error_code']}"
+        )
+    if not str(result["content"]).strip():
+        raise SystemExit(f"{name} 返回内容为空")
+    tool_names = {
+        str(call.get("tool_name"))
+        for call in result["tool_calls"]
+        if isinstance(call, dict)
+    }
+    if not tool_names.intersection(result["expected_tools"]):
+        raise SystemExit(
+            f"{name} 未调用预期工具：expected={result['expected_tools']} "
+            f"actual={tool_names}"
+        )
+    events = result["events"]
+    if not str(events.headers.get("content-type") or "").startswith("text/event-stream"):
+        raise SystemExit(f"{name} SSE Content-Type 异常：{events.headers.get('content-type')}")
+    if "event: done" not in events.text:
+        raise SystemExit(f"{name} SSE 未返回 done")
+    actions = {str(row["action_type"]) for row in result["audits"]}
+    if "agent.tool_executed" not in actions:
+        raise SystemExit(f"{name} 未记录工具审计")
+    if "agent.run_completed" not in actions:
+        raise SystemExit(f"{name} 未记录运行完成审计")
+
+
+async def main() -> None:
+    app = create_app()
+    auth = {"Authorization": f"Bearer {token(RISK_USER)}"}
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=app),
+        base_url="http://test",
+        timeout=120,
+    ) as client:
+        for case in CASES:
+            result = await run_case(client, auth, case)
+            validate_result(result)
+            print(
+                f"{result['name']} PASSED "
+                f"run_id={result['run_id']} "
+                f"tools={[call.get('tool_name') for call in result['tool_calls']]} "
+                f"content={result['content'][:120]!r}"
+            )
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
@@ -0,0 +1,112 @@
+"""风控 Agent 的真实 Agent Run、工具调用和 SSE 验收。"""
+
+from __future__ import annotations
+
+import asyncio
+import datetime as dt
+import json
+import uuid
+from pathlib import Path
+
+import httpx
+import jwt
+from sqlalchemy import text
+
+from app.core.config import get_settings
+from app.infrastructure.db import SessionFactory
+from app.main import create_app
+from app.worker.runtime import WorkerRuntime
+
+PRIVATE_KEY = Path("config/jwt/jwt-private.pem").read_text(encoding="utf-8")
+RISK_USER = "9002"
+AGENT_TYPE = "risk"
+INTENT = "risk_overview"
+TOOL = "get_risk_overview"
+MESSAGE = "请查看当前风险概览"
+
+
+def token(subject: str) -> str:
+    now = dt.datetime.now(dt.UTC)
+    return jwt.encode(
+        {
+            "sub": subject,
+            "iss": get_settings().jwt_issuer,
+            "aud": get_settings().jwt_audience,
+            "exp": now + dt.timedelta(minutes=30),
+            "nbf": now - dt.timedelta(seconds=5),
+            "jti": str(uuid.uuid4()),
+        },
+        PRIVATE_KEY,
+        algorithm="RS256",
+    )
+
+
+async def audit_rows(trace_id: str) -> list[dict[str, object]]:
+    async with SessionFactory() as session:
+        rows = (
+            await session.execute(
+                text(
+                    "SELECT action_type, detail FROM interaction_audit "
+                    "WHERE JSON_UNQUOTE(JSON_EXTRACT(detail,'$.trace_id'))=:trace "
+                    "ORDER BY id"
+                ),
+                {"trace": trace_id},
+            )
+        ).mappings().all()
+    result = []
+    for row in rows:
+        detail = row["detail"]
+        if isinstance(detail, str):
+            detail = json.loads(detail)
+        result.append({"action_type": row["action_type"], "detail": detail})
+    return result
+
+
+async def main() -> None:
+    app = create_app()
+    auth = {"Authorization": f"Bearer {token(RISK_USER)}"}
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=app),
+        base_url="http://test",
+        timeout=120,
+    ) as client:
+        accepted = await client.post(
+            "/api/v1/agent-runs",
+            headers=auth,
+            json={
+                "agent_type": AGENT_TYPE,
+                "message": MESSAGE,
+                "session_id": f"risk-e2e-{uuid.uuid4()}",
+                "idempotency_key": uuid.uuid4().hex,
+            },
+        )
+        if accepted.status_code != 202:
+            raise SystemExit(f"受理失败：{accepted.status_code} {accepted.text}")
+        data = accepted.json()["data"]
+        run_id = str(data["run_id"])
+        trace_id = str(data["trace_id"])
+        await WorkerRuntime().execute(run_id)
+        detail = (await client.get(f"/api/v1/agent-runs/{run_id}", headers=auth)).json()["data"]
+        events = await client.get(
+            f"/api/v1/agent-runs/{run_id}/events",
+            headers={**auth, "Accept": "text/event-stream"},
+        )
+    audits = await audit_rows(trace_id)
+    tool_calls = (detail.get("result") or {}).get("tool_calls") or {}
+    calls = tool_calls.get("calls") or []
+    print(f"run_id={run_id}")
+    print(f"status={detail.get('status')} error_code={detail.get('error_code')}")
+    print(f"tool_calls={json.dumps(calls, ensure_ascii=False)}")
+    print(f"audit_actions={[row['action_type'] for row in audits]}")
+    print(f"sse_content_type={events.headers.get('content-type')}")
+    print(f"sse_has_done={'event: done' in events.text}")
+    if detail.get("status") != "succeeded":
+        raise SystemExit(f"风控 Agent 运行失败：{detail}")
+    if not any(call.get("tool_name") == TOOL for call in calls):
+        raise SystemExit("未记录 get_risk_overview 工具调用")
+    if not any(row["action_type"] == "agent.tool_executed" for row in audits):
+        raise SystemExit("未记录工具审计")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
@@ -0,0 +1,123 @@
+"""创建风控 Agent 本地验收所需的模型端点和发布配置。"""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import json
+from datetime import UTC, datetime
+
+from sqlalchemy import text
+
+from app.infrastructure.db import SessionFactory
+
+ADMIN_USER_ID = 9003
+ENDPOINT_CODE = "deepseek-flash"
+RELEASE_NO = "risk-agent-local-v1"
+TOOL_CONFIGS = (
+    ("risk:risk_overview", {"allowed_tools": ["get_risk_overview"]}),
+    ("risk:risk_search", {"allowed_tools": ["search_risk_alerts"]}),
+    ("risk:risk_evidence", {"allowed_tools": ["get_alert_evidence"]}),
+    ("risk:general", {"allowed_tools": []}),
+)
+INTENT_CONFIGS = (
+    ("risk_overview", "风险概览", ["请查看当前风险概览", "当前有多少高风险预警"], ["get_risk_overview"]),
+    ("risk_search", "风险查询", ["查询高风险预警", "查看规则 RW-007 命中的预警"], ["search_risk_alerts"]),
+    ("risk_evidence", "预警证据", ["查询预警编号 ALERT-001 的证据", "查看这条预警的证据链"], ["get_alert_evidence"]),
+    ("general", "通用风险咨询", ["奶龙风控智能助手能做什么", "说明你的功能边界"], []),
+)
+
+
+def checksum(value: dict) -> str:
+    payload = json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+    return hashlib.sha256(payload.encode("utf-8")).hexdigest()
+
+
+async def seed() -> None:
+    now = datetime.now(UTC).replace(tzinfo=None)
+    async with SessionFactory() as session:
+        await session.execute(text("""
+            INSERT INTO model_endpoint_config
+            (endpoint_code, provider, model_name, base_url, secret_ref, capabilities,
+             allowed_data_levels, context_window, timeout_ms, status, created_by,
+             reviewer_id, reviewed_at, created_at, updated_at)
+            VALUES
+            (:endpoint_code, 'deepseek', 'deepseek-chat', 'https://api.deepseek.com',
+             'env:DEEPSEEK_API_KEY', :capabilities, :data_levels, 64000, 30000,
+             'active', :admin_id, :admin_id, :now, :now, :now)
+            ON DUPLICATE KEY UPDATE
+              provider=VALUES(provider), model_name=VALUES(model_name),
+              base_url=VALUES(base_url), secret_ref=VALUES(secret_ref),
+              capabilities=VALUES(capabilities), allowed_data_levels=VALUES(allowed_data_levels),
+              context_window=VALUES(context_window), timeout_ms=VALUES(timeout_ms),
+              status='active', reviewer_id=:admin_id, reviewed_at=:now, updated_at=:now
+        """), {
+            "endpoint_code": ENDPOINT_CODE,
+            "capabilities": json.dumps(["chat", "intent_classification", "risk_answer"]),
+            "data_levels": json.dumps(["internal"]),
+            "admin_id": ADMIN_USER_ID,
+            "now": now,
+        })
+
+        release_id = await session.scalar(
+            text("SELECT id FROM config_release WHERE status='active' LIMIT 1")
+        )
+        if release_id is None:
+            result = await session.execute(text("""
+                INSERT INTO config_release
+                (release_no, title, change_summary, status, created_by, reviewer_id,
+                 reviewed_at, activated_at, created_at, updated_at)
+                VALUES (:release_no, '奶龙风控智能助手本地配置',
+                        '发布 risk Agent 工具白名单和意图配置', 'active',
+                        :admin_id, :admin_id, :now, :now, :now, :now)
+            """), {"release_no": RELEASE_NO, "admin_id": ADMIN_USER_ID, "now": now})
+            release_id = int(result.lastrowid)
+
+        for config_key, value in TOOL_CONFIGS:
+            await session.execute(text("""
+                INSERT INTO platform_config_item
+                (release_id, namespace, config_key, value_json, schema_version, checksum, created_at)
+                VALUES (:release_id, 'agent_tools', :config_key, :value_json, '1', :checksum, :now)
+                ON DUPLICATE KEY UPDATE
+                  value_json=VALUES(value_json), schema_version=VALUES(schema_version),
+                  checksum=VALUES(checksum)
+            """), {
+                "release_id": release_id,
+                "config_key": config_key,
+                "value_json": json.dumps(value, ensure_ascii=False),
+                "checksum": checksum(value),
+                "now": now,
+            })
+
+        for intent_code, intent_name, examples, allowed_tools in INTENT_CONFIGS:
+            await session.execute(text("""
+                INSERT INTO agent_intent_config
+                (agent_type, intent_code, intent_name, description, examples,
+                 classifier_instruction, confidence_threshold, max_clarification_rounds,
+                 transfer_on_failure, allowed_tools, priority, version, status,
+                 effective_at, created_by, reviewer_id, reviewed_at, created_at, updated_at)
+                VALUES
+                ('risk', :intent_code, :intent_name, :description, :examples,
+                 :instruction, 0.6500, 2, 1, :allowed_tools, 100, 1, 'active',
+                 :now, :admin_id, :admin_id, :now, :now, :now)
+                ON DUPLICATE KEY UPDATE
+                  intent_name=VALUES(intent_name), description=VALUES(description),
+                  examples=VALUES(examples), classifier_instruction=VALUES(classifier_instruction),
+                  allowed_tools=VALUES(allowed_tools), status='active',
+                  effective_at=:now, reviewer_id=:admin_id, reviewed_at=:now, updated_at=:now
+            """), {
+                "intent_code": intent_code,
+                "intent_name": intent_name,
+                "description": f"奶龙风控智能助手：{intent_name}",
+                "examples": json.dumps(examples, ensure_ascii=False),
+                "instruction": "只处理风控只读查询、分析和边界说明，不执行人工处置。",
+                "allowed_tools": json.dumps(allowed_tools, ensure_ascii=False),
+                "admin_id": ADMIN_USER_ID,
+                "now": now,
+            })
+        await session.commit()
+    print(f"risk_agent_config_ready release_id={release_id} endpoint={ENDPOINT_CODE}")
+
+
+if __name__ == "__main__":
+    asyncio.run(seed())