1. 为什么我们需要给编码代理套上一层“工程控制层”

1.1 从“能跑”到“跑得明白”的鸿沟

过去大半年，我一直在折腾各种编码代理（Coding Agent）的落地。从最初的新鲜感，到后来在真实项目里被反复折磨，中间隔着的不是模型能力，而是一整套工程化的缺失。Pi Coding Agent 本身在代码生成、任务拆解上的表现已经相当能打，但一旦把它放进一个需要连续运行数小时、涉及几十个文件改动、还要跟 CI/CD 打交道的真实场景里，问题就全冒出来了。

最典型的三个痛点：第一，黑盒感。代理在后台跑了什么、调了哪些工具、哪一步开始跑偏，你只能等它最后吐出一个结果，中间过程完全不可见。第二，断点即灾难。网络抖一下、进程被系统回收、或者你手滑关了个终端，整个任务就得从头再来，前面烧掉的 token 和时间全打水漂。第三，编排靠人肉。多个代理之间怎么分工、谁先谁后、失败了怎么重试、结果怎么汇总，全靠写死脚本，改一次逻辑就得动一次代码。

Pi-Harness 这个项目，就是冲着这三个问题去的。它不是什么新模型，也不是什么新框架，而是一层工程控制层——套在 Pi Coding Agent 外面的“仪表盘 + 黑匣子 + 调度台”。用一句话说清楚：它让编码代理从“一个能跑的函数”变成“一个可观测、可恢复、可编排的服务”。

1.2 谁适合看这套东西

如果你只是偶尔用代理写个脚本、改个 bug，那 Pi-Harness 可能有点重。但如果你符合下面任意一条，这套控制层的价值会非常直接：

• 你在把编码代理接入到持续集成流程里，需要它稳定跑完整个流水线；
• 你在做多代理协作，比如一个负责写、一个负责审、一个负责跑测试；
• 你需要对代理的每一次工具调用、每一次文件改动做审计和回溯；
• 你被“跑一半挂了”搞到心态爆炸，想要一个能断点续跑的方案。

我接下来的内容，会从整体设计思路讲到具体实现细节，再到实操中踩过的坑。所有代码和配置都是可以直接抄作业的，参数选择我也会把背后的计算逻辑讲清楚。你不需要是分布式系统专家，但最好对 Python 异步编程和基本的进程管理有点概念。

2. 整体设计思路：三层解耦，各管一摊

2.1 为什么是“控制层”而不是“新框架”

市面上不少方案喜欢把可观测、恢复、编排全塞进代理本体里，改得代理越来越臃肿。Pi-Harness 走的是另一条路：代理只负责干活，控制层负责记录、调度和兜底。这两者之间通过一个标准化的“事件流”协议通信，代理每做一步就往外吐一个结构化事件，控制层订阅这些事件做文章。

这么设计的好处很实在。第一，代理本身可以保持轻量，升级模型或者换代理实现时，控制层不用大改。第二，可观测和编排逻辑可以独立演进，今天你只想看日志，明天想加自动重试，后天想接告警，都是加订阅者的事，不用动代理代码。第三，出问题时排查边界清晰——是代理决策错了，还是控制层调度错了，一看事件流就知道。

我试过把控制逻辑直接写进代理的 prompt 里，让模型自己“记得”上报状态。实测下来非常不稳，模型经常忘了上报，或者上报格式五花八门。把这件事下沉到工程层，用代码而不是用提示词来保证，才是靠谱的做法。

2.2 三个核心能力的职责边界

Pi-Harness 的三个关键词——可观测、可恢复、可编排——不是三个独立模块，而是共享同一套事件底座的三层能力。理解它们的边界很重要，不然容易设计过度。

可观测解决的是“看得见”。它订阅代理的所有事件，包括思考过程、工具调用、文件读写、错误抛出，然后做结构化存储和实时展示。核心指标有三个：任务进度（完成了多少子步骤）、资源消耗（token 数、耗时、工具调用次数）、异常信号（连续失败、循环调用、超时）。

可恢复解决的是“挂得起”。它依赖可观测层产出的检查点（Checkpoint）。每完成一个原子步骤，控制层就把当前状态快照下来——包括代理的上下文、已改动的文件清单、执行到哪一步。恢复时从最近的检查点重建现场，而不是从头再来。这里的关键是原子步骤的粒度划分，太粗了恢复代价高，太细了快照开销大。

可编排解决的是“管得住”。它把多个代理或者多个任务阶段组织成一张有向图，节点是执行单元，边是依赖关系。编排层负责按拓扑顺序调度、处理失败重试、汇总结果。它不关心代理内部怎么干活，只关心“这个节点该不该跑、跑完没有、要不要重跑”。

2.3 事件流协议：控制层的“神经系统”

整套东西能转起来，靠的是一个统一的事件流协议。我定义了一个基础事件结构，所有代理动作都往这个结构上套：

{ "event_id": "uuid", "task_id": "任务唯一标识", "step_index": 12, "event_type": "tool_call | file_write | thought | error | checkpoint", "timestamp": 1710000000.123, "payload": { ... }, "parent_event_id": "可选，用于追踪因果链" }

这个结构看着简单，但有几个设计点值得说。step_index是单调递增的，恢复时靠它定位断点。parent_event_id用来串因果链，比如一个工具调用失败触发了重试，重试事件就指向原事件，排查时能一眼看出重试了几次。event_type用枚举而不是自由文本，是为了让控制层能做确定性处理，不用去解析自然语言。

代理侧只需要在关键节点调用一个emit_event()函数，剩下的序列化、传输、存储全由控制层接管。我实测下来，这种“代理主动上报 + 控制层被动订阅”的模式，比控制层去轮询代理状态要可靠得多，延迟也低。

3. 可观测层落地：把黑盒变成玻璃盒

3.1 事件采集的三种接入方式

代理往外吐事件，接入方式得看你的代理是怎么跑的。我总结了三种常见场景，各有各的接法。

第一种，代理是独立进程。这种情况最省事，让代理把事件写到标准输出或者一个本地 socket，控制层起一个采集器去读。标准输出的好处是零依赖，坏处是跟代理的正常日志混在一起，需要加个前缀区分。我一般用[HARNESS_EVENT]作为前缀，采集器只认这个前缀的行。

第二种，代理是库函数，跑在同一进程里。那就直接用一个内存队列，代理调emit_event()往队列里塞，控制层起一个消费者协程去取。这种方式延迟最低，但要注意队列满了会阻塞代理，得设个上限和丢弃策略。

第三种，代理跑在容器里。那就走网络，代理把事件 POST 到一个本地 HTTP 端点，控制层接收后落盘。这种方式隔离性好，但网络抖动会导致事件丢失，需要加本地缓冲和重传。

我自己的项目里用的是第二种加第三种混合：本地开发用内存队列，线上跑容器用 HTTP 上报，控制层统一做去重和排序。

3.2 结构化存储：别再用 print 打日志了

可观测的核心不是“有日志”，而是“日志能被机器读懂”。我见过太多项目，代理跑完吐出一大坨文本，出问题了只能人肉 grep。Pi-Harness 的做法是，所有事件先落成 JSON Lines 格式，一行一个事件，然后按需导入到查询引擎里。

存储选型上，我推荐分两层。热数据用 SQLite 或者 DuckDB，单文件、零运维，查询快，适合最近几小时的事件回溯。冷数据定期归档到对象存储，需要长期审计时再拉回来。别一上来就上 Elasticsearch，那玩意儿运维成本太高，小团队扛不住。

表结构我设计得很简单，就一张事件表加一张检查点表：

CREATE TABLE events ( event_id TEXT PRIMARY KEY, task_id TEXT NOT NULL, step_index INTEGER NOT NULL, event_type TEXT NOT NULL, timestamp REAL NOT NULL, payload TEXT, parent_event_id TEXT ); CREATE INDEX idx_task_step ON events(task_id, step_index); CREATE TABLE checkpoints ( checkpoint_id TEXT PRIMARY KEY, task_id TEXT NOT NULL, step_index INTEGER NOT NULL, created_at REAL NOT NULL, state_blob BLOB NOT NULL );

state_blob里存的是序列化后的代理上下文和文件快照。这里有个坑：文件快照别存全量，存 diff 就行。我一开始图省事存全量，一个中等项目跑下来快照文件几个 G，恢复时加载慢得要死。改成存相对上一个检查点的 diff 之后，体积降了两个数量级。

3.3 实时看板：三个必须盯住的指标

可观测不是把数据堆在那儿就完事，得让人一眼看出问题。我做了个简单的终端看板，只盯三个指标，但每个都直击要害。

进度指标：当前 step_index 除以预估总步数。预估总步数怎么来？代理启动时先让它输出一个任务分解，数一下子任务数量，乘以一个经验系数（我一般用 1.5，因为实际执行往往比计划多几步）。这个指标低于预期时，说明代理卡住了或者跑偏了。

消耗指标：累计 token 数和累计耗时。这两个要设阈值告警。我的经验值是，单个子任务 token 超过 50k 或者耗时超过 5 分钟，大概率是代理陷入了循环，该人工介入了。

异常指标：连续失败次数和重复调用次数。连续失败超过 3 次，基本可以判定这条路走不通，需要换策略。重复调用同一个工具且参数相同超过 2 次，说明代理在“鬼打墙”，得强制打断。

看板用rich库在终端里画就行，不用搞 Web 界面。终端看板的优势是跟开发环境在一起，不用切窗口，刷新频率设成 1 秒一次，对性能几乎没影响。

4. 可恢复层实现：断点续跑的工程细节

4.1 检查点粒度：多细才算合适

可恢复的核心是检查点，而检查点的核心是粒度。粒度太粗，恢复时回退太多，浪费算力；粒度太细，快照开销大，拖慢正常执行。我试过几种粒度，最后定在原子工具调用级别。

具体来说，代理每完成一次工具调用（比如读一个文件、写一个文件、跑一次测试），控制层就触发一次检查点。为什么选这个粒度？因为工具调用是代理执行的最小可观测单元，也是副作用产生的地方。文件写入这种操作，要么没发生，要么发生了，不存在中间态，快照起来干净。

但这里有个优化：不是每次工具调用都存全量快照。我做了个判断，只有产生副作用的工具调用（写文件、执行命令）才存文件 diff，只读操作（读文件、搜索）只存代理上下文。这样既保证了恢复的完整性，又控制了快照体积。

4.2 状态重建：恢复时到底要还原什么

恢复不是简单地把代理重新启动，而是要还原到一个“可以继续往下跑”的状态。我梳理了一下，需要还原的东西有四类：

第一，代理的对话上下文。这是最关键的，代理得知道之前干了什么、为什么这么干。我把上下文序列化成消息列表存进检查点，恢复时直接喂回去。

第二，文件系统的状态。代理改过的文件得还原到检查点时刻的样子。这里用 diff 链来重建：从初始状态开始，依次应用每个检查点的 diff，直到目标检查点。

第三，执行进度。当前跑到第几步、哪些子任务完成了、哪些还没开始。这个用 step_index 和任务分解计划对照就能算出来。

第四，外部副作用的状态。比如代理提交了一个 PR、发了一个通知，这些操作没法回滚，得记录下来，恢复时跳过或者做补偿。我一般用一个“已执行副作用清单”来管理，恢复时先检查这个清单。

4.3 恢复触发：自动还是手动

恢复的触发方式有两种：自动和手动。自动恢复适合那些明确的、可判定的失败，比如进程崩溃、网络超时。手动恢复适合那些需要人判断的场景，比如代理连续失败、输出质量明显下降。

我的做法是设一个恢复策略表，按失败类型决定触发方式：

自动恢复要加个熔断，重试超过上限就停下来等人。我踩过的坑是，不加熔断的话，一个必然失败的任务会无限重试，把资源全耗光。

5. 可编排层：多代理协作的调度逻辑

5.1 任务图：把工作流画成有向无环图

编排的本质是把一个大任务拆成一张有向无环图（DAG），节点是执行单元，边是依赖关系。Pi-Harness 里，我用一个简单的 YAML 来描述这张图：

nodes: - id: analyze agent: pi-coding prompt: "分析代码库结构，输出模块依赖图" - id: implement agent: pi-coding prompt: "根据依赖图实现新功能" depends_on: [analyze] - id: test agent: pi-coding prompt: "为新功能编写并运行测试" depends_on: [implement] - id: review agent: pi-coding prompt: "审查代码改动，输出审查意见" depends_on: [implement]

这个图里，analyze是起点，implement依赖它，test和review都依赖implement但彼此独立，可以并行跑。编排层要做的事就是：解析这张图，按拓扑顺序调度，处理并行和依赖。

5.2 调度器：拓扑排序加并发控制

调度器的核心逻辑不复杂，就是拓扑排序加一个并发池。但有几个细节决定了它好不好用。

并发度控制：不是所有能并行的节点都该并行。我设了一个全局并发上限（默认 3），防止同时跑太多代理把机器压垮。每个节点还可以单独设并发上限，比如跑测试的节点比较重，就限制它一次只跑一个。

依赖传递：下游节点怎么拿到上游节点的产出？我的做法是，每个节点执行完把结果写到一个共享的“上下文存储”里，下游节点启动时从存储里读自己依赖的那些节点的输出。这样解耦了节点之间的直接通信，调度器只管顺序，不管数据怎么传。

失败传播：一个节点失败了，它的下游节点怎么办？我给了两种策略：fail_fast和continue_on_error。前者是下游全部取消，后者是下游照跑，但标记为“依赖失败”。默认用fail_fast，因为依赖失败的结果往往没意义。

5.3 结果汇总：别让编排变成信息黑洞

多个节点跑完，结果得汇总成一个可读的报告。我设计了一个汇总器，把每个节点的状态、耗时、token 消耗、产出摘要收集起来，生成一份 Markdown 报告。报告里最关键的是失败节点的错误上下文，包括失败时的最后几步事件、当时的代理上下文、以及建议的排查方向。

汇总器还负责做冲突检测。多个节点如果改了同一个文件，汇总时要标出来，提示人工检查。我遇到过两个并行节点同时改一个配置文件，结果互相覆盖，最后跑出来的东西谁都不对。加了冲突检测之后，这种情况在汇总阶段就能发现。

6. 实操全流程：从零搭起一套控制层

6.1 环境准备与依赖安装

先把基础环境搭起来。我用的 Python 3.11，依赖不多，核心就几个：

pip install asyncio aiohttp rich duckdb pyyaml

asyncio和aiohttp负责异步事件处理，rich做终端看板，duckdb做事件存储和查询，pyyaml解析任务图。版本上没什么特别要求，都是成熟库，装最新稳定版就行。

目录结构我建议这样组织：

pi-harness/ ├── core/ │ ├── event_bus.py # 事件总线 │ ├── checkpoint.py # 检查点管理 │ └── scheduler.py # 调度器 ├── adapters/ │ └── pi_agent.py # Pi Coding Agent 适配器 ├── storage/ │ └── event_store.py # 事件存储 ├── dashboard/ │ └── monitor.py # 终端看板 └── configs/ └── task_graph.yaml # 任务图配置

这个结构的好处是控制层和代理适配器分离，换代理只改adapters目录，核心逻辑不动。

6.2 事件总线与检查点的代码实现

事件总线是整个系统的中枢，我用一个异步队列加一组订阅者来实现：

import asyncio from collections import defaultdict class EventBus: def __init__(self): self.queue = asyncio.Queue(maxsize=10000) self.subscribers = defaultdict(list) def subscribe(self, event_type, handler): self.subscribers[event_type].append(handler) async def publish(self, event): await self.queue.put(event) async def run(self): while True: event = await self.queue.get() handlers = self.subscribers.get(event["event_type"], []) handlers += self.subscribers.get("*", []) for handler in handlers: asyncio.create_task(handler(event))

这里有个细节：maxsize=10000是防止代理疯狂吐事件把内存撑爆。满了之后put会阻塞，相当于给代理一个背压信号，让它慢点。订阅者用create_task异步执行，避免一个慢订阅者拖垮整个总线。

检查点管理器的核心是快照和恢复两个方法：

import json import hashlib class CheckpointManager: def __init__(self, store): self.store = store self.file_states = {} # 当前文件状态缓存 def snapshot(self, task_id, step_index, context, changed_files): diff = {} for path, content in changed_files.items(): old = self.file_states.get(path, "") if old != content: diff[path] = {"old": old, "new": content} self.file_states[path] = content state = { "context": context, "file_diff": diff, "step_index": step_index } blob = json.dumps(state).encode() checkpoint_id = hashlib.sha256(blob).hexdigest()[:16] self.store.save_checkpoint(checkpoint_id, task_id, step_index, blob) return checkpoint_id

file_states缓存了当前所有文件的内容，这样算 diff 时不用去读磁盘，快很多。快照 ID 用内容哈希，天然去重，同样的状态不会存两份。

6.3 任务图配置与调度执行

任务图用 YAML 配置，调度器读进来后先做拓扑排序，然后按层执行：

import yaml from collections import deque class Scheduler: def __init__(self, graph_config, max_concurrency=3): self.nodes = {n["id"]: n for n in graph_config["nodes"]} self.max_concurrency = max_concurrency self.semaphore = asyncio.Semaphore(max_concurrency) def topological_layers(self): in_degree = {nid: 0 for nid in self.nodes} for node in self.nodes.values(): for dep in node.get("depends_on", []): in_degree[node["id"]] += 1 layers = [] ready = deque([nid for nid, d in in_degree.items() if d == 0]) while ready: layer = list(ready) layers.append(layer) ready.clear() for nid in layer: for other in self.nodes.values(): if nid in other.get("depends_on", []): in_degree[other["id"]] -= 1 if in_degree[other["id"]] == 0: ready.append(other["id"]) return layers async def execute(self, task_id): layers = self.topological_layers() results = {} for layer in layers: tasks = [self.run_node(task_id, nid, results) for nid in layer] layer_results = await asyncio.gather(*tasks, return_exceptions=True) for nid, res in zip(layer, layer_results): results[nid] = res return results async def run_node(self, task_id, node_id, results): async with self.semaphore: node = self.nodes[node_id] deps_output = {d: results.get(d) for d in node.get("depends_on", [])} return await self.invoke_agent(task_id, node, deps_output)

按层执行的好处是，同一层的节点天然可以并行，层与层之间天然有依赖，不用额外做同步。semaphore控制全局并发，防止层内节点太多把机器压垮。

6.4 终端看板的快速搭建

看板用rich的Live组件，每秒刷新一次：

from rich.live import Live from rich.table import Table from rich.console import Console class Dashboard: def __init__(self, event_store): self.store = event_store self.console = Console() def render(self, task_id): table = Table(title=f"Task {task_id}") table.add_column("指标") table.add_column("值") stats = self.store.get_stats(task_id) table.add_row("进度", f"{stats['progress']:.1%}") table.add_row("Token 消耗", str(stats["tokens"])) table.add_row("耗时", f"{stats['elapsed']:.1f}s") table.add_row("连续失败", str(stats["consecutive_failures"])) table.add_row("重复调用", str(stats["duplicate_calls"])) return table async def run(self, task_id): with Live(self.render(task_id), refresh_per_second=1) as live: while True: await asyncio.sleep(1) live.update(self.render(task_id))

看板只读事件存储，不干扰主流程。刷新频率 1 秒是权衡后的选择，再快没必要，再慢反应迟钝。

7. 常见问题与排查技巧实录

7.1 事件丢失与乱序怎么破

事件丢失最常见的原因是代理进程崩溃时缓冲区里的事件还没发出去。我的解法是本地先落盘再发送：代理每产生一个事件，先追加写到本地的一个 WAL（预写日志）文件，然后再发到控制层。控制层收到后回一个 ack，代理收到 ack 才删 WAL 里的记录。这样即使发送失败，重启后从 WAL 重放就行。

乱序问题主要出现在多进程并发上报时。解法是给每个事件带上step_index，控制层收到后不立即处理，而是按step_index排序后再处理。如果某个 step_index 迟迟不到，设一个超时（我设 30 秒），超时后标记为“缺失”继续往下走，避免一个丢失的事件卡住整个流程。

7.2 检查点恢复后代理“失忆”怎么办

这是恢复层最隐蔽的坑。代理恢复后，虽然上下文还原了，但它可能不记得自己之前为什么做某个决策。原因是上下文里只有“做了什么”，没有“为什么做”。我的解法是在检查点里额外存一个决策摘要，每次代理做出关键决策时，让它用一句话总结决策理由，存进检查点。恢复时把决策摘要也喂回去，代理就能接上思路。

另一个原因是工具调用的副作用没还原。比如代理之前创建了一个临时文件，恢复时这个文件不存在了，代理再引用它就报错。解法是检查点里记录所有副作用的清单，恢复时先重建这些副作用。

7.3 编排层死锁的排查思路

死锁通常表现为任务图卡在某一层不动了。排查步骤我总结成一张表：

我遇到过一次信号量泄漏，原因是run_node里抛异常时没有正确释放 semaphore。后来改成用async with上下文管理器，异常时自动释放，问题就没了。

7.4 性能调优的几个关键参数

跑了一段时间后，我调过几个参数，效果比较明显：

事件批量写入：单条写 DuckDB 很慢，改成攒 100 条或 1 秒批量写，吞吐提升 10 倍以上。

检查点压缩：state_blob用 zstd 压缩后再存，体积降 70%，恢复时解压开销可以忽略。

看板降频：从每秒刷新改成每 2 秒，CPU 占用降了一半，肉眼几乎感觉不到差别。

并发上限：从 3 调到 5 再调回 3，发现 5 的时候机器负载太高，代理响应变慢，整体耗时反而增加。最终定在 3，是这台机器的甜点值。

8. 我踩过的坑和几条实在建议

8.1 别把控制层做成第二个代理

我一开始犯的错，是让控制层去“理解”代理的输出，比如解析代理的自然语言回复来判断任务是否完成。结果控制层越来越复杂，最后变成了一个用规则写的蹩脚代理。后来我改了原则：控制层只处理结构化事件，不碰自然语言。代理要上报状态，就用结构化字段，别用自然语言描述。这样控制层逻辑简单、确定、好维护。

8.2 检查点不是越多越好

我一度以为检查点越密越安全，每个工具调用都存全量快照。结果跑一个大任务，快照文件几十个 G，恢复时加载慢到怀疑人生。后来改成只存 diff、只对副作用操作存快照，体积降了两个数量级，恢复速度从分钟级降到秒级。检查点的价值在于能恢复，不在于数量多。

8.3 编排的复杂度要匹配任务规模

不是所有任务都值得上 DAG 编排。如果就是一个线性任务，写个顺序执行的脚本就够了，上编排层是过度设计。我一般用这个标准判断：任务有并行分支、有失败重试需求、有多个代理协作，三者占其一才上编排。否则就是给自己找麻烦。

8.4 可观测数据要定期清理

事件表涨得很快，一个中等任务跑下来几万条事件。如果不清理，DuckDB 文件几个月就上 G。我的做法是设一个保留策略：热数据保留 7 天，7 天前的归档到冷存储，30 天前的直接删。归档和删除用定时任务做，别手动清，容易忘。

8.5 恢复策略要能人工覆盖

自动恢复很方便，但有时候自动恢复会陷入“恢复-失败-再恢复”的循环。我加了一个人工覆盖开关，检测到同一个任务自动恢复超过 3 次，就停下来等人介入。人介入时可以手动指定从哪个检查点恢复、要不要跳过某个失败节点。这个开关救过我好几次，避免了一个坏任务把资源全耗光。

这套 Pi-Harness 我陆陆续续迭代了两个多月，从最初只能看日志，到现在能自动恢复、能编排多代理，中间踩的坑比写代码的时间还多。但每次看到代理挂掉后能自己从断点续上，或者多个代理并行跑完一个复杂任务，就觉得这些功夫没白费。如果你也在把编码代理往生产环境推，建议从可观测层开始做，先把“看得见”解决了，再考虑恢复和编排。顺序反了，后面会很难受。