1. 为什么需要给 Pi Coding Agent 加一层工程控制1.1 从“能跑”到“敢用”的鸿沟Pi Coding Agent 这类编码智能体刚上手的时候确实惊艳丢一段需求它能自己读文件、改代码、跑命令甚至把测试也顺手补上。但真把它放进日常工程流里跑几天问题就冒出来了。最典型的三类第一它跑到一半卡住了你不知道它卡在哪一步、上下文里到底塞了什么第二它改崩了代码你想回滚却发现中间状态根本没留痕第三你想让它同时处理“修 bug”和“补文档”两件事结果两个任务互相踩脚文件锁冲突、上下文串味。这三个问题分别对应三个能力缺口可观测、可恢复、可编排。Pi-Harness 要补的就是这一层。它不是去改 Pi Coding Agent 的推理内核而是在它外面套一层工程控制层把 Agent 的每一次思考、每一次工具调用、每一次文件写入都变成可记录、可回放、可干预的事件流。我自己的判断是一个编码 Agent 能不能进生产不取决于它单次任务跑得多漂亮而取决于它出错的时候你能不能三分钟内定位、五分钟内恢复。Pi-Harness 的价值就在这。1.2 谁适合看这套东西如果你只是拿 Pi Coding Agent 写写脚本、做做 demo那这层控制层对你来说偏重。但如果你符合下面任意一条这套东西值得你花时间你要把编码 Agent 接进 CI/CD让它自动处理 issue、自动提 PR你团队里多个人共用一套 Agent 服务需要隔离和审计你跑的是长任务比如跨几十个文件的重构中途失败重来的成本很高你需要向别人解释“这个 Agent 到底干了什么”也就是要有可追溯的执行记录。说白了Pi-Harness 面向的是把 Agent 当“工程组件”而不是“玩具”的人。1.3 整体设计思路控制层与推理层解耦Pi-Harness 的核心设计原则只有一条推理归推理控制归控制。Pi Coding Agent 负责“想”Harness 负责“记、管、编排”。两者之间通过一个标准化的事件总线通信Agent 每产生一个动作思考、调用工具、写文件、执行命令就向总线发一个事件Harness 订阅这些事件做持久化、做状态机推进、做并发调度。这么设计的好处是Agent 内核可以独立升级Harness 不用跟着改反过来 Harness 想换存储、换调度策略也不影响 Agent。坏处是多了一层序列化开销但对编码任务这种秒级粒度的场景这点开销可以忽略。提示不要试图把控制逻辑塞进 Agent 的 prompt 里。用 prompt 做编排任务一复杂就会失控而且完全不可观测。控制层必须是代码不是提示词。2. 可观测层把 Agent 的黑盒变成事件流2.1 事件模型怎么设计可观测的第一步是定义清楚“什么算一个事件”。Pi-Harness 里我把事件分成五类覆盖 Agent 执行的全生命周期事件类型触发时机关键字段session.start一次任务开始session_id, task_desc, repo_pathstep.thinkAgent 产生一次推理step_id, thought, token_usagetool.call调用工具前后tool_name, args, result, durationfile.write写文件前后path, diff, before_hash, after_hashsession.end任务结束status, total_steps, error这里有个容易踩的坑tool.call一定要记录调用前和调用后两个时间点而不是只记结果。因为排查问题时你经常需要知道“这个命令跑了多久”“是不是超时了”。只记结果的话超时和快速失败你分不出来。file.write里的before_hash和after_hash是恢复能力的基础后面讲恢复时会展开。diff字段建议存 unified diff 格式体积小、可读性好回放的时候直接能看。2.2 事件持久化为什么选追加写而不是数据库事件存储我最终选了追加写的 JSONL 文件而不是直接上 SQLite 或 Postgres。理由有三条第一编码 Agent 的事件是天然有序的追加写完美契合这个特性不需要事务、不需要锁。第二JSONL 可以直接tail -f看排查问题时不用起数据库客户端运维成本极低。第三出问题时文件就是证据可以直接打包发给别人不依赖任何环境。当然 JSONL 也有代价查询能力弱。所以我的做法是双写——JSONL 作为 source of truth同时异步写一份到 SQLite 做索引支持按 session、按文件路径、按时间范围查询。查询走 SQLite回放和审计走 JSONL。import json, time, hashlib from pathlib import Path class EventStore: def __init__(self, session_dir: Path): self.session_dir session_dir self.session_dir.mkdir(parentsTrue, exist_okTrue) self.log_path session_dir / events.jsonl def emit(self, event_type: str, payload: dict): event { ts: time.time(), type: event_type, payload: payload, } with self.log_path.open(a, encodingutf-8) as f: f.write(json.dumps(event, ensure_asciiFalse) \n) return event这段代码看着简单但有两个细节值得说。ensure_asciiFalse是为了让中文注释、中文 commit message 在日志里可读不然全是\uXXXX。另外ts用浮点秒而不是格式化字符串是为了后续做时间差计算方便展示的时候再格式化。2.3 实时观测把事件流接到终端光有日志还不够长任务跑起来你总得知道现在到哪了。Pi-Harness 提供了一个轻量的 TUI 观测器订阅事件流并实时渲染。核心就是一个 tail 解析 渲染的循环def watch(log_path: Path): with log_path.open(r, encodingutf-8) as f: f.seek(0, 2) # 跳到文件末尾 while True: line f.readline() if not line: time.sleep(0.2) continue event json.loads(line) render(event)seek(0, 2)这个操作是让它只关注新事件不回放历史。如果你想从头看去掉这行就行。render函数按事件类型做不同展示tool.call显示工具名和耗时file.write显示文件路径和增删行数step.think折叠显示因为思考内容通常很长。实测下来这个观测器对调试帮助极大。以前 Agent 卡住我只能干等现在能立刻看到它是在等某个命令返回还是在反复读同一个文件——后者通常意味着它陷入了循环需要人工干预。2.4 观测数据的三个实用场景事件流存下来不是摆设我实际用得最多的是三个场景。场景一复盘失败任务。任务失败了直接grep status: error找到出错的那一步然后往前翻 20 条事件基本能还原出错的上下文。比让 Agent 自己解释“我为什么失败”靠谱得多因为 Agent 的自我解释经常是编的。场景二统计工具使用分布。把tool.call事件按tool_name聚合你能看出 Agent 大部分时间花在哪。我统计过一次发现它 60% 的时间在read_file说明我的任务描述太模糊它需要反复读文件找线索。后来我把关键文件路径直接写进任务描述效率提升明显。场景三成本核算。step.think里的token_usage累加起来就是一次任务的真实成本。多 Agent 场景下这个数据能帮你判断哪个 Agent 是“吞金兽”该优化它的 prompt 还是该限制它的步数。3. 可恢复层让中断的任务能接着跑3.1 恢复的本质是状态快照“可恢复”听起来玄乎本质就一句话在关键节点把状态存下来恢复时从最近的快照继续。难点在于“状态”包含什么。对编码 Agent 来说状态至少有三部分对话状态Agent 的上下文消息列表这是它“记得什么”文件状态工作区里所有被改过的文件内容这是它“改了什么”执行状态当前进行到第几步、哪些步骤已完成这是它“走到哪了”。三者缺一不可。只存对话恢复后文件对不上只存文件恢复后 Agent 不知道自己干过什么只存执行进度前两者都没了等于重来。3.2 快照策略全量还是增量全量快照简单粗暴每 N 步把整个工作区打包。问题是编码项目动辄几百 MB每步都全量存磁盘扛不住。所以 Pi-Harness 用的是增量快照 定期全量的混合策略。增量快照基于file.write事件里的 diff。每次写文件把 diff 追加到一个 patch 文件里。恢复时从最近一次全量快照开始依次应用后续的 patch就能还原到任意一步。class SnapshotManager: def __init__(self, repo: Path, snapshot_dir: Path): self.repo repo self.snapshot_dir snapshot_dir self.patch_file snapshot_dir / patches.diff def record_write(self, path: str, before: str, after: str): diff unified_diff( before.splitlines(keependsTrue), after.splitlines(keependsTrue), fromfilefa/{path}, tofilefb/{path}, ) with self.patch_file.open(a, encodingutf-8) as f: f.writelines(diff) def restore_to(self, step: int): base self._latest_full_before(step) self._apply_full(base) self._apply_patches_until(step)全量快照的触发条件我设了两个每 50 步一次或者累计 patch 体积超过 10MB 一次。前者保证恢复粒度不会太粗后者防止 patch 文件无限膨胀。注意patch 应用一定要用git apply而不是自己写解析器。unified diff 的边界情况空文件、二进制文件、换行符差异非常多自己写迟早翻车。用 git 现成的能力稳。3.3 恢复流程与幂等性恢复的时候有个隐蔽的坑工具调用的副作用。比如 Agent 执行了git commit恢复后如果重放这一步会多出一个 commit。所以恢复不能简单重放必须做幂等处理。我的做法是给每个工具调用打一个idempotency_key通常是session_id step_id tool_name的哈希。恢复时先查这个 key 有没有执行过执行过就跳过直接读缓存的结果。对于有副作用的工具写文件、执行命令这个机制能避免重复执行。def execute_tool(tool_name, args, step_id, session_id): key hashlib.sha256( f{session_id}:{step_id}:{tool_name}.encode() ).hexdigest() cached cache.get(key) if cached is not None: return cached result dispatch(tool_name, args) cache.set(key, result) return result这个缓存表本身也要持久化不然进程重启就丢了。我把它和事件流存在同一个 session 目录下用 SQLite 存key 做唯一索引。3.4 恢复的边界哪些情况救不回来得诚实说不是所有中断都能恢复。下面几种情况我踩过基本只能重来外部状态被改了Agent 依赖的远程分支被别人 force push 了本地 patch 应用不上非文件副作用Agent 调了某个 API 创建了资源恢复时这个资源已经存在重复创建会报错上下文超限恢复时对话历史太长塞不进模型上下文窗口只能截断截断后 Agent 行为可能漂移。所以我的建议是恢复能力是兜底不是常态。真正重要的是把任务拆小让单次执行时间短到不容易中断。Pi-Harness 的编排层就是干这个的。4. 可编排层多 Agent 协作与任务调度4.1 为什么单 Agent 不够用单 Agent 处理复杂任务时最大的问题是上下文污染。让它先修 bug 再写文档修 bug 过程中读的一堆源码会占满上下文写文档时它脑子里全是代码细节写出来的东西又干又硬。而且单 Agent 是串行的修 bug 和写文档本来可以并行串行就是浪费时间。多 Agent 编排解决的就是这两个问题隔离上下文和并行执行。每个 Agent 有自己的上下文窗口互不干扰调度器负责把任务分给合适的 Agent并管理它们的依赖关系。4.2 任务图用 DAG 描述依赖Pi-Harness 用有向无环图DAG来描述任务依赖。每个节点是一个子任务边表示“A 完成后 B 才能开始”。比如“重构模块”这个任务可以拆成parse_ast ── refactor_core ── run_tests ── update_docs │ └── update_changelogrun_tests依赖refactor_coreupdate_docs和update_changelog都依赖run_tests通过。调度器按拓扑序执行同一层的节点可以并行。from collections import deque def schedule(dag): indegree {n: 0 for n in dag.nodes} for n in dag.nodes: for m in dag.successors(n): indegree[m] 1 ready deque([n for n in dag.nodes if indegree[n] 0]) while ready: batch list(ready) ready.clear() results run_parallel(batch) # 同层并行 for n, ok in results.items(): if not ok: raise TaskFailed(n) for m in dag.successors(n): indegree[m] - 1 if indegree[m] 0: ready.append(m)这个调度器有个细节同层并行但不同层严格串行。为什么不干脆全并行因为依赖关系必须尊重run_tests没通过就写文档写出来的文档可能和实际代码对不上。4.3 Agent 隔离文件锁与工作区多 Agent 并行最大的风险是文件冲突。两个 Agent 同时改同一个文件后写的覆盖先写的先写的白干。Pi-Harness 用两层防护第一层是声明式文件锁。每个子任务在定义时声明它会碰哪些文件调度器在派发前检查这些文件有没有被其他运行中的任务占用有就排队。第二层是工作区隔离。每个 Agent 在自己的 git worktree 里干活改完再合并回主工作区。这样即使锁没拦住冲突也会在合并时暴露出来而不是静默覆盖。# 为每个 agent 创建独立 worktree git worktree add ../agent-1-worktree -b agent-1-branch git worktree add ../agent-2-worktree -b agent-2-branch用 worktree 而不是 clone是因为 worktree 共享同一个.git目录省磁盘而且分支管理更清晰。合并的时候用git merge --no-ff保留每个 Agent 的提交历史方便追溯。4.4 编排层的失败处理多 Agent 场景下失败处理比单 Agent 复杂得多。一个 Agent 挂了依赖它的下游全得停。Pi-Harness 的策略是快速失败 局部重试节点失败立即标记该节点及其所有下游为blocked不再派发失败节点本身允许重试 N 次默认 2 次重试时用恢复层的能力从最近快照继续如果重试仍失败整个 DAG 暂停等人工介入。这里有个经验重试次数不要设太多。我一开始设了 5 次结果一个必然失败的任务比如依赖的外部服务挂了重试 5 次浪费了十几分钟。后来改成 2 次快速暴露问题人工介入反而更快。5. 实操从零搭一套 Pi-Harness5.1 环境准备与依赖安装先把基础环境弄好。Pi-Harness 本身是 Python 写的依赖不多主要是 git 和 Python 3.10。# 确认 git 版本worktree 需要 2.5 git --version # 确认 python 版本 python3 --version # 创建虚拟环境 python3 -m venv .venv source .venv/bin/activate # Windows 用 .venv\Scripts\activate # 安装依赖 pip install pydantic sqlite-utils richpydantic用来定义事件和任务的 schema做校验sqlite-utils简化 SQLite 操作rich用来做终端渲染。这三个是核心依赖其他都是可选的。提示如果你在 Windows 上跑git worktree 的路径处理要注意反斜杠问题。建议统一用pathlib.Path它会自动处理平台差异。我踩过一次坑硬编码的../agent-1在 Windows 上创建到了错误的位置。5.2 初始化 Harness 与配置初始化就是建目录结构、写配置文件。目录结构我设计成这样.pi-harness/ ├── config.yaml ├── sessions/ │ └── session_id/ │ ├── events.jsonl │ ├── patches.diff │ ├── snapshots/ │ └── cache.db └── worktrees/配置文件用 YAML主要配这几项agent: max_steps: 100 timeout_seconds: 600 model: your-model-name snapshot: full_interval: 50 patch_size_limit_mb: 10 orchestration: max_parallel: 4 retry_times: 2max_parallel别设太大。我试过设 8结果机器 CPU 打满Agent 之间抢资源反而比串行还慢。4 是个比较稳的值具体看你机器配置。5.3 接入 Pi Coding Agent接入的核心是实现一个适配器把 Pi Coding Agent 的输出转成 Harness 的事件。不同版本的 Agent 接口可能不一样但思路一致拦截它的工具调用包装成事件。class PiAgentAdapter: def __init__(self, agent, store: EventStore): self.agent agent self.store store def run(self, task: str): self.store.emit(session.start, {task_desc: task}) try: for step in self.agent.iter_steps(task): if step.kind think: self.store.emit(step.think, { thought: step.content, token_usage: step.tokens, }) elif step.kind tool: self.store.emit(tool.call, { tool_name: step.name, args: step.args, result: step.result, duration: step.duration, }) elif step.kind write: self.store.emit(file.write, { path: step.path, diff: step.diff, }) self.store.emit(session.end, {status: ok}) except Exception as e: self.store.emit(session.end, { status: error, error: str(e), }) raise这个适配器的关键是异常也要发session.end事件。不然任务崩了事件流里没有结束标记观测器会一直等恢复逻辑也不知道任务已经终止。5.4 跑一个多 Agent 任务配置好之后定义一个 DAG 就能跑。下面是一个“修 bug 补测试 更新文档”的三节点任务dag DAG() dag.add_node(fix_bug, agentcoder, files[src/core.py]) dag.add_node(add_test, agenttester, files[tests/test_core.py]) dag.add_node(update_doc, agentwriter, files[docs/core.md]) dag.add_edge(fix_bug, add_test) dag.add_edge(add_test, update_doc) harness Harness(config.pi-harness/config.yaml) harness.run(dag)跑起来之后观测器会实时显示每个节点的状态。fix_bug和add_test因为文件不冲突理论上可以并行但这里我加了依赖边所以是串行。如果你确认它们不冲突去掉add_edge就能并行。5.5 恢复一个中断的任务假设任务跑到add_test时机器断电了。重启后恢复pi-harness resume --session session_id --from-step 37Harness 会做这几件事读events.jsonl找到最后一条成功事件从snapshots/找最近的全量快照应用patches.diff到第 37 步然后从第 38 步继续。整个过程对 Agent 是透明的它以为自己一直在跑。实测下来恢复一个 50 步的任务大概花 3-5 秒主要是应用 patch 的时间。如果 patch 特别多可以考虑更频繁地做全量快照用空间换时间。6. 常见问题与排查实录6.1 事件流排查速查表现象可能原因排查方法任务卡住不动工具调用超时看最后一条tool.call的 duration恢复后行为异常上下文截断检查恢复时的 token 数文件冲突锁没生效查file.write事件的 path 重叠重复提交幂等 key 失效查 cache.db 是否有对应记录事件流断档进程被 kill看session.end是否存在这张表是我实际排查时总结的覆盖了 80% 的问题。剩下 20% 通常是环境问题比如 git 版本太低、磁盘满了。6.2 几个踩过的坑坑一git worktree 的清理。任务结束后如果不删 worktreegit worktree list会越来越长而且分支残留。我写了个清理脚本在session.end时自动执行git worktree remove。但要注意如果任务失败了worktree 要保留方便事后排查。坑二patch 的换行符。Windows 上写的文件是 CRLFLinux 上是 LFpatch 应用时会报错。解决办法是在仓库根目录放一个.gitattributes强制* textauto eollf统一换行符。坑三SQLite 并发写。多个 Agent 同时写 cache.db 会锁表。我一开始用默认配置频繁报database is locked。后来开了 WAL 模式问题解决conn.execute(PRAGMA journal_modeWAL) conn.execute(PRAGMA busy_timeout5000)WAL 模式允许读写并发busy_timeout让写操作等待而不是立即失败。这两个配置一加并发问题基本消失。坑四上下文窗口的估算。恢复时判断上下文是否超限不能简单按字符数算。中英文 token 比例不一样代码和自然语言也不一样。我的经验值是代码按 3 字符/token中文按 1.5 字符/token英文按 4 字符/token。混着算的时候取保守值宁可截断早一点也别超限报错。6.3 性能调优的几个参数跑久了会发现几个参数对性能影响很大值得单独调snapshot.full_interval调小恢复快但占磁盘调大反之。50 是个平衡点任务步数少的话可以调到 20。orchestration.max_parallel不是越大越好受 CPU 和 IO 限制。先用 4 跑观察 CPU 利用率没打满再往上加。agent.max_steps设太小任务跑不完设太大跑飞了浪费。我一般按任务复杂度估简单任务 30复杂重构 150。这些参数没有万能值得根据你的任务类型和机器配置调。建议先跑几个典型任务看事件流里的实际数据再定参数。6.4 安全与审计的注意事项最后说几个安全相关的点。事件流里会记录文件内容和命令如果涉及敏感信息密钥、内部地址要注意脱敏。我的做法是在emit之前过一层过滤器匹配到敏感模式就替换成***。另外多 Agent 场景下要限制每个 Agent 的权限。比如writer类型的 Agent 不应该有执行 shell 命令的能力只给它文件读写权限。这个在适配器层做给不同 agent 配不同的工具白名单。审计方面events.jsonl本身就是审计日志。谁在什么时候改了什么文件、执行了什么命令一目了然。如果要满足更严格的合规要求可以定期把事件流归档到对象存储保留期限按你的规范来。这套东西我用了大半年最大的体会是Agent 的能力上限由模型决定但 Agent 能不能用得放心由控制层决定。Pi-Harness 这层控制不复杂但缺了它Agent 就永远停留在“玩具”阶段。把可观测、可恢复、可编排这三件事做扎实编码 Agent 才真正算得上工程组件。