1. 为什么需要给 Pi Coding Agent 加一层工程控制1.1 从“能跑”到“跑得稳”之间的鸿沟Pi Coding Agent 这类工具刚上手的时候最让人兴奋的是它真的能自己读代码、改文件、跑命令。你给它一个任务它就开始一轮一轮地思考、调用工具、观察结果看起来像模像样。但只要你把它放到稍微真实一点的场景里问题马上就暴露出来了。我最早用它做一个小重构任务是“把项目里所有requests调用替换成httpx并保证测试通过”。前几轮它干得挺漂亮读文件、改 import、跑测试。但跑到第七八轮的时候它开始重复读同一个文件改完又改回去测试报错信息它看了三遍还在原地打转。最后 token 烧完了任务没完成我连它到底改了哪些文件、为什么卡住都说不清楚。这就是典型的“能跑但不可控”。Agent 的循环本质是一个 while 循环思考、行动、观察、再思考。没有外部约束的话它可能陷入死循环、可能越改越乱、可能在中途因为一次网络抖动就整个崩掉而且崩了之后你没法从断点恢复只能从头再来。Pi-Harness 要解决的就是这个问题。它不是去改 Pi Coding Agent 的模型能力而是在它外面套一层工程控制层把“可观测、可恢复、可编排”这三件事补齐。说白了模型负责聪明Harness 负责靠谱。1.2 三个关键词到底指什么先把这三个词拆开讲清楚不然后面全是空话。可观测指的是 Agent 每一步在干什么、花了多少 token、调用了哪个工具、输入输出是什么、耗时多久这些都要能被记录下来并且能回放。没有可观测性调试 Agent 就像在黑箱里猜谜。你只能看到最终结果中间发生了什么全靠脑补。可恢复指的是 Agent 跑到一半挂了或者你主动中断了下次能从上次的状态继续而不是从零开始。这要求把 Agent 的运行时状态对话历史、已修改文件、工具调用记录持久化下来并且能重新加载。可编排指的是多个 Agent 或者多个任务之间能协同工作。比如一个 Agent 负责写代码一个负责 review一个负责跑测试它们之间怎么传递上下文、怎么串行或并行、怎么处理依赖关系这些都需要编排层来管。这三个能力单独看都不新鲜Git 本身就有版本控制的可恢复性CI 系统有可观测性工作流引擎有编排能力。但把它们组合起来专门服务于 Coding Agent 的运行时这就是 Pi-Harness 的价值所在。1.3 谁适合参考这套思路如果你只是偶尔用 Agent 写个小脚本那确实不需要这么重的控制层。但如果你符合下面任意一条这套东西就值得认真看你要用 Agent 处理真实项目代码量大、依赖多、测试套件跑一次要几分钟你需要 Agent 长时间运行比如批量重构、批量补测试、批量修 lint你要把 Agent 接入 CI/CD让它自动处理 issue 或者 PR你需要多个 Agent 协作完成一个复杂任务你被 Agent 的“跑飞”坑过想搞清楚它到底在哪一步出了问题我自己的经验是一旦 Agent 的单次任务超过 5 分钟或者超过 20 轮工具调用没有 Harness 基本就是在赌运气。2. 整体架构设计与核心思路拆解2.1 为什么选择“外挂控制层”而不是改 Agent 内核这里有个关键的设计决策Pi-Harness 是套在 Pi Coding Agent 外面的而不是改它的源码。这个选择背后有几个考虑。第一解耦。Agent 内核的迭代速度很快模型、prompt、工具集都可能变。如果把控制逻辑写进内核每次内核升级你都要重新适配。外挂层通过标准接口通常是 stdin/stdout 的 JSON 流或者 HTTP API和 Agent 通信内核怎么变都不影响控制层。第二可替换。今天用 Pi Coding Agent明天可能换别的 Agent。控制层只要定义好抽象接口底层 Agent 是可以替换的。这就像 Git 的 remote 机制不管你连的是哪个远端本地的操作逻辑是一样的。第三可测试。控制层独立出来之后你可以用 mock 的 Agent 来测试控制逻辑不用真的跑模型。这对调试编排逻辑特别重要不然每次测试都要烧 token。我踩过的一个坑是早期我图省事直接把重试逻辑写在了 Agent 的 prompt 里让它“如果失败就重试”。结果模型有时候会忘记重试有时候会无限重试。后来把重试逻辑抽到 Harness 层用代码控制才稳定下来。能用代码控制的逻辑绝对不要交给模型去判断这是我在 Agent 工程里最重要的心得。2.2 运行时状态机的设计Harness 的核心是一个状态机。Agent 的每一轮循环在 Harness 看来就是一次状态转移。我把它抽象成这么几个状态状态含义触发下一步的条件IDLE空闲等待任务收到任务描述THINKINGAgent 正在生成下一步动作收到工具调用请求或最终答案ACTING正在执行工具调用工具返回结果OBSERVING处理工具返回结果结果注入完成CHECKPOINT保存状态快照快照写入成功DONE任务完成-FAILED任务失败-这个状态机看起来简单但它是可恢复性的基础。每次状态转移Harness 都会把当前状态、对话历史、工具调用记录写到一个 checkpoint 文件里。如果进程挂了重启后从最后一个 checkpoint 加载就能继续跑。checkpoint 的粒度是个需要权衡的点。太粗恢复后要重跑很多太细写文件的开销大。我的经验是每完成一次工具调用就 checkpoint 一次因为工具调用往往是有副作用的改了文件、跑了命令重跑代价高。而 THINKING 阶段是纯计算重跑代价低不需要 checkpoint。2.3 可观测性的数据模型可观测性不是简单地打日志而是要有一套结构化的数据模型。我设计的事件结构大概是这样{ event_id: evt_20240115_001, timestamp: 2024-01-15T10:23:45.123Z, session_id: sess_abc123, step: 7, event_type: tool_call, tool_name: edit_file, tool_input: {path: src/main.py, old: ..., new: ...}, tool_output: {success: true, lines_changed: 3}, tokens: {prompt: 1523, completion: 89}, duration_ms: 2340, state_before: THINKING, state_after: OBSERVING }这个结构的关键在于它把每一步的因果关系都记录下来了。你能看到第 7 步调用了 edit_file输入是什么输出是什么花了多少 token耗时多久状态怎么转移的。有了这些数据你就能做很多分析哪一步最耗时、哪一步 token 消耗最大、哪一步开始出现重复调用。我实测下来用这套数据模型定位一个“Agent 卡住”的问题从原来的半小时缩短到五分钟。直接看事件流找到第一个重复的 tool_call问题基本就在那附近。2.4 编排层的抽象编排层要解决的是“多个 Agent 怎么协作”。我把它抽象成三个概念任务Task、角色Role、流水线Pipeline。任务是原子单位比如“修改文件 A 的函数 B”。角色是执行任务的 Agent 配置比如“写代码的 Agent”和“review 的 Agent”有不同的 prompt 和工具集。流水线是任务的依赖图定义了任务之间的先后顺序和数据传递。这里有个设计上的取舍是用 DAG有向无环图还是用更简单的线性流水线DAG 更灵活但实现复杂而且 Agent 任务往往有不确定性你很难提前把依赖关系定死。我的选择是线性流水线为主支持条件分支。大部分 Coding Agent 的场景是“改代码 → 跑测试 → 如果失败就修 → 再跑测试”这种用线性流水线加循环就够了。3. 核心细节解析与实操要点3.1 Checkpoint 的存储格式与恢复逻辑Checkpoint 存什么、怎么存直接决定了恢复能力的天花板。我试过几种方案最后落在一个比较务实的组合上。存储内容分三部分对话历史完整的 messages 数组包括 system、user、assistant、tool 各种角色文件系统快照不是全量快照而是记录 Agent 修改过的文件列表和每个文件的 diff运行时元数据当前 step、状态、token 累计消耗、开始时间为什么文件系统用 diff 而不是全量快照因为全量快照太占空间一个中等项目几百 MB每步都存一遍磁盘直接爆。用 diff 的话只存 Agent 改动的部分通常几 KB 到几十 KB。恢复的时候逻辑是反过来的先加载对话历史然后按顺序应用所有 diff把文件系统恢复到中断时的状态最后从记录的 step 继续。def restore_from_checkpoint(checkpoint_path, workspace): ckpt load_json(checkpoint_path) # 1. 恢复对话历史 messages ckpt[messages] # 2. 按顺序应用 diff for diff_record in ckpt[file_diffs]: apply_diff(workspace, diff_record) # 3. 恢复运行时状态 state ckpt[runtime_state] return messages, state注意diff 的应用顺序很重要必须严格按照记录的顺序来。如果 Agent 对同一个文件改了三次你乱序应用就会得到错误的结果。我在早期版本就踩过这个坑后来在 diff 记录里加了单调递增的序号才解决。3.2 工具调用的幂等性处理可恢复性有个隐藏的前提工具调用必须是幂等的或者至少能被安全地重放。但现实是很多工具调用不幂等。比如git commit你重放一次就多一个 commit比如rm file重放一次就报错。Harness 的处理策略是给每个工具调用打标记只读工具read_file、list_dir、grep天然幂等重放无副作用可重放工具edit_file、write_file通过 diff 记录实现幂等重放时先检查当前状态是否已经是目标状态不可重放工具git commit、npm install、rm执行前先记录“意图”执行后记录“结果”恢复时如果发现意图已记录但结果未记录就标记为“不确定状态”需要人工介入这个分类不是绝对的你可以根据自己的工具集调整。关键是要在工具注册的时候就声明它的幂等性而不是等到出问题才想。我遇到过一个很典型的问题Agent 执行git commit之后进程挂了恢复时它又执行了一次git commit结果多了一个空 commit。后来我在 Harness 里加了一个规则对于不可重放工具执行前先写一个 pending 标记执行成功后改成 done恢复时遇到 pending 就暂停并提示。这个改动之后类似问题再没出现过。3.3 可观测性的采样与聚合全量记录所有事件数据量会很大。一个跑一小时的 Agent 任务可能产生几万条事件。如果每条都实时写磁盘IO 会成为瓶颈。我的做法是分层采样关键事件全量记录状态转移、工具调用、错误、checkpoint这些必须全记思考过程采样记录Agent 的中间推理按 10% 采样或者只在 token 消耗超过阈值时记录心跳事件聚合每 5 秒一次的心跳聚合成一条统计记录而不是每次都写聚合的时候我关心的是几个指标累计 token、累计耗时、工具调用次数分布、错误率。这些指标能快速告诉你 Agent 的健康状况。class MetricsAggregator: def __init__(self, window_seconds5): self.window window_seconds self.buffer [] def record(self, event): self.buffer.append(event) if time.time() - self.last_flush self.window: self.flush() def flush(self): summary { window_start: self.last_flush, event_count: len(self.buffer), total_tokens: sum(e[tokens][total] for e in self.buffer), tool_calls: Counter(e[tool_name] for e in self.buffer if e[event_type] tool_call), errors: [e for e in self.buffer if e.get(error)] } write_metrics(summary) self.buffer [] self.last_flush time.time()3.4 编排层的上下文传递多 Agent 协作最容易出问题的地方是上下文传递。Agent A 改完代码Agent B 要 reviewB 需要知道 A 改了什么、为什么改。如果只是把 A 的最终输出丢给 B信息损失很大。我的方案是结构化上下文包{ task_id: task_001, from_role: coder, to_role: reviewer, artifacts: { changed_files: [src/main.py, src/utils.py], diff: ..., test_results: {passed: 12, failed: 2}, reasoning_summary: 将 requests 替换为 httpx因为项目需要异步支持 }, constraints: [不要修改测试文件, 保持 API 兼容] }这个包比纯文本输出信息密度高得多reviewer Agent 能快速抓住重点。而且结构化的东西方便程序处理比如你可以写规则自动检查“changed_files 是否超过 10 个”超过就拒绝进入 review 阶段。提示上下文包里的 reasoning_summary 不要用 Agent 的原始输出那个太啰嗦。让 Agent 在完成任务时额外生成一个 50 字以内的摘要专门用于传递。这个摘要的质量直接影响下游 Agent 的表现。4. 实操过程与核心环节实现4.1 环境准备与依赖安装先把基础环境搭起来。Pi-Harness 本身是个 Python 项目依赖不多主要是标准库加几个辅助库。# 创建虚拟环境 python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate # 安装依赖 pip install pydantic rich httpx这里解释下为什么选这几个pydantic用来定义事件、checkpoint、上下文包的数据模型自带校验能避免很多低级错误rich终端里的可观测性展示实时看 Agent 状态很直观httpx如果 Agent 是通过 HTTP 通信的用它比 requests 更现代支持异步Git 是必须的因为文件快照和 diff 都依赖它。确保git --version能正常输出。如果你在 Windows 上建议用 Git Bash 而不是 CMD路径处理会少很多坑。git config --global user.name your-name git config --global user.email your-email这两行配置很多人觉得无所谓但 Agent 执行git commit的时候如果没配会直接失败。我见过好几次因为这个问题导致整个流水线卡住。4.2 核心模块的代码结构项目结构我建议这样组织pi-harness/ ├── harness/ │ ├── __init__.py │ ├── state_machine.py # 状态机 │ ├── checkpoint.py # 检查点 │ ├── observability.py # 可观测性 │ ├── orchestrator.py # 编排 │ └── tools/ │ ├── __init__.py │ ├── base.py # 工具基类 │ └── git_tools.py # Git 相关工具 ├── configs/ │ └── default.yaml └── tests/状态机是核心先把它写出来from enum import Enum from pydantic import BaseModel class State(str, Enum): IDLE idle THINKING thinking ACTING acting OBSERVING observing CHECKPOINT checkpoint DONE done FAILED failed class StateMachine(BaseModel): current: State State.IDLE step: int 0 max_steps: int 100 def transition(self, event_type: str): transitions { (idle, task_received): State.THINKING, (thinking, tool_call): State.ACTING, (thinking, final_answer): State.DONE, (acting, tool_result): State.OBSERVING, (observing, processed): State.CHECKPOINT, (checkpoint, saved): State.THINKING, } key (self.current.value, event_type) if key not in transitions: raise ValueError(f非法转移: {key}) self.current transitions[key] self.step 1 if self.step self.max_steps: self.current State.FAILEDmax_steps这个限制非常重要。没有它Agent 可能无限循环。我一般设 100复杂任务设 200。超过就强制失败然后人工看事件流分析原因。4.3 Checkpoint 的写入与恢复实操Checkpoint 模块的关键是原子写入。如果写到一半进程挂了checkpoint 文件损坏恢复就失败了。import os import json import tempfile def save_checkpoint(path, data): # 先写临时文件 dir_name os.path.dirname(path) fd, tmp_path tempfile.mkstemp(dirdir_name) try: with os.fdopen(fd, w) as f: json.dump(data, f, ensure_asciiFalse) f.flush() os.fsync(f.fileno()) # 强制刷盘 # 原子替换 os.replace(tmp_path, path) except Exception: os.unlink(tmp_path) raiseos.fsync这行很多人会漏。不刷盘的话操作系统可能把数据缓存在内存里进程挂了数据就丢了。我早期版本没加这行测试的时候故意 kill 进程十次有三次 checkpoint 是空的。加上之后就没再出过问题。恢复逻辑def load_latest_checkpoint(checkpoint_dir): files sorted( [f for f in os.listdir(checkpoint_dir) if f.endswith(.json)], keylambda x: int(x.split(_)[1].split(.)[0]) ) if not files: return None latest os.path.join(checkpoint_dir, files[-1]) with open(latest) as f: return json.load(f)按 step 编号排序取最新的。这里假设 checkpoint 文件名格式是ckpt_step.json。4.4 可观测性面板的搭建光有数据不够要能直观地看。我用 rich 做了一个终端面板实时刷新from rich.live import Live from rich.table import Table from rich.console import Console def render_dashboard(metrics): table Table(titlePi-Harness 运行时状态) table.add_column(指标, stylecyan) table.add_column(值, stylegreen) table.add_row(当前状态, metrics[state]) table.add_row(当前步数, str(metrics[step])) table.add_row(累计 Token, str(metrics[total_tokens])) table.add_row(工具调用次数, str(metrics[tool_calls])) table.add_row(错误数, str(metrics[errors])) table.add_row(运行时长, f{metrics[duration]:.1f}s) return table console Console() with Live(consoleconsole, refresh_per_second2) as live: while not done: live.update(render_dashboard(get_metrics())) time.sleep(0.5)这个面板看起来简单但实战中价值巨大。你能一眼看到 Agent 是不是卡住了步数不涨、是不是在烧 tokentoken 涨得快但步数不涨、是不是在报错错误数增加。我有个习惯Agent 跑长任务的时候旁边就开着这个面板。一旦发现异常立刻中断看事件流比等它跑完再分析效率高得多。4.5 多 Agent 编排的落地案例举个实际例子批量修复 lint 错误。任务是把项目里所有 ESLint 报错修掉。流水线设计扫描 Agent跑eslint --format json生成错误列表分组按文件分组每个文件一个子任务修复 Agent可并行每个子任务一个 Agent修完一个文件验证 Agent跑eslint确认错误减少跑测试确认没引入新问题汇总生成报告编排代码大概长这样async def fix_lint_pipeline(project_path): # 1. 扫描 errors await scan_agent.run(project_path) # 2. 分组 groups group_by_file(errors) # 3. 并行修复 tasks [fix_agent.run(g) for g in groups] results await asyncio.gather(*tasks, return_exceptionsTrue) # 4. 验证 verify_result await verify_agent.run(project_path) # 5. 汇总 return summarize(results, verify_result)这里用asyncio.gather做并行但要注意并行度控制。如果同时跑 50 个 AgentAPI 限流、磁盘 IO、内存都会爆。我一般设max_concurrent5用信号量控制。sem asyncio.Semaphore(5) async def run_with_limit(agent, task): async with sem: return await agent.run(task)注意并行 Agent 修改同一个文件会冲突。分组的时候要保证每个文件只被一个 Agent 处理。如果两个错误在同一个文件合并成一个子任务不要拆开。5. 常见问题与排查技巧实录5.1 Agent 陷入循环怎么破这是最常见的问题。表现是步数一直涨但任务没进展工具调用模式重复。排查思路看事件流找到第一个重复的 tool_call看那个 tool_call 的输入和输出判断为什么 Agent 觉得需要重试检查是不是工具返回了模糊的结果导致 Agent 不确定是否成功常见原因和解决现象可能原因解决反复读同一个文件上下文里文件内容被截断增大上下文窗口或分段读取反复改同一个地方测试一直失败但错误信息不明确让测试输出更详细的错误反复跑同一个命令命令有副作用但 Agent 没意识到在工具描述里明确副作用在两个方案间反复横跳prompt 里有矛盾的约束检查 prompt消除歧义我的兜底方案是在 Harness 层加一个重复检测器如果连续 3 次 tool_call 的 (tool_name, input_hash) 相同就强制中断把控制权交回给人。class LoopDetector: def __init__(self, threshold3): self.threshold threshold self.history [] def check(self, tool_name, tool_input): key (tool_name, hash(json.dumps(tool_input, sort_keysTrue))) self.history.append(key) if len(self.history) self.threshold: recent self.history[-self.threshold:] if len(set(recent)) 1: return True # 检测到循环 return False5.2 Checkpoint 恢复后状态不一致这个问题的表现是恢复后 Agent 的行为和中断前对不上比如它以为某个文件还没改但其实已经改了。根因通常是 checkpoint 和实际文件系统不同步。可能的原因checkpoint 写入成功但文件修改还没落盘文件被 Agent 之外的东西改了比如你手动改了diff 应用顺序错了解决方法是在 checkpoint 里记录文件哈希恢复时校验def verify_workspace(checkpoint, workspace): for path, expected_hash in checkpoint[file_hashes].items(): actual hash_file(os.path.join(workspace, path)) if actual ! expected_hash: raise WorkspaceMismatch( f{path} 哈希不匹配期望 {expected_hash}实际 {actual} )校验失败就拒绝恢复提示人工介入。宁可停下来也不要带着错误状态继续跑。5.3 Token 消耗失控Agent 跑着跑着 token 消耗飙升一个任务烧掉几十万 token。这通常是因为上下文无限增长。对话历史每轮都追加如果不做压缩很快就爆了。我的策略是滑动窗口 摘要保留最近 N 轮完整对话N 一般设 10更早的对话压缩成摘要工具调用的详细输出只保留最近几次更早的只留结论def compress_history(messages, keep_recent10): if len(messages) keep_recent: return messages old messages[:-keep_recent] recent messages[-keep_recent:] summary summarize(old) # 用一个小模型或规则生成摘要 return [{role: system, content: f历史摘要{summary}}] recent摘要的质量很关键。我试过用规则摘要只保留工具调用和结果也试过用小模型摘要。规则摘要便宜但信息损失大小模型摘要贵但保留信息多。折中方案是规则摘要为主关键决策点用小模型补充。5.4 Git 相关的疑难杂症Agent 操作 Git 的时候问题特别多。列几个我踩过的问题一git commit报 “nothing to commit”。原因是 Agent 改了文件但没git add。解决是在 commit 工具里自动 add或者明确要求 Agent 先 add。问题二分支切换后文件冲突。Agent 在分支 A 改了文件切到分支 B 时冲突。解决是每次任务开始前检查工作区是否干净不干净就 stash 或拒绝执行。问题三git diff输出太大。Agent 改了很多文件diff 几万行塞不进上下文。解决是限制 diff 输出只给统计信息加关键文件的 diff。# 只看统计 git diff --stat # 只看某个文件 git diff -- src/main.py # 限制行数 git diff | head -500问题四SSH 认证失败。Agent 执行git push时如果 SSH key 没配好会卡在认证。解决是提前配好免密或者用 token 方式。测试方法ssh -T gityour-git-host能正常返回就说明认证没问题。5.5 常见问题速查表问题快速定位解决方向Agent 卡住不动看步数是否增长检查是否在等工具返回循环调用看 tool_call 重复率加循环检测器Token 飙升看上下文长度压缩历史恢复后行为异常校验文件哈希修复 checkpoint 逻辑工具调用失败看工具返回的 error检查工具实现和权限并行冲突看是否有文件被多 Agent 改加锁或重新分组Git 操作失败看 git status检查工作区状态和认证6. 一些实操心得与扩展方向6.1 关于“可观测”的度可观测性不是越多越好。我早期版本记录了一切结果日志文件几个 GB查问题的时候反而找不到重点。后来我定了个原则记录能回答“为什么”的信息而不是“是什么”的信息。比如“Agent 调用了 edit_file”这是“是什么”价值不大。“Agent 调用 edit_file 是因为上一步测试失败失败原因是断言不匹配”这是“为什么”价值大。前者是事件后者是因果链。Harness 要重点记录因果链。6.2 关于“可恢复”的边界不是所有状态都能恢复。比如 Agent 调用了一个外部 API 改了远程数据这个副作用没法回滚。所以可恢复性的边界是本地状态可恢复远程副作用需人工确认。我在 Harness 里加了一个“副作用登记”机制工具执行前声明它有没有远程副作用有的话执行前暂停等人确认。这个机制救过我好几次避免 Agent 误删远程资源。6.3 关于“可编排”的复杂度编排层最容易过度设计。我见过有人上来就搞 DAG 引擎、搞分布式调度结果简单任务跑不起来。我的建议是从线性流水线开始遇到真实需求再加复杂度。大部分 Coding Agent 场景线性流水线加条件分支就够了。真正需要 DAG 的场景很少通常是任务之间有复杂的依赖关系而这种场景往往更适合人工拆解而不是让 Agent 自动编排。6.4 后续可以扩展的方向这套 Harness 搭起来之后有几个自然的扩展方向接入 CI把 Harness 包装成一个 CI jobAgent 自动处理 issue、自动修 lint、自动补测试多模型路由简单任务用小模型复杂任务用大模型Harness 层做路由经验回放把成功的任务轨迹存下来遇到类似任务时作为 few-shot 示例成本控制设置 token 预算超预算自动降级或暂停我个人最看好的是经验回放。Agent 每次成功完成任务把轨迹存下来下次遇到类似任务把轨迹作为参考注入上下文。实测下来这能显著提升 Agent 的成功率尤其是那些有固定套路的任务比如“给某个函数补单元测试”。最后分享一个小技巧Harness 的配置文件用 YAML但关键参数比如 max_steps、token 预算支持环境变量覆盖。这样本地调试和 CI 运行可以用不同的配置不用改文件。# configs/default.yaml agent: max_steps: ${MAX_STEPS:-100} token_budget: ${TOKEN_BUDGET:-500000} checkpoint: dir: ${CHECKPOINT_DIR:-./checkpoints} interval: 1${VAR:-default}这个语法在 shell 里很常见但在 YAML 里需要自己解析。写个小函数处理一下就行比引入复杂的配置库轻量得多。