From 7a3058d8b54e1dd543ddd008f857261b6d8f2d06 Mon Sep 17 00:00:00 2001 From: xiaoxixi Date: Fri, 17 Jul 2026 17:45:22 +0800 Subject: [PATCH] docs: finalize streaming turn architecture --- AGENTS.md | 16 ++++-- README.md | 23 +++++--- docs/ARCHITECTURE.md | 49 +++++++++++++--- docs/STREAMING_TURN_DESIGN.md | 56 +++++++------------ .../about-picobot/references/architecture.md | 24 ++++++-- .../skills/about-picobot/references/config.md | 4 ++ .../about-picobot/references/db-schema.md | 12 ++-- src/agent/agent_loop.rs | 2 +- src/agent/context_compressor.rs | 2 +- src/providers/anthropic.rs | 14 ++++- src/providers/mod.rs | 4 +- src/providers/openai.rs | 9 ++- src/providers/stream.rs | 43 +++++++++++--- 13 files changed, 175 insertions(+), 83 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 0366a39..44eab25 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -46,6 +46,8 @@ Channel → MessageBus.inbound → Gateway processor → SessionManager → per- ↑ │ └── Channel ← OutboundDispatcher ← per-conversation lane ← MessageBus.outbound +AgentLoop → TurnEvent → Session TurnController → latest TurnSnapshot → DeliveryCoordinator → per-turn TurnSink → Channel + WebSocket/Channel → MessageBus.control → Gateway processor → SessionManager (dialog operations) Scheduler → SessionManager scheduled execution → AgentLoop → Scheduler delivery policy → SessionManager/MessageBus ``` @@ -58,9 +60,10 @@ Scheduler → SessionManager scheduled execution → AgentLoop → Scheduler del | `client` | TUI rendering, WebSocket client for CLI chat | `App`, `run()` | | `channels` | External integrations (Feishu, CLI chat) | `ChannelManager`, `Channel` trait | | `bus` | Bounded async queues and ordered outbound delivery lanes | `MessageBus`, `OutboundDispatcher`, `InboundMessage`, `OutboundMessage`, `ControlMessage` | -| `session` | Conversation lifecycle, dialog operations, per-session serialization, persistence coordination | `SessionManager`, `Session` | -| `agent` | LLM call loop, tool execution, context compression | `AgentLoop` | -| `providers` | LLM API clients (OpenAI-compatible, Anthropic) | `LLMProvider` trait, factory `create_provider()` | +| `session` | Conversation lifecycle, dialog operations, per-session serialization, Turn state, persistence coordination | `SessionManager`, `Session`, `TurnController` | +| `agent` | LLM call loop, tool execution, context compression, semantic Turn events | `AgentLoop`, `TurnEvent` | +| `providers` | Native LLM streams normalized into text/reasoning/tool/usage chunks | `LLMProvider`, `ProviderChunk`, `create_provider()` | +| `delivery` | Snapshot projection, latest-wins throttling, terminal retry, per-turn sink lifecycle | `DeliveryCoordinator`, `TurnDeliveryService`, `PresentationPolicy` | | `tools` | Agent tools (bash, file ops, http, web, get_skill) | `ToolRegistry`, `Tool` trait | | `skills` | Skills loading, management, and prompt building | `SkillsLoader`, `Skill` | | `storage` | SQLite persistence for sessions and messages | `Storage`, `SessionMeta`, `MessageMeta` | @@ -74,9 +77,11 @@ Scheduler → SessionManager scheduled execution → AgentLoop → Scheduler del ### Functional Boundaries -- **Channels** only send/receive messages via `MessageBus`; they know nothing about sessions or LLM +- **Channels** publish inbound messages through `MessageBus`; outbound writes arrive through `OutboundDispatcher` or a per-turn `TurnSink`. They know nothing about sessions or LLM - **MessageBus** owns bounded inbound/outbound/control queues; outbound routing and retries belong to `OutboundDispatcher`, not the queue - **SessionManager** owns session state, dialog operations, context construction, per-session work queues, and persistence coordination +- **TurnController** is the only owner of active Turn state; snapshots are complete latest-wins values, not token queues, and `Completed` is published only after atomic persistence succeeds +- **DeliveryCoordinator** projects active Turn snapshots without mutating history; it owns `TurnSink` lifecycle but no platform message IDs, which remain private to each sink - **WorkManager** owns the single active plan per session, item state transitions, plan versions, and plan-change events; plans are optional and absent from ordinary chat context - **Scheduler** supports legacy direct-delivery jobs and managed `task`/`monitor` jobs; managed agents cannot call `send_message`, and `on_alert` suppresses only healthy informational results - **AgentLoop** is stateless across turns; it receives prepared history, calls LLM providers, executes tools, and returns one result @@ -86,12 +91,15 @@ Scheduler → SessionManager scheduled execution → AgentLoop → Scheduler del - **WebUI/TUI file transfer** streams bytes over authenticated HTTP and sends only short-lived upload IDs/attachment metadata over WebSocket; messages persist local media paths without guaranteeing later availability, and client responses must never expose those paths - **WebUI authentication** protects every management API and `/ws`; only static pairing assets, public health/status, and pairing submission may bypass device auth. Pair-code issuance requires both a real loopback peer and the filesystem-held admin token; never put bearer tokens in URLs or logs - **Providers** are pure HTTP clients; no bus/session/channel awareness +- **Provider reasoning state** is private replay data: persist it, replay it only to the matching provider, and never expose it to clients, channels, or logs - **Tools** are executed by `AgentLoop`; they receive raw arguments and normally return text. Tools that produce model-consumable media use the structured `execute_with_media` side channel; model capability checks and provider content-block serialization stay outside tools ### Concurrency and Lifecycle Invariants - Messages in one session are processed serially through a bounded queue; different sessions may run concurrently - Outbound messages are ordered per `(channel, chat_id)`; a slow destination must not block unrelated destinations +- Active Turn delivery and ordinary outbound delivery share the same per-`(channel, chat_id)` write lock; never enqueue token deltas into MessageBus +- Slow Turn consumers may skip intermediate snapshots but must receive an explicit bounded terminal delivery; shutdown must call sink abort so platform cleanup remains possible - Never hold a Session mutex across model, network, or database I/O unless a documented invariant requires it - Slow work derived from session state must validate `worker_generation`/`state_version` before committing results - Related durable mutations use Storage transaction APIs; persistence failure must not leave silent memory/database divergence diff --git a/README.md b/README.md index 4e8d3a3..5242aff 100644 --- a/README.md +++ b/README.md @@ -9,9 +9,10 @@ PicoBot 是一个用 Rust 编写的个人 AI 助手运行时。它在本地启 ## 适合做什么 - 在终端里和本地 AI 助手持续对话。 -- 在浏览器中聊天，并查看日志、任务和记忆，修改运行配置与助手档案。 +- 在 TUI 或浏览器中实时查看正文、思考过程和工具执行状态，并在完成后收敛到持久化历史。 +- 在浏览器中查看日志、任务和记忆，修改运行配置与助手档案。 - 复杂任务可创建 session 级 Todo 计划，把不同子项并行委托给多个子 Agent；聊天页侧栏实时显示进度。 -- 将同一套 Agent 能力接入飞书/Lark。 +- 将同一套 Agent 能力接入飞书/Lark，并可选用单张卡片实时更新回复。 - 让 Agent 使用本地文件、Shell、搜索、HTTP、浏览器、MCP 工具完成任务。 - 把长期偏好、事实和历史摘要存成可检索记忆。 - 用 Cron 定时执行任务，并把结果发回目标渠道。 @@ -111,7 +112,7 @@ picobot pair WebUI 随二进制嵌入，不需要 Node.js、npm 或单独部署静态文件，提供： -- 在线聊天、会话创建/切换、历史回放、Markdown 消息、可折叠工具调用卡片，以及基于 Gateway 实时命令清单的 `/` 斜杠命令补全。 +- 在线聊天、会话创建/切换、历史回放、流式 Markdown、独立思考区、实时工具状态、可折叠历史工具调用卡片，以及基于 Gateway 实时命令清单的 `/` 斜杠命令补全。 - 文件选择、拖放和剪贴板图片上传；消息中的附件可预览或下载。附件按服务端路径引用，原文件移动或删除后历史附件可能不可用。 - 可持久化的浅色/深色主题，首次访问时跟随系统偏好。 - Cron 定时任务、最近运行记录和后台子任务状态。 @@ -157,7 +158,7 @@ picobot service uninstall unit 位于 `~/.config/systemd/user/picobot.service`，以执行 `service install` 时的当前目录作为初始工作目录。服务异常退出时由 systemd 自动重启；`stop` 和 `restart` 会通过 SIGTERM 触发 Gateway 的有界优雅关停。 -TUI 会把一个随机客户端标识保存到 `~/.picobot/tui_client_id`，因此关闭并重新打开客户端后会恢复同一组 dialog 和最近使用的会话。界面支持历史回放、会话列表与归档筛选、命令补全、Unicode/中文编辑、括号粘贴、多行输入和文件传输。 +TUI 会把一个随机客户端标识保存到 `~/.picobot/tui_client_id`，因此关闭并重新打开客户端后会恢复同一组 dialog 和最近使用的会话。界面支持流式正文、独立思考与工具状态、历史回放、会话列表与归档筛选、命令补全、Unicode/中文编辑、括号粘贴、多行输入和文件传输。 常用快捷键： @@ -177,11 +178,11 @@ WebUI/TUI 上传文件默认保存到 `~/.picobot/media/cli_chat`，单文件上 ## 运行时数据流 -用户消息进入 PicoBot 后，会被转换为统一的 inbound message，经由 MessageBus 交给 SessionManager。SessionManager 选择当前 dialog、组装上下文、调用 AgentLoop；AgentLoop 调用模型和工具，最终响应通过 outbound bus 回到原渠道。 +用户消息进入 PicoBot 后，会被转换为统一的 inbound message，经由 MessageBus 交给 SessionManager。SessionManager 选择当前 dialog、组装上下文并创建活动 Turn；AgentLoop 消费 Provider 原生流、执行工具并发出结构化事件，TurnController 将它们归约成可丢中间帧的完整快照。DeliveryCoordinator 把快照投影给 TUI、WebUI 或 Channel，最终消息在 SQLite 原子提交成功后才进入 `Completed`。 详细时序和失败语义见 [架构文档：消息与控制数据流](docs/ARCHITECTURE.md#4-消息与控制数据流)。 -同一 session 的普通消息由专属有界队列串行处理，不同 session 可以并发；出站消息按 `(channel, chat_id)` 分 lane 保序，慢渠道不会阻塞其他目标。Gateway 的长生命周期任务统一由 `TaskSupervisor` 取消和限时回收。 +同一 session 的普通消息由专属有界队列串行处理，不同 session 可以并发；活动 Turn 使用 latest-wins 快照，慢展示端只跳过中间状态，不反压模型。普通出站消息按 `(channel, chat_id)` 分 lane 保序，两条投递路径共享目标写锁。Gateway 的长生命周期任务统一由 `TaskSupervisor` 取消和限时回收。 核心边界： @@ -191,7 +192,8 @@ WebUI/TUI 上传文件默认保存到 `~/.picobot/media/cli_chat`，单文件上 | `bus` | 异步消息队列，承载 inbound、outbound、control 三类消息 | | `session` | 管理会话生命周期、dialog 操作、上下文、记忆召回、压缩和持久化 | | `agent` | 执行无状态 LLM/tool 循环，处理模型响应和工具调用 | -| `providers` | OpenAI 兼容接口和 Anthropic Messages API 客户端 | +| `providers` | OpenAI 兼容接口和 Anthropic Messages API 的原生流解析与回放 | +| `delivery` | 活动 Turn 的展示过滤、latest-wins 节流、终态投递与 TurnSink 生命周期 | | `tools` | Agent 可调用工具集合 | | `storage` | SQLite schema、CRUD、消息和任务持久化 | | `scheduler` | 领取定时任务，执行普通/巡检 Agent，并按投递策略记录或发送结果 | @@ -209,6 +211,8 @@ WebUI/TUI 上传文件默认保存到 `~/.picobot/media/cli_chat`，单文件上 | `cli_chat` | Ratatui 终端客户端，通过 WebSocket 连接 Gateway | | `feishu` | 飞书/Lark 消息、反应、文件上传下载和媒体引用 | +飞书默认只发送终态结果。设置 `channels.feishu.live_updates=true` 后会创建一张卡片并持续编辑；`live_update_interval_ms` 默认 500ms，运行时限制在 250–5000ms。外部渠道始终不会收到模型 reasoning。 + ### 会话 Session ID 使用三段式： @@ -310,6 +314,8 @@ Skill 是包含 `SKILL.md` 的目录。加载优先级从高到低： | `memory.timeline_retention_days` | `90` | | `mcp.tool_timeout_secs` | `180` | | `browser.enabled` | `false` | +| `channels.feishu.live_updates` | `false` | +| `channels.feishu.live_update_interval_ms` | `500` | 更完整的配置字段说明见 [resources/skills/about-picobot/references/config.md](resources/skills/about-picobot/references/config.md)。 @@ -341,7 +347,7 @@ Inbound 消息类型： | `get_slash_commands` | 无 | | `ping` | 无 | -Outbound 消息类型包括 `assistant_response`、`error`、`session_established`、`session_created`、`session_list`、`session_loaded`、`session_history`、`session_renamed`、`session_archived`、`session_deleted`、`history_cleared`、`slash_commands_list`、`pong`、`command_executed` 和 `system_notification`。其中异步 `assistant_response` / `system_notification` 可携带 `session_id`，客户端应避免把迟到结果显示到其他 dialog。 +Outbound 消息类型包括活动 Turn 使用的 `turn_updated`，以及 `assistant_response`、`error`、`session_established`、`session_created`、`session_list`、`session_loaded`、`session_history`、`session_renamed`、`session_archived`、`session_deleted`、`history_cleared`、`slash_commands_list`、`pong`、`command_executed` 和 `system_notification`。`turn_updated` 每次携带完整快照和单调 revision，客户端只替换当前 session 的活动 Turn；`assistant_response` 保留给独立完整消息。历史消息可包含 reasoning、turn/iteration、completion status 和结构化工具元数据，但不会暴露 Provider 私有回放状态。 ## 测试 @@ -370,6 +376,7 @@ src/ channels/ CLI chat 和飞书/Lark 集成 client/ Ratatui 终端 UI config/ 配置加载、环境变量替换、路径展开 + delivery/ 活动 Turn 快照投影、节流与 TurnSink 生命周期 gateway/ Axum HTTP/WebSocket server 和 GatewayState 装配 mcp/ MCP 客户端连接和工具包装 memory/ 记忆管理和记忆类型 diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 595fbb8..47ae2f4 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -2,7 +2,7 @@ 本文档描述 PicoBot 当前实现的运行时边界、数据流、并发模型和演进约束。它面向维护者和后续参与改进的 Agent，是代码架构的主入口；行为细节仍以代码和测试为最终依据。 -拟议中的流式模型输出、reasoning 展示、活动 Turn 快照和 Channel 实时投递架构见 [STREAMING_TURN_DESIGN.md](STREAMING_TURN_DESIGN.md)。该文档是尚未实现的设计，不代表当前运行时行为。 +流式模型输出、reasoning 展示、活动 Turn 快照和 Channel 实时投递的详细设计与取舍见 [STREAMING_TURN_DESIGN.md](STREAMING_TURN_DESIGN.md)。 ## 1. 设计目标 @@ -43,6 +43,8 @@ flowchart LR Agent --> Providers[LLM providers] Agent --> Tools[ToolRegistry / MCP] Sessions <--> Storage[(SQLite)] + Sessions -->|TurnSnapshot| Delivery[DeliveryCoordinator] + Delivery -->|TurnSink| Channels Sessions -->|OutboundMessage| Bus Scheduler[Scheduler] --> Sessions Bus --> Dispatcher[OutboundDispatcher] @@ -61,8 +63,9 @@ flowchart LR | `channels` | 外部协议适配、权限检查、媒体收发 | 会话选择、LLM 调用 | | `bus` | 三条有界异步队列与出站投递协调 | 会话路由、业务状态 | | `session` | dialog 路由、会话状态、串行工作队列、上下文和持久化协调 | 外部渠道协议 | -| `agent` | 单次无状态模型/工具循环、上下文压缩、子 Agent | 持有 dialog 生命周期 | -| `providers` | 把统一请求映射到模型 API | Session、Bus 或 Channel 感知 | +| `agent` | 单次无状态模型/工具循环、上下文压缩、子 Agent、Turn 语义事件 | 持有 dialog 生命周期 | +| `providers` | 把统一请求映射为原生模型流，并归一化正文、reasoning、工具和 usage | Session、Bus 或 Channel 感知 | +| `delivery` | 活动 Turn 快照投影、latest-wins 节流、终态重试和 TurnSink 生命周期 | Provider 协议、会话历史、平台 API 细节 | | `tools` / `mcp` | 工具定义、注册和执行适配 | 隐式修改会话路由 | | `storage` | SQLite schema、迁移、原子 CRUD | 运行时调度策略 | | `memory` | Knowledge/Timeline 的存取与召回 | 直接驱动消息发送 | @@ -87,7 +90,9 @@ sequenceDiagram participant G as Message processor participant S as SessionManager participant W as Per-session worker - participant A as AgentLoop + participant A as AgentLoop / Provider + participant T as TurnController + participant L as DeliveryCoordinator participant D as OutboundDispatcher C->>B: publish InboundMessage @@ -95,9 +100,18 @@ sequenceDiagram G->>S: handle_message S->>W: try_send AgentTask S-->>G: AgentProcessing - W->>A: process(history) - A-->>W: final response - W->>B: publish OutboundMessage + W->>T: start Turn + W->>L: subscribe latest snapshots + W->>A: process_streaming(history) + A-->>T: reasoning/text/tool events + T-->>L: complete TurnSnapshot + L->>C: TurnSink update (best effort) + A-->>W: emitted messages + W->>W: atomic persistence + W->>T: Completed + T-->>L: terminal snapshot + L->>C: TurnSink finish (bounded retry) + W->>B: independent messages/fallback only B->>D: consume outbound D->>C: Channel::send ``` @@ -108,6 +122,20 @@ sequenceDiagram - 每个 session 有一条容量为 32 的队列，同一 session 串行处理，不同 session 的 worker 可并发执行。 - 队列满时明确拒绝新消息，不允许无界积压。 - Slash command 不进入 Agent 队列，由 `SessionManager` 直接执行，因此 `/stop` 等控制操作不会排在长模型调用之后。 +- Session 为每个 Agent 请求创建一个 `TurnController`。Provider 向 AgentLoop 发 delta，AgentLoop 发结构化 TurnEvent，只有 TurnController 能把事件归约为有序 block 和单调 revision 的完整快照。 +- Turn 快照经 Tokio `watch` 发布，语义为 latest-wins；慢客户端或慢渠道跳过中间状态，不反压 Provider。终态明确编码在快照中，不依赖 sender 关闭。 +- Agent 本轮消息原子持久化成功后才发布 `Completed`。取消或失败若已有可见正文，则保存为 `cancelled`/`interrupted` partial；只有 reasoning 时不创建 assistant 历史。 + +### 活动 Turn 投递 + +`DeliveryCoordinator` 与普通出站投递并列： + +- `TurnDeliveryService` 根据 Channel 创建本轮独占的 `TurnSink`；sink 私有保存远端消息 ID 和清理资源。 +- `PresentationPolicy` 在快照离开 Gateway 核心前过滤内容。TUI/WebUI 展示独立 reasoning 和详细工具状态；外部 Channel 不接收 reasoning，只接收紧凑工具状态；无人值守投递只保留正文。 +- `LivePolicy::Snapshot` 按渠道间隔发送最新运行态；`FinalOnly` 忽略运行态，只处理终态。终态绕过节流并只对明确的瞬态错误重试。 +- `cli_chat` 将同一 `turn_updated` 快照发给 TUI 和 WebUI。客户端只保留当前 session 中 revision 更新的 `active_turn`，终态随后由持久化历史校准。 +- 飞书默认 `FinalOnly`；开启 `live_updates` 后，第一个可见快照创建卡片，后续编辑同一卡片，终态编辑失败则发送完整结果兜底。reaction 清理在 finish、abort 和 Gateway shutdown 中幂等执行。 +- DeliveryCoordinator 与 OutboundDispatcher 共享 `(channel, chat_id)` 写锁，避免活动 Turn 终态与独立消息并发写入同一目标。 ### 出站投递 @@ -164,7 +192,7 @@ SessionManager 负责组装会话上下文：系统提示、Skills、召回的 K - 5 秒 busy timeout。 - schema version 迁移。 -持久化范围包括 sessions、messages、memories、task plans/items、scheduled jobs、job runs 和 background tasks。修改 schema 时应： +持久化范围包括 sessions、messages、memories、task plans/items、scheduled jobs、job runs 和 background tasks。消息保存可展示 `reasoning_content`、Provider 私有回放状态、`turn_id`、iteration 和 completion status；私有 Provider 状态不进入 WebSocket/Channel，且只允许回放给同一 Provider。Provider 诊断和失败审计只记录模型、消息数、工具数等请求摘要，不保存正文、reasoning 或签名 payload。修改 schema 时应： 1. 更新集中式 schema/迁移逻辑。 2. 保留已有数据库的升级路径。 @@ -181,7 +209,7 @@ SessionManager 负责组装会话上下文：系统提示、Skills、召回的 K ## 7. 后台任务与生命周期 -`TaskSupervisor` 是 Gateway 内部后台任务的统一所有者。message processor、outbound dispatcher、scheduler、session workers、outbound lanes、通知消费者和子 Agent 后台任务都应通过它注册。 +`TaskSupervisor` 是 Gateway 内部后台任务的统一所有者。message processor、outbound dispatcher、scheduler、session workers、Turn delivery、outbound lanes、通知消费者和子 Agent 后台任务都应通过它注册。 两种注册方式： @@ -197,6 +225,8 @@ SessionManager 负责组装会话上下文：系统提示、Skills、召回的 K WebSocket 每个客户端的 writer task 是连接局部任务，由连接 handler 自己限时回收；它不跨越连接生命周期。 +Turn delivery 使用 `spawn_graceful`。全局取消发生时先停止读取快照，再在共享目标写锁下有界调用 `TurnSink::abort`，使平台 reaction 等私有资源能在 Supervisor 宽限期内清理。 + ### WebUI 与管理 API Gateway 在 `/` 提供随二进制编译的 HTML/CSS/JavaScript，不依赖外部 CDN 或前端运行服务。前端源码位于 `webui/`，由 Svelte 5 + Vite 构建，Bits UI 提供无样式的可访问组件原语。`build.rs` 监听前端源码、锁文件和构建配置，增量地将生产资源生成到 Cargo `OUT_DIR`，Rust 再从该目录编译嵌入；前端产物不进入仓库，最终用户使用发布二进制时不需要 Node.js。浏览器聊天继续使用 `/ws` 和 `cli_chat` 渠道，因此复用现有 dialog scope、每会话串行 worker、历史持久化和出站 lane；会话历史帧保留工具调用 ID、名称、参数和工具结果角色，WebUI 在本轮完成后刷新历史并将其渲染为默认折叠的工具卡片；聊天页的 Todo 侧栏默认隐藏，按 session 保存快照和未读状态，并通过结构化 `session_plan`/`plan_updated` 帧刷新，计划变化不会写入聊天历史；斜杠命令补全通过 `get_slash_commands` 获取 Gateway 的实时命令与别名清单，不在前端重复定义；WebUI 不直接调用 Provider 或 SessionManager。 @@ -249,6 +279,7 @@ WebUI/TUI 文件字节通过受鉴权的 HTTP 接口流式传输，WebSocket 只 3. 将可重试错误表示为 `ConnectionError`/`SendError`，永久错误使用其他类型。 4. 为 start/stop 幂等性、取消建连、投递失败和媒体边界增加测试。 5. 不要从 Channel 直接调用 SessionManager 或 Provider。 +6. 若支持活动 Turn，实现 `live_policy`、`presentation_policy` 和每 Turn 一个实例的 `open_turn`；sink 必须消费完整快照而不是拼接 token，并使 finish/abort 清理幂等。 ### 新增 Tool diff --git a/docs/STREAMING_TURN_DESIGN.md b/docs/STREAMING_TURN_DESIGN.md index 0d75218..81a700f 100644 --- a/docs/STREAMING_TURN_DESIGN.md +++ b/docs/STREAMING_TURN_DESIGN.md @@ -1,12 +1,12 @@ # 流式 Turn、Reasoning 展示与 Channel 投递设计 -> 状态：拟议设计，尚未实现。 +> 状态：已实现（2026-07）。 > -> 本文定义 PicoBot 未来的流式模型输出、reasoning 展示、工具过程展示和 Channel 实时投递架构。当前运行时行为仍以 `docs/ARCHITECTURE.md` 和代码为准。实施本设计时允许重做内部、WebSocket、TUI、WebUI 和 Channel 协议；只要求 SQLite 数据库提供可靠迁移路径。 +> 本文记录 PicoBot 流式模型输出、reasoning 展示、工具过程展示和 Channel 实时投递的设计依据与架构决策。当前运行时总览见 `docs/ARCHITECTURE.md`，具体行为以代码和测试为准。 ## 1. 背景 -当前 Provider 只提供一次性 `chat()` 调用。AgentLoop 等待完整响应，将 `content`、`reasoning_content` 和 tool calls 组装为 `ChatMessage`，Session 在 AgentLoop 完成后原子持久化本轮消息，再通过 `OutboundMessage` 发送最终正文。 +改造前 Provider 只提供一次性 `chat()` 调用。AgentLoop 等待完整响应，将 `content`、`reasoning_content` 和 tool calls 组装为 `ChatMessage`，Session 在 AgentLoop 完成后原子持久化本轮消息，再通过 `OutboundMessage` 发送最终正文。 这个模型具有清晰的持久化语义，但无法表达： @@ -17,7 +17,7 @@ - 不支持编辑的 Channel 自动降级为只发送最终结果； - 取消、失败、慢消费者和投递失败时的确定行为。 -现有 `Channel::send_delta(chat_id, delta)` 没有 Turn 身份、消息身份、reasoning/text 分类、工具边界、终态和取消语义，也绕过现有出站排序机制，不适合作为后续架构基础。 +旧 `Channel::send_delta(chat_id, delta)` 没有 Turn 身份、消息身份、reasoning/text 分类、工具边界、终态和取消语义，也绕过出站排序机制，因此已被 `TurnSink` 取代。 ## 2. 参考实现结论 @@ -351,12 +351,7 @@ thinking 文本进入 `reasoning`，签名和原始 block 进入 `provider_state ### 7.4 reasoning-only -Provider 不擅自把 reasoning 提升为正文。AgentLoop 在最终轮发现只有 reasoning、没有正文且没有工具调用时，执行统一产品策略： - -- 默认生成一个明确的空正文终态，并在 UI 保留 reasoning；或 -- 后续配置允许把 reasoning 作为正文 fallback。 - -首版建议不泄漏 reasoning 为正文，由 UI 显示“模型未返回最终正文”。 +Provider 不擅自把 reasoning 提升为正文。最终轮只有 reasoning、没有正文且没有工具调用时，AgentLoop 保存空正文 assistant 消息及其 reasoning，Turn 正常进入 Completed；交互 UI 仍可显示 reasoning，但不会把它冒充最终答案。 ## 8. AgentLoop 与 TurnController @@ -565,12 +560,12 @@ WebUI 对 Running Markdown 可以按动画帧或快照频率渲染，Completed ## 13. WebSocket 协议 -不保留旧 `assistant_response` 流程。活动 Turn 使用一个统一 frame： +Agent 主 Turn 不再通过 `assistant_response` 发送最终正文，活动与终态都使用统一 frame。`assistant_response` 仅保留给不属于活动 Turn 的完整独立消息： ```json { "type": "turn_updated", - "turn": { + "snapshot": { "id": "...", "session_id": "...", "message_id": "...", @@ -604,7 +599,7 @@ Provider 完成 ### 14.1 取消 -建议首版语义： +采用以下语义： - 没有 Assistant 正文：不持久化 assistant 消息，Turn 标记 Cancelled； - 已向用户展示部分正文：持久化部分正文并标记 `completion_status=cancelled`； @@ -621,7 +616,7 @@ Provider 完成 ## 15. SQLite 迁移 -建议为 `messages` 增加： +schema v4 为 `messages` 增加： ```text turn_id TEXT NULL @@ -640,7 +635,7 @@ provider_state TEXT NULL -- JSON，Provider 私有回放状态 - Session 加载继续修复 tool-call chains； - 原子提交覆盖完整 Turn 的所有 emitted messages。 -是否新增 `turns` 表首版暂不需要。运行中 Turn 只存在内存，历史可通过 messages.turn_id 分组。如果未来要跨 Gateway 重启恢复运行态，再单独设计 durable turn lease/state。 +当前不新增 `turns` 表。运行中 Turn 只存在内存，历史可通过 messages.turn_id 分组。如果未来要跨 Gateway 重启恢复运行态，再单独设计 durable turn lease/state。 ## 16. 生命周期与并发不变量 @@ -765,32 +760,19 @@ src/protocol.rs - cancelled/interrupted 消息恢复； - 完整 Turn 原子提交失败不产生部分历史。 -## 20. 实施顺序 +## 20. 实施记录 -1. 增加数据库迁移和新的消息 reasoning/provider state 语义。 -2. 引入 ProviderChunk 和流式优先 Provider trait。 -3. 实现 OpenAI-compatible streaming 和 collect helper。 -4. 实现 TurnEvent、TurnController、TurnSnapshot 及纯单元测试。 -5. AgentLoop 接入 TurnEmitter，但暂时只使用最终结果投递。 -6. 增加 DeliveryCoordinator、LivePolicy 和 TurnSink。 -7. 重做 cli_chat WebSocket 为统一 `turn_updated` 快照。 -8. 重做 TUI/WebUI 为 `history + active_turn` 渲染模型。 -9. 实现 Anthropic streaming 和完整 provider_state 回放。 -10. 实现 FeishuTurnSink 的创建、编辑、终态和 fallback。 -11. 删除旧 `send_delta`、旧 reasoning 特例和 Channel 侧 think 清理。 -12. 更新 `ARCHITECTURE.md`、README、AGENTS.md 和运行时产品知识。 +实现按可独立验证的里程碑完成：SQLite 消息语义、Turn 状态机、OpenAI 原生流、Agent/Session 生命周期、DeliveryCoordinator、WebSocket/TUI/WebUI、Anthropic 签名回放、FeishuTurnSink，最后删除过渡适配并同步运行时文档。每个里程碑均保持非流式最终回复可用，且没有为旧增量协议保留双栈。 -每一步都必须保持非流式最终回复可用；但不为旧协议保留双栈兼容代码。 +## 21. 已采用的产品策略 -## 21. 实施前需确认的产品策略 +这些选择不改变架构，但决定默认产品行为： -这些选择不改变架构，但必须在实现前确定默认值： - -1. `/stop` 后是否持久化已展示的部分正文。本文建议持久化并标记 cancelled。 -2. TUI/WebUI reasoning 默认折叠还是隐藏。本文建议折叠。 -3. 外部 Channel reasoning 默认策略。本文建议隐藏。 -4. reasoning-only 是否允许提升为正文。本文建议不提升。 -5. provider_state 的保留期限和 LLM 原始响应审计策略。 +1. `/stop` 后持久化已展示的部分正文并标记 `cancelled`；只有 reasoning 时不创建 assistant 历史。 +2. TUI/WebUI 在独立区域展示 reasoning；WebUI 默认折叠，TUI 直接显示。 +3. 外部 Channel 默认隐藏 reasoning，工具只显示紧凑状态。 +4. reasoning-only 不提升为正文。 +5. `provider_state` 随 assistant 消息保留，用于同 Provider 精确回放；不下发客户端或 Channel，损坏时安全忽略。 ## 22. 架构验收标准 diff --git a/resources/skills/about-picobot/references/architecture.md b/resources/skills/about-picobot/references/architecture.md index 560e41b..59dc669 100644 --- a/resources/skills/about-picobot/references/architecture.md +++ b/resources/skills/about-picobot/references/architecture.md @@ -7,6 +7,8 @@ Channel → MessageBus.inbound → Gateway processor → SessionManager → per- ↑ │ └── Channel ← OutboundDispatcher ← per-conversation lane ← MessageBus.outbound +AgentLoop → TurnEvent → TurnController → latest TurnSnapshot → DeliveryCoordinator → TurnSink → Channel + WebSocket/Channel → MessageBus.control → Gateway processor → SessionManager (dialog 操作) Scheduler → SessionManager.handle_cron_message → AgentLoop → send_message ``` @@ -19,9 +21,10 @@ Scheduler → SessionManager.handle_cron_message → AgentLoop → send_message | `client` | TUI 聊天客户端 | | `channels` | 外部集成（飞书、CLI），仅收发消息 | | `bus` | 有界 inbound/outbound/control 队列；出站 dispatcher 与分目标 lane | -| `session` | 会话生命周期、dialog 操作、每 session 串行队列、上下文与持久化协调 | -| `agent` | LLM 调用循环、工具执行、上下文压缩、媒体处理、子 Agent | -| `providers` | LLM API 客户端（OpenAI 兼容、Anthropic） | +| `session` | 会话生命周期、dialog 操作、每 session 串行队列、Turn 状态、上下文与持久化协调 | +| `agent` | LLM 调用循环、工具执行、上下文压缩、媒体处理、子 Agent、Turn 语义事件 | +| `providers` | OpenAI/Anthropic 原生流解析，统一正文、reasoning、工具、usage 与私有回放状态 | +| `delivery` | 完整 Turn 快照的展示过滤、latest-wins 节流、终态投递和 TurnSink 生命周期 | | `tools` | Agent 工具（bash、文件操作、搜索、HTTP、web、browser、memory、delegate 等） | | `skills` | Skill 加载、管理和 prompt 构建 | | `storage` | SQLite 持久化 | @@ -36,11 +39,14 @@ Scheduler → SessionManager.handle_cron_message → AgentLoop → send_message ## 功能边界 -- Channels 仅收发消息，不感知 session 或 LLM +- Channels 通过 MessageBus 发布入站消息，通过 OutboundDispatcher 或每 Turn 一个的 TurnSink 接收出站写入，不感知 session 或 LLM - MessageBus 本体持有三条有界队列；出站路由、顺序和重试由 `OutboundDispatcher` 负责 - SessionManager 拥有 session 状态、dialog 路由、上下文构建和每 session worker，并通过 worker 创建 AgentLoop +- TurnController 是活动 Turn 状态的唯一 owner；Session 在消息原子提交成功后才发布 Completed - AgentLoop 跨轮无状态，接收已准备的 history 调用 LLM、执行工具并返回一次结果 -- Providers 是纯 HTTP 客户端，无 bus/session/channel 感知 +- Providers 是纯 HTTP 流客户端，无 bus/session/channel 感知；签名 reasoning 状态只回放给匹配 Provider，不下发客户端或 Channel +- DeliveryCoordinator 只投影完整快照，不修改会话历史；慢消费者跳过中间 revision，终态显式、有界投递 +- 每个活动 Turn 独占一个 TurnSink；平台 message ID 和 reaction 清理状态只存在于 sink 内 - Tools 接收原始参数，返回字符串结果 - MCP 工具在 Gateway 初始化时连接服务器、发现工具，并包装成普通 Tool 注册到 ToolRegistry - 子 Agent 由 `delegate` 工具创建，复用 provider 配置和按需过滤后的工具集；后台任务结果通过 MessageBus 发回原会话 @@ -58,6 +64,8 @@ Scheduler → SessionManager.handle_cron_message → AgentLoop → send_message - `browser` 工具只有在 `browser.enabled=true` 时注册，依赖 Chrome/Chromium 与 WebDriver - 同一 session 的普通消息串行处理，不同 session 可并发；session 队列容量为 32，满时明确拒绝 - 出站消息按 `(channel, chat_id)` 分 lane 保序；lane 容量为 64，慢目标不阻塞其他目标 +- 活动 Turn 与普通出站消息共享 `(channel, chat_id)` 写锁；禁止把 token delta 放入 MessageBus +- `cli_chat` 向 TUI/WebUI 发送统一 `turn_updated` 完整快照；飞书默认 FinalOnly，开启 `live_updates` 后编辑同一卡片 - 长生命周期后台任务由 TaskSupervisor 管理；连接局部任务由其 owner 限时 join 或 abort - 外部建连、重试等待和关停 join 必须可取消且有硬超时 - 不得记录 API Key、Authorization header 或包含临时凭据的完整连接 URL @@ -138,6 +146,12 @@ Worker 的处理原则： 4. Session 持久化由独立 `persistence_lock` 串行化；批量消息使用原子写入，失败时精确回滚内存后缀。 5. 上下文溢出时按 Provider 返回的真实限制重新压缩并重试。 +### 活动 Turn + +每个主 Agent 请求会创建一个内存 Turn。Provider delta 经 AgentLoop 转换为 reasoning、正文、工具开始/完成等语义事件，TurnController 归约为有序 block 和单调 revision 的完整快照。TUI/WebUI 使用 `history + active_turn` 渲染，不自行拼接 token；中间帧可丢，下一快照会自动收敛。 + +展示策略在 Gateway 核心出口应用：交互客户端可显示 reasoning 和详细工具状态；外部渠道隐藏 reasoning、工具仅显示紧凑状态；无人值守投递只保留正文。运行态不逐 token 入库，完成、取消或中断时才原子保存消息及 completion status。 + ### 会话恢复 从 Storage 恢复 session 时： diff --git a/resources/skills/about-picobot/references/config.md b/resources/skills/about-picobot/references/config.md index b6c3dfb..76742a5 100644 --- a/resources/skills/about-picobot/references/config.md +++ b/resources/skills/about-picobot/references/config.md @@ -97,6 +97,10 @@ Gateway WebUI 的“配置”页可以编辑实际加载的配置文件。读取 | `agent` | string | - | 使用的 agent 名称 | | `media_dir` | string | ~/.picobot/media/feishu | 配置默认值；Gateway 注册渠道时会覆盖为 `{workspace}/media/feishu` | | `reaction_emoji` | string | "Typing" | 回复意向表达的表情 | +| `live_updates` | bool | false | 是否用单张卡片实时编辑活动 Turn；关闭时只发送终态 | +| `live_update_interval_ms` | int | 500 | 卡片更新最小间隔，运行时限制在 250–5000ms | + +飞书属于外部渠道：无论是否开启实时卡片，都不会接收模型 reasoning；工具只显示紧凑状态。配置修改需重启 Gateway 生效。 ## mcp 字段 diff --git a/resources/skills/about-picobot/references/db-schema.md b/resources/skills/about-picobot/references/db-schema.md index 57d6b1f..7887098 100644 --- a/resources/skills/about-picobot/references/db-schema.md +++ b/resources/skills/about-picobot/references/db-schema.md @@ -2,7 +2,7 @@ 数据库为 SQLite，默认位于 workspace 下的 `picobot.db`。 -连接启用 WAL、`synchronous=NORMAL`、foreign keys、5 秒 busy timeout，连接池最多 8 个连接。当前 `PRAGMA user_version=1`；启动时会在事务内补齐旧库字段和索引，遇到比程序更新的 schema version 会拒绝启动。 +连接启用 WAL、`synchronous=NORMAL`、foreign keys、5 秒 busy timeout，连接池最多 8 个连接。当前 `PRAGMA user_version=4`；启动时会在事务内补齐旧库字段和索引，遇到比程序更新的 schema version 会拒绝启动。 ## sessions 表 @@ -41,7 +41,11 @@ | `tool_calls` | TEXT | 工具调用参数 JSON | | `source` | TEXT | 消息来源（跨会话消息时标记来源 session_id） | | `created_at` | INTEGER | 创建时间（Unix 毫秒） | -| `reasoning_content` | TEXT | provider 返回的推理内容（如有） | +| `reasoning_content` | TEXT | 可展示的模型 reasoning（如有） | +| `provider_state` | TEXT | Provider 私有回放状态 JSON；只回放给匹配 Provider，不下发客户端或 Channel | +| `turn_id` | TEXT | 产生该消息的活动 Turn ID | +| `iteration` | INTEGER | Agent 工具循环中的迭代序号 | +| `completion_status` | TEXT | `completed` / `cancelled` / `interrupted`，旧数据默认 completed | `(session_id, seq)` 有唯一索引，防止并发写入重复序号。删除 session 会通过外键级联删除 messages。 @@ -137,9 +141,9 @@ Scheduler 使用原子 `UPDATE ... RETURNING` 领取到期任务。任务结果 | `created_at` | INTEGER | 调用时间 | | `provider` | TEXT | 提供商类型 | | `model` | TEXT | 模型名称 | -| `request_body` | TEXT | 请求体 JSON | +| `request_body` | TEXT | 请求摘要 JSON；旧记录可能是完整请求体 | | `response_body` | TEXT | 响应体 JSON | | `error` | TEXT | 错误信息 | | `duration_ms` | INTEGER | 耗时（毫秒） | -`request_body`/`response_body` 可能包含用户内容，排障和导出数据库时应按敏感数据处理。 +旧数据中的 `request_body`/`response_body` 可能包含用户内容，排障和导出数据库时应按敏感数据处理。新 Provider 请求的 `request_body` 只保存模型、消息数、工具数和 stream 标志等摘要；错误响应仍可能包含服务端回显内容。 diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index b3032d4..4519115 100644 --- a/src/agent/agent_loop.rs +++ b/src/agent/agent_loop.rs @@ -1248,7 +1248,7 @@ mod tests { cache_creation_input_tokens: None, }, }; - Ok(crate::providers::provider_stream_from_response(response)) + Ok(crate::providers::provider_stream_for_test(response)) } fn ptype(&self) -> &str { diff --git a/src/agent/context_compressor.rs b/src/agent/context_compressor.rs index 927ba5e..14ba089 100644 --- a/src/agent/context_compressor.rs +++ b/src/agent/context_compressor.rs @@ -703,7 +703,7 @@ mod tests { &self, _request: ChatCompletionRequest, ) -> Result { - Ok(crate::providers::provider_stream_from_response( + Ok(crate::providers::provider_stream_for_test( ChatCompletionResponse { id: "mock".into(), model: "mock".into(), diff --git a/src/providers/anthropic.rs b/src/providers/anthropic.rs index 5ff2429..8659156 100644 --- a/src/providers/anthropic.rs +++ b/src/providers/anthropic.rs @@ -545,8 +545,16 @@ impl LLMProvider for AnthropicProvider { req_builder = req_builder.header(key.as_str(), value.as_str()); } - let req_body_str = serde_json::to_string_pretty(&body).unwrap_or_default(); - tracing::debug!(req_body = %req_body_str, "LLM request"); + let request_summary = super::stream::diagnostic_request_summary( + &self.model_id, + body.messages.len(), + body.tools.as_ref().map_or(0, Vec::len), + ); + tracing::debug!( + message_count = body.messages.len(), + tool_count = body.tools.as_ref().map_or(0, Vec::len), + "Anthropic streaming request" + ); let resp = req_builder.json(&body).send().await.inspect_err(|e| { let is_timeout = e.is_timeout(); @@ -586,7 +594,7 @@ impl LLMProvider for AnthropicProvider { .append_llm_call( &self.name, &self.model_id, - &req_body_str, + &request_summary, Some(&body_text), Some(&error_msg), start.elapsed().as_millis() as u64, diff --git a/src/providers/mod.rs b/src/providers/mod.rs index 9ebff78..beecd14 100644 --- a/src/providers/mod.rs +++ b/src/providers/mod.rs @@ -7,9 +7,11 @@ pub use self::anthropic::AnthropicProvider; pub use self::openai::OpenAIProvider; use crate::config::LLMProviderConfig; +#[cfg(test)] +pub use stream::provider_stream_for_test; pub use stream::{ DynProviderError, FinishReason, ProviderChunk, ProviderResponseAccumulator, ProviderStream, - ProviderStreamItem, collect_provider_stream, provider_stream_from_response, + ProviderStreamItem, collect_provider_stream, }; pub use traits::{ ChatCompletionRequest, ChatCompletionResponse, LLMProvider, Message, Tool, ToolCall, diff --git a/src/providers/openai.rs b/src/providers/openai.rs index c550211..9efef17 100644 --- a/src/providers/openai.rs +++ b/src/providers/openai.rs @@ -592,8 +592,11 @@ impl LLMProvider for OpenAIProvider { req_builder = req_builder.header(key.as_str(), value.as_str()); } - let req_body_str = serde_json::to_string_pretty(&body).unwrap_or_default(); - tracing::debug!(req_body = %req_body_str, "LLM request"); + let request_summary = super::stream::diagnostic_request_summary( + &self.model_id, + body["messages"].as_array().map_or(0, Vec::len), + body["tools"].as_array().map_or(0, Vec::len), + ); let resp = req_builder.json(&body).send().await.inspect_err(|e| { let is_timeout = e.is_timeout(); @@ -625,7 +628,7 @@ impl LLMProvider for OpenAIProvider { .append_llm_call( &self.name, &self.model_id, - &req_body_str, + &request_summary, Some(&text), Some(&error), start.elapsed().as_millis() as u64, diff --git a/src/providers/stream.rs b/src/providers/stream.rs index 2013053..c01da82 100644 --- a/src/providers/stream.rs +++ b/src/providers/stream.rs @@ -2,7 +2,7 @@ use std::collections::BTreeMap; use std::error::Error; use std::pin::Pin; -use futures_util::{Stream, StreamExt, stream}; +use futures_util::{Stream, StreamExt}; use serde::{Deserialize, Serialize}; use crate::bus::ProviderReasoningState; @@ -13,6 +13,20 @@ pub type DynProviderError = Box; pub type ProviderStreamItem = Result; pub type ProviderStream = Pin + Send>>; +pub(crate) fn diagnostic_request_summary( + model: &str, + message_count: usize, + tool_count: usize, +) -> String { + serde_json::json!({ + "model": model, + "message_count": message_count, + "tool_count": tool_count, + "stream": true, + }) + .to_string() +} + /// Incremental framing shared by SSE-based providers. It accepts arbitrary /// byte/UTF-8 boundaries and returns only joined `data:` payloads. #[derive(Default)] @@ -212,11 +226,8 @@ pub async fn collect_provider_stream( Ok(accumulator.finish()) } -/// Adapt a complete response to the stream-first provider contract. -/// -/// This is intentionally a compatibility bridge for providers while their -/// native streaming parser is implemented; consumers still use one interface. -pub fn provider_stream_from_response(response: ChatCompletionResponse) -> ProviderStream { +#[cfg(test)] +pub fn provider_stream_for_test(response: ChatCompletionResponse) -> ProviderStream { let finish_reason = if response.tool_calls.is_empty() { FinishReason::Stop } else { @@ -248,12 +259,30 @@ pub fn provider_stream_from_response(response: ChatCompletionResponse) -> Provid } chunks.push(ProviderChunk::Usage(response.usage)); chunks.push(ProviderChunk::Done(finish_reason)); - Box::pin(stream::iter(chunks.into_iter().map(Ok))) + Box::pin(futures_util::stream::iter(chunks.into_iter().map(Ok))) } #[cfg(test)] mod tests { use super::*; + use futures_util::stream; + + #[test] + fn diagnostic_summary_contains_counts_without_message_content() { + let summary = diagnostic_request_summary("model", 3, 2); + + assert_eq!( + serde_json::from_str::(&summary).unwrap(), + serde_json::json!({ + "model": "model", + "message_count": 3, + "tool_count": 2, + "stream": true + }) + ); + assert!(!summary.contains("reasoning")); + assert!(!summary.contains("signature")); + } #[tokio::test] async fn collector_assembles_interleaved_tool_argument_fragments() {