契约(OpenAI 兼容 chunk,AI SDK / fetch-event-source 可直接接): 首帧 meta(session_id/trace_id/disclaimer)→ delta → finish_reason=stop → [DONE]。 前端侧:免责声明由首帧下发、前端常驻渲染;落库文本仍按原口径拼尾部。 设计拍板: 1. 新增独立端点,原 POST /api/chat 契约与既有测试零影响; 2. 鉴权/限流/输入防护全部在返回 StreamingResponse 之前完成(SSE 一开就改不了 状态码),401/403/404/409/429/400 仍是普通 JSON; 3. 整轮一次性落库:中途异常/断连不落消息(Tool 留痕已落可审计),不产生 半截内容污染历史窗口。 实现: - agent_service:抽 needs_disclaimer/_degraded_reply/_base_state;新增 stream_chat 生成器(Tool 节点同步跑完再推 LLM 文本,无 key 走降级整块); - api/chat:抽 _guard_request(准入→空白→限流→注入拦截)与 _prepare_turn (归属+会话解析/创建),同步与流式共用,守卫零偏差; - session_repository:新增 insert_turn——user+assistant 同事务落库 + 事务内 取 seq,修掉评审 P0(落库失败会半截落且前端收不到 [DONE] 挂起)与 P1 (两次写非原子);同步端点一并改用。 测试:新增 9 例(契约/免责/降级/异常不落库/落库失败/鉴权边界/超长/closed/Tool 留痕),全量 pytest 494→503 绿。遗留:无心跳帧(长生成空隙靠反代 timeout 配置), 断连留空会话待清理策略。
477 lines
19 KiB
Python
477 lines
19 KiB
Python
"""对话接口(T-06 最小闭环 · FLOW §2):四 Agent 统一 chat 入口。
|
||
|
||
链路:X-Agent-Type 分流 + Agent 准入(deps.assert_agent_access,手册 §5.4)
|
||
→ T-03 输入防护(input_guard:注入短语 / 超长,命中即拒 + input_guard_log
|
||
留痕,fail-fast 在会话解析前)→ customer 归属固定本人 / 代理人等指定客户走
|
||
G-01 归属校验(A-01 语义)→ SessionGuard(会话存在、actor/agent_type 一致
|
||
AUTH_403_SESSION_AGENT、active 状态,手册 §9)→ memory_service 窗口 →
|
||
agent_service(T-07 图 + T-04 Tool 节点:意图→归属校验→Core RO 只读查询)
|
||
→ user/assistant 双消息落 MySQL + Redis 窗口 → 响应 {session_id, reply,
|
||
has_disclaimer, trace_id}。
|
||
落库:agent_session/agent_message(同 trace_id);agent_tool_call 由 Tool
|
||
节点落(T-04,success/blocked/error 全留痕)。审计:鉴权失败/越权经
|
||
deps.deny 双写留痕;输入防护拒绝经 T-03 落 input_guard_log。
|
||
方案 B(前端拉侧):GET /sessions(本人会话分页列表)、
|
||
GET /sessions/{id}/messages(历史消息升序分页)、POST /sessions/{id}/close
|
||
(active→closed,重复关闭 409);与 POST "" 共用入口守卫 + SessionGuard。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import time
|
||
from collections.abc import Iterator
|
||
from typing import Any
|
||
from uuid import uuid4
|
||
|
||
from fastapi import APIRouter, Depends, Query, Request
|
||
from fastapi.responses import StreamingResponse
|
||
from pydantic import BaseModel, Field
|
||
|
||
from app.api.deps import (
|
||
AGENT_TYPES,
|
||
AuthContext,
|
||
assert_agent_access,
|
||
assert_customer_access,
|
||
deny,
|
||
get_auth_context,
|
||
)
|
||
from app.repository.core_ro import CoreReadOnlyRepository
|
||
from app.repository.risk_repository import RiskRepository
|
||
from app.repository.session_repository import SessionRepository
|
||
from app.service import agent_service, input_guard, memory_service
|
||
from app.utils.exceptions import ApiError, StateConflict
|
||
from app.utils.trace import current_trace, new_trace
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
router = APIRouter(prefix="/api/chat", tags=["chat"])
|
||
|
||
# Pydantic 硬顶(仅防 DoS 的超宽上限):业务上限 4000 由 input_guard
|
||
# (T-03)判定——在 guard 层拦截才能落 input_guard_log 留痕;
|
||
# Pydantic 层直接 422 会绕过留痕(F-03「拦截记录」要求)。
|
||
MESSAGE_HARD_CEILING = 20000
|
||
|
||
|
||
def _repo() -> RiskRepository:
|
||
"""审计仓储(deny 留痕用;测试 monkeypatch 点)。"""
|
||
return RiskRepository()
|
||
|
||
|
||
def _session_repo() -> SessionRepository:
|
||
"""会话仓储(测试 monkeypatch 点)。"""
|
||
return SessionRepository()
|
||
|
||
|
||
def _core_ro() -> CoreReadOnlyRepository:
|
||
"""归属校验仓储(测试 monkeypatch 点)。"""
|
||
return CoreReadOnlyRepository()
|
||
|
||
|
||
class ChatRequest(BaseModel):
|
||
session_id: str | None = Field(None, description="缺省新建会话;传入则续聊")
|
||
message: str = Field(..., min_length=1, max_length=MESSAGE_HARD_CEILING)
|
||
customer_id: str | None = Field(
|
||
None, description="目标客户:customer 角色忽略(强制本人);advisor/risk/analyst 可指定(过归属校验)"
|
||
)
|
||
title: str | None = Field(None, max_length=256)
|
||
|
||
|
||
def _primary_role(auth: AuthContext) -> str:
|
||
"""会话主角色(agent_session.actor_role):按 Agent 边界优先序取(评审 P3-11,
|
||
多角色 token 落库稳定)。"""
|
||
for role in ("customer", "advisor", "analyst", "risk_officer", "compliance", "ops"):
|
||
if role in auth.roles:
|
||
return role
|
||
return auth.roles[0] if auth.roles else "unknown"
|
||
|
||
|
||
def _resolve_customer_id(
|
||
auth: AuthContext, agent_type: str, requested: str | None
|
||
) -> str | None:
|
||
"""会话关联客户:customer 强制本人;其余角色指定时过 G-01 归属校验。"""
|
||
if agent_type == "customer":
|
||
if requested and requested != auth.customer_id:
|
||
deny(auth, "AUTH_403_NOT_OWNER", _repo(), customer_id=requested, agent_type=agent_type)
|
||
return auth.customer_id
|
||
if requested:
|
||
assert_customer_access(auth, requested, core_ro=_core_ro(), risk_repo=_repo())
|
||
return requested
|
||
return None
|
||
|
||
|
||
def _resolve_agent_type(request: Request) -> str:
|
||
"""X-Agent-Type 解析(POST 与查询/关闭端点共用;debug 通道在此强制)。"""
|
||
agent_type = request.headers.get("X-Agent-Type", "").strip()
|
||
if not agent_type:
|
||
raise ApiError(401, "AUTH_401_MISSING_AGENT_TYPE", "missing X-Agent-Type header")
|
||
if agent_type not in AGENT_TYPES:
|
||
raise ApiError(400, "BAD_REQUEST", f"invalid X-Agent-Type: {agent_type}")
|
||
return agent_type
|
||
|
||
|
||
def _assert_chat_entry(auth: AuthContext, agent_type: str) -> None:
|
||
"""对话线入口守卫(四端点共用):矩阵准入 + risk_manager 显式拒绝。
|
||
|
||
C5 前置(PRD 4A.1):对话线不放行 risk_manager——HTTP 台账才放行,保住
|
||
FR-6 冻结口径。矩阵放行解决 HTTP 通道,chat 层显式拒绝兜底(manager 根本
|
||
进不了对话线,Tool 层 assert_tool_access 天然 fail-closed)。会话查询/
|
||
关闭端点同属对话线数据面,沿用同一口径(方案 B 拍板)。
|
||
"""
|
||
assert_agent_access(auth, agent_type, risk_repo=_repo())
|
||
if agent_type == "risk" and "risk_manager" in auth.roles:
|
||
deny(auth, "AUTH_403_ROLE", _repo(), message="对话线仅限 risk_officer,请走 HTTP 台账")
|
||
|
||
|
||
def _guard_session(auth: AuthContext, agent_type: str, session_id: str) -> dict:
|
||
"""SessionGuard(手册 §9):会话存在 + actor/agent_type 一致,他人会话 fail-closed。"""
|
||
session = _session_repo().get_session(session_id)
|
||
if session is None:
|
||
raise ApiError(404, "NOT_FOUND", f"session not found: {session_id}")
|
||
if session["actor_id"] != auth.actor_id or session["agent_type"] != agent_type:
|
||
deny(
|
||
auth,
|
||
"AUTH_403_SESSION_AGENT",
|
||
_repo(),
|
||
customer_id=session.get("customer_id"),
|
||
message="session belongs to another actor or agent",
|
||
agent_type=agent_type,
|
||
)
|
||
return session
|
||
|
||
|
||
def _guard_request(req: ChatRequest, request: Request, auth: AuthContext) -> tuple[str, str]:
|
||
"""同步/流式共用前置守卫(方案 C 抽):返回 (agent_type, 清洗后消息)。
|
||
|
||
顺序与既有口径严格一致:Agent 准入 → 空白 → 限流 429 → 注入/超长 400。
|
||
必须在返回 StreamingResponse **之前**跑完——SSE 一旦 200 就改不了状态码,
|
||
所以 401/403/429/400 一律走普通 JSON 响应体。
|
||
"""
|
||
agent_type = _resolve_agent_type(request)
|
||
_assert_chat_entry(auth, agent_type)
|
||
|
||
message = req.message.strip()
|
||
if not message:
|
||
raise ApiError(400, "BAD_REQUEST", "message is blank")
|
||
|
||
# T-03 限流(actor 级固定窗口,拍板 30 次/分):先于内容防护——计数
|
||
# 覆盖全部请求(含将被注入拦截的),重复攻击者快速收敛到 429,不再
|
||
# 逐条扫描+留痕;Redis 异常 fail-open(可用性保护非安全边界)。
|
||
if not input_guard.check_rate_limit(agent_type, auth.actor_id):
|
||
try:
|
||
_repo().insert_input_guard_log(
|
||
trace_id=current_trace() or new_trace(),
|
||
agent_type=agent_type,
|
||
actor_id=auth.actor_id,
|
||
guard_type=input_guard.GUARD_RATE_LIMIT,
|
||
action="blocked",
|
||
raw_excerpt=f"rate limited: {auth.actor_id}",
|
||
session_id=req.session_id,
|
||
)
|
||
except Exception:
|
||
logger.warning(
|
||
"rate limit log failed (degraded): actor=%s", auth.actor_id, exc_info=True
|
||
)
|
||
raise ApiError(429, "GUARD_RATE_LIMITED", "rate limit exceeded, retry later")
|
||
|
||
# T-03 输入防护(F-03/G-03):准入后、会话解析前 fail-fast——被拒输入
|
||
# 不建会话、不落消息表。命中即拒(拍板:宁可误拒不可漏放);留痕失败
|
||
# 降级 warning,拒绝语义优先(与 deps 401/403 留痕降级同口径)。
|
||
verdict = input_guard.inspect_message(message)
|
||
if verdict.blocked:
|
||
try:
|
||
_repo().insert_input_guard_log(
|
||
trace_id=current_trace() or new_trace(),
|
||
agent_type=agent_type,
|
||
actor_id=auth.actor_id,
|
||
guard_type=verdict.guard_type or input_guard.GUARD_INJECTION,
|
||
action="blocked",
|
||
raw_excerpt=message[:1024],
|
||
session_id=(req.session_id or "")[:64], # P3 评审:未校验字段先截断再落审计库
|
||
)
|
||
except Exception:
|
||
logger.warning(
|
||
"input guard log failed (degraded): actor=%s type=%s",
|
||
auth.actor_id,
|
||
verdict.guard_type,
|
||
exc_info=True,
|
||
)
|
||
code = (
|
||
"GUARD_BLOCKED_OVERSIZE"
|
||
if verdict.guard_type == input_guard.GUARD_OVERSIZE
|
||
else "GUARD_BLOCKED_INJECTION"
|
||
)
|
||
raise ApiError(400, code, "message rejected by input guard")
|
||
return agent_type, message
|
||
|
||
|
||
def _prepare_turn(req: ChatRequest, auth: AuthContext, agent_type: str) -> tuple[str, str | None]:
|
||
"""会话解析/创建(同步/流式共用):返回 (session_id, customer_id)。
|
||
|
||
归属解析 → 续聊走 SessionGuard(404/403/409)→ 新建走 create_session。
|
||
"""
|
||
customer_id = _resolve_customer_id(auth, agent_type, req.customer_id)
|
||
session_repo = _session_repo()
|
||
|
||
if req.session_id:
|
||
session = _guard_session(auth, agent_type, req.session_id)
|
||
if session["status"] != "active":
|
||
raise ApiError(409, "STATE_CONFLICT", f"session is {session['status']}")
|
||
sid = session["session_id"]
|
||
else:
|
||
sid = f"sess-{uuid4().hex[:16]}"
|
||
session_repo.create_session(
|
||
session_id=sid,
|
||
trace_id=current_trace(),
|
||
agent_type=agent_type,
|
||
actor_id=auth.actor_id,
|
||
actor_role=_primary_role(auth),
|
||
customer_id=customer_id,
|
||
advisor_id=auth.actor_id if agent_type == "advisor" else None,
|
||
title=req.title,
|
||
)
|
||
return sid, customer_id
|
||
|
||
|
||
@router.post("")
|
||
def chat_api(req: ChatRequest, request: Request, auth: AuthContext = Depends(get_auth_context)) -> dict:
|
||
agent_type, message = _guard_request(req, request, auth)
|
||
sid, customer_id = _prepare_turn(req, auth, agent_type)
|
||
|
||
history = memory_service.get_recent(agent_type, sid)
|
||
result = agent_service.chat(
|
||
agent_type,
|
||
history,
|
||
message,
|
||
session_id=sid,
|
||
trace_id=current_trace(),
|
||
actor={"actor_id": auth.actor_id, "roles": auth.roles, "token_type": auth.token_type},
|
||
customer_id=customer_id,
|
||
)
|
||
|
||
# 落盘:user + assistant 同步写(同事务,异步化归后续);同 trace_id 贯通
|
||
trace_id = current_trace()
|
||
_session_repo().insert_turn(
|
||
session_id=sid,
|
||
trace_id=trace_id,
|
||
user_content=message,
|
||
assistant_content=result["reply"],
|
||
has_disclaimer=bool(result["has_disclaimer"]),
|
||
)
|
||
memory_service.append_window(
|
||
agent_type,
|
||
sid,
|
||
[
|
||
{"role": "user", "content": message},
|
||
{"role": "assistant", "content": result["reply"]},
|
||
],
|
||
)
|
||
|
||
return {
|
||
"session_id": sid,
|
||
"agent_type": agent_type,
|
||
"customer_id": customer_id,
|
||
"reply": result["reply"],
|
||
"has_disclaimer": result["has_disclaimer"],
|
||
"trace_id": trace_id,
|
||
}
|
||
|
||
|
||
# ---------- 方案 B:前端「拉」侧只读接口(会话列表 / 历史消息 / 关闭会话) ----------
|
||
#
|
||
# 前端对话页三件套,与 POST "" 共用同一套入口守卫(_assert_chat_entry:
|
||
# 矩阵准入 + risk_manager 显式拒绝)与 SessionGuard(_guard_session:
|
||
# 仅本人会话 + agent_type 一致,越权 403 留痕)。纯读/状态流转,不改表、
|
||
# 不碰 Tool 契约;manager 与对话线保持同口径 deny(见 _assert_chat_entry)。
|
||
|
||
|
||
@router.get("/sessions")
|
||
def list_sessions_api(
|
||
request: Request,
|
||
limit: int = Query(20, ge=1, le=100),
|
||
offset: int = Query(0, ge=0),
|
||
auth: AuthContext = Depends(get_auth_context),
|
||
) -> dict:
|
||
"""当前登录人的会话列表(created_at 倒序分页;agent_type 经 X-Agent-Type 头指定)。"""
|
||
agent_type = _resolve_agent_type(request)
|
||
_assert_chat_entry(auth, agent_type)
|
||
items, total = _session_repo().list_sessions(
|
||
actor_id=auth.actor_id, agent_type=agent_type, limit=limit, offset=offset
|
||
)
|
||
return {"items": items, "total": total, "limit": limit, "offset": offset}
|
||
|
||
|
||
@router.get("/sessions/{session_id}/messages")
|
||
def list_messages_api(
|
||
session_id: str,
|
||
request: Request,
|
||
limit: int = Query(50, ge=1, le=200),
|
||
offset: int = Query(0, ge=0),
|
||
auth: AuthContext = Depends(get_auth_context),
|
||
) -> dict:
|
||
"""指定会话的历史消息(seq_no 升序分页;closed 会话历史仍可读)。"""
|
||
agent_type = _resolve_agent_type(request)
|
||
_assert_chat_entry(auth, agent_type)
|
||
_guard_session(auth, agent_type, session_id)
|
||
items, total = _session_repo().list_messages_page(session_id, limit=limit, offset=offset)
|
||
return {
|
||
"session_id": session_id,
|
||
"items": items,
|
||
"total": total,
|
||
"limit": limit,
|
||
"offset": offset,
|
||
}
|
||
|
||
|
||
@router.post("/sessions/{session_id}/close")
|
||
def close_session_api(
|
||
session_id: str,
|
||
request: Request,
|
||
auth: AuthContext = Depends(get_auth_context),
|
||
) -> dict:
|
||
"""关闭会话(active → closed + closed_at);重复关闭/非 active 409。"""
|
||
agent_type = _resolve_agent_type(request)
|
||
_assert_chat_entry(auth, agent_type)
|
||
session = _guard_session(auth, agent_type, session_id)
|
||
if session["status"] != "active":
|
||
raise ApiError(409, "STATE_CONFLICT", f"session is {session['status']}")
|
||
# 条件更新(WHERE status='active'):并发双击时后到者 rowcount=0 —— 不静默
|
||
# 返回 200,转 409 与“重复关闭”同语义(评审 P1)。
|
||
if not _session_repo().close_session(session_id):
|
||
raise ApiError(409, "STATE_CONFLICT", "session is closed")
|
||
return {"session_id": session_id, "status": "closed"}
|
||
|
||
|
||
# ---------- 方案 C:SSE 流式对话(POST /api/chat/stream) ----------
|
||
#
|
||
# 契约(OpenAI 兼容 chunk 格式,fetch-event-source / AI SDK 可直接接):
|
||
# 首帧 data: {"...","choices":[{"delta":{"role":"assistant"}}],"meta":{...}}
|
||
# 中间 data: {"...","choices":[{"delta":{"content":"文本块"}}]}
|
||
# 结束 data: {"...","choices":[{"delta":{},"finish_reason":"stop"}],"meta":{...}}
|
||
# data: [DONE]
|
||
# 异常 data: {"error":{"code":...,"message":...}} → data: [DONE]
|
||
# meta 为同层扩展字段(session_id/trace_id/disclaimer/has_disclaimer),
|
||
# OpenAI 标准无此键,不破坏 delta 兼容。
|
||
#
|
||
# 两条硬约束(设计拍板):
|
||
# 1) 鉴权/限流/防护全部在返回 StreamingResponse 之前完成——SSE 一旦 200
|
||
# 就改不了状态码,故 401/403/429/400 仍是普通 JSON;
|
||
# 2) 消息落库在收完 done 后一次性写;中途异常/断连整轮不落(Tool 留痕
|
||
# 已落可审计),不产生半截内容污染历史窗口。
|
||
|
||
_SSE_DONE = "data: [DONE]\n\n"
|
||
|
||
|
||
def _sse(payload: dict) -> str:
|
||
"""单帧编码(ensure_ascii=False 保中文直出;每帧以空行结束)。"""
|
||
return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
|
||
|
||
|
||
def _chunk(
|
||
trace_id: str,
|
||
delta: dict,
|
||
finish_reason: str | None = None,
|
||
meta: dict | None = None,
|
||
) -> str:
|
||
"""OpenAI 兼容 chunk 帧;meta 仅首帧/结束帧携带。"""
|
||
payload: dict[str, Any] = {
|
||
"id": trace_id,
|
||
"object": "chat.completion.chunk",
|
||
"created": int(time.time()),
|
||
"model": "deepseek-chat",
|
||
"choices": [{"index": 0, "delta": delta, "finish_reason": finish_reason}],
|
||
}
|
||
if meta is not None:
|
||
payload["meta"] = meta
|
||
return _sse(payload)
|
||
|
||
|
||
@router.post("/stream")
|
||
def chat_stream_api(
|
||
req: ChatRequest, request: Request, auth: AuthContext = Depends(get_auth_context)
|
||
) -> StreamingResponse:
|
||
"""流式对话(方案 C):与 POST "" 同守卫,逐块推送 LLM 文本。"""
|
||
agent_type, message = _guard_request(req, request, auth)
|
||
sid, customer_id = _prepare_turn(req, auth, agent_type)
|
||
trace_id = current_trace() or ""
|
||
has_disclaimer = agent_service.needs_disclaimer(agent_type)
|
||
history = memory_service.get_recent(agent_type, sid)
|
||
|
||
def _events() -> Iterator[str]:
|
||
# 首帧:meta 先下发 session_id(前端刷新后可续聊)+ 免责声明文本
|
||
# (customer/risk 线合规要求:流式正文先出,声明不能等到最后)。
|
||
meta: dict[str, Any] = {
|
||
"session_id": sid,
|
||
"agent_type": agent_type,
|
||
"customer_id": customer_id,
|
||
"trace_id": trace_id,
|
||
"has_disclaimer": has_disclaimer,
|
||
"disclaimer": agent_service.CHAT_DISCLAIMER if has_disclaimer else None,
|
||
}
|
||
yield _chunk(trace_id, {"role": "assistant"}, meta=meta)
|
||
full: list[str] = []
|
||
try:
|
||
for kind, text in agent_service.stream_chat(
|
||
agent_type,
|
||
history,
|
||
message,
|
||
session_id=sid,
|
||
trace_id=trace_id,
|
||
actor={"actor_id": auth.actor_id, "roles": auth.roles, "token_type": auth.token_type},
|
||
customer_id=customer_id,
|
||
):
|
||
if kind == "delta":
|
||
full.append(text)
|
||
yield _chunk(trace_id, {"content": text})
|
||
except Exception as exc: # 生成中异常:整轮不落库,结构化错误收尾
|
||
logger.warning("chat stream failed (no message persisted): %s", exc, exc_info=True)
|
||
yield _sse({"error": {"code": "STREAM_FAILED", "message": "生成失败,请重试"}})
|
||
yield _SSE_DONE
|
||
return
|
||
reply = "".join(full)
|
||
if has_disclaimer:
|
||
reply = f"{reply}\n\n{agent_service.CHAT_DISCLAIMER}"
|
||
|
||
# 落盘(与同步同口径):user + assistant 同事务一次性写,同 trace_id 贯通。
|
||
# 落库失败 → 整轮不落(不出现 user 落、assistant 未落的半截历史),
|
||
# 并补发 error 帧收尾——否则前端收不到 [DONE] 会一直挂着(评审 P0)。
|
||
try:
|
||
_session_repo().insert_turn(
|
||
session_id=sid,
|
||
trace_id=trace_id,
|
||
user_content=message,
|
||
assistant_content=reply,
|
||
has_disclaimer=has_disclaimer,
|
||
)
|
||
except Exception as exc:
|
||
logger.warning("chat stream persist failed: %s", exc, exc_info=True)
|
||
yield _sse({"error": {"code": "PERSIST_FAILED", "message": "消息保存失败,请重试"}})
|
||
yield _SSE_DONE
|
||
return
|
||
memory_service.append_window(
|
||
agent_type,
|
||
sid,
|
||
[
|
||
{"role": "user", "content": message},
|
||
{"role": "assistant", "content": reply},
|
||
],
|
||
)
|
||
yield _chunk(
|
||
trace_id,
|
||
{},
|
||
finish_reason="stop",
|
||
meta={"session_id": sid, "trace_id": trace_id, "has_disclaimer": has_disclaimer},
|
||
)
|
||
yield _SSE_DONE
|
||
|
||
return StreamingResponse(
|
||
_events(),
|
||
media_type="text/event-stream",
|
||
headers={
|
||
"Cache-Control": "no-cache",
|
||
"Connection": "keep-alive",
|
||
"X-Accel-Buffering": "no", # 关掉反代缓冲,否则前端收不到增量
|
||
},
|
||
)
|