做 Agent 框架工程化,我最大的体会是:模型的能力边界,一半在权重里,另一半在“它到底能触达什么样的环境”里。我最近在持续打磨一套开源推理模型配套的 Harness 评估框架,它要解决的核心问题,是评估一个会动手的编码类 Agent 时最容易翻车的三个环节——过程不可见、结果不可复现、行为不可审计。整套设计就两根柱子:工具能力全部插件化,会话过程完整落盘成可回放的日志。这篇文章把这两根柱子的设计逻辑、实现细节,以及我在实际接入中踩过的坑都拆开讲一遍。适合三类人看:正在搭 Agent 评测系统的后端工程师、研究 Agent 可观测性的人,以及想理解大模型应用怎么工程化落地的读者。
1. 整体设计思路拆解:为什么 Harness 非做不可
1.1 一个会动手的模型,靠什么来评估
传统 NLP 评测是“输入-输出”式的:给模型一段文本,拿它与标准答案比对,算准确率、算 F1。但编码类 Agent 完全不是这个玩法。给它一个仓库、一个 issue、一个初始 commit,它要自己去读代码、定位缺陷、改动文件、执行测试、观察输出、再迭代修改。整个过程是多轮长链路、带环境状态的,最终能不能通过测试,只反映结果的一部分;更关键的是过程——它在哪一步跑偏了、为什么反复修改同一个文件、是不是在靠暴力尝试碰运气,这些都要有据可查。
这决定了 Harness 不能只是“一个发请求拿结果的客户端”,它必须同时具备三个能力。第一,提供一个真实、可控、可隔离的执行环境,让模型想干嘛就干嘛,但干坏事了不会伤到宿主机和其他任务;第二,把模型的“动手动作”抽象成一组稳定、可枚举、带 schema 的工具接口,让模型是通过 API 而不是靠碰运气的方式操作环境;第三,把每一次观测、思考、调用、结果全部记录下来,形成一条可以被事后重放的完整轨迹。这三个需求,正好对应 Harness 的三大模块:沙箱环境、工具插件、会话日志。我之所以在设计时把这三大模块分开,是因为它们各自的演进节奏完全不同——环境镜像更新频繁,插件列表随评估任务变化,日志格式则要尽量保持稳定。
1.2 插件化不是“为了优雅”,而是为了活下来
刚开始做工具接入时,我踩过一个经典的坑:直接在核心引擎里写死工具调用。今天要加一个“执行 Python 脚本”,就往引擎里塞一个函数;明天要加“读文件”,再塞一个。看起来很快,但三个评估任务之后,引擎代码就变成了一锅粥。工具的入参校验、错误处理、超时控制、日志记录全都耦合在一起,改一个工具,十几个地方跟着改。
后来我把工具全部插件化,思路其实非常简单,就像家里的电源插座——墙上的接口是固定的,你插什么电器都行,但每个电器的内部实现完全独立。插件化之后,核心引擎只认一个稳定的接口:插件名、参数 schema、执行入口、返回结果。剩下的事全部交给插件自己处理。模型侧只看到一份工具清单(名称、描述、参数定义),框架侧只维护一个插件注册表,新增能力就是写一个插件类然后注册,删除能力就是注销一行。这样做最大的收益不是“代码好看”,而是评估任务的可组合性——同一个 Harness 可以同时承载代码修复、数据爬取、系统运维三种完全不同的 Agent 任务,只需要在配置里挂不同的插件集合。
1.3 会话日志:评测链路的黑匣子
飞机有黑匣子,Agent 评测也需要自己的黑匣子。真实评测中我遇到过太多“看起来失败了但不知道死在哪”的情况:模型输出了一段错误的 JSON,工具调用抛了个异常,沙箱环境重置后复现不出来……如果没有完整日志,排查全靠猜。
会话日志要承担三个角色。第一是复现依据:同一个任务、同一个初始状态、同一份日志,应该能重建出几乎一致的执行过程;第二是审计凭证:评测结果要公平,就必须能回答“Agent 当时确实执行过这些命令、看到过这些输出”,日志就是证据;第三是调优素材:模型走偏的每一帧、工具报错的每一条信息,都是 Prompt 和工具设计的改进依据。我见过不少团队在评测时只存“最终结果”,省空间是省了,但出了问题两眼一抹黑。会话日志这件事,千万不能省。
2. 核心架构与插件机制详解
2.1 工具插件:把模型的“念头”翻译成环境动作
模型通过工具调用与外部世界交互,这在今天已经是标准范式。Harness 要做的是把这个范式工程化:模型输出一个结构化动作(比如“执行命令 ls -la”),框架负责解析、路由、执行、回收结果,然后原样返回给模型。插件就是这个链条上的执行单元。
一个工具插件,我建议最少由四个部分组成。名字,唯一且语义清晰;描述,说明工具在什么场景下用、有哪些注意事项,这直接决定模型会不会乱用;参数 schema,用 JSON Schema 描述入参结构和约束,框架在路由前先做校验;执行函数,接收解析好的参数和执行上下文,返回结构化的结果。下面是最小实现:
class ToolPlugin: name = "" description = "" parameters = {} # JSON Schema def execute(self, params: dict, ctx: ExecContext) -> ToolResult: raise NotImplementedError注册表负责收集和管理插件,同时对外暴露两份数据:一份给模型看(工具清单),一份给引擎用(路由表)。
class PluginRegistry: def __init__(self): self._plugins = {} def register(self, plugin: ToolPlugin): self._plugins[plugin.name] = plugin def tool_schemas(self): return [{ "name": p.name, "description": p.description, "parameters": p.parameters } for p in self._plugins.values()] def invoke(self, name: str, params: dict, ctx: ExecContext): plugin = self._plugins[name] return plugin.execute(params, ctx)这里有一个我在实战中反复强调的点:插件执行函数里一定要做超时控制和输出截断,不是每个工具都能在规定步数内正常返回的。一个卡住的命令,如果不 kill,整条评测链路都会跟着卡死;一个几 MB 的命令输出,如果不截断,模型下一个回合的上下文窗口可能直接被撑爆。插件是面向“模型”的,但它的背面是真实操作系统,一切按生产标准来写。
2.2 沙箱执行环境:给每个 Agent 一间隔离的工作间
插件负责“动作”,沙箱负责“场所”。刚开始我试过让 Agent 直接在宿主机上跑命令,后果可以想象:一次误写的删除命令就足以让整个评测环境瘫痪。后来我改用容器化方案,每个任务独立启动一个容器,里面装好代码仓库、依赖库、评测工具。容器镜像只读、可复现、随建随拆,宿主机安全,任务之间互不干扰。
沙箱设计有几个细节值得展开。一是镜像要提前构建,不能等评测开始时在线装依赖,否则网络抖动一次,整个评测队列全废;二是资源配额必须显式声明,内存、CPU、磁盘、超时时间都要在任务配置里写清楚,防止一个失控 Agent 拖垮同节点的其他任务;三是网络策略要按需开放,大多数编码类任务不需要公网,把外网关闭能避免模型“跑出去”下载一些不可控的东西,也可以大幅提升评测安全性。我见过不少框架为了省事让沙箱随用随连公网,结果评测结果根本不可复现——同一个任务,昨天能跑通,今天依赖一更新就挂了,这种教训一次就够。
2.3 执行引擎与并发调度的取舍
引擎是 Harness 的中枢,它做的事情可以概括成四步循环:取模型输出,解析出工具调用,交给插件执行,把结果连同观测状态写回日志。这个循环本身不复杂,复杂的是并发和调度。
小规模调试时,串行执行最稳:一个任务跑完再跑下一个,日志顺序清晰,问题好定位。但到了需要几百个任务同时评估的场景,串行就撑不住了,得让沙箱承担并行任务的单元。实操中我采用的方案是把每个沙箱封装成独立的执行服务,通过 HTTP 接口对外提供工具调用能力。这样并行调度层(我用的是容器编排平台的任务队列)只负责“把任务塞给某个沙箱”,不关心 Agent 内部状态;Agent 控制器只管跟沙箱的 API 通信。所有交互都走接口,日志自然也在接口层统一收集,这比在每个 Agent 内部单独埋点要可靠得多。
这里要提醒一个容易踩的坑:并行度不是越高越好。评测任务通常都有资源上限,沙箱数量开得太多,反而会因为 CPU 争抢导致工具执行变慢、超时频发,最后评测结果里混进大量“非模型因素”的噪声。我一般在单机上先把资源基准测一遍,再按“每个任务独享配额”倒推并行度。
3. 可回放会话日志的工程实现
3.1 日志要记录到“每一帧”
什么叫“可回放”?不是把模型说的话记下来就完了,而是要把一次会话里所有能影响后续决策的信息都记下来。我自己最终落地的日志是一个按事件追加的 JSON 文件,每个事件有全局递增的序号、类型、时间戳和载荷。事件类型至少包括:任务元信息、模型输出、工具调用请求、工具执行结果、环境状态变更、异常事件。
{ "session_id": "task-001", "task": { "repo": "some-repo", "base_commit": "a1b2c3", "prompt": "修复登录接口的越权问题" }, "events": [ { "step": 1, "type": "model_output", "content": "我先看一下项目结构...", "ts": 1710000000.123 }, { "step": 2, "type": "tool_call", "tool": "bash", "params": {"command": "find . -type f -name '*.py' | head -50"}, "ts": 1710000000.150 }, { "step": 3, "type": "tool_result", "exit_code": 0, "stdout": "src/auth.py\nsrc/server.py\n...", "stderr": "", "duration_ms": 42, "ts": 1710000000.192 } ] }还有一个细节很重要:文件快照。Agent 改过哪些文件、改了什么内容,是编码类评测里最关键的证据。我的做法是在每个工具调用回合之间,对工作目录做一次增量变更记录;如果条件允许,还会把改动前的文件内容一起存下来,作为附件放在对应事件里。这样回放时不仅能看它说了什么、做了什么,还能精确看到它对代码库产生了什么影响。
3.2 回放功能的三个层次
日志存下来了,回放才有意义。我把回放分成三个层次,由浅入深。
第一个层次是文本回放:把 JSON 日志渲染成一份类似聊天记录的 Markdown 或 HTML 文档。模型每一步说了什么、调用了什么工具、工具返回了什么,按时间顺序一字排开。这个层次主要服务于人的快速阅读——评测完了先扫一眼,大概就知道 Agent 的节奏对不对。第二个层次是状态回放:把某一步时刻的完整环境状态还原出来。比如日志里记录了第 12 步之前的文件快照,我就可以把工作目录恢复到那一刻,手动执行查看、修改、调试,相当于“回到过去”检查现场。这个层次对疑难问题定位非常有用,因为很多问题只有在特定上下文里才能复现。第三个层次是指标回放:从日志里自动抽取结构化指标,比如总步数、工具调用次数、各工具耗时分布、异常事件数、首次成功修改文件的步数。这些指标可以批量喂给报告系统,做任务级或模型级的横向对比。
三个层次的实现成本是递增的,但收益也递增。小团队如果时间紧张,我建议至少把第一和第三层次做扎实:文本回放用于人工排查,指标抽取用于自动评估。
3.3 回放一致性的三个关键细节
保证“回放出来跟原跑一致”,比想象中难。我整理出三个最容易出问题的环节。
第一,事件顺序必须全局单调。不同来源的日志(比如模型输出来自网络回调、工具结果来自沙箱进程)可能因为异步到达而乱序。我最后的处理办法是统一的中心化日志写入,所有事件在写入时分配全局递增序号,禁止各模块自行落盘后再合并排序。第二,输出必须“完整捕获”而不是“参考捕获”。工具执行结果如果被截断,日志里存的应该同时包含截断后的展示内容和指向完整内容的附件路径;绝不能只存截断后的片段,否则回放时模型看到的上下文和真实情况对不上。第三,有状态的操作要记录副作用。比如某个命令悄悄修改了一个系统配置文件、某个环境变量被导出、某个后台进程被启动——这些不会直接出现在工具返回里,但会影响后续回合。我的做法是在一个工具调用结束后,主动做一次“状态指纹”采集,记录关键文件的哈希和进程列表,作为事件的附属字段。
4. 实操:从零搭建可复现的最小 Harness
4.1 最小插件注册器与 Bash 工具实现
纸上谈兵不如直接动手。我下面给一个最小可跑的 Harness 核心,不依赖任何重框架,用标准库就能跑通。先看插件侧:
import subprocess class BashPlugin: name = "bash" description = "在沙箱内执行一条 shell 命令" parameters = { "type": "object", "properties": { "command": { "type": "string", "description": "要执行的完整命令" } }, "required": ["command"] } def __init__(self, timeout=10, max_output_bytes=8192): self.timeout = timeout self.max_output_bytes = max_output_bytes def execute(self, params, ctx): result = subprocess.run( params["command"], shell=True, capture_output=True, text=True, timeout=self.timeout, ) return { "exit_code": result.returncode, "stdout": result.stdout[:self.max_output_bytes], "stderr": result.stderr[:self.max_output_bytes], }注意 execute 里我加了两个默认参数:timeout 和 max_output_bytes。这两个看起来不起眼,实战里救命的次数最多——没有超时,一个 sleep 命令就能挂死整个评测任务;没有截断,一次全文输出就能让日志文件膨胀到几 MB。
文件读写插件我平时单独拆一个,不并进 bash 里,主要原因是权限边界不同:bash 理论上能碰所有文件,文件插件则应限制在任务工作目录内。这种“最小权限”的插件划分,既方便审计,也避免模型用一招鲜的工具干所有事。
4.2 会话日志落盘与 Markdown 回放实现
日志模块我给一个足够简单但能用的版本:
import json import time import os class SessionLogger: def __init__(self, session_id, output_dir="traces"): self.session_id = session_id self.path = os.path.join(output_dir, f"{session_id}.json") self.events = [] os.makedirs(output_dir, exist_ok=True) def log(self, etype, payload, step=None): event = { "step": step if step else len(self.events) + 1, "type": etype, "payload": payload, "ts": time.time(), } self.events.append(event) with open(self.path, "w", encoding="utf-8") as f: json.dump({"session_id": self.session_id, "events": self.events}, f, ensure_ascii=False, indent=2)这里有一个我踩过坑后的心得:每次写入都全量 dump 而不是增量 append,看起来效率低,但换来的是日志文件永不损坏——进程中途崩溃时,文件里始终是“上一次完整状态”,解析不会因为半行 JSON 而挂掉。评测任务本身频率不高,这点开销完全可以接受。
回放渲染也很直接,遍历 events 按类型拼 Markdown:
def render_markdown(log_path): with open(log_path, encoding="utf-8") as f: data = json.load(f) lines = [f"# Session {data['session_id']}"] for ev in data["events"]: if ev["type"] == "model_output": lines.append(f"#### 模型(step {ev['step']})\n\n{ev['payload']['content']}") elif ev["type"] == "tool_call": lines.append(f"#### 调用 `{ev['payload']['tool']}`(step {ev['step']})\n\n```bash\n{ev['payload']['params'].get('command', '')}\n```") elif ev["type"] == "tool_result": lines.append(f"**返回码** {ev['payload']['exit_code']}\n\n```\n{ev['payload']['stdout']}\n```") return "\n\n".join(lines)这个生成逻辑很简单,但足够让一个评测结果变成可读的图文报告。要接 HTML 模板、加时间轴、加 diff 视图,都是在这个基础上扩展。
4.3 完整接入一个 Agent 的流程与配置
把上面的模块串起来,跑一次完整评测的流程大概是:
- 准备沙箱镜像。基于容器引擎基础镜像,安装好评测仓库和依赖,保存为固定标签。
- 启动沙箱服务。把仓库挂载进去,启动执行服务的 HTTP 端口,记录访问地址。
- 写任务配置。用 YAML 声明模型接口、插件列表、步数上限、最大 token、日志目录。
- 运行评测主循环。控制器把任务 prompt 发给模型,解析模型返回,若含工具调用就走注册表 invoke,结果写日志并拼回模型上下文;到达最大步数或模型认为任务完成则结束。
- 生成回放报告。调用 render_markdown 输出,供人审阅。
一个最小配置大概是:
task: repo: "sample-repo" base_commit: "a1b2c3" prompt: "修复登录接口的越权问题" agent: model_endpoint: "http://localhost:8000/v1" max_steps: 30 sandbox: image: "eval-base:v1" memory_limit: "4g" network: "offline" plugins: - bash - file_read - file_write logs: output_dir: "traces"这套流程跑通之后,你会发现一个很大的红利:因为日志是全量可回放的,评测的“审查”环节变成了纯纯地读文件,不再需要任何人在线盯场。谁在什么时间做了什么、模型看到什么输出才做了决定,全部可追溯。这个能力对团队协作和结果复现都非常关键。
5. 常见问题排查与避坑记录
5.1 高频问题速查表
实操里最常踩的问题,我整理成一个速查表,方便对照排查:
| 问题现象 | 可能原因 | 排查与解决思路 |
|---|---|---|
| 沙箱内执行命令超时 | 任务依赖网络下载,或命令本身阻塞 | 预构建镜像缓存依赖;给命令显式超时并 kill 进程 |
| 日志文件迅速膨胀 | 工具输出没有截断,或回放反复写全量 JSON | 在插件层加输出上限;日志每 N 次事件压缩一次 |
| 回放结果与原跑不一致 | 存在非确定性操作(随机数、时间、外部请求) | 记录环境变量、随机种子、系统时间;禁用公网依赖 |
| 多个任务互相干扰 | 共享了挂载目录或同一套临时文件路径 | 每个任务独立命名空间和工作目录,禁止共享写路径 |
| 模型反复调用同一工具 | 工具描述不清晰,或返回结果让模型误解 | 检查工具 description,补充使用场景和失败示例 |
还有一些比较隐蔽的:插件注册顺序影响工具清单展示顺序,进而可能影响模型的选择倾向;JSON Schema 里忘了写 required,模型就会理直气壮地少传参数;工具返回的 stderr 被吞掉,模型完全看不到报错,会一直原地打转。
5.2 三个值得展开的坑
第一个坑:日志“看起来完整”但没法精确回放。我早期用事件轮询拉取模型输出,网络抖动时事件会乱序,导致回放时模型“先说后想”,逻辑对不上。后来痛定思痛,改成在 Harness 进程内做单点写入,先入先出,永不在各子模块本地缓存日志。这个改动让回放可信度大幅提升。
第二个坑:为了省空间截断工具输出。截断本身没错,错的是只截断不保留全量。有一次排查一个分析型任务,Agent 明明输出了完整分析却判断失误,结果一查日志,工具返回被前端截断到 2KB,模型根本没有看到完整数据。从那以后我规定:日志层永远保留完整输出的附件或哈希,展示层才截断。数据可以按需删,但证据链不能断。
第三个坑:并发评测时资源争抢。起初我把并行度拉到机器核心数的两倍,结果工具执行时间翻倍、超时率暴涨,评测分数被污染。后来改成先压测基准:单个任务独占 2 核 4G 内存时平均工具延迟是多少,再按 80% 资源水位倒推最大并行度。评测环境要的是“稳定”而不是“榨干”,这个观念得摆正。
我把这套东西从最初的“写死工具的脚本”一路改到现在的全插件化加可回放日志,最直接的感受是:框架的复杂度并没有消失,而是被转移到了更可控的地方——从“散落在引擎各处的分支逻辑”转移到了“一个个独立插件和一份份结构化日志”里。如果你也在搭 Agent 评测,我建议先别追求功能多,先把“工具插件化”和“会话可回放”这两根柱子立起来,其他能力都是在这两根柱子上长出来的。最后再分享一个小技巧:每次改完框架,都拿一个历史任务的日志做一遍回放,对比关键事件是否一致——这个动作我坚持了很久,替我挡掉过至少三次让全量评测结果失真的回归。