聊天机器人智能体系统架构

1. 一条消息的完整旅程

当用户通过 POST /{agentId}/sessions/{sessionId}/chat 发送一条消息后，系统会经历以下完整流程：

┌──────────┐    HTTP POST     ┌────────────────────┐
│  客户端   │ ──────────────→  │  Controller 层     │
│          │                  │  路由 & 参数校验     │
└──────────┘                  └─────────┬──────────┘
                                        │
                              ┌─────────▼──────────┐
                              │  AgentChatService   │
                              │  构建 AgentMessage   │
                              └─────────┬──────────┘
                                        │ submit()
                              ┌─────────▼──────────┐
                              │  AgentMessageQueue  │
                              │  (Memory/Redis/MQ)  │
                              └─────────┬──────────┘
                                        │ poll()
                              ┌─────────▼──────────┐
                              │  Centralized        │
                              │  Dispatcher         │
                              │  (dispatcher线程)    │
                              └─────────┬──────────┘
                                        │ workerPool.execute()
                              ┌─────────▼──────────┐
                              │  Worker 线程        │
                              │  获取 session 锁    │
                              │  获取响应通道       │
                              └─────────┬──────────┘
                                        │
                              ┌─────────▼──────────┐
                              │  AutoPlanningAgent  │
                              │  意图→规划→执行→回答 │
                              └─────────┬──────────┘
                                        │ StreamHandler 回调
                              ┌─────────▼──────────┐
                              │  响应通道            │
                              │  SSE 流式推送        │
                              │  或 回调返回结果     │
                              └─────────┬──────────┘
                                        │
                              ┌─────────▼──────────┐
                              │  客户端收到响应      │
                              └──────────────────────┘

1.1 详细步骤分解

Step 1 — Controller 接收请求

AiChatbotAgentController 收到 POST 请求后：

校验用户身份（SecurityUtils.getUser()）
校验 Agent 是否存在
如果 sessionId 为 "new"，创建新会话；否则验证会话有效性
加载知识库配置到会话

Step 2 — 消息入队

AgentChatServiceImpl.sendMessage() 将用户消息封装为 AgentMessage，通过 CentralizedAgentMessageDispatcher.submit() 提交到 AgentMessageQueue。方法立即返回 messageId，不等待处理完成。

java

AgentMessage msg = AgentMessage.userInput(sessionId, agentId, userId, message);
centralizedAgentMessageDispatcher.submit(msg);  // 非阻塞
return msg.getMessageId();

Step 3 — Dispatcher 取出消息

CentralizedAgentMessageDispatcher 内部的 dispatcher 线程持续运行 dispatchLoop()：

java

while (running) {
    AgentMessage message = messageQueue.poll(1, TimeUnit.SECONDS);  // 阻塞等待
    Semaphore lock = sessionLocks.computeIfAbsent(sessionId, k -> new Semaphore(1));
    workerPool.execute(() -> {
        lock.acquire();       // 保证同一 session 顺序执行
        processMessage(msg);
        lock.release();
    });
}

Step 4 — Worker 处理消息

processMessage() 的执行逻辑：

从 ChatSessionManager 获取 session，验证是否有效、是否超过最大轮数
从 AgentResponseChannelRegistry 获取该 session 的响应通道
根据通道类型选择执行模式：
- SSE 通道 → 流式执行：autoPlanningAgent.autonomousPlanning(..., streamHandler)
- 回调通道（QQ Bot 等）→ 同步执行：autoPlanningAgent.autonomousPlanningSync(...)，完成后回调 channel.deliverResponse()

Step 5 — AutoPlanningAgent 执行推理

Agent 运行 LangGraph 状态机，依次经过多个节点（详见第 3 节）。

Step 6 — 结果推送给客户端

通过 StreamHandler 的回调方法（onToken、onNodeComplete、onComplete 等），实时推送到客户端的 SSE 连接。

2. 核心模块总览

┌─────────────────────────────────────────────────────────────────────────┐
│                           Controller 层                                 │
│                     AiChatbotAgentController                            │
│           POST /chat  |  GET /sse/connect  |  GET /disconnect           │
└──────────────────────────────┬──────────────────────────────────────────┘
                               │
┌──────────────────────────────▼──────────────────────────────────────────┐
│                           Service 层                                    │
│                      AgentChatServiceImpl                               │
│         sendMessage() | connectSession() | disconnectSession()          │
└────┬─────────────┬───────────────┬──────────────────┬──────────────────┘
     │             │               │                  │
     ▼             ▼               ▼                  ▼
┌─────────┐ ┌───────────┐ ┌──────────────┐ ┌──────────────────────┐
│ Message │ │ Session   │ │ SSE 连接     │ │ Channel              │
│ Queue   │ │ Manager   │ │ Manager      │ │ Registry             │
│         │ │           │ │              │ │                      │
│ Memory  │ │ L1:内存   │ │ SseEmitter   │ │ SSE / Callback       │
│ Redis   │ │ L2:Redis  │ │ 心跳保活     │ │ 通道管理              │
│ RocketMQ│ │ L3:数据库 │ │              │ │                      │
└────┬────┘ └───────────┘ └──────────────┘ └──────────────────────┘
     │
     ▼
┌──────────────────────────────────────────────────────────────────────┐
│                   CentralizedAgentMessageDispatcher                   │
│        1 dispatcher 线程  +  4~16 worker 线程池                       │
│        per-session Semaphore 保证同 session 顺序执行                   │
└────────────────────────────────┬─────────────────────────────────────┘
                                 │
                                 ▼
┌──────────────────────────────────────────────────────────────────────┐
│                        AutoPlanningAgent                              │
│     load_config → intent_recognition → planning → agentic_loop       │
│                                                                      │
│  ┌──────────────┐ ┌─────────────┐ ┌──────────────┐ ┌─────────────┐  │
│  │ToolCallNode  │ │ RagNode     │ │SkillCallNode │ │LlmGenerate  │  │
│  │McpCallNode   │ │             │ │              │ │SubAgentCall │  │
│  └──────────────┘ └─────────────┘ └──────────────┘ └─────────────┘  │
└──────────────────────────────────────────────────────────────────────┘

3. 各模块详解

3.1 ChatSessionManager — 会话管理器

管理所有聊天会话的生命周期，采用 三级缓存架构：

层级	存储	特点
L1	`ConcurrentHashMap`（内存）	最快，进程内访问
L2	Redis	分布式共享，优雅降级
L3	Database	持久化存储

核心职责：

创建会话 createSession()：同步写入 L1 + L2，L3 延迟持久化（等第一条消息时触发）
读取会话 getSession()：L1 → L2 → L3 逐级查找
消息管理 addMessage() / addMessages()：同步更新 L1 + L2，异步通过 dirty queue 批量持久化到 L3
生命周期 incrementLoopCount()、pauseSession()、resumeSession()
定时任务：每 1 分钟批量持久化脏数据，每 5 分钟清理过期会话

ChatSession 数据模型：

ChatSession
├── sessionId, agentId, userId
├── messages: List<ChatMessage>      ← 对话历史
├── context: Map<String, Object>     ← 任意运行时状态
├── status: ACTIVE / PAUSED / ENDED / EXPIRED
├── loopCount / maxLoops             ← 轮数控制
├── createTime / lastActiveTime / expireMinutes
└── datasetList: List<KnowledgeBaseDTO>  ← 知识库缓存

3.2 SseConnectionManager — SSE 连接管理器

管理所有 SSE 长连接（sessionId → SseEmitter），提供连接的创建、查询、关闭和心跳保活。

连接生命周期：

create(sessionId, timeout)
  │  创建 SseEmitter，注册 onTimeout / onError / onCompletion 回调
  ▼
startHeartbeat(sessionId)
  │  每 30 秒发送 SSE comment 事件（:heartbeat），防止反向代理断开
  ▼
[消息处理期间，SseStreamHandler 通过 emitter 推送各种事件]
  ▼
remove(sessionId)
  │  停止心跳，关闭 emitter，移除映射
  ▼
连接释放

心跳机制：反向代理（nginx 等）会在连接空闲超过一定时间后主动断开。心跳通过定期发送 SSE 注释事件（不触发客户端 onmessage）保持连接活跃。

3.3 AgentMessageQueue — 消息队列

统一的消息队列抽象，CentralizedAgentMessageDispatcher 通过此接口收发消息。

java

public interface AgentMessageQueue {
    boolean send(AgentMessage message);                              // 发送消息
    AgentMessage poll(long timeout, TimeUnit unit);                  // 取消息（阻塞）
    int size();                                                      // 队列长度
    void shutdown();                                                 // 关闭
}

三种实现，通过 ai.agent.mq.type 配置切换：

实现	配置值	适用场景
`MemoryAgentMessageQueue`	`memory`（默认）	单实例部署，最快
`RedisAgentMessageQueue`	`redis`	分布式部署，Redis Stream
`RocketMQAgentMessageQueue`	`rocketmq`	企业级高可用

3.4 CentralizedAgentMessageDispatcher — 集中式消息调度器

系统的"引擎"，实现 SmartLifecycle，Spring 容器启动时自动运行。

线程模型：

                    AgentMessageQueue
                          │
                          │ poll()
                          ▼
               ┌─── dispatcher 线程 (1个) ───┐
               │   从队列取消息并分发          │
               │   检查 poison pill           │
               └──────────┬──────────────────┘
                          │ workerPool.execute()
               ┌──────────▼──────────────────┐
               │    Worker 线程池 (4~16个)     │
               │                              │
               │  ┌─ session-A ─ Semaphore ─┐ │
               │  │  acquire → process → release
               │  └─────────────────────────┘ │
               │  ┌─ session-B ─ Semaphore ─┐ │
               │  │  acquire → process → release
               │  └─────────────────────────┘ │
               └──────────────────────────────┘

关键特性：

per-session Semaphore(1)：同一 session 的消息严格顺序执行，不同 session 之间并行
CallerRunsPolicy：worker 队列满时，由 dispatcher 线程直接执行，实现背压
双通道支持：根据 AgentResponseChannel 类型自动选择流式或同步执行模式

3.5 AgentResponseChannel — 响应通道

抽象不同渠道的响应方式，使 Agent 执行逻辑与输出渠道解耦。

AgentResponseChannel (接口)
├── SseResponseChannel          ← 浏览器 SSE 流式推送
│   └── getStreamHandler() → new SseStreamHandler(emitter)
│
└── CallbackResponseChannel     ← 第三方渠道（QQ Bot、钉钉等）
    └── getStreamHandler() → NullStreamHandler.INSTANCE
    └── deliverResponse(messageId, response)  ← 同步回调

AgentResponseChannelRegistry 是全局注册表，维护 sessionId → AgentResponseChannel 映射，线程安全。

通道选择逻辑（在 Dispatcher 中）：

java

AgentResponseChannel channel = channelRegistry.get(sessionId);

if (channel instanceof CallbackResponseChannel callback) {
    // 同步模式：完成后回调
    String response = agent.autonomousPlanningSync(...);
    callback.deliverResponse(messageId, response);
} else {
    // 流式模式：实时推送
    StreamHandler handler = channel.getStreamHandler();
    agent.autonomousPlanning(..., handler);
}

3.6 StreamHandler — 流式事件推送

定义了 Agent 执行过程中所有可推送的事件类型：

事件类别	方法	说明
节点	`onNodeStart / onNodeComplete`	工作流节点开始/完成
Token	`onToken`	LLM 生成的文本片段
RAG	`onRagStart / onRagComplete / onReferences`	知识库检索
思维链	`onThinkingStageStart / onThinkingContent`	推理过程展示
任务进度	`onTaskStepStart / onTaskStepProgress / onOverallProgress`	多步骤任务进度
结构化数据	`onStructuredBlockStart / onStructuredBlockField`	结构化内容块
生命周期	`onError / onComplete`	错误和完成

两个实现：

SseStreamHandler：将事件转为 SSE event 推送，token 每 200ms 批量发送以节省带宽
NullStreamHandler：空实现（Null Object 模式），用于回调通道等不需要流式的场景

3.7 AutoPlanningAgent — 自主规划 Agent

基于 LangGraph4j 构建的状态机，实现了完整的自主规划执行循环。

工作流图：

START
  │
  ▼
load_config ─── 从数据库加载 Agent 配置、工具、技能、知识库
  │
  ▼
intent_recognition ─── LLM 分析用户意图，判断是否需要 RAG，提取关键词
  │
  ▼
planning ─── LLM 生成执行计划，判断是否有需要执行的步骤
  │
  ├── 有执行步骤 ──→ agentic_loop ──→ task_end ──→ END
  │                    (迭代执行)       (生成最终回答)
  │
  └── 无执行步骤 ──→ task_end ──→ END
                     (直接回答)

各节点职责：

节点	类名	职责
load_config	AutoPlanningAgent 内部	加载 Agent 实体、工具列表、技能列表、知识库
intent_recognition	`IntentRecognitionNode`	意图识别、RAG 判断、关键词提取、技能筛选
planning	`PlanningNode`	生成自然语言执行计划，决定走执行循环还是直接回答
agentic_loop	`AgenticTaskExecutionNode`	核心执行循环（见下文）
task_end	`TaskEndNode`	汇总执行结果，生成最终回答

3.8 Agentic Loop — 核心执行循环

AgenticTaskExecutionNode 是系统最复杂的节点，实现了 LLM 驱动的迭代执行：

┌──────────────────────────────────────────────┐
│              Agentic Loop                     │
│                                              │
│  ┌─────────────────────────────────────────┐ │
│  │ 1. 评估当前进度（对比目标与执行历史）      │ │
│  │ 2. LLM 决策下一步动作                    │ │
│  │ 3. 执行动作                              │ │
│  │ 4. 记录结果到执行历史                    │ │
│  │ 5. 更新 TaskTodoManager                  │ │
│  │ 6. 判断是否继续循环                      │ │
│  └─────────┬───────────────────┬───────────┘ │
│            │                   │              │
│    继续循环 ↗                   ↘ 终止条件     │
│                                              │
│  终止条件：                                   │
│  - LLM 返回 done 动作                        │
│  - 连续失败 2 次                              │
│  - 达到最大迭代次数                           │
└──────────────────────────────────────────────┘

LLM 可选择的动作类型：

动作	执行节点	说明
`tool_call`	`ToolCallNode`	调用外部工具（HTTP API）
`mcp_call`	`McpCallNode`	调用 MCP 远程工具
`rag`	`RagNode`	知识库检索
`skill_call`	`SkillCallNode`	调用预定义技能
`sub_agent_call`	`SubAgentCallNode`	调用子 Agent
`llm_generate`	`LlmGenerateNode`	纯文本生成
`done`	—	任务完成，退出循环

核心原则：

单步原子化：每次迭代仅执行一个主动作
update_todo_list 是辅助操作，不计入迭代次数
禁止重复执行已成功的动作
连续失败 2 次必须终止循环

3.9 TaskTodoManager — 任务进度管理

纯 POJO，管理 Agent 在 Agentic Loop 中的任务清单。LLM 通过 update_todo_list 动作更新清单内容，用于追踪多步骤任务的完成进度。

TaskTodoManager
├── 最多 20 个任务项
├── 三种状态: pending → in_progress → completed
├── 同一时间仅允许 1 个 in_progress
├── update(): 全量更新清单
├── render(): 渲染为可读文本，注入到 LLM Prompt 中
└── toSerializable() / fromSerializable(): 状态序列化

4. SSE 连接与消息处理的时序

对于推荐的异步流式模式，客户端需要两步操作：

客户端                    服务端
  │
  │──── GET /sse/connect ──────────────→│
  │                                     │ 1. 创建 SseEmitter
  │                                     │ 2. 注册 SseResponseChannel
  │                                     │ 3. 启动心跳（30s）
  │←── SSE 流建立 ─────────────────────│
  │                                     │
  │──── POST /chat ────────────────────→│
  │                                     │ 4. 构建 AgentMessage
  │                                     │ 5. 提交到 MessageQueue
  │←── { messageId } ─────────────────│
  │                                     │
  │     [Dispatcher 取出消息]            │
  │     [Worker 执行 Agent 推理]         │
  │                                     │
  │←── SSE: node_start ───────────────│
  │←── SSE: thinking_stage_start ─────│
  │←── SSE: token (批量, 200ms) ──────│
  │←── SSE: token ────────────────────│
  │←── SSE: rag_start ────────────────│
  │←── SSE: rag_complete ─────────────│
  │←── SSE: references ───────────────│
  │←── SSE: token ────────────────────│
  │←── SSE: node_complete ────────────│
  │←── SSE: complete ─────────────────│
  │                                     │
  │──── GET /disconnect ───────────────→│
  │                                     │ 6. 注销通道
  │                                     │ 7. 关闭连接 + 心跳
  │                                     │ 8. 清理 session 锁

5. 第三方渠道接入（以 QQ Bot 为例）

QQ 用户 @Bot 发消息
       │
       ▼
QqBotTest.onAtMessageCreate(message)
       │
       │  agentChatService.sendMessage(agentId, sessionId, content, userId)
       ▼
AgentMessage 入队 → Dispatcher 消费
       │
       │  因为通道是 CallbackResponseChannel
       │  所以走同步模式: autonomousPlanningSync()
       ▼
Agent 推理完成，返回 response 字符串
       │
       │  callbackChannel.deliverResponse(messageId, response)
       ▼
回调执行: replyToQq(messageId, response)
       │
       │  调用 QQ Bot API 回复消息
       ▼
QQ 用户收到回复

接入新渠道只需三步：

创建会话：agentChatService.createSystemSession(agentId)
注册回调通道：channelRegistry.register(sessionId, new CallbackResponseChannel(...))
发送消息：agentChatService.sendMessage(...)

6. 关键设计决策

为什么用全局单队列而不是 per-session 队列？

旧版架构为每个 session 创建独立的队列和消费线程，存在以下问题：

线程数随 session 数线性增长，高并发时资源消耗大
每个消费线程大部分时间空闲（Agent 推理耗时长，一个 session 同时只处理 1 条消息）

新版采用全局单队列 + 共享 worker 线程池，通过 per-session Semaphore 保证顺序，线程复用率高。

为什么把 AgentMessageQueue 抽象为接口？

虽然单实例部署只需要 MemoryAgentMessageQueue（一个 LinkedBlockingQueue），但抽象为接口后：

切换到 Redis/RocketMQ 只需修改配置 ai.agent.mq.type=redis
分布式部署时多个实例共享消息队列
不影响 Dispatcher 和 Agent 的任何代码

为什么 StreamHandler 用回调而不是返回值？

Agent 推理是长时间运行的过程（可能几十秒到几分钟），采用回调模式可以：

实时推送中间状态（thinking、节点进度、RAG 检索）
客户端立即看到响应，不用等全部完成
用 NullStreamHandler（Null Object 模式）统一处理不需要流式的场景，避免 null 检查

前端chat新的api使用方式，以前是一步走，现在是两步走

// 1. 先创建会话 const session = await fetch(/v1/chatbot/${agentId}/sessions, { method: 'POST' });

// 2. 建立 SSE 连接 const es = new EventSource(/v1/chatbot/${agentId}/sessions/${sessionId}/connect);

es.addEventListener('connected', (e) => { console.log('连接已建立', JSON.parse(e.data)); });

es.addEventListener('token', (e) => { // 处理流式 token });

es.addEventListener('error', (e) => { // 处理错误 });

// 3. 发送消息（sse=false） const result = await fetch(/v1/chatbot/${agentId}/sessions/${sessionId}/chat?sse=false, { method: 'POST', body: JSON.stringify({ message: '你好' }) }); // result.data.messageId — 消息已入队，SSE 流将通过 EventSource 推送结果

聊天机器人智能体系统架构 ​

1. 一条消息的完整旅程 ​

1.1 详细步骤分解 ​

2. 核心模块总览 ​

3. 各模块详解 ​

3.1 ChatSessionManager — 会话管理器 ​

3.2 SseConnectionManager — SSE 连接管理器 ​

3.3 AgentMessageQueue — 消息队列 ​

3.4 CentralizedAgentMessageDispatcher — 集中式消息调度器 ​

3.5 AgentResponseChannel — 响应通道 ​

3.6 StreamHandler — 流式事件推送 ​

3.7 AutoPlanningAgent — 自主规划 Agent ​

3.8 Agentic Loop — 核心执行循环 ​

3.9 TaskTodoManager — 任务进度管理 ​

4. SSE 连接与消息处理的时序 ​

5. 第三方渠道接入（以 QQ Bot 为例） ​

6. 关键设计决策 ​

为什么用全局单队列而不是 per-session 队列？ ​

为什么把 AgentMessageQueue 抽象为接口？ ​

为什么 StreamHandler 用回调而不是返回值？ ​

聊天机器人智能体系统架构

1. 一条消息的完整旅程

1.1 详细步骤分解

2. 核心模块总览

3. 各模块详解

3.1 ChatSessionManager — 会话管理器

3.2 SseConnectionManager — SSE 连接管理器

3.3 AgentMessageQueue — 消息队列

3.4 CentralizedAgentMessageDispatcher — 集中式消息调度器

3.5 AgentResponseChannel — 响应通道

3.6 StreamHandler — 流式事件推送

3.7 AutoPlanningAgent — 自主规划 Agent

3.8 Agentic Loop — 核心执行循环

3.9 TaskTodoManager — 任务进度管理

4. SSE 连接与消息处理的时序

5. 第三方渠道接入（以 QQ Bot 为例）

6. 关键设计决策

为什么用全局单队列而不是 per-session 队列？

为什么把 AgentMessageQueue 抽象为接口？

为什么 StreamHandler 用回调而不是返回值？