docs: finalize streaming turn architecture
This commit is contained in:
parent
c6e022f6cb
commit
7a3058d8b5
16
AGENTS.md
16
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
|
||||
|
||||
23
README.md
23
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/ 记忆管理和记忆类型
|
||||
|
||||
@ -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
|
||||
|
||||
|
||||
@ -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. 架构验收标准
|
||||
|
||||
|
||||
@ -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 时：
|
||||
|
||||
@ -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 字段
|
||||
|
||||
|
||||
@ -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 标志等摘要；错误响应仍可能包含服务端回显内容。
|
||||
|
||||
@ -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 {
|
||||
|
||||
@ -703,7 +703,7 @@ mod tests {
|
||||
&self,
|
||||
_request: ChatCompletionRequest,
|
||||
) -> Result<crate::providers::ProviderStream, crate::providers::DynProviderError> {
|
||||
Ok(crate::providers::provider_stream_from_response(
|
||||
Ok(crate::providers::provider_stream_for_test(
|
||||
ChatCompletionResponse {
|
||||
id: "mock".into(),
|
||||
model: "mock".into(),
|
||||
|
||||
@ -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,
|
||||
|
||||
@ -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,
|
||||
|
||||
@ -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,
|
||||
|
||||
@ -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<dyn Error + Send + Sync>;
|
||||
pub type ProviderStreamItem = Result<ProviderChunk, DynProviderError>;
|
||||
pub type ProviderStream = Pin<Box<dyn Stream<Item = ProviderStreamItem> + 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::<serde_json::Value>(&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() {
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user