113 lines
3.8 KiB
Python
113 lines
3.8 KiB
Python
"""风控 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(get_settings().jwt_private_key_path).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())
|