zhanghongyu_0626
3ad244add0
- Updated the `dashboard` function in `analyst.py` to include additional metrics for different user roles, improving data visibility for analysts, customers, advisors, and risk officers. - Introduced a new `prepare_customer_stream` function in `customer_service.py` to facilitate streaming responses for customer interactions, enhancing the chat experience. - Added new API endpoints in `analyst.ts` for fetching dashboard metrics and managing analyst assets, streamlining data handling and user interactions. - Updated frontend components to support new dashboard features and asset management, ensuring a cohesive user experience across the application. This update significantly improves the functionality and usability of the analyst and customer service features, providing users with enhanced tools for data analysis and interaction.
531 lines
22 KiB
Python
531 lines
22 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.auth_adapter import host_auth_for_customer_service
|
||
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.service.customer_service import prepare_customer_stream, run_customer_chat
|
||
from app.utils.compliance_guard import RISK_DISCLAIMER
|
||
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)
|
||
end_session: bool = Field(default=False, description="customer 显式结束会话（触发归档）")
|
||
|
||
|
||
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
|
||
|
||
|
||
def _assistant_content_for_persist(body: str, has_disclaimer: bool) -> str:
|
||
"""落库 / Redis 窗口 assistant 正文：尾部拼免责声明（README / 流式同口径）。
|
||
|
||
guard_node 在同步路径可能已拼过；此处幂等，避免 sync/stream 落盘口径漂移。
|
||
"""
|
||
if not has_disclaimer:
|
||
return body
|
||
marker = agent_service.CHAT_DISCLAIMER
|
||
trimmed = body.rstrip()
|
||
if trimmed.endswith(marker):
|
||
return body
|
||
sep = "" if not trimmed else "\n\n"
|
||
return f"{trimmed}{sep}{marker}"
|
||
|
||
|
||
@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)
|
||
|
||
trace_id = current_trace()
|
||
intent: str | None = None
|
||
transfer = False
|
||
|
||
if agent_type == "customer":
|
||
host_ctx = host_auth_for_customer_service(auth, trace_id=trace_id or new_trace())
|
||
reply, has_disclaimer, intent, transfer = run_customer_chat(
|
||
host_ctx,
|
||
message,
|
||
sid,
|
||
customer_id or auth.customer_id or auth.actor_id,
|
||
req.end_session,
|
||
)
|
||
assistant_content = _assistant_content_for_persist(reply, has_disclaimer)
|
||
else:
|
||
history = memory_service.get_recent(agent_type, sid)
|
||
result = agent_service.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,
|
||
)
|
||
has_disclaimer = bool(result["has_disclaimer"])
|
||
assistant_content = _assistant_content_for_persist(result["reply"], has_disclaimer)
|
||
|
||
_session_repo().insert_turn(
|
||
session_id=sid,
|
||
trace_id=trace_id,
|
||
user_content=message,
|
||
assistant_content=assistant_content,
|
||
has_disclaimer=has_disclaimer,
|
||
)
|
||
memory_service.append_window(
|
||
agent_type,
|
||
sid,
|
||
[
|
||
{"role": "user", "content": message},
|
||
{"role": "assistant", "content": assistant_content},
|
||
],
|
||
)
|
||
|
||
return {
|
||
"session_id": sid,
|
||
"agent_type": agent_type,
|
||
"customer_id": customer_id,
|
||
"reply": assistant_content,
|
||
"has_disclaimer": has_disclaimer,
|
||
"trace_id": trace_id,
|
||
"intent": intent,
|
||
"transfer_to_human": transfer,
|
||
}
|
||
|
||
|
||
# ---------- 方案 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 new_trace()
|
||
has_disclaimer = agent_service.needs_disclaimer(agent_type)
|
||
customer_prep: dict[str, Any] | None = None
|
||
history: list[dict] = []
|
||
if agent_type == "customer":
|
||
host_ctx = host_auth_for_customer_service(auth, trace_id=trace_id)
|
||
cust_id = customer_id or auth.customer_id or auth.actor_id
|
||
customer_prep = prepare_customer_stream(
|
||
host_ctx, message, sid, cust_id, req.end_session
|
||
)
|
||
has_disclaimer = bool(customer_prep.get("has_disclaimer", False))
|
||
else:
|
||
history = memory_service.get_recent(agent_type, sid)
|
||
|
||
def _events() -> Iterator[str]:
|
||
meta_disclaimer = (
|
||
RISK_DISCLAIMER
|
||
if agent_type == "customer" and has_disclaimer
|
||
else agent_service.CHAT_DISCLAIMER if has_disclaimer else None
|
||
)
|
||
meta: dict[str, Any] = {
|
||
"session_id": sid,
|
||
"agent_type": agent_type,
|
||
"customer_id": customer_id,
|
||
"trace_id": trace_id,
|
||
"has_disclaimer": has_disclaimer,
|
||
"disclaimer": meta_disclaimer,
|
||
}
|
||
yield _chunk(trace_id, {"role": "assistant"}, meta=meta)
|
||
full: list[str] = []
|
||
try:
|
||
if agent_type == "customer":
|
||
assert customer_prep is not None
|
||
for text in customer_prep["chunks"]:
|
||
full.append(text)
|
||
yield _chunk(trace_id, {"content": text})
|
||
else:
|
||
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
|
||
body = customer_prep["reply"] if customer_prep else "".join(full)
|
||
reply = _assistant_content_for_persist(body, has_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", # 关掉反代缓冲，否则前端收不到增量
|
||
},
|
||
)
|