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 恢复触发自动还是手动恢复的触发方式有两种自动和手动。自动恢复适合那些明确的、可判定的失败比如进程崩溃、网络超时。手动恢复适合那些需要人判断的场景比如代理连续失败、输出质量明显下降。我的做法是设一个恢复策略表按失败类型决定触发方式失败类型触发方式最大重试次数回退步数进程崩溃自动30从最近检查点续工具调用超时自动21回退一步重试连续失败手动-由人指定输出质量差手动-由人指定自动恢复要加个熔断重试超过上限就停下来等人。我踩过的坑是不加熔断的话一个必然失败的任务会无限重试把资源全耗光。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 pyyamlasyncio和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(maxsize10000) 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))这里有个细节maxsize10000是防止代理疯狂吐事件把内存撑爆。满了之后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_idfile_states缓存了当前所有文件的内容这样算 diff 时不用去读磁盘快很多。快照 ID 用内容哈希天然去重同样的状态不会存两份。6.3 任务图配置与调度执行任务图用 YAML 配置调度器读进来后先做拓扑排序然后按层执行import yaml from collections import deque class Scheduler: def __init__(self, graph_config, max_concurrency3): 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_exceptionsTrue) 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(titlefTask {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_second1) 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 编排层死锁的排查思路死锁通常表现为任务图卡在某一层不动了。排查步骤我总结成一张表现象可能原因排查方法某层一直不执行依赖未满足检查上游节点是否完成节点一直 running代理卡住看该节点的事件流最后一条并发数上不去信号量泄漏检查 semaphore 是否被正确释放层与层之间卡住拓扑排序有环用图算法检测环我遇到过一次信号量泄漏原因是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 我陆陆续续迭代了两个多月从最初只能看日志到现在能自动恢复、能编排多代理中间踩的坑比写代码的时间还多。但每次看到代理挂掉后能自己从断点续上或者多个代理并行跑完一个复杂任务就觉得这些功夫没白费。如果你也在把编码代理往生产环境推建议从可观测层开始做先把“看得见”解决了再考虑恢复和编排。顺序反了后面会很难受。
