1. 先搞清楚Agent Harness到底是什么
这两年做AI Agent,稍微深入一点就会碰到一个词:Agent Harness。第一次看到的人,很容易把它理解成某个框架或者SDK,其实它更像一个“运行骨架”:Agent负责动脑,Harness负责保证这个脑子在受控的环境里把事情做完。
我自己一开始做Agent的时候,也走过一段弯路。最早的版本就是一个Python脚本,拿到用户问题,拼prompt,调模型接口,模型返回一段话里有工具调用就解析一下,然后执行,再把结果拼回去继续调。功能是能跑的,但项目一多就出问题:有的Agent要访问数据库,有的要操作内部工单系统,有的要调文件服务,每个脚本里重复的“调用模型、解析工具、处理上下文、失败重试”逻辑越来越多,改一个公共逻辑要动七八个文件。后来我把公共部分抽出来,做成了一个独立的执行层,从那个时候开始,我才意识到我其实是在写一个Agent Harness。
Harness的核心职责,说白了就是四件事:控制Agent的主循环,管理上下文,统一调用工具,记录运行过程。它不决定Agent“想什么”,但它决定Agent“怎么走”,以及“走歪了怎么拉回来”。
这套东西适合谁来参考?如果你正在搭一个内部Agent平台,或者想把一个个散装脚本升级成可维护的Agent执行环境,又或者只是自己写一个带工具的Agent发现很难调试,那这篇文章应该能给你一个可以直接落地的骨架。我后面讲的内容不绑定某个具体模型服务,也不依赖某个特定框架,核心思路和代码结构可以照搬到自己的项目里。
1.1 Harness不是Agent,是Agent的“运行壳”
很多人会把Harness和Agent本身混在一起。Agent本身是那套决策逻辑:你给它一个目标,它决定下一步是该回答用户、查资料还是执行动作。Harness则是承载这套决策逻辑的壳:它负责把模型输出的“想法”翻译成真实系统里的调用,把执行结果带回来,再决定让模型继续想还是收尾。
打比方的话,模型像司机,Harness是车。司机决定去哪里,车负责提供方向盘、油门、刹车、仪表盘和异常报警。没有车,司机再厉害也只能原地站着。Harness也一样,没有它,模型就算知道该调工具,也只是在回复文本里“口头约定”一下,代码层面没人替它真正执行。
这个区分很重要,因为它决定了设计边界。如果Harness里塞满了业务决策,比如“用户说这个关键词就调用那个接口”,那Harness就退化成业务代码了。反过来,如果Agent逻辑里写了太多执行细节,比如自己拼HTTP请求、自己起线程,那Agent就没法复用,换个环境就得改一遍。我踩过这个坑:早期我把工具调用逻辑直接写在Agent的system prompt和函数里,后来要支持新的工具协议,几乎把整个Agent重写了一遍。Harness的价值就在这里——把“决策”和“执行”解耦,让模型永远只面向统一的工具接口。
1.2 什么场景需要自己写Harness
市面上的通用框架已经很多了,为什么不直接拿来用?不是不能用,而是很多时候通用框架提供的抽象和你团队的实际需求不匹配。
比如一个典型场景:公司内部有大量老系统,有的是REST接口,有的是消息队列,有的只能通过命令行脚本触发。通用框架默认工具调用是HTTP JSON,你要把每一种老系统都包一层适配器,适配器本身没问题,但框架为了做通用性,会在协议转换、事件传播、配置加载上引入很多约定,团队学习成本不低。
再比如你需要对Agent的每一步做严格审计。哪些人触发了哪个工具,参数是什么,执行结果是什么,失败原因是什么,这些都要落库。通用框架通常有日志,但日志格式和审计要求经常对不上。与其在框架外面再做一层采集,不如在Harness层面直接定义事件,让审计系统订阅这些事件。
还有一个更实际的原因:当你需要精细控制Agent的步数、token预算、重试策略、上下文压缩策略,甚至是暂停和恢复时,通用框架往往会给你一个“半黑盒”。你只能在固定回调点上做事情,稍微越界一点就得改源码。自己写一个轻量Harness,这些控制点全部开放,规则也全部由你定。
当然,我并不是建议大家什么都从零造。如果团队刚起步,业务也简单,先用现成框架快速验证,完全合理。但一旦你发现自己开始跟框架的抽象做斗争,或者为了一个控制点不得不绕过框架内部,那就是该考虑自研Harness的信号了。
2. 架构拆分:把Harness拆成四个可替换的模块
Harness听起来是个大词,真拆开看,核心模块没有想象中多。我现在的做法是拆成四个部分:编排引擎、上下文管理器、工具注册表、可观测层。模型调用不是单独一层,它被编排引擎调用,只是外部依赖。
它们的协作关系大致是这样:用户请求进入编排引擎,引擎从上下文管理器拿历史消息,组装出请求参数,调用模型服务。模型返回如果带工具调用,引擎去工具注册表查出对应工具,做参数校验,执行工具,把结果追加到上下文,再进入下一轮。这个循环一直重复,直到模型返回纯文本回答,或者达到步数上限。
注意一个关键点:工具执行后的结果不是直接丢回给模型,而是先经过上下文管理器。因为工具返回的内容可能很大,可能包含敏感信息,也可能需要被截断或转成结构化的“观察结果”。Harness要在这个地方做第一道过滤,而不是让原始工具结果直接进入下一轮模型的上下文。
2.1 核心是状态机,不是流水线
很多人第一次设计Harness,会把它想成一条流水线:用户输入 -> 调模型 -> 执行工具 -> 再调模型。这个思路单看没问题,但真正实现会发现不够用。模型调用超时怎么办?工具执行到一半用户取消了怎么办?某一步连续失败要不要进入降级模式?这些都是分支,不是一条直线。
所以我在自己项目里把Harness主体实现成了一个显式的状态机。状态最少有:等待输入、模型调用中、工具执行中、等待继续、完成、失败、已取消。每个状态下,Harness只做这个状态允许的事情,其他事件一律排队或者拒绝。
这样做的好处是调试的时候非常直观。线上出问题,只需要看当前实例停在哪个状态,就知道问题出在哪一段。如果模型返回一直没解析出合法结果,状态机会在“工具执行中”和“等待继续”之间反复切换,指标上立刻能看出来。如果没有状态机,所有逻辑都用if嵌套,出了问题只能靠log逐行猜。
状态转移的核心规则也很简单:每一步都从“等待继续”状态开始,没有挂起的工具调用就表示可以调模型,有挂起的工具调用则必须先清掉。我见过一个问题是工具结果还没追加上去,下一次模型调用就发出去了,导致模型“看不见”自己刚执行过的工具结果,然后开始重复调同一个工具。状态机天然杜绝了这种问题。
2.2 上下文管理器要可插拔
上下文是Agent的短期记忆,也是Harness里最容易被低估的模块。很多Agent效果差,不是模型不行,而是上下文管理太粗糙:要么塞太多无关历史,把模型搞晕;要么为了省token过度截断,把关键信息丢了。
我设计的上下文管理器是一个独立接口,核心方法就是:读取当前消息列表、追加一条消息、按策略压缩。具体怎么压缩是可替换的。最简单的策略是滑动窗口,只保留最近N轮;再复杂一点,把早期消息用摘要模型压缩成一段概述;更复杂的是按相关性做检索,只把和当前目标相关的历史片段送回模型。
这个模块必须可插拔,原因是不同任务的上下文规律差别很大。比如客服场景,历史对话可能很长,但真正重要的是当前问题和最近的几轮;数据分析场景,工具返回的表结构信息往往比用户原始描述更重要;多步骤操作场景,中间步骤的结果必须完整保留,不能只是摘要。Harness不应该替Agent决定用哪种策略,但必须提供一个地方让开发者嵌入策略。
2.3 工具注册表:所有能力都要过一道检查
工具注册表是Harness的“能力地图”。每个工具在注册表里都有一份元信息:名称、描述、参数JSON Schema、执行函数、超时时间、权限等级、是否允许被Agent自动调用。
为什么需要注册表而不是直接写个函数调一下?因为模型调用工具时给的是结构化参数,Harness必须按Schema校验参数类型和必填项。这一步能拦住大量低级错误。我统计过自己项目里的失败案例,工具参数错误导致的失败占比相当高,比如模型把日期传成字符串、把数字字段传成null、甚至把工具名拼错。注册表统一校验之后,这些错误可以在进入业务代码之前就被识别出来,要么修正,要么直接告诉模型“参数不对,请重新给”。
注册表还负责权限控制。同一个工具,不同来源的对话可能有不同权限。比如普通用户触发的Agent只能读自己名下的数据,管理员触发的Agent才能写配置。这个判断必须在Harness层做,不能留给工具函数自己判断。工具函数只负责执行,不应假设调用者已经通过鉴权,否则容易出现越权。
2.4 可观测层:事件先于日志
可观测性在设计Harness时就要想,不能等出了问题再加。但可观测不等于“多打日志”。我在实践里发现,事件才是最稳定的观测单元,日志只是事件的渲染形式之一。
Harness每完成一个关键动作,就抛出一个结构化事件,例如agent.started、task.completed、tool.invoking、tool.completed、tool.failed、context.compacted。每个事件都带sessionId、stepId、timestamp、相关工具名或模型名。日志只是把这些事件输出到文件,监控系统也可以直接消费同一份事件流。这样,同一个Harness,既可以接日志平台,也可以接指标系统,还可以做在线排障回放,而且事件格式是稳定的,不随日志库变化。
这个设计帮我解决过一个很头疼的问题:线上有个Agent每次跑到第三步就中断,日志里只能看到个别日志片段,很难还原完整上下文。后来我加了事件流,把每一步的消息数量、token数、工具结果大小都记录下来,才定位到是某个工具返回了一个超大JSON,把上下文撑爆导致请求超时。没有事件级的观测,这种问题排查会非常痛苦。
3. 实现一个最小可用Harness:代码怎么落地
前面讲了架构思路,下面我把一个最小可用的Harness实现骨架贴出来。这个实现不依赖具体模型服务,模型接入层用一个抽象接口代替,你自己换成实际的SDK调用即可。
3.1 先定义消息与工具调用类型
我会先用数据类把核心类型定下来,而不是先写逻辑。类型定清楚了,后面实现不容易跑偏。
from dataclasses import dataclass, field from typing import Any, Callable, Optional @dataclass class Message: role: str # system / user / assistant / tool content: str tool_call_id: Optional[str] = None name: Optional[str] = None @dataclass class ToolCall: id: str name: str arguments: dict @dataclass class ModelResponse: content: str tool_calls: list[ToolCall] = field(default_factory=list)这里的Message兼容最常见的Agent协议:助手消息如果带了工具调用,就通过tool_call_id和后续的tool消息配对。这里有个细节:工具执行结果返回给模型时,role应当是tool,而且必须带tool_call_id,否则很多模型服务会报错或者无法关联结果。
工具的定义我单独用一个ToolSpec,它描述工具是什么,以及怎么调用:
@dataclass class ToolSpec: name: str description: str parameters: dict function: Callable[..., Any] timeout: float = 10.0parameters字段直接给JSON Schema格式,因为主流模型服务都支持这种格式的function calling。工具的实际执行函数用function保存,后面校验完参数再传入。
3.2 注册工具和参数校验
工具注册表的实现不需要花哨,核心是两个方法:注册和按名获取。
class ToolRegistry: def __init__(self): self._tools: dict[str, ToolSpec] = {} def register(self, spec: ToolSpec): if spec.name in self._tools: raise ValueError(f"duplicated tool name: {spec.name}") self._tools[spec.name] = spec def get(self, name: str) -> Optional[ToolSpec]: return self._tools.get(name) def schemas(self) -> list[dict]: return [ { "type": "function", "function": { "name": spec.name, "description": spec.description, "parameters": spec.parameters, }, } for spec in self._tools.values() ]参数校验可以简单做,不需要自己写完整Schema解析。如果项目依赖少,可以直接用现成的JSON Schema校验库;如果不想引入依赖,也要做基本检查:必填字段是否存在,类型是否匹配。完整Schema解析是后面才需要考虑的事,第一版能拦住空参数和明显错类型就够用了。
有一点提醒:schemas()返回给模型时,工具描述一定要写清楚,参数示例也要给。大多数情况下,模型调错参数不是因为它笨,而是描述没写明白。比如一个工具叫query_user,描述只写“查用户”,模型很容易猜错参数。写成query_user(user_id: string, include_resources: bool)并附上说明,错误率会明显下降。
3.3 主循环:控制Agent的每一步
接下来是核心主循环。这一段是Agent反复“思考-执行工具-再看结果”的循环。我把它写成同步的,方便理解;异步版本逻辑一样,只是把模型调用和工具执行换成async版本。
class Harness: def __init__(self, model, registry: ToolRegistry, max_steps: int = 8): self.model = model self.registry = registry self.max_steps = max_steps def run(self, task: str) -> str: messages = [Message(role="user", content=task)] for step in range(self.max_steps): response = self.model.generate( messages=messages, tools=self.registry.schemas(), ) if not response.tool_calls: return response.content for call in response.tool_calls: result = self._execute_tool(call) messages.append( Message( role="tool", content=result, tool_call_id=call.id, name=call.name, ) ) raise RuntimeError("agent exceeded max steps")这个循环里有几个自己写的时候要注意的点。
第一,response.tool_calls可能为空列表,也可能是一个非空列表。为空要直接返回文本;非空则每个工具调用都要执行并且把结果追加进messages。有些模型一次会返回多个工具调用,如果其中一个失败,不要把整个循环打断,而是要把错误信息作为那个工具的结果返回给模型,让它自己决定下一步。这比直接抛异常对用户友好得多。
第二,max_steps不是总的消息轮数,而是“模型被调用”的次数。实操中可以把默认值设小一点,比如8次,避免模型在一个简单任务上反复横跳。如果你发现正常的Agent经常超过步数,那更可能是工具设计得太细,需要合并工具,而不是无脑调大步数上限。
第三,循环结束之后抛异常不是唯一选择。更温和的做法是返回一段字符串“抱歉,步骤超限未完成”。到底抛异常还是返回失败文本,取决于上层是任务式调用还是对话式调用。如果是对话,返回文本更自然;如果是异步任务,抛异常更容易被任务系统捕获。我倾向于把“步数超限”当成一种结果而不只是错误,毕竟它反映了Agent运行到了边界。
3.4 把工具错误喂回模型,而不是直接吞掉
工具执行的实现里有一个很关键的决策:工具抛出的异常,Harness要捕获并转成可读消息,返回给模型。
def _execute_tool(self, call: ToolCall) -> str: spec = self.registry.get(call.name) if spec is None: return f"Error: unknown tool named '{call.name}'" try: result = spec.function(**call.arguments) return str(result) except Exception as exc: return f"Error: tool '{call.name}' failed: {exc}"这段代码看起来简单,表现力很强。模型如果要求调用一个不存在的工具,下一轮它会看到明确的错误信息,然后自行纠正。工具执行抛错,只要能转为文本,模型也能根据错误信息调整参数或换一种方法。
这里有一个容易踩的坑:不要直接把Python异常的完整堆栈返回给模型。一方面,堆栈信息暴露内部路径和依赖结构,有安全风险;另一方面,模型看到堆栈反而可能抓住无关细节开始“分析错误”,浪费步数。应该把异常信息精简成一句话,再附上参数里哪些字段是必填、哪些类型不对。这个“给模型看的错误消息”和“给开发者看的错误日志”应当是两套内容。
我还习惯在_execute_tool里加一个耗时统计,把执行时长记录到事件流里。这个数据很有用,能看出是模型决策慢还是工具执行慢。
4. 可观测性怎么做:日志、指标、回放
前面架构部分提到了事件流,这里展开讲实现和踩坑。
4.1 用事件钩子代替埋点散落
我最初实现可观测性时,直接在关键位置print日志。后来发现print出来的信息要么太多,要么太少,根本无法回答“Agent为什么这么做”。改成事件钩子之后,每个关键节点只往外发一个事件,消费方自己决定怎么处理。
class Harness: def __init__(self, model, registry, max_steps=8): self._listeners = [] def add_listener(self, fn): self._listeners.append(fn) def _emit(self, event: str, data: dict): for fn in self._listeners: fn(event, data)在run方法里,就可以在合适的地方调用_emit。比如模型要调用工具前发一个tool.invoking,工具返回结果后发一个tool.completed,上下文被压缩时发一个context.compacted。这些事件统一带会话ID,按时间排开,就是一份比日志更好用的“现场记录”。
事件钩子在测试里也很有用。我可以写一个测试监听器,断言某个事件必然出现、某个工具被连续调用了多少次。这比检查最终输出文本可靠得多,因为Agent可能最后回答的内容一样,但过程完全不同,过程是否合理只能靠事件验证。
4.2 把Trace串起来:sessionId与stepId
多轮Agent问题排查时,缺少关联ID是最痛苦的。一次用户请求,可能产生了多个模型调用、多个工具调用,如果日志里没有统一ID,根本没法把链路串起来。
我给每个run实例生成一个session_id,给每个主循环生成一个step_id,每个工具调用再用tool_call_id和模型侧对齐。事件格式大致是:
{ "session_id": "s_20250301_xxxx", "step_id": 3, "event": "tool.completed", "timestamp": 1740760000000, "tool_name": "query_user", "duration_ms": 120 }这个结构不复杂,但排查问题效率提升非常明显。有一次线上Agent反馈“有时候回答得不对”,我让用户把会话ID发过来,直接按session_id拉出全部事件,一眼就看到某个工具返回了空结果,后续模型基于空结果继续回答,自然就错了。如果没有这个ID,问题根本无从下手。
4.3 配置快照:让过程可以回放
只记录事件还不够,Agent行为还受配置影响。模型版本、温度参数、system prompt版本、工具Schema版本、上下文压缩策略,这些都是影响输出的变量。同一个任务今天跑出一个结果、明天跑出另一个结果,不一定是随机的,很可能是因为配置变了。
我在Harness启动时记录一份配置快照,把它和本次运行的session_id绑定。快照包含模型名、模型参数、工具列表、系统提示词版本号、Harness版本号。后续做数据回放时,按快照重建运行环境,才能复现问题。
这一步不增加多少代码,但需要养成习惯。每次改动Harness代码,版本号要记得累加。否则快照里的版本号永远一样,也就失去参考价值。
5. 常见问题与排查技巧实录
Harness写得多了,会遇到几类高频问题。我整理一下自己碰过的坑和对应的排查思路。
5.1 模型不按预期调用工具
表现:该调工具的时候不调,直接凭记忆回答;或者反复调用同一个工具,参数却没什么变化。
先看工具Schema。常见问题是描述太模糊,参数示例缺失,模型不知道怎么填。另一个常见原因是System Prompt里对工具的用法没有说明,比如“必须先调用查询工具才能作答”,这个指令通常需要写在系统提示词里,模型才会认真遵守。
如果Schema没问题,再看模型的temperature。这个参数过高会让模型“自由发挥”,容易跳过工具。工具调用型的任务,我会把温度调低,一般取0或者接近0。
有时模型也会因为前面的工具错误消息太奇怪而“绕开”工具。比如模型想查用户,工具返回“Error: unknown tool named 'query_users'”,模型可能认为自己没有这个工具,然后开始瞎编。所以工具错误信息一定要消息灵通:告诉模型“你有这个工具,但参数格式不对”,而不是让它产生误解。
5.2 上下文被工具结果撑爆
表现:跑几步之后请求变慢,然后超时;或者后面几步模型开始重复之前的内容、忘记用户原始意图。
排查第一步,看工具返回结果的大小。很多工具直接返回整个数据库表结构或完整对象,而这些结果模型根本不需要。处理办法是在工具函数内部做结果裁剪:只返回最关键的字段,或者用摘要替代原始内容。
第二步,看上下文管理策略。如果只做滑动窗口,历史里的关键信息被过早挤掉,模型就会“失忆”。可以考虑分层策略:用户目标始终保留,最近的中间步骤详细保留,更早的历史压成一段摘要。
补偿办法是把max_steps设置小一点,减少上下文累积。但这是治标不治本,根本还是控制工具返回体量。
5.3 模型陷入“假成功”
表现:模型没有真正完成任务,但返回了一大段看似正常的文本,比如“操作已完成”。
这个问题的根本原因是Harness缺少结果校验。模型认为完成了,不等于真完成了。比如一个任务要求“创建订单并发送通知”,模型可能只调用了创建订单工具,然后在文本里说“通知已发送”,但实际上并没有调用通知工具。
解决办法是在任务目标里给出明确的完成标准,并在Harness里增加一个可选的校验器:当模型返回最终文本前,校验器检查关键前置工具是否都调用成功。没有校验器的Harness,就像一个没有验收标准的项目,Agent自己说了算。这种问题在对话场景不明显,但在自动化任务里会酿成大祸。
5.4 工具执行抛异常之后Agent乱套
表现:工具刚开始正常,中间抛了个异常,下一轮模型就开始反复尝试同一个错误参数,甚至道歉来道歉去,不干正事。
我排查过好几次这种问题,根因都是一样的:异常信息没有给模型足够多的“修正依据”。只告诉它“失败了”,它不知道怎么改。正确格式应该包含:哪个参数不合法、合法格式长什么样、最简单的修正方式是什么。例如:
Error: query_user requires string field 'user_id', got None. Correct example: {"user_id": "U123456"}这会让下一轮模型大概率直接修正参数。如果错误信息含糊,模型只能瞎猜,次数全浪费在重试上。
6. 从Demo到生产:几个容易漏掉的设计
最后一个部分,聊聊把Harness从能跑变成能稳定跑的细节。
6.1 并发隔离:每个任务独享一个Harness实例
Harness内部有可变状态:当前消息列表、当前步数、事件监听器、上下文管理器状态。这些都不能跨任务共享。我见过有人把Harness设计成单例,结果两个用户同时用,消息列表互相污染,A用户的问题跑到B用户的上下文里。这个bug极难排查。
生产环境我建议每个任务创建独立的Harness实例,共享的只有工具注册表和模型客户端。工具注册表本身是只读的,可以安全共享;模型客户端如果有连接池,也可以共享。实例之间不要共享任何运行期状态。
如果你的平台对性能要求特别高,可以引入对象池复用Harness实例,但复用前必须做一次完整的reset(),确保消息列表、步数、事件缓冲全部清空。这个reset如果不彻底,比每次新建实例更容易出问题。
6.2 权限边界:工具不能对Agent透明开放
Harness一旦接上真实工具,权限问题就会迎面而来。最稳妥的原则是:Agent能看到的工具列表,本身就是过滤过的。普通用户的任务,不允许注册表里出现写操作工具;高权限任务才能看到全部工具。
这样设计的好处是模型永远“不知道自己不知道”。如果注册表里有写工具,但Harness在权限层拦截拒绝,模型可能会反复尝试,浪费步数。更好的做法是权限层直接不把这些工具加入schemas(),模型连调用机会都没有。对于偶发需要“尝试”的跨权限操作,可以给工具加一个denied描述,但这种情况越少越好。
另外,工具函数内部也要做鉴权,不能只依赖注册表。因为工具可能被其他模块直接调用,绕过Harness。Harness层的权限是“第一道门”,工具函数自身的鉴权才是底线。
6.3 压测与回放:Harness本身也要被测试
Agent的效果很难断言,但Harness可以也应该被测试。测试分两层:单元测试覆盖主循环、状态转移、错误处理;回放测试覆盖真实历史数据。
我做回放时会记录真实线上运行的输入、事件流和最终输出。每次Harness代码更新,就拿一批历史输入重新跑,对比输出的结构和工具调用序列。不需要完全一致,但要能看出明显退化。比如同一个任务,改动前会查数据库再回答,改动后变成了直接编,这种情况回放测试能很快发现。
压测方面,重点看两件事:一是高并发下模型调用和工具执行的耗时分布,二是上下文管理策略在长对话中的内存和token消耗。Harness本身不应该成为性能瓶颈,但如果事件监听器里做了同步IO,比如每个事件都写数据库,高并发下就会拖慢主流程。事件消费尽量异步化,或者至少批量写入。
最后再补一个我常用的做法:给Harness留一个“直通模式”。出问题时,可以临时关掉工具调用,让模型只基于用户输入回答,用来快速判断问题是出在工具链还是模型决策上。这个开关在生产环境能省很多排查时间。Agent Harness不是一锤子买卖,它会在实际运行中不断暴露出边界情况,守住状态机、事件流、权限和可复现性这几条主线,即使后续换模型、加工具,骨架也不需要推倒重来。