本文引用的文件
本文面向 Agent 对话模块的开发者与维护者,说明前端如何发起对话、后端 AgentChatController 与 AgentChatService 的职责边界,以及从前端到后端的完整 SSE(Server-Sent Events)流式交互链路。重点包括:
chatApi.ts 暴露接口,由 sse.ts 使用 fetch + ReadableStream 处理事件流。chatStream.ts 作为 Pinia store,负责会话生命周期、乐观插入用户消息、增量聚合助手回复、工具调用时序和 UI 渲染。AgentChatServiceImpl:校验工作空间、建立会话级互斥、驱动 Agent 流事件、按事件类型生成 SSE 帧、持久化消息并发送心跳。AgentChatSession、AgentMessage、Insight 会话在对话生命周期中的关系,并给出“选择模型后发送文本”“查看工具调用日志”“中断或重连 SSE”等典型流程。本功能横跨前后端两个子工程:
| 层级 | 关键路径 | 职责 |
|---|---|---|
| 前端 API 层 | ai-electron/frontend/src/ai/api/chatApi.ts |
封装会话、消息、图谱复原及流式对话入口;将解析器转发给 sse.ts |
| 前端 SSE 实现 | ai-electron/frontend/src/ai/api/sse.ts |
通用 SSE 帧解析器、fetch 流读取、未登录/非 SSE 响应处理、中止信号透传 |
| 前端状态管理 | ai-electron/frontend/src/ai/store/chatStream.ts |
会话列表、消息列表、流式对话状态机、工具步骤与思考过程累积、UI 富内容块增量扫描 |
| 前端类型与常量 | ai-electron/frontend/src/ai/api/types.ts、ai-electron/frontend/src/ai/utils/constants.ts |
会话/消息/工具/Insight 契约、getChatStreamUrl() 地址拼接 |
| 后端控制器 | ai-server/.../controller/AgentChatController.java |
/chat 下 REST 路由;/chat/stream 返回 Flux<ServerSentEvent<String>> |
| 后端服务接口 | ai-server/.../service/AgentChatService.java |
会话 CRUD、消息 CRUD、流式对话抽象 |
| 后端服务实现 | ai-server/.../service/impl/AgentChatServiceImpl.java |
工作空间校验、会话锁、Agent 流事件映射、SSE 帧组装、心跳、落库 |
| 后端实体 | AgentChatSession.java、AgentMessage.java |
数据库表 agent_chat_session、agent_message 的字段映射 |
graph TB
subgraph "前端"
CA["chatApi.ts<br/>API 入口"]
ST["sse.ts<br/>SSE 解析与 fetch 流"]
CS["chatStream.ts<br/>Pinia 状态与流式状态机"]
TP["types.ts / constants.ts<br/>类型与地址"]
end
subgraph "后端"
AC["AgentChatController.java<br/>/chat 路由"]
AS["AgentChatService.java<br/>服务接口"]
ASI["AgentChatServiceImpl.java<br/>流式逻辑实现"]
DB["agent_chat_session / agent_message<br/>PostgreSQL JSONB"]
end
CA --> ST
CA --> CS
CS --> CA
ST --> TP
AC --> AS
AS --> ASI
ASI --> DB
ST --> AC
图表来源
章节来源
chatApi.tschatApi.ts 是聊天相关 HTTP 接口的统一入口。它不直接实现 SSE 细节,而是把流式能力转发给 sse.ts:
createParser、streamSse、streamChat,保持调用方 chatStream.ts 零改动。关键设计点:
defHttp(即 aiHttp)。ReadableStream。章节来源
sse.tssse.ts 是通用 SSE 基础设施,同时供 AI 数据分析与 Insight 研判复用。其职责包括:
帧解析器 createParser
\n\n 与 \r\n\r\n 分隔。: 开头)。data: 字段。event 分发到 onIntent、onThinking、onToolCall、onToolInput、onToolResult、onHeartbeat、onGraphData、onInsight、onToken、onDone、onError。flush(),防止最后一帧没有结尾空行时丢失末事件。streamSse 连接层
fetch 发起 POST,携带 x-token、Accept: text/event-stream。response.body.getReader() 逐块解码并喂给解析器。AbortError 透传给调用方;其他异常转 transportError。streamChat 适配器
getChatStreamUrl(),即 /chat/stream。streamSse。章节来源
chatStream.tschatStream.ts 是整个对话 UI 的核心状态机。它不是简单的网络回调包装,而是承担了:
abortController。重要状态字段:
| 字段 | 含义 |
|---|---|
workspaceId |
当前案件/工作空间 |
sessions |
会话列表 |
activeSessionId |
当前选中会话 |
messages |
当前会话消息数组 |
streaming |
是否正在流式输出 |
pendingExpert |
来自 Insight 页面的预选研判方案 |
thinking |
模型思考正文 |
trace |
思考与工具的真实到达顺序链 |
toolSteps |
工具调用步骤列表 |
intent / suggestions |
意图识别与下一步引导 |
lastActivityAt / lastHeartbeatAt |
用于区分“模型在输出”和“连接断掉” |
章节来源
AgentChatController.java控制器只承担路由和参数绑定职责,不包含业务编排逻辑:
/chat/sessions、标题更新、删除、置顶。/chat/messages、收藏。/chat/graph/restore。/chat/attachments。POST /chat/stream,返回 Flux<ServerSentEvent<String>>。控制器注释明确描述了 SSE 事件顺序:
tool_calltool_inputtool_resulttokendoneerror章节来源
AgentChatService.java该接口定义了聊天域的所有对外方法,包括:
stream(ChatRequestDTO req)。它不关心具体事件如何映射,只声明契约。
章节来源
AgentChatServiceImpl.java这是流式对话的实际实现。核心行为:
HarnessAgent.streamEvents,将 Agent 事件映射为 SSE 帧。heartbeat。章节来源
下图展示一次完整的 Agent 对话请求从前端到后端的控制流和数据流:
sequenceDiagram
participant UI as "聊天页面"
participant Store as "chatStream.ts"
participant Api as "chatApi.ts"
participant Sse as "sse.ts"
participant Ctrl as "AgentChatController.java"
participant Svc as "AgentChatServiceImpl.java"
participant Db as "agent_chat_session / agent_message"
UI->>Store: "选择模型并发送文本"
Store->>Store: "乐观插入用户消息与占位助手消息"
Store->>Api: "streamChat(params)"
Api->>Sse: "streamSse(getChatStreamUrl(), params, handlers)"
Sse->>Ctrl: "POST /chat/stream"
Ctrl->>Svc: "stream(req)"
Svc->>Db: "校验会话与工作空间"
Svc->>Svc: "会话级互斥检查"
Svc->>Svc: "启动 Agent 流事件"
loop "SSE 事件流"
Svc-->>Sse: "tool_call / tool_input / tool_result / token / thinking / heartbeat"
Sse-->>Store: "对应 on* 回调"
Store->>Store: "累积 content / toolSteps / trace / thinking"
end
alt "正常完成"
Svc-->>Sse: "done"
Svc->>Db: "持久化助手消息"
else "错误或中断"
Svc-->>Sse: "error / 连接关闭"
Svc->>Db: "持久化错误或部分消息"
end
Sse-->>Store: "finally 清理流状态"
Store->>Store: "结束 streaming、刷新会话元信息"
图表来源
前端调用链如下:
chatStream.ts 的 send 方法判断是否需要新建会话。chatApi.streamChat。chatApi.ts 转发到 sse.ts 的 streamChat。sse.ts 拼接 /chat/stream 地址并发起 fetch。TextDecoder + SseFrameParser 逐帧解析。将事件分发给 chatStream.ts 注册的 handlers。
flowchart TD
Start["用户发送消息"] --> Draft{"是否存在有效会话?"}
Draft -->|否| Create["创建会话"]
Draft -->|是| UseExisting["复用现有会话"]
Create --> InsertUser["乐观插入用户消息"]
UseExisting --> InsertUser
InsertUser --> InsertAssistant["插入占位助手消息"]
InsertAssistant --> StreamApi["chatApi.streamChat"]
StreamApi --> SseCore["sse.ts streamSse"]
SseCore --> Fetch["fetch POST /chat/stream"]
Fetch --> Parse["SseFrameParser 解析帧"]
Parse --> Handlers["chatStream handlers 更新状态"]
Handlers --> Render["UI 增量渲染"]
图表来源
章节来源
后端 AgentChatServiceImpl 将 AgentScope 的 AgentEvent 转换为 SSE 帧:
| Agent 阶段 | SSE event | 前端处理器 | 状态变化 |
|---|---|---|---|
| 工具开始 | tool_call |
onToolCall |
新增 ToolStep{status:'running'} |
| 工具入参完成 | tool_input |
onToolInput |
写入 step.input |
| 工具结果完成 | tool_result |
onToolResult |
写入 step.result、status='done' |
| 模型 token | token |
onToken |
追加 assistant.content |
| 思考开始 | thinking.start |
onThinking |
开启 thinkingActive,创建 trace think 节点 |
| 思考增量 | thinking.delta |
onThinking |
追加 thinking 与 trace.text |
| 思考结束 | thinking.end |
onThinking |
关闭 thinkingActive |
| 完成 | done |
onDone |
结束 streaming,持久化消息 |
| 错误 | error |
onError |
标记 messageType='error' |
| 心跳 | heartbeat |
onHeartbeat |
更新 lastHeartbeatAt |
心跳流与业务事件流是独立的:心跳通过 Flux.interval(10s) 生成,并在连接结束时停止。
章节来源
chatStream.ts 对每个事件的处理策略:
onIntent:保存意图识别结果。onSuggestions:保存扁平建议与方向分组。onThinking:分段累积思考,维护 thinkingActive 与 trace。onToolCall:新增工具步骤,记录 trace tool 节点。onToolInput:归一化工具入参文本。onToolResult:写入结果并标记完成。onToken:追加正文,并按 100ms 节流执行富内容块增量扫描。onHeartbeat:仅更新时间戳,不影响 content。onGraphData:回填 get_case_graph 的全量图谱。onDone:结束消息 streaming,但整体 finally 仍会清理流状态。onError / onTransportError:设置 messageType 为 error 或 interrupted。章节来源
AgentChatSession| 字段 | 类型 | 含义 |
|---|---|---|
id |
Long | 主键 |
agent_id |
Long | 绑定的研判方案 |
model_id |
Long | 会话默认模型 |
case_id |
Long | 工作空间/案件 |
title |
String | 会话标题 |
context |
JSONB | 上下文 |
message_count |
Integer | 消息数 |
total_tokens |
Long | 累计 token |
last_message_at |
LocalDateTime | 最后消息时间 |
pinned |
Boolean | 是否置顶 |
attachments |
JSONB | 附件摘要 |
create_at |
LocalDateTime | 创建时间 |
update_at |
LocalDateTime | 更新时间 |
AgentMessage| 字段 | 类型 | 含义 |
|---|---|---|
id |
Long | 主键 |
session_id |
Long | 所属会话 |
role |
String | user / assistant / system |
content |
String | 正文 |
content_type |
String | 内容类型 |
token_count |
Integer | token 计数 |
metadata |
JSONB | 思考过程、附件摘要、错误标记等 |
tool_events |
JSONB | 工具调用事件序列 |
parent_id |
Long | 父消息 |
message_type |
String | text / error / interrupted |
reply_to_message_id |
Long | 回复目标 |
starred |
Boolean | 是否收藏 |
create_at |
LocalDateTime | 创建时间 |
erDiagram
AGENT_CHAT_SESSION {
long id PK
long agent_id
long model_id
long case_id
string title
jsonb context
int message_count
long total_tokens
datetime last_message_at
boolean pinned
jsonb attachments
datetime create_at
datetime update_at
}
AGENT_MESSAGE {
long id PK
long session_id FK
string role
string content
string content_type
int token_count
jsonb metadata
jsonb tool_events
long parent_id
string message_type
long reply_to_message_id
boolean starred
datetime create_at
}
INSIGHT_SESSION {
long id PK
string name
jsonb context
datetime create_at
datetime update_at
}
AGENT_CHAT_SESSION ||--o{ AGENT_MESSAGE : "包含"
图表来源
在对话生命周期中:
AgentChatSession 表示一次 Agent 对话会话,绑定工作空间、研判方案和模型。AgentMessage 表示会话中的一条消息,可以是用户提问、助手回答、错误提示或中断结果。insight 帧;前端在 chatStream.ts 中通过 onInsight 处理。章节来源
chatStream.ts 的 send(text, modelId) 被调用。streamChat({ sessionKey, message, modelId })。sessionKey 为会话 ID,校验工作空间,建立 Agent 流。done 后结束流式态,刷新会话标题、消息数和最后消息时间。注意:modelId 在前端类型中要求按字符串传递,避免雪花 ID 精度丢失。
章节来源
工具调用日志保存在 AgentMessage.tool_events,前端将其反序列化为 ToolStep[]:
| 阶段 | 事件 | ToolStep 变化 |
|---|---|---|
| 开始 | tool_call |
新增步骤,状态 running |
| 入参 | tool_input |
填充 input |
| 结果 | tool_result |
填充 result,状态 done |
| 图谱 | graph_data |
回填 graphPayload |
前端还会对历史消息中的 get_case_graph 调用进行图谱复原:如果数据库中只存了查询条件,则调用 /chat/graph/restore 拉取全量图。
章节来源
中断:
chatStream.ts 调用 stop()。stop() 清除定时器并调用 abortController.abort()。sse.ts 捕获 AbortError 并向上抛出。chatStream.ts catch 分支忽略 AbortError,进入 finally。重连:
AbortController 和新的 fetch 连接。onTransportError 中暴露重连策略,而不是无限重试同一个 reader。章节来源
graph LR
ChatPage["聊天页面组件"] --> ChatStream["chatStream.ts"]
ChatStream --> ChatApi["chatApi.ts"]
ChatApi --> Sse["sse.ts"]
Sse --> Constants["constants.ts"]
ChatStream --> Types["types.ts"]
Sse --> Types
图表来源
graph LR
Controller["AgentChatController"] --> Service["AgentChatService"]
Service --> Impl["AgentChatServiceImpl"]
Impl --> SessionMapper["AgentChatSessionMapper"]
Impl --> MessageMapper["AgentMessageMapper"]
Impl --> AgentService["AgentService"]
Impl --> ModelService["AgentModelService"]
Impl --> UsageStore["UsageStore"]
Impl --> GraphService["GraphService"]
Impl --> CaseAccessGuard["CaseAccessGuard"]
图表来源
章节来源
chatStream.ts 对每次 token 追加并不立即重新扫描全文,而是通过 RESAN_THROTTLE_MS 节流,默认约 100ms。这避免了长回答中反复扫描 Markdown/代码/ECharts/Insight 围栏导致的 O(n²) 成本。
已完成块会被冻结复用,进一步降低重复解析压力。
章节来源
sse.ts 维护一个字符串缓冲区,逐块拼接后再按 \n\n 切分。对于超大响应:
TextDecoder 以流模式解码,减少一次性转换开销。优化建议:
章节来源
AgentChatServiceImpl 使用 ConcurrentHashMap<Long, Semaphore> 保证同一会话只有一条流。这避免:
注意:这是进程内锁,多实例部署时需要扩展为分布式锁。
章节来源
现象:
原因:
排查要点:
lastHeartbeatAt 是否持续更新。lastActivityAt 也很久没变,大概率是连接断了。章节来源
现象:
text/event-stream。原因:
Content-Type,再尝试解析 JSON 中的 code=401。处理:
handleUnauthorizedRedirect 跳转。章节来源
现象:
原因:
处理:
sse.ts 会尝试读取响应体并提取 message 或 error。章节来源
现象:
原因:
chatStream.ts 中 target.content 不断拼接 token。sse.ts 的 buffer 累积未解析文本。tool_steps 中 result 可能包含大量文本。缓解措施:
章节来源
现象:
原因:
处理:
persistPartialMessage 会写入 messageType=interrupted。章节来源
Agent 对话与 SSE 流式交互是一个典型的前后端实时协作场景:
chatApi.ts 暴露稳定接口,sse.ts 处理低层流协议,chatStream.ts 承载业务状态机。AgentChatController 只做路由,AgentChatServiceImpl 负责会话互斥、Agent 事件映射、SSE 帧生产和消息持久化。AgentChatSession、AgentMessage 与 Insight 会话的区别,有助于避免把不同业务会话混为一谈。