ooodc 6756a3d0ae feat: 添加 OutboundDispatcher 模块，重构消息分发逻辑，优化渠道消息处理

Co-authored-by: Copilot <copilot@github.com>

2026-04-28 14:52:33 +08:00

22 KiB

Raw Blame History

PicoBot

PicoBot 是一个用 Rust 构建的多通道 Agent 网关。它把消息接入、LLM 调用、工具执行、会话持久化、长期记忆、技能系统和计划任务整合到同一个运行时里，适合做本地 Agent、团队机器人或带长期上下文的自动化助手。

当前代码库已经实现以下核心能力：

基于 Gateway 的统一消息入口，支持 WebSocket CLI 与飞书通道
面向工具调用的 Agent 循环，支持多轮 tool calling
SQLite 持久化会话、消息、长期记忆、技能事件和调度任务
基于用户维度的长期记忆检索与写入机制
基于 SKILL.md 的项目级 / 用户级技能加载与运行时管理
定时任务系统，支持延迟、周期、绝对时间和 cron 调度
超长上下文压缩与历史摘要
持久化 Agent 配置文件注入与周期性重注入

1. 项目定位

PicoBot 的设计目标不是“只会聊天”的单进程 Bot，而是一个可持续运行的 Agent 基础设施：

消息从不同渠道进入统一总线
SessionManager 负责会话路由和运行时服务编排，AgentExecutionService 负责上下文准备、AgentLoop 执行、结果持久化和回复生成
SQLite 作为事实来源保存跨重启状态
Agent 在每轮推理时可以读取文件、执行命令、发 HTTP 请求、读写记忆、管理技能和调度任务

如果你需要一个“带长期记忆、可调用工具、可接入飞书、可计划执行任务”的机器人，这个项目的实现面基本够用。

2. 核心架构

整体链路可以概括为：

Channel 接收外部消息
MessageBus 将消息送入统一的 inbound 队列
Gateway 启动的 InboundProcessor 调用 SessionManager 定位目标 Session
AgentExecutionService 准备上下文、运行 AgentLoop、执行工具调用并收集结果
生成的 user / assistant / tool / system 消息按真实顺序写入 SQLite
OutboundDispatcher 将结果投递到目标通道

主要模块如下：

src/gateway：网关入口、HTTP 健康检查、WebSocket 控制面、Session 池、CLI 会话服务、OutboundDispatcher 与 Agent 执行编排
src/bus：消息总线队列与消息结构定义，不包含渠道投递逻辑
src/agent：AgentLoop 与上下文压缩器
src/providers：不同 LLM Provider 的统一抽象，当前支持 openai 和 anthropic
src/tools：内置工具集合
src/storage：SQLite 持久化实现
src/channels：渠道适配层，当前已有 CLI 与飞书通道
src/scheduler：数据库驱动的计划任务调度器
src/skills：技能发现、加载与运行时管理
src/client / src/cli：本地 CLI 客户端和交互命令

3. 消息机制

PicoBot 的消息不是简单的“用户文本 + 助手文本”，而是完整的会话事件流。

3.1 消息类型

系统中的 ChatMessage 至少包含以下角色：

user：用户输入
assistant：模型回复
system：系统提示、Agent Prompt、调度附加提示、历史压缩摘要
tool：工具执行结果

此外，assistant 消息还能携带：

tool_calls：本轮准备调用哪些工具
reasoning_content：模型的补充推理文本

用户消息也支持附带媒体引用，当前消息结构中已经支持文本与图片 URL 形式的多模态块，以及本地媒体文件路径引用。

3.2 消息总线

MessageBus 维护两条异步队列：

inbound：Channel -> Agent
outbound：Agent -> Dispatcher -> Channel

这样做的好处是：

渠道接入和 Agent 执行解耦
可在网关内部统一处理重试、记录、路由和扩展
后续增加新渠道时不需要改动核心 Agent 流程

3.3 工具调用消息流

当模型发起工具调用时，PicoBot 会按顺序保存：

一条 assistant 消息，记录 tool_calls
一条或多条 tool 消息，记录每个工具返回结果
最终 assistant 汇总回复

这意味着即使进程重启，系统仍然可以恢复：

当时模型调用了什么工具
工具入参和返回结果是什么
最终结论如何形成

4. 会话与上下文机制

4.1 会话标识

PicoBot 为不同渠道统一维护持久化会话：

CLI：session_id 等于 chat_id
非 CLI 渠道：session_id = channel_name:chat_id

这样可以让：

CLI 显式创建、切换和管理多个会话
飞书等渠道把同一个 chat 稳定映射到同一条持久化会话

4.2 Agent Prompt 注入

PicoBot 会在 ~/.picobot/agent/AGENT.md 维护一份持久化 Agent 画像文件。

行为规则：

如果目录不存在会自动创建
如果文件不存在会自动写入默认内容
当前活动对话为空或刚 reset 时，会先把 AGENT.md 作为第一条 system 消息注入
每经过指定数量的 user 轮次，会在下一条用户消息进入前重新注入最新 AGENT.md

4.3 上下文压缩

上下文压缩不是在每次请求前同步执行，而是在一轮消息处理完成、assistant / tool 消息已经按真实顺序落库之后，再异步安排后台压缩任务。这样可以先保证当前回复链路完成，再处理历史瘦身。

完整机制如下：

系统先对当前活动历史做一个近似 token 估算。估算规则不是调用 tokenizer，而是按“约每 4 个字符约等于 1 token，并再乘以 1.2 安全系数”计算。
当估算结果超过模型上下文窗口的 50% 时，压缩器才认为“需要压缩”。这里的上下文窗口来自 agent 对应模型配置里的 context_window_tokens；未配置时按 128000 估算。
即使超过阈值，如果当前历史里的 user turn 数量不超过保留阈值，也不会压缩。当前默认会完整保留最近 3 个 user turn。
一旦满足条件，压缩器会先按 user 消息切分 turn，再确定“旧历史”和“最近保留段”的分界点。
在旧历史里，不是所有 system 消息都会被折叠：带有 agent_prompt 和 scheduled_system_prompt 标记的 system 消息会被原样保留，避免丢失 Agent 基本设定和调度任务附加约束。
除这些必须保留的 system 消息外，分界点之前的其余消息会被整理成一段待摘要 transcript。 transcript 会保留角色信息；tool 消息还会带上工具名，方便摘要时保留关键操作和结果。
如果 transcript 本身已经过长，会先按 context_summary_max_chars 截断，再交给模型总结。
摘要调用和主对话使用同一个 provider，提示词明确要求保留：用户请求、关键标识符、文件路径、URL、工具调用、命令、结果、错误、决策、偏好和当前任务状态。如果摘要调用失败，会退化为直接截断 transcript，而不会中断主流程。
摘要结果会被包装成一条新的 system 消息，并打上 SYSTEM_CONTEXT_HISTORY_COMPACTION 标记，内容前缀为 [Compressed History]。
后台提交阶段不会直接修改旧消息，而是向消息表尾部追加一段“新的活动段”：依次写入保留的关键 system 消息、压缩摘要消息、最近保留的消息，以及在压缩快照之后新产生的 delta 消息。
提交成功后，sessions.reset_cutoff_seq 会被推进到压缩前的最大 seq。这样旧消息仍然留在数据库里用于审计或全量导出，但默认恢复到运行时上下文时，只会加载新的活动段。
为避免并发覆盖，压缩提交前会检查快照是否过期：如果 reset_cutoff_seq 已变化，或者压缩期间又有更新导致快照不再匹配，本次压缩会跳过，不会覆盖较新的上下文。
压缩提交成功后，Session 会重新加载当前 chat 的活动历史，后续轮次看到的就是“关键 system 消息 + 压缩摘要 + 最近若干完整 turn”的新上下文。

这套机制的目标不是简单删历史，而是把“远端历史变成可恢复摘要”，同时保证：

Agent Prompt 和调度提示不丢
最近对话细节仍然完整
工具调用形成的重要事实被保留下来
数据库里仍有完整原始消息流水

5. 持久化与存储机制

PicoBot 目前使用 SQLite 作为主要持久化后端，默认数据库路径为：

~/.picobot/storage/sessions.db

数据库会保存以下核心实体：

sessions：会话元数据
messages：消息流水
memories：长期记忆
skill_events：技能发现与使用事件
scheduler_jobs：计划任务定义与运行状态

设计原则是：

内存对象负责运行时性能与临时状态
SQLite 负责跨重启、跨进程恢复

详细表结构和持久化说明可参考 docs/PERSISTENCE.md。

6. 长期记忆机制

长期记忆是 PicoBot 和普通聊天机器人最不一样的部分之一。

6.1 记忆存储模型

每条记忆至少包含这些维度：

scope_kind：当前实现为 user
scope_key：由 channel_name:sender_id 组成
namespace：记忆命名空间，例如 profile、preferences、tasks
memory_key：命名空间内的键
content：记忆正文
source_session_id / source_message_id / source_message_seq：来源追踪信息

这意味着记忆默认按“渠道中的某个用户”隔离，而不是按单个 chat 隔离。

6.2 记忆读取与写入工具

内置了两类记忆工具：

memory_search：只读检索，支持 search / get / list
memory_manage：写操作，支持 put / update / delete

推荐模式是：

先用 memory_search 回忆用户长期信息
只在确实识别到高价值长期信息时，再用 memory_manage 写入或更新

6.3 记忆机制的时机

长期记忆不会像会话历史那样在每轮请求开始时自动拼进上下文；它的使用时机是显式、按需、分阶段发生的。

完整时序如下：

用户新消息到达后，系统先恢复当前 chat 的活动历史、system 消息和 Agent Prompt。这一步恢复的是会话上下文，不会自动把 memories 表里的所有长期记忆直接注入本轮请求。
AgentLoop 开始推理后，如果模型判断当前问题依赖用户历史，例如语言偏好、长期任务、身份事实、历史决策或稳定工作方式，就应主动调用 memory_search。因此“读记忆”的时机通常发生在本轮推理前段或中段，而不是网关层强制预加载。
memory_search 的作用域不是当前 chat，而是当前 channel_name:sender_id 对应的用户范围。这意味着同一个用户在同一渠道下的不同 chat，可以共享长期记忆。
当模型已经确认某条信息值得跨会话保留时，才调用 memory_manage。因此“写记忆”的时机通常发生在本轮推理中后段：先理解当前问题，再决定是否沉淀为长期记忆。
memory_manage 写入时会记录当前工具上下文中的 session_id、message_id、message_seq、channel_name 和 chat_id。所以一旦写入成功，这条记忆会立刻持久化到 SQLite，并能追溯来源。
已写入的记忆不会自动回流到“当前这一轮”已经构建好的提示上下文。它主要影响后续轮次中模型再次调用 memory_search 的结果，以及后续记忆维护后生成的托管摘要。

可以把两类上下文区分为：

会话历史：默认参与当前 chat 的直接推理
长期记忆：只有模型主动检索时才进入推理过程

6.4 记忆维护机制

PicoBot 不只是“把记忆存进去”，还内置了记忆整理逻辑。

系统会把长期记忆粗分为几类：

用户事实
偏好
行为模式
其他信息

随后通过记忆维护流程：

去重低价值记忆
合并重复或可归并的记忆项
识别冲突信息
生成一份受控的“用户记忆摘要” Markdown

维护时机和生效方式是：

SessionManager 会先按 scope_key 取出该用户范围下的全部长期记忆。
然后把候选记忆整理成结构化计划，交给模型做归纳和裁剪。
模型输出会包含用户事实、偏好、行为模式、需要合并的条目、低价值条目，以及一份 managed_markdown 摘要。
系统据此回写 memories 表：合并项会重新 put_memory，低价值项会 delete_memory，自动整理产生的记录会标记 source_type = memory_maintenance。
所有 scope 的 managed_markdown 会再被合并，并写入 ~/.picobot/agent/AGENT.md 的托管记忆区块。

这意味着记忆维护不是为了立刻改变当前一轮回复，而是为了影响后续新对话、后续 reset 后的新活动段，以及未来每次重新注入的 Agent Prompt。

6.5 定时记忆维护

Scheduler 默认内置一个任务：

builtin.memory_maintenance_daily
每天本地时间 03:00 运行

这个任务以 internal_event 的方式触发 memory_maintenance 事件。触发后，系统会遍历所有已有用户 scope，逐个执行记忆维护，并在全部处理完成后统一更新 AGENT.md 中的“用户记忆摘要”托管区块。

7. 技能机制

PicoBot 支持基于文件系统的技能系统，用来给 Agent 注入某一类任务的专门说明。

7.1 技能发现位置

项目级技能：.picobot/skills/*/SKILL.md
用户级技能：~/.picobot/skills/*/SKILL.md

7.2 最小 SKILL.md 格式

---
description: 用于总结 Rust 项目架构
---

这里写详细说明。

约束：

frontmatter 中必须有 description
name 可省略，省略时使用目录名
非法技能文件会被跳过，并记录 warning 日志

7.3 技能注入方式

运行时会发生两件事：

AgentLoop 动态注入“技能索引”系统提示，列出当前可用技能的名称和描述
模型可以通过工具读取具体技能内容并按技能说明执行任务

7.4 技能管理工具

内置工具：

skill_list：只读列出技能
skill_manage：运行时创建、更新、删除、读取和重载技能

skill_manage 支持的 action：

list
get
create
update
delete
reload

skills 配置示例：

{
	"skills": {
		"enabled": true,
		"sources": ["project", "user"],
		"max_index_chars": 4000,
		"max_listed_skills": 32
	}
}

8. 工具机制

PicoBot 的 Agent 是围绕工具调用构建的。当前默认注册的工具包括：

calculator：简单数学计算
file_read：读取文件
file_write：写文件
file_edit：编辑文件
memory_search：读取长期记忆
memory_manage：写入 / 更新 / 删除长期记忆
scheduler_manage：管理调度任务
skill_list：列出技能
skill_manage：管理技能
bash：执行 shell 命令
http_request：发起 HTTP 请求
web_fetch：抓取网页正文

其中：

文件工具适合做代码库和文档操作
记忆工具适合维持长期用户画像
scheduler_manage 允许 Agent 自主创建后续计划任务
bash / http_request / web_fetch 让 Agent 具备更强的外部交互能力

9. 调度器机制

PicoBot 带有一个基于 SQLite 的调度器，而不是纯内存或 JSON 文件驱动的任务系统。

9.1 支持的调度类型

delay：延迟执行一次
interval：固定间隔执行
at：某个绝对时间执行一次
cron：cron 表达式调度

9.2 支持的任务类型

internal_event：内部事件
outbound_message：直接向目标通道发消息
agent_task：构造一次合成用户输入，复用完整 Agent 流程执行
silent_agent_task：在独立后台会话中执行 Agent 流程，成功不推送，失败回主 chat 通知

agent_task 会复用正常链路中的这些能力：

历史加载
Agent Prompt 注入
工具调用
会话持久化
渠道消息下发

silent_agent_task 和 agent_task 使用同一套 Agent 执行能力，但路由语义不同：

target.chat_id 仍表示用户通知目标
target.session_chat_id 表示后台任务实际写入的会话；如果省略，会稳定派生为 scheduler/{job_id}
成功执行后不会向用户发送正常结果
执行失败时会向主 chat 发送一条失败通知，便于用户感知异常
后台任务的历史、压缩和会话内上下文会留在独立会话中，不污染主会话

9.3 运行时管理

通过 scheduler_manage 可以进行：

list
get
put
delete
pause
resume

任务定义和状态都会写进 SQLite，所以重启后仍可恢复。

调度器配置示例：

{
	"scheduler": {
		"enabled": true,
		"tick_resolution_ms": 1000,
		"misfire_policy": "skip",
		"jobs": [
			{
				"id": "session.cleanup",
				"kind": "internal_event",
				"schedule": {
					"type": "interval",
					"seconds": 300
				},
				"payload": {
					"event": "session_cleanup"
				}
			},
			{
				"id": "agent.daily_summary",
				"kind": "agent_task",
				"schedule": {
					"type": "cron",
					"expression": "30 18 * * *"
				},
				"target": {
					"channel": "feishu",
					"chat_id": "oc_xxx"
				},
				"payload": {
					"prompt": "请总结今天的项目进展，并列出明天的优先事项",
					"agent": "default",
					"fresh_session": true,
					"system_prompt": "你是日报助手，输出时先给摘要，再给待办。",
					"sender_id": "scheduler-daily-summary",
					"metadata": {
						"job_type": "daily_summary",
						"source": "scheduler"
					}
				}
				},
				{
					"id": "agent.weekly_report.background",
					"kind": "silent_agent_task",
					"schedule": {
						"type": "cron",
						"expression": "0 8 * * 1"
					},
					"target": {
						"channel": "feishu",
						"chat_id": "oc_xxx",
						"session_chat_id": "scheduler/agent.weekly_report.background"
					},
					"payload": {
						"prompt": "请后台整理上周项目进展，输出结构化周报草稿，重点标出风险、阻塞项和下周优先级。",
						"agent": "default",
						"fresh_session": false,
						"system_prompt": "你是周报助手，只在后台整理，不要面向用户寒暄。",
						"sender_id": "scheduler-weekly-report",
						"metadata": {
							"job_type": "weekly_report_background",
							"source": "scheduler"
						}
					}
				}
			}
		]
	}
}

推荐场景：

agent_task：用户需要直接收到结果，例如日报提醒、定时播报、定时外发通知
silent_agent_task：任务需要长期积累独立上下文或后台整理材料，但不应污染主会话，例如周报草稿整理、周期性资料汇总、后台分析任务

10. 渠道与运行方式

10.1 当前支持的通道

WebSocket CLI 客户端
飞书通道

10.2 Gateway 接口

网关当前暴露：

/health：健康检查
/ws：CLI 客户端连接入口

10.3 CLI 使用方式

程序提供两个主命令：

cargo run -- gateway
cargo run -- agent

含义：

gateway：启动网关服务
agent：以 WebSocket 方式连接网关，进入本地 CLI 会话

CLI 中已实现的交互命令包括：

/new [title]
/reset
/sessions
/use
/rename

22 KiB Raw Blame History Unescape Escape