euphon/tori
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor agent runtime: state machine, feedback processing, execution log

Browse Source 
- Add state.rs with AgentState/Step/StepStatus/AgentPhase as single source of truth
- Extract prompts to markdown files loaded via include_str!
- Replace plan_steps table with execution_log + agent_state_snapshots
- Implement user feedback processing with docker-build-cache plan diff:
  load snapshot → LLM revise_plan → diff (title, description) → invalidate from first mismatch → resume
- run_agent_loop accepts optional initial_state for mid-execution resume
- Broadcast plan step status (done/running/pending) to frontend on step transitions
- Rewrite frontend types/components to match new API (ExecutionLogEntry, PlanStepInfo with status)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Fam Zheng
2026-03-02 08:54:43 +00:00
parent 7f6dafeab6
commit 46424cfbc4
16 changed files with 910 additions and 992 deletions
Download Patch File Download Diff File Expand all files Collapse all files

442
doc/context.md
View File

@@ -1,392 +1,86 @@
# Agent Runtime 设计
# Agent Context 构建
## Context 管理现状与设计
AgentState 如何变成 LLM API call 的 messages 数组。定义见 `src/state.rs`。
## 现状
当前没有做 context 长度限制，存在超过 model token limit 的风险。
### 已有的缓解机制
1. **Phase transition 时 clear**：`step_messages` 在 planning→executing 和 step→step 切换时会 `clear()`，避免跨阶段累积
2. **单条 tool output 截断**：bash 输出限制 8000 bytes，read_file 超长时也截断
3. **Step context 摘要**：已完成步骤只保留 summary（`step_summaries`），不带完整输出
### 风险场景
- 一个 execution step 内 tool call 轮次过多（反复 bash、read_file），`step_messages` 无限增长
- 每轮 LLM 的 assistant message + tool result 都 push 进 `step_messages`，没有上限
- 最终整个 messages 数组超过模型 context window
## 方案设计
### 策略：滑动窗口 + 早期消息摘要
当 `step_messages` 长度超过阈值时，保留最近 N 轮完整对话，早期的 tool call/result 对折叠为一条摘要消息。
```
[system prompt]
[user: step context]
[summary of early tool interactions] ← 压缩后的历史
[recent assistant + tool messages] ← 完整保留最近 N 轮
```
### 具体实现
1. **Token 估算**：用字符数粗估（1 token ≈ 3-4 chars 中英混合），不需要精确 tokenizer
2. **阈值**：可配置，默认如 80000 chars（约 20k-25k tokens），给 system prompt 和 response 留余量
3. **压缩触发**：每次构建 messages 时检查总长度，超过阈值则压缩
4. **压缩方式**：
- 简单版：直接丢弃早期 tool call/result 对，替换为 `[已执行 N 次工具调用，最近结果见下文]`
- 进阶版：用 LLM 生成摘要（额外一次 API 调用，但质量更好）
5. **不压缩的部分**：system prompt、user context、最近 2-3 轮完整交互
### 实现位置
在 `run_agent_loop` 中构建 messages 之后、调用 LLM 之前，插入压缩逻辑：
## AgentState
```rust
// agent.rs run_agent_loop 内，约 L706-L725
let (mut messages, tools) = match &state.phase { ... };
// 压缩 context
compact_messages(&mut messages, MAX_CONTEXT_CHARS);
```
`compact_messages` 函数：从前往后扫描，保留 system/user 头部，计算总长度，超限时将早期 assistant+tool 消息替换为摘要。
---
## 执行中用户反馈处理 (Plan-Centric Feedback)
### 核心原则
Agent loop 是 **以 plan 为中心线性推进** 的。用户反馈的处理不应该是简单的"中断/不中断"二元决策，而应该根据反馈落在 plan 的哪个位置来决定行为：
```
Plan: [step1 ✓] [step2 ✓] [step3 🔄 executing] [step4 ○] [step5 ○]
↑
当前执行位置
用户反馈 → 影响哪里？
├─ 影响已完成部分 (step1/step2) → 回退：中断当前执行，replan from affected step
├─ 改了需求本身 → 回退：中断当前执行，全部 replan
├─ 影响未完成部分 (step4/step5) → 前进：不中断，调整后续 plan steps
└─ 影响当前步骤 (step3) → 视情况：注入到当前 step context 或中断重做
```
### 与"中断"的关系
"中断"只是一个实现手段，不是目的。真正的决策是：
1. **反馈是否影响当前或已完成的工作？** → 必须中断（继续执行是浪费）
2. **反馈只影响未来的步骤？** → 不中断，修改 plan 里的 pending steps，执行到那里时自然生效
这意味着 cancel token 之类的机制仍然需要（作为中断的执行手段），但决策逻辑是 plan-aware 的。
### 决策流程
```
用户 comment 到达
│
├─ workflow idle (done/failed)
│ └─ 当前逻辑：以 comment 为 context 重新执行
│
└─ workflow executing (step N)
│
├─ 1. 分类 LLM 调用（轻量，不阻塞执行）
│ 输入：原始需求、当前 plan、当前步骤进度、用户 comment
│ 输出：{ impact: "past" | "current" | "future" | "requirement_change" }
│
├─ impact = "past" 或 "requirement_change"
│ → cancel 当前执行
│ → replan（带上已完成步骤的摘要 + 用户反馈）
│
├─ impact = "current"
│ → 注入到 step_messages（当前步骤的 context）
│ → agent 在下一次 LLM 调用时自然看到
│ → 如果反馈是否定性的（"这样不行"），cancel 当前步骤，重做
│
└─ impact = "future"
→ 修改 plan 中 pending steps（update/insert/delete）
→ 不中断当前执行
→ 执行到对应步骤时自然生效
```
### 实现要点
#### 1. Plan 作为可变数据结构
当前 plan steps 存在 DB 里，但 agent 执行时是线性推进的，没有"修改后续步骤"的能力。需要：
- plan steps 支持运行时增删改（不仅是状态更新）
- agent 每执行完一步，从 DB 重新读 plan（而不是用内存快照），这样外部修改能被感知
#### 2. 并发：分类 LLM 与执行 LLM 并行
分类调用不应该阻塞执行。用 `tokio::spawn` 做分类，分类结果通过 channel 或 shared state 回传。如果分类结果是"需要中断"，设置 cancel token。
```rust
// 伪代码
tokio::spawn(async move {
let impact = classify_feedback(&llm, &plan, current_step, &comment).await;
match impact {
Impact::Past | Impact::RequirementChange => {
cancel_token.cancel();
// replan 会在 cancel 生效后由主循环触发
}
Impact::Current => {
// 注入到 step context
step_messages_tx.send(comment).await;
}
Impact::Future => {
// 调用 LLM 修改后续 plan steps
replan_future_steps(&llm, &plan, current_step, &comment).await;
}
}
});
```
#### 3. 中断后的 Context 衔接
中断执行后重启时，需要构建的 context：
- 已完成步骤的摘要（已有 `step_summaries` 机制）
- 被中断步骤的部分执行记录
- 用户反馈内容
- "请根据用户反馈重新规划剩余步骤"的指令
这和上面 context 压缩的滑动窗口方案互补——中断后重启本质上就是一次强制的 context 压缩。
### 与 context 压缩的统一视角
两个问题其实是同一件事的两面：
| | Context 压缩 | 用户反馈 |
|---|---|---|
| 触发条件 | token 数超阈值 | 用户发了 comment |
| 本质操作 | 摘要化早期历史，保留近期 | 根据反馈位置，决定哪些历史需要保留/丢弃/重做 |
| 共享机制 | step_summaries、compact_messages | 同上 + plan 的增删改 |
两者都需要一个健壮的"摘要化已完成工作"的基础能力。先把这个基础能力做好，上面两个场景都能受益。
### Plan-Centric Context 优先级
Context 管理也应该围绕 plan 展开。Plan 本身是结构化的、紧凑的，基本不会超。撑爆 context 的是执行日志（一个 step 内反复 tool call 的历史）。这就有了清晰的层级和主次：
```
优先级（从高到低，压缩时从低往高砍）：
1. System prompt + 需求 ← 不可压缩，恒定大小
2. Plan 全貌（步骤标题 + 状态） ← 不可压缩，O(步骤数) 很短
3. 用户 comments ← 保留原文，通常很短
4. 当前步骤的最近 N 轮 tool call ← 完整保留（agent 需要看到最近的执行结果）
5. 当前步骤的早期 tool call ← 第一优先压缩目标，摘要化
6. 已完成步骤的执行摘要 ← 已经是一句话/步骤，极紧凑
```
关键洞察：**Plan 是骨架，执行日志是血肉。** 骨架永远保留，血肉按新鲜度裁剪。这和人类记忆一样——你记得"今天做了什么"（plan），但不记得每一步的具体命令输出（执行日志）。
压缩算法因此变得很简单：
1. 算总 token 估计
2. 如果超阈值，从"当前步骤的早期 tool call"开始砍，替换为 `[前 N 次工具调用已折叠，关键结果：...]`
3. 如果还超，压缩已完成步骤的摘要（合并多步为一段）
4. Plan 和 system prompt 永远不动
---
## 状态管理、并发与一致性
### 系统中的参与者和他们各自的"世界观"
一致性问题的本质是：多个参与者对同一份状态有各自的视图，这些视图会因为延迟和并发而产生分歧。
| 参与者 | 看到的世界 | 更新方式 | 延迟 |
|---|---|---|---|
| **Agent loop** | 内存 `AgentState` | 直接修改 | 0（权威源） |
| **LLM** | 上一次调用时传入的 messages | 每次调用时重建 | 一次 LLM 调用的时间（数秒） |
| **DB** | 持久化的 plan_steps, workflow.status | agent 写入 | 取决于 agent 何时 flush |
| **前端** | 最近一次 WS 推送 / API 响应 | WS broadcast | 网络延迟 + WS 推送时机 |
| **用户** | 看着前端 + 自己的判断 | 发 comment | 人类反应时间（秒~分钟） |
问题出在哪：
```
时间线：
t0 agent 开始执行 step 3
t1 前端显示 step 3 running（WS 推送）
t2 用户看到 step 3，觉得方向不对，开始打字
t3 agent 完成 step 3，开始 step 4
t4 前端显示 step 4 running
t5 用户发出 comment："step 3 不对，应该这样做"
→ 但此时 step 3 已经做完了，step 4 在跑了
→ 用户的反馈是基于 t1 时的世界观
```
这种"用户看到的是过去"是不可消除的（人类反应时间），关键是系统怎么处理这个 gap。
### 核心设计：统一状态 + 单写者
#### 状态 Schema
一个 project 的完整运行时状态 = 一个 `ProjectState` 对象：
```rust
struct ProjectState {
workflow_id: String,
requirement: String,
status: WorkflowStatus, // Idle / Executing / Done / Failed
plan: Vec<PlanStep>, // 有序步骤，每步有 status
current_step: Option<usize>, // 正在执行哪一步
step_summaries: Vec<StepSummary>, // 已完成步骤的紧凑摘要
step_messages: Vec<ChatMessage>, // 当前步骤的 tool call 历史（临时，会膨胀）
memo: String, // agent 备忘
iteration: u32, // LLM 调用轮次
cancel: CancellationToken, // 中断信号
struct AgentState {
phase: AgentPhase, // Planning | Executing { step } | Completed
steps: Vec<Step>, // 执行计划，每个 step 有 status + optional summary
current_step_chat_history: Vec<ChatMessage>, // 当前步骤内的多轮对话，step 切换时 clear
scratchpad: String, // LLM 的跨步骤工作区
}
```
用 `Arc<RwLock<ProjectState>>` 持有。
整个结构体 `Serialize/Deserialize`，可 JSON 直接存 DB。
#### 单写者规则
**只有 agent loop 修改 `ProjectState`。** 没有例外。
其他所有参与者（反馈分类器、API handler、WS handler）都是读者或事件发送者：
- 想看状态 → read lock，clone 出快照，立即释放
- 想改状态 → 发 event 到 agent loop 的 channel，由 agent loop 决定怎么改
这彻底消除了写-写竞争。不需要事务，不需要 CAS，不需要冲突解决。
#### 数据流
## Step 生命周期
```
用户 comment ─→ API handler ─→ mpsc channel ─→ agent loop (唯一写者)
│
write lock ──→ ProjectState
│
write DB (flush)
│
broadcast WS ──→ 前端
Pending → Running → Done(summary) 或 Failed
```
### 一致性分析：三对关系
summary 在 step 完成时由 LLM 填入，作为后续步骤的压缩上下文。
#### 1. 内存 ↔ DB 一致性
## Messages 组装
当前问题：`AgentState`（内存）和 `plan_steps`（DB）是两套独立数据，各管各的。
### Planning 阶段
**解法：内存是 master，DB 是 checkpoint。**
```
[ system(planning_prompt), user(requirement), ...current_step_chat_history ]
```
- Agent loop 修改 `ProjectState`（内存），在关键节点 flush 到 DB
- 前端 API 读内存（read lock），不读 DB（运行时）
- DB 只在重启恢复时读
### Executing 阶段
Flush 时机（从密到疏）：
| 事件 | 写 DB | 原因 |
```
[ system(execution_prompt), user(step_context), ...current_step_chat_history ]
```
`step_context` 由 `build_step_context()` 拼接：
```
## 需求
{requirement}
## 计划概览
1. 分析代码结构 done
2. 实现核心逻辑 >> current
3. 写测试 FAILED
4. 集成测试
## 当前步骤（步骤 2）
标题：实现核心逻辑
描述：...
## 已完成步骤摘要 ← 从 steps 中 filter Done，取 summary
- 步骤 1: ...
## Scratchpad ← LLM 自己维护的跨步骤笔记
...
```
## 持久化
### DB 表
- **agent_state_snapshots** — step 切换时 insert 一行 AgentState JSON 快照（追加，不覆盖，保留历史）。恢复时取最新一行。
- **execution_log** — tool call 的输入输出记录（不可变历史），前端展示 + report 生成用。
Plan 步骤只从 AgentState JSON 读，不再单独写表。
### Executing 阶段 text response
LLM 返回纯文本时不隐含"workflow 结束"，写 execution_log 显示给用户。只有显式调 `advance_step` 才推进步骤。
## 待做：Context 压缩
当前无长度限制，`current_step_chat_history` 在单步 tool call 轮次过多时会无限增长。
压缩按优先级从低到高砍：
| 优先级 | 内容 | 处理 |
|---|---|---|
| workflow.status 变更 | 立即 | 重启恢复必需 |
| plan 变更（新建/修改步骤） | 立即 | 重启恢复 + 持久记录 |
| 步骤完成（summary 产出） | 立即 | 重启恢复的关键数据 |
| step_messages 增长 | 不写 | 临时数据，中断时才摘要化 |
| memo 更新 | 步骤切换时 | 丢了问题不大 |
重启恢复 = 从 DB 重建 `ProjectState`，`step_messages` 为空（丢失可接受，summaries 提供足够上下文）。
#### 2. 后端 ↔ 前端 一致性
前端通过两个渠道获取状态：
- **WS 推送**：agent loop 每次修改状态后 broadcast
- **API 拉取**：前端 mount 时、切换 project 时
一致性模型：**最终一致**。前端可能落后几百毫秒，但不会出现永久分歧。
需要保证的不变量：
- WS 推送的顺序和状态变更顺序一致（单写者天然保证）
- 前端收到 WS 消息后，可以安全地 patch 本地状态（不需要全量刷新）
- 如果 WS 断连，重连后拉一次全量状态即可恢复
#### 3. Agent loop ↔ LLM ↔ 用户反馈 一致性
这是最微妙的。三方的时间线：
```
Agent loop: ──[step3 执行中]──[step3 完成]──[step4 开始]──...
LLM: ──[thinking...]──[返回 tool calls]──
用户: ──[看到 step3]──[打字中...]──[发送 comment]──
```
三个视角可能同时看到不同的状态。处理方式：
**a) LLM 的"冻结视角"**
每次 LLM 调用时，传入的 messages 是一个 point-in-time 快照。LLM 返回的 tool calls 是基于那个快照的。在 LLM 思考期间，状态可能已经变了（比如用户发了 comment），但 LLM 不知道。
这没关系——LLM 返回后，agent loop 先检查 cancel flag，再决定是否执行 tool calls。如果 cancel 了，丢弃 LLM 的响应，不浪费。
**b) 用户反馈的"过时性"**
用户发 comment 时看到的可能是几秒前的状态。反馈里提到的"step 3"可能已经完成了。
处理：反馈分类器拿当前快照（read lock），对比用户反馈和当前进度，判断反馈是关于：
- 已完成的步骤 → 需要 revert/redo
- 当前步骤 → 注入或中断
- 未来步骤 → 修改 plan
- 需求级别 → 全部 replan
分类结果作为 event 发给 agent loop。Agent loop 收到时再次基于最新状态做最终决策（双重检查）。
**c) 反馈分类器的并发**
反馈分类器 `tokio::spawn` 出去，和 agent loop 并行：
```
Agent loop: ──[step4 执行中]──────────[step4 完成]──[处理 feedback event]──
↑
Feedback classifier: ──[拿 snapshot]──[LLM 分类...]──[发 event]─┘
```
分类器拿的 snapshot 可能已经过时了（分类期间 agent 又推进了一步）。但因为分类结果是建议性的，agent loop 收到后基于最新状态做最终决策，所以不会出错。最坏情况：建议无效，忽略。
### Agent loop 内部的事件处理
Agent loop 不再是"拿 event → 跑完整个 run_agent_loop → 拿下一个 event"。而是在 tool call 循环内部也能感知新事件：
```rust
for iteration in 0..MAX_ITERATIONS {
// 1. 检查中断
if state.cancel.is_cancelled() {
// 摘要化当前 step_messages，存 DB
break;
}
// 2. 检查是否有 pending feedback events（非阻塞）
while let Ok(event) = feedback_rx.try_recv() {
handle_feedback(&mut state, event);
// 可能修改了 plan、设置了 cancel、注入了 step_messages
}
// 3. 构建 messages，调 LLM
let messages = build_messages(&state);
let response = llm.chat(messages).await?;
// 4. 执行 tool calls，更新 state
// ...
}
```
关键：`feedback_rx.try_recv()` 是非阻塞的。在每次 LLM 调用之间检查一次。如果有 feedback，立即在 agent loop 内处理（修改 plan / 设置 cancel / 注入 context），然后继续循环。
这保持了**单写者**——所有状态变更都在这个循环里发生。
### 实现顺序
1. **定义 `ProjectState`** — 集中状态，`Arc<RwLock<_>>`。Agent loop 内部用起来，行为不变。
2. **DB flush 策略** — 关键节点写 DB（status 变更、步骤完成），前端 API 改为读内存。
3. **Step 摘要持久化** — 步骤完成时写 summary 到 DB，重启时重建 ProjectState。
4. **Context 压缩** — step_messages 滑动窗口。
5. **CancellationToken + feedback channel** — agent loop 内部 `try_recv`，单写者处理反馈。
6. **Feedback 分类器** — `tokio::spawn`，read lock 拿快照，LLM 分类，结果发回 agent loop。
| 高 | system prompt / requirement / plan 概览 | 不压缩 |
| 中 | 当前步骤最近 N 轮 tool call | 完整保留 |
| 低 | 当前步骤早期 tool call | 替换为摘要 |