kimi-cli 为何具有远超 crush 的持续工作能力：七层架构对比分析

> 日期: 2026-04-18 > 对标项目: kimi-cli (Python/asyncio) > 基准项目: github.com/linkerlin/crush (Go) > 分析方法: 逐文件源码审查，含具体行号引用 --- ## 核心结论 kimi-cli 的"持续工作能力"并非来自单一功能，而是**七层嵌套容错机制**的协同效果。crush 在大多数层面要么缺失，要么设计更粗糙。差距从内到外递进。 --- ## 第一层：Tool 级容错 | | kimi-cli | crush | |---|---|---| | **工具异常** | 捕获 → 返回 `ToolRuntimeError` 给 LLM → LLM 自行调整重试 (`toolset.py:190-218`) | 同：工具错误作为 tool result 返回给模型 | | **Hook 异常** | **Fail-open** — hook 引擎崩溃时默认放行 (`engine.py:240-242`) | Runner 层吞没 hook 错误 (`runner.go:101-155`)，仅 log | | **差异** | 两者在此层相当 | | --- ## 第二层：LLM 调用级重试（**差距已部分弥合**） | | kimi-cli | crush | |---|---|---| | **重试框架** | `@tenacity.retry` — 指数退避+jitter（0.3s→5s），最多 N 次 (`kimisoul.py:842-854`) | `errorx.Classify()` + 指数退避重试（1s→2s→4s，最多 3 次），集成于 `query_engine_impl.go` 主循环 | | **可重试判定** | `_is_retryable_error()` — 精确区分 429/500/502/503/504/超时/空响应 (`kimisoul.py:1050-1061`) | `errorx.Classify()` — 15 种 FailoverReason（rate_limit/overloaded/server_error/timeout 等），含 `Retryable` 和 `ShouldCompress` 恢复提示 | | **连接恢复** | `_run_with_connection_recovery()` — 三层恢复：① 401→OAuth刷新重试 ② 连接断开→provider.on_retryable_error() ③ 二次失败→标记耗尽阻止无限重试 (`kimisoul.py:1063-1134`) | coordinator.go 有 401→OAuth 刷新重试一次，errorx 有 `ShouldCompress` 自动压缩恢复，但**无连接级恢复（provider.on_retryable_error）、无耗尽标记** | | **Compaction 重试** | 压缩操作本身也有独立 tenacity 重试 (`kimisoul.py:978-990`) | **无独立重试**（压缩失败直接上抛） | **影响**：crush 已通过 `errorx` 分类器实现了对 rate_limit/overloaded/server_error 等瞬态错误的自动重试（指数退避，最多 3 次），并在 `ShouldCompress` 场景下自动触发上下文压缩恢复。但仍缺少**连接级恢复**（provider 重建）和**耗尽标记**防止无限重试。压缩操作本身也无独立重试。与 kimi-cli 的差距已从"完全缺失"缩小到"缺少连接恢复层和压缩操作重试"。 ### kimi-cli 的具体实现 **可重试错误判定** (`kimisoul.py:1050-1061`): ```python @staticmethod def _is_retryable_error(exception: BaseException) -> bool: if isinstance(exception, (APIConnectionError, APITimeoutError)): return not bool(getattr(exception, "_kimi_recovery_exhausted", False)) if isinstance(exception, APIEmptyResponseError): return True return isinstance(exception, APIStatusError) and exception.status_code in ( 429, 500, 502, 503, 504, ) ``` **连接恢复** (`kimisoul.py:1063-1134`): ```python async def _run_with_connection_recovery(self, name, operation, *, chat_provider=None, _auth_retried=False): try: return await operation() except APIStatusError as error: if error.status_code != 401 or _auth_retried: raise # OAuth 刷新 + 重试一次 await self._runtime.oauth.ensure_fresh(self._runtime, force=True) return await self._run_with_connection_recovery(..., _auth_retried=True) except (APIConnectionError, APITimeoutError) as error: if not isinstance(chat_provider, RetryableChatProvider): raise recovered = chat_provider.on_retryable_error(error) if not recovered: raise try: return await operation() except (APIConnectionError, APITimeoutError) as second_error: second_error._kimi_recovery_exhausted = True # 标记耗尽 raise ``` --- ## 第三层：Checkpoint + D-Mail 时间回溯（**独有能力**） | | kimi-cli | crush | |---|---|---| | **Checkpoint** | 每步前 `_checkpoint()` 在 context.jsonl 追加 `{"role": "_checkpoint", "id": N}` (`kimisoul.py:721`) | **无** | | **Revert** | `revert_to(checkpoint_id)` — 文件旋转+逐行回放，精确恢复到任意历史检查点 (`context.py:135-200`) | **无** | | **D-Mail** | 工具可发送 D-Mail 到过去 checkpoint → `_step()` 抛 `BackToTheFuture` 异常 → `_agent_loop()` 捕获 → `revert_to()` 回溯 → 注入新消息 → **继续循环** (`denwarenji.py`, `kimisoul.py:908-931,724,773-775`) | **无** | | **asyncio.shield** | `_grow_context()` 用 shield 防止 Ctrl+C 破坏上下文一致性 (`kimisoul.py:889`) | **无** | **这是 kimi-cli 最独特的设计**。D-Mail 机制允许 Agent **在运行中主动回溯并修正方向**，而非在错误累积到无法恢复时才崩溃退出。 ### D-Mail 完整链路 ``` 工具调用 SendDMail → DenwaRenji.send_dmail(dmail) [denwarenji.py:21-29] → 存入 _pending_dmail _step() 执行完毕后： → self._denwa_renji.fetch_pending_dmail() [kimisoul.py:908] → 取出 _pending_dmail → raise BackToTheFuture(checkpoint_id, messages) [kimisoul.py:914] _agent_loop() 捕获： → except BackToTheFuture as e: [kimisoul.py:724] → back_to_the_future = e 循环底部处理： → await self._context.revert_to(checkpoint_id) [kimisoul.py:773] → await self._checkpoint() [kimisoul.py:774] → await self._context.append_message(dmail_messages) [kimisoul.py:775] → 继续循环 ``` ### Context.revert_to 实现 (`context.py:135-200`) ```python async def revert_to(self, checkpoint_id: int): # 文件旋转（非原地修改，旧数据不丢失） rotated_file_path = await next_available_rotation(self._file_backend) await aiofiles.os.replace(self._file_backend, rotated_file_path) # 内存状态完全重置 self._history.clear() self._token_count = 0 self._next_checkpoint_id = 0 # 从旋转后文件逐行回放至目标 checkpoint async with ( aiofiles.open(rotated_file_path, ...) as old_file, aiofiles.open(self._file_backend, "w", ...) as new_file, ): async for line in old_file: line_json = self._parse_context_line(line, ...) if line_json.get("role") == "_checkpoint" and line_json.get("id") == checkpoint_id: break # 到目标即停 keep_line = self._apply_context_record(line_json, ...) if keep_line: await new_file.write(line) ``` --- ## 第四层：Auto-Compaction（**差距已大幅缩小**） | | kimi-cli | crush | |---|---|---| | **触发方式** | 每步前 `should_auto_compact()` 双条件检查（比例触发 OR 预留空间不足）(`kimisoul.py:701-718`) | `maybeCompress()` **三点触发**：① preflight（循环开始前预检）② 每轮工具执行后检查 ③ errorx `ShouldCompress` 触发的压缩恢复 | | **压缩算法** | `SimpleCompaction` — 保留最近 N 条消息 + LLM 摘要旧历史 (`compaction.py:103-189`) | `StructuredCompressor` — 结构化摘要（Goal/Progress/KeyDecisions/RelevantFiles/RemainingWork），支持 FocusTopic 引导、增量压缩（基于 PreviousSummary）、`pruneOldToolResults` 预处理、tool pair 消毒 | | **压缩后恢复** | clear → write_system_prompt → **checkpoint** → append compacted messages (`kimisoul.py:960-1047`) | 压缩后替换历史消息 + 保留尾部 token 预算内的近期消息，但**无 checkpoint 锚点** | | **压缩自身重试** | 压缩有独立 tenacity 重试 + 连接恢复 (`kimisoul.py:978-990`) | **无独立重试** | | **失败行为** | 压缩失败 → raise → 上层异常处理接管 | 摘要失败 → 错误上抛 | **差异**：crush 的压缩已从"循环后补救"升级为**循环内多点触发**（preflight + per-turn + error-recovery），算法也从简单文本摘要升级为 LLM 驱动的结构化增量压缩。但仍缺少 **checkpoint 锚点**（压缩后无法回滚）和**压缩操作独立重试**。 ### kimi-cli compact_context 流程 (`kimisoul.py:960-1047`) ``` compact_context() ├─ wire_send(CompactionBegin) ├─ _compact_with_retry() ← tenacity.retry + connection_recovery │ └─ SimpleCompaction.compact() │ ├─ 保留最近 N 条 user/assistant 消息 │ └─ 旧历史喂给 LLM 生成摘要 ├─ context.clear() ← 文件旋转，旧数据不丢失 ├─ context.write_system_prompt() ├─ context.checkpoint() ← 新安全锚点 ├─ context.append_message(压缩后消息) └─ wire_send(CompactionEnd) ``` --- ## 第五层：Ralph Loop / FlowRunner（**独有能力**） | | kimi-cli | crush | |---|---|---| | **自动迭代循环** | `ralph_loop()` 构建有环流图：BEGIN→执行→决策(CONTINUE/STOP)→循环。`max_ralph_iterations=-1` 时无限循环。LLM 在决策节点**自评是否完成** (`kimisoul.py:1175-1249`) | **无**。Agent 执行一次 user prompt → 回复 → 结束 | | **动态注入** | Plan mode 每 N 轮注入约束提醒，Yolo mode 注入非交互提醒，**防止长循环中行为漂移** (`dynamic_injections/plan_mode.py`, `yolo_mode.py`) | **无** | **这是持续工作能力的核心**：kimi-cli 的 Agent 不是"回答一次问题"而是"持续执行直到任务完成"。Ralph Loop 让 LLM 自己判断任务是否完成，未完成则自动继续。crush 没有这种循环机制——用户必须不断发送消息才能让 agent 继续工作。 ### Ralph Loop 实现 (`kimisoul.py:1175-1249`) ```python @staticmethod def ralph_loop(user_message, max_ralph_iterations) -> FlowRunner: total_runs = max_ralph_iterations + 1 if max_ralph_iterations < 0: total_runs = 1000000000000000 # effectively infinite nodes = { "BEGIN": FlowNode(id="BEGIN", label="BEGIN", kind="begin"), "END": FlowNode(id="END", label="END", kind="end"), "R1": FlowNode(id="R1", label=prompt_content, kind="task"), "R2": FlowNode(id="R2", label="...automated loop...", kind="decision"), } outgoing = { "BEGIN": [FlowEdge(src="BEGIN", dst="R1", ...)], "R1": [FlowEdge(src="R1", dst="R2", ...)], "R2": [ FlowEdge(src="R2", dst="R2", label="CONTINUE"), # 自环 FlowEdge(src="R2", dst="END", label="STOP"), ], } return FlowRunner(flow, max_moves=total_runs) ``` FlowRunner 执行： ``` BEGIN → R1(执行任务) → R2(LLM决策: CONTINUE or STOP) ├─ CONTINUE → R1(继续) └─ STOP → END ``` --- ## 第六层：Shell 层全局兜底（**设计差距**） | | kimi-cli | crush | |---|---|---| | **主循环** | `while True` REPL 永不退出 (`shell/__init__.py:481`) | Bubble Tea Model 的 `Update()` 消息循环，永不退出 | | **MaxStepsReached** | 捕获 → 提示"Send another message to continue" → `return False` → 主循环继续 (`shell/__init__.py:915-920`) | budget 耗尽 → grace call（注入系统消息要求简洁总结）→ 自然终止 → agent 回到空闲 | | **错误分类** | 7 种精确分类（LLMNotSet/LLMNotSupported/401/402/403/Connection/Timeout/EmptyResponse/MaxSteps/Cancel）(`shell/__init__.py:868-931`) | **引擎层**：`errorx.Classify()` 有 15 种 FailoverReason（auth/billing/rate_limit/overloaded/server_error/timeout/context_overflow 等），含自动重试和压缩恢复。**UI 层**：仍较粗（cancel/permission/provider），其余统一 InfoTypeError | | **消息队列排空** | 主回合结束后排空运行期间排队消息，每个作为新回合执行，20代安全阀 (`shell/__init__.py:810-866`) | `messageQueue` 在 agent.Run 末尾递归处理，但 Cancel 时**直接丢弃** (`agent.go:865-868`) | | **后台任务完成后自动续行** | 背景任务完成 → 自动触发 `<system-reminder>Background tasks completed...</system-reminder>` → agent 继续处理 (`shell/__init__.py:493-497`) | **无** | **关键差异**：kimi-cli 的 `MaxStepsReached` 是**软限制**——用户发下一条消息即从断点续接。crush 的 budget 耗尽虽然也允许继续，但没有这种显式的"从断点续接"提示。更重要的是，kimi-cli 有消息队列排空和后台任务自动续行——**Agent 不会因为"闲下来"就停止工作**。 ### kimi-cli Shell 错误处理 (`shell/__init__.py:868-931`) ```python try: await run_soul(self.soul, user_input, ...) return True except LLMNotSet: console.print('[red]LLM not set, send "/login" to login[/red]') except LLMNotSupported as e: console.print(f"[red]{e}[/red]") except APIStatusError as e: if e.status_code == 401: console.print("[red]Authorization failed[/red]") elif e.status_code == 402: console.print("[red]Membership expired[/red]") elif e.status_code == 403: console.print("[red]Quota exhausted[/red]") except APIConnectionError: console.print("[red]Network error[/red]") except APITimeoutError: console.print("[red]Timeout[/red]") except APIEmptyResponseError: console.print("[red]Empty response[/red]") except MaxStepsReached as e: console.print(f"[yellow]{e}[/yellow]\n" "[dim]Send another message to continue where it left off.[/dim]") except RunCancelled: console.print("[red]Interrupted by user[/red]") except Exception as e: console.print(f"[red]Unexpected error: {e}[/red]") raise return False # ← 不退出主循环 ``` --- ## 第七层：会话级恢复（**独有能力**） | | kimi-cli | crush | |---|---|---| | **/undo** | 交互式回滚到任意历史回合 → fork 新会话 → `Reload` 切换 (`slash.py:725-787`) | **无**。无分支、无回滚 | | **/fork** | 从指定回合分叉新会话 (`session_fork.py:215-281`) | **无** | | **wire.jsonl** | 所有 Wire 消息持久化为 JSONL，可精确重建任意时刻的 UI 状态 | SQLite messages 表持久化，但无 checkpoint/回滚 | | **context.jsonl** | append-only JSONL，支持 checkpoint/revert_to 文件旋转 | SQLite session + messages | | **进程重启恢复** | `recover()` 扫描后台任务，将心跳过期的标记为 lost，状态一致性恢复 (`background/manager.py:397-458`) | **无对应机制** | --- ## 总结：差距矩阵 ``` kimi-cli 七层容错 crush 对应能力 ───────────────────────────────────────────────────────── 1. Tool 异常 → ToolRuntimeError ✅ 有（同等） 2. tenacity + 连接恢复 + 耗尽标记 ⚠️ 大部分有（errorx 分类器+指数退避重试+压缩恢复已接入，缺连接恢复层和耗尽标记） 3. Checkpoint + D-Mail 时间回溯 ❌ 完全缺失 4. Auto-Compaction + 独立重试 ⚠️ 大部分有（循环内三点触发+结构化增量压缩+tool pair 消毒，缺 checkpoint 锚点和压缩操作独立重试） 5. Ralph Loop 自动迭代 ❌ 完全缺失 6. Shell 全局兜底 + 队列排空 ⚠️ 部分有（引擎层 15 种错误分类+grace call，但 UI 层分类粗、Cancel 时丢弃队列） 7. /undo + /fork + 进程重启恢复 ❌ 完全缺失 ``` --- ## 对 crush 演进的启示 若 crush 想获得类似的持续工作能力，优先级应为： 1. **P0** — 引入 Checkpoint 机制（为回溯、Cancel 保留和 D-Mail 打基础） 2. **P0** — 引入 Ralph Loop 式自动迭代（让 Agent 自评完成度并继续） 3. **P1** — 补齐连接恢复层 + 耗尽标记 + 压缩操作独立重试（errorx 重试框架已有，需补最后一环） 4. **P1** — MaxStepsReached 后的软续接提示 + Cancel 时保留队列 5. **P2** — D-Mail / BackToTheFuture 时间回溯 6. **P2** — UI 层错误分类精细化（引擎层已有 15 种分类，需透传到 UI） **一句话总结**：kimi-cli 的持续工作能力来自"**任何失败都不是终点，而是可恢复的中间状态**"这一设计哲学。crush 已通过 `errorx` 分类器+结构化压缩器大幅缩小了第 2、4 层差距，但在 Checkpoint（第 3 层）、Ralph Loop（第 5 层）和会话级恢复（第 7 层）方面仍完全缺失——这三者是"持续工作体验"的核心差异化所在。 --- **维护状态**: 2026-04-18 代码校准更新（修正第 2、4、6 层 crush 能力评估，反映 errorx/StructuredCompressor 等已接入能力）