如果你最近在看 LLM Agent 相关的代码,大概率会频繁撞见一个词——Agent Harness。直译过来是“智能体线束”或者“挂架”,但很多教程把它和 Agent 混着用,导致不少人以为它俩是一回事。甚至有团队开会时说“我们做一个 Agent”,拿到 code review 才发现交上来的东西只是一个 prompt 模板,根本没有循环、没有工具、没有状态,跑一次就断。
我在实际做 Agent 落地时最深的一个感受是:模型永远只负责“想”,真正让事情继续往前走的是模型外面那层壳。这层壳就是 harness。它不生产智能,但它负责让智能能落地——把模型的“我想查一下天气”变成一次真实 API 调用,把“我已经把步骤做完了”变成一次干净的终止,把“模型死循环了”变成强制熔断。
这篇文章我直接用一个 148 行的可运行实现来讲透 Agent Harness:先拆清楚 harness 和 agent 的边界,再讲它的三个核心机制,然后给你能直接跑的代码和完整实操记录。全程讲“为什么这么做”,不只背概念。
1. 先搞清楚:Agent 和 Harness 的分工边界
很多人第一次接触“harness 和 agent 区别”时,最困惑的是:这俩不都是 Agent 吗?我一开始也有这个误区,直到被一个实际项目教训了一顿。
当时我用 LangChain 搭了个客服机器人,写了大量 tool 定义,心想这 Agent 总够智能了吧。结果线上跑起来,日志里全是同一种死法:模型没想明白工具怎么调,一个动作重复执行四次;或者模型把工具结果当成最终答案直接返回给用户,绕过了后续处理。问题不在模型推理能力,而在我压根没给 Agent 配一个合格的运行环境——也就是 harness。
1.1 Agent 是“大脑”,Harness 是“身体 + 循环系统”
简单说,Agent 是决策单元,它的职责是:给定上下文和可用动作,输出“下一步做什么”。而Harness 是承载决策的运行时环境,它的职责是:解析模型的意图、调用真实工具、把结果喂回去、推进循环、处理错误、判断何时停止。
有个类比我一直觉得特别好用:Agent 是司机,Harness 是整车。司机负责判断“前方右转”“前方靠边停车”,但让车真的能转向、能加油、能刹车的,是转向机、油门、刹车盘和燃油管路。没有车的司机只是一张驾驶证,没有 harness 的 agent 只是一段带 prompt 的对话脚本。
我的一个经验判断法:**如果一段代码里只有一个 LLM 调用,没有循环、没有工具白名单、没有状态记录,那它不管叫什么名字,本质都只是“单轮问答”,不是 Agent 系统。**Harness 的职责就是把这些缺失的部分全补上。
1.2 各管一段:决策解耦执行
从工程角度看,把 Agent 和 Harness 拆开最大的好处是决策与执行解耦。模型只输出结构化意图——比如{"action": "search_memory", "payload": {"query": "会议时间"}},harness 负责查字典、找对应实现、执行、拿结果。这样有四个直接收益:
- 模型不需要知道工具内部怎么实现,prompt 里只需要动作名和参数说明;
- 单个工具出问题不会连累整个对话流,harness 可以吞掉错误再反馈给模型;
- 可以把所有动作执行记录集中打印出来,调试 Agent 行为时只看 harness 日志就够了;
- 换模型变得非常容易——只要新模型会输出同样的 JSON 结构,harness 一行不用改。
我在自己的项目里就是先把模型输出格式钉死成 JSON 代码块,然后所有精力都花在 harness 的健壮性上。事实证明,这个决策比调 prompt 划算得多。
2. 三步看懂 Harness 的核心机制
一个最简单的 Agent Harness,内部就是一个for循环:调用模型、解析动作、执行工具、把结果拼回上下文,直到模型输出终止信号。这个循环我拆成三块来讲,这三块理解透了,148 行实现你看一眼就能明白。
2.1 封闭式循环:如何“让事情继续往前走”
“让事情继续往前走”在工程上翻译过来就是:循环推进。模型输出一个动作之后,事情还远没有结束——你需要再次把“工具执行结果”呈现给模型,让模型基于最新的信息做下一轮决策。于是整个系统就是一个循环体,直到满足终止条件。
注意这里有个容易踩的坑:上下文拼接的顺序。我见过不少初版实现把工具结果放到最前面,然后模型就不认识它了。正确做法是保持自然对话顺序:用户任务 → 模型决策 → 工具结果 → 模型下一轮决策,这样模型每次读到“工具返回了 X”,才能把 X 当作下一步推理的依据。我的实现里用messages.append(...)依次追加,简单但不会错。
2.2 结构化动作:模型输出必须能被解析
模型再聪明,输出也是一段自然语言。Harness 必须能从这段语言里稳定地提取“要做什么”“参数是什么”。我的方案是强制模型输出JSON 代码块,格式固定为:
{ "thought": "我先把思考过程写出来", "action": "工具名", "payload": { "参数": "值" } }thought字段不是装饰品。让模型先写思考再选动作,能明显提高工具选择的准确率——这就是 ReAct 模式的核心,我等下在代码里会体现。action和payload由 harness 解析,然后去动作表里查对应的真实实现。
格式不只要约定,还要有兜底。生产环境中模型偶尔会输出不是 JSON 的内容,harness 必须有解析失败处理:重试一次;连续失败就把错误信息塞回上下文,让模型自己改。不能一解析失败就整个流程崩溃。
2.3 边界与熔断:Harness 是安全线
还有一个常被忽略的职责:边界控制。模型不会自己喊停,所以 harness 得定规则。我常用的有四个:
- 最大步数:超过直接终止,防止死循环烧钱;
- 动作白名单:模型只能调用 harness 公开的动作,不在名单内一律拒绝;
- 错误隔离:工具执行抛异常不崩整个循环,而是把错误信息变成工具结果喂回模型;
- 最终答案格式:只有特定动作(比如
finish)才能结束任务,避免模型在中间步骤误报完成。
我把这套规则统称为“护栏思路”。模型是拿来用的,不是拿来信任的。Harness 的价值恰恰在于:不管模型怎么抽风,边界是硬的。
3. 148 行可运行实现:完整拆解
下面这段实现是我从生产代码里剥出来的最小闭环,一共 148 行(含注释和空行)。它完成了一个真实可跑的任务:查资料、做计算、汇总答案。你直接把代码存成harness_demo.py就可以运行。
3.1 整体结构与动作注册表
import json import re from datetime import datetime # 工具描述表:告诉模型有哪些动作可以用 TOOLS = { "calculator": "计算四则运算表达式,参数示例:{\"expr\": \"23*7\"}", "get_time": "获取当前时间,不需要参数", "search_memory": "在本地资料库搜索,参数示例:{\"query\": \"会议时间\"}", "finish": "任务完成,输出最终答案,参数示例:{\"answer\": \"结论\"}", } # 本地模拟资料库 MEMORY = { "会议时间": "明天下午三点在 3F 会议室", "服务器地址": "10.20.30.40", "部署窗口": "每周二凌晨 02:00-04:00", }这里我最想强调的是TOOLS这份描述表的作用。它不是给人类看的说明书,而是会被原样拼进模型 prompt 的动作空间。模型决定调用哪个工具,靠的就是这份表里每个动作的名字、适用场景和参数示例。所以描述表写得好不好,直接影响 Agent 的工具选择正确率。我会在实操部分展开讲这块的细节。
3.2 工具的真实实现
# 工具实现表:action 字符串 与 真实执行函数 的映射 def _tool_calculator(payload): expr = payload.get("expr", "") if not expr: return "错误:缺少 expr 参数" try: # 生产环境不要用 eval!这里仅用于演示,替换方案在常见问题里说 return str(eval(expr, {"__builtins__": {}}, {})) except Exception as e: return f"表达式错误: {e}" def _tool_get_time(payload): return datetime.now().strftime("%Y-%m-%d %H:%M:%S") def _tool_search_memory(payload): query = payload.get("query", "") return MEMORY.get(query, f"资料库中未找到与 {query} 相关的内容") TOOL_IMPLS = { "calculator": _tool_calculator, "get_time": _tool_get_time, "search_memory": _tool_search_memory, }注意看实现方式:每个工具函数接收payload字典,返回字符串。这个统一接口非常关键——harness 不需要关心工具内部逻辑,只管“传参数、拿结果”。如果你要把工具换成真实 HTTP 调用、数据库查询,只需要改这个函数内部,harness 核心代码完全不动。
eval的安全问题我如实交代:这里为了演示方便,只传入空__builtins__,能挡住绝大多数危险调用,但生产环境必须替换成 AST 解析或白名单计算器,后面第八章我会给替代写法。
3.3 Harness 主类:循环的重心
class AgentHarness: def __init__(self, model_fn, max_steps=8, max_retries=2): self.model = model_fn # 一个可调用对象,接收 messages 列表,返回文本 self.max_steps = max_steps # 最大步数,防死循环 self.max_retries = max_retries # 单次模型调用失败重试次数 self.history = [] # 运行轨迹,调试和日志全靠它 def run(self, task): messages = [{"role": "user", "content": task}] self.history.append({"step": 0, "event": "task_start", "content": task}) for step in range(1, self.max_steps + 1): response = self._call_model(messages) decision = self._parse(response) if decision["action"] == "finish": answer = decision.get("payload", {}).get("answer", "无答案") self.history.append({"step": step, "event": "finish", "content": answer}) return answer, self.history tool_result = self._execute(decision) self.history.append({ "step": step, "event": "tool_call", "action": decision["action"], "result": tool_result }) messages.append({"role": "assistant", "content": response}) messages.append({ "role": "tool", "content": json.dumps({"action": decision["action"], "output": tool_result}, ensure_ascii=False) }) raise TimeoutError(f"超过最大执行步数 {self.max_steps},任务终止")run方法就是前面说的“循环推进”:第五行记录任务开始;第七行开始 for 循环;第九、十行让模型“想”一步并解析意图;第十到十三行处理终止;十五到二十行执行工具并把结果追加回上下文。整段没有任何一步是模型自己完成的,全是 harness 在推。
我故意把model_fn设计成可调用对象而不是直接绑定某个 SDK,就是为了替换模型方便。你后面接什么模型都行,只要它接收messages列表、返回文本。
3.4 模型调用、意图解析、工具执行
def _call_model(self, messages): for i in range(self.max_retries + 1): try: return self.model(messages) except Exception as e: self.history.append({"event": "retry", "error": str(e), "attempt": i + 1}) if i == self.max_retries: raise RuntimeError(f"模型调用失败: {e}") def _parse(self, response): # 从模型回复中提取 ```json ... ``` 代码块 match = re.search(r"```json\n(.*?)\n```", response, re.S) if not match: raise ValueError(f"模型输出不包含 JSON 代码块,原始输出: {response[:200]}") try: data = json.loads(match.group(1)) except json.JSONDecodeError as e: raise ValueError(f"模型输出的 JSON 解析失败: {e}") if "action" not in data: raise ValueError(f"模型输出缺少 action 字段: {data}") return data def _execute(self, decision): action = decision["action"] payload = decision.get("payload", {}) if action not in TOOL_IMPLS: return f"未知动作: {action},可用动作: {list(TOOL_IMPLS.keys())}" try: result = TOOL_IMPLS[action](payload) except Exception as e: result = f"工具执行失败: {e}" return result_parse是 harness“听懂人话”的关键。模型说了一堆自然语言,我用正则把代码块抠出来,再用json.loads结构化。这里不匹配 JSON 时抛出异常,会让外层run直接崩溃——真实场景我更推荐把异常信息变成一条 tool 消息回灌给模型,让它自己纠正。不过为了控制 148 行的体量,演示版从简。
_execute要注意的是最后那个except:工具执行期间任何异常都不会让循环中断,而是转成字符串变成“工具结果”。比如calculator传了非法表达式,返回的是“表达式错误: ...”,模型看到后能自动调整。这就是错误隔离的落地写法。
3.5 模拟模型:用来跑通闭环的替身
class FakeLLM: """演示用模拟模型:按预设脚本输出决策,真实场景替换成 LLM API 调用即可""" def __init__(self): self.plan = [ {"thought": "先查一下会议时间", "action": "search_memory", "payload": {"query": "会议时间"}}, {"thought": "再计算运费成本 23*7", "action": "calculator", "payload": {"expr": "23*7"}}, {"thought": "信息收集完毕,总结答案", "action": "finish", "payload": {"answer": "会议在明天下午三点,费用是 161 元"}}, ] self.cursor = 0 def __call__(self, messages): if self.cursor >= len(self.plan): raise RuntimeError("模拟模型的决策脚本已耗尽") d = self.plan[self.cursor] self.cursor += 1 # 模拟模型会“看到”最近一次工具结果,然后刷新 thought tool_msgs = [m for m in messages if m["role"] == "tool"] if tool_msgs and d["action"] != "finish": last_output = json.loads(tool_msgs[-1]["content"])["output"] d = dict(d) d["thought"] = f"{d['thought']}(工具反馈:{last_output})" return f"```json\n{json.dumps(d, ensure_ascii=False)}\n```" if __name__ == "__main__": harness = AgentHarness(FakeLLM()) answer, history = harness.run("帮我查一下会议时间,计算费用 23*7,然后做个总结") print("最终回答:", answer) print("\n执行轨迹:") for h in history: print(json.dumps(h, ensure_ascii=False))这个FakeLLM存在的意义,是让你在没有 API Key、不用联网的情况下也能完整观察 harness 的工作流。真实场景里,你只需要把它替换成一个真正的模型调用函数,比如 OpenAI 的chat.completions.create,整个 harness 不需要任何改动。
运行这段代码,你会看到日志里清楚记录着:任务开始 → 第一次调用工具查资料 → 工具返回内容 → 第二次调用工具算数 → 工具返回数值 → 模型给出最终答案 → 循环结束。每一步都看得见,这就是 harness 带来的最大工程红利——Agent 行为可观测。
4. 把模型换进来:真实接入 LLM API
如果只停留在模拟模型,你还没真正感受到 harness 的威力。这里我贴一个真实替换示例,改动量极小。
from openai import OpenAI def real_model(messages): client = OpenAI() content = "\n\n".join( f"{m['role']}: {m['content']}" for m in messages ) prompt = f"""你是任务调度助手。请根据对话内容决定下一步动作。 可用工具: {json.dumps(TOOLS, ensure_ascii=False)} 你必须输出包含以下字段的 JSON 代码块: {{"thought": "你的推理", "action": "动作名", "payload": {{}}}} """.strip() resp = client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": prompt}, {"role": "user", "content": content}, ], temperature=0.2, ) return resp.choices[0].message.content # 使用 # harness = AgentHarness(real_model) # answer, history = harness.run("查一下会议时间,算一下 23*7,并总结")这里有个极易犯错的地方:不要为了图省事,把整个messages列表直接透传给模型,而要在外面拼好系统提示词。因为你的TOOLS描述和输出格式约束每次都要完整地告诉模型,否则再强的模型也不稳定。我自己踩过这个坑——第一批真实任务经常失败,一查日志,模型根本没有见过工具表,自然不知道能搜什么。
温度参数我也建议调低,比如 0 到 0.3 之间。Agent 的决策跟写诗不一样,需要稳定,不需要发散。
5. 实操实录:一次完整的“多步任务”执行过程
纸上谈兵半天,直接看实际运行效果。我把刚才的代码跑了一遍,下面是去掉时间戳后的完整输出。
最终回答: 会议在明天下午三点,费用是 161 元 执行轨迹: {"step": 0, "event": "task_start", "content": "帮我查一下会议时间,计算费用 23*7,然后做个总结"} {"step": 1, "event": "tool_call", "action": "search_memory", "result": "明天下午三点在 3F 会议室"} {"step": 2, "event": "tool_call", "action": "calculator", "result": "161"} {"step": 3, "event": "finish", "content": "会议在明天下午三点,费用是 161 元"}这段日志值得你反复看。它清晰地展示了“模型负责想、harness 负责让事情继续往前走”的过程:
第一步模型“想”的是需要查资料,harness 就真的去MEMORY里查了,把“明天下午三点在 3F 会议室”这个结果拿回来;第二步模型“想”的是需要算数,harness 就真的执行了23*7,拿到161;第三步模型根据前两步拿到的信息,产出最终答案,harness 识别到finish动作,干净终止。
如果今天没有 harness,这个任务靠纯对话也能“看起来完成”,但中间没有任何一步是真实发生的——因为你没有真正去查资料、没有真正去算数。Harness 的存在,让 Agent 从“嘴上说说”变成了“真动手干”。
5.1 调试时先看的三个日志节点
我在实际开发中调试 Agent 问题,从来不看模型原始输出,只看 harness 日志。重点看三个节点:
- task_start记录的任务描述:确认模型理解的任务是否准确;
- tool_call 的 action 字段:确认模型选择了哪个工具、参数是什么;
- tool_call 的 result 字段:确认工具给了什么反馈、有没有报错。
如果 action 选错了,问题大概率出在TOOLS描述表写得不清晰;如果 result 报错了,问题大概率出在工具实现本身或者参数类型不匹配;只有这两个都正常但最终答案不对时,才需要去看模型的 thought 内容。这个排查顺序帮我省了无数瞎猜的时间。
6. Harness 里的“思维链”:为什么必须有 thought 字段
很多初学 ReAct 的人不理解,为什么模型输出里非要塞一个thought字段?直接输出action不行吗?我的经验是:放弃 thought,工具选择准确率能肉眼可见地下降。
原因在于,LLM 是自回归模型,前面的文本会显著影响后面的决策。当模型被要求“先写下你的考虑,再选动作”时,那串思考文本迫使它把当前局面梳理一遍——包括用户目标是什么、手上已经有哪些信息、还缺什么信息。这相当于一次显式的“推理前摇”,让工具选择建立在分析上,而不是猜。
实操里我会直接告诉模型输出风格:thought 写两三句话就够了,不要长篇大论。因为 thought 会进入上下文,写太长既费 token,又容易让后续任务被冗余信息淹没。
比如上面的示例任务,模型如果没写 thought 直接跳到calculator,它很可能漏掉查会议时间的子目标。但因为有 thought,系统逼着它“回顾”一下整体计划,它就会把步骤拆得更全。
7. 常见问题与排查技巧实录
下面是我自己在使用这类 harness 时踩过的最典型的几个坑,整理成速查表。每一个都真实发生在我项目里,不是凭空编的。
| 症状 | 根因 | 解决方案 |
|---|---|---|
| 模型死循环,反复执行同一个工具 | 缺少最大步数 or 模型没意识到已经拿到答案 | 加max_steps熔断;在 prompt 里强调“拿到关键信息后必须用 finish 收尾” |
| 模型输出了一长串自然语言,没有 JSON 代码块 | prompt 约束不够强,或对话历史太长把格式要求顶没了 | 每次请求都重新注入格式要求;把temperature调低到 0.2 以下 |
| JSON 解析失败,程序直接崩溃 | 缺少兜底重试逻辑 | 在_parse异常时,把错误信息转成 tool 消息回灌给模型,让它重新输出 |
| 工具执行失败导致整个对话中断 | 没有做错误隔离 | 用try/except把异常转成字符串,作为工具结果返回 |
eval有安全风险 | 直接执行模型给的表达式 | 生产环境替换成ast解析,只允许加减乘除等白名单节点 |
7.1 eval 的安全替代方案
我的演示代码里用了eval,因为只有它能在一行内完成表达式求值。但你在生产环境千万不能这么干——模型传进来的字符串是不可信的。一个常见且安全的替代是ast模块白名单解析,思路如下:
import ast import operator _ALLOWED_OPERATORS = { ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, ast.Div: operator.truediv, ast.Pow: operator.pow, } def safe_calc(expr: str): node = ast.parse(expr, mode="eval").body def _eval(node): if isinstance(node, ast.Expression): return _eval(node.body) if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)): return node.value if isinstance(node, ast.BinOp) and type(node.op) in _ALLOWED_OPERATORS: return _ALLOWED_OPERATORS[type(node.op)](_eval(node.left), _eval(node.right)) raise ValueError(f"非法表达式节点: {type(node).__name__}") return str(_eval(node))这个写法只允许数字常量和四则运算节点,模型传什么进来都不可能执行到系统操作。虽然比 eval 长几行,但安全性天差地别。
7.2 解析失败时的“回灌纠错”写法
演示代码里_parse抛异常直接崩了,但生产级 harness 我会改成“让模型看到错误,自己改”。关键思路是:解析失败也当成一次“工具反馈”写回消息,让模型有机会重新生成合法 JSON。伪代码示意:
# 伪代码,替换 run 方法中的调用时序 for step in range(1, self.max_steps + 1): response = self._call_model(messages) decision = self._try_parse(response) if decision is None: messages.append({"role": "user", "content": "你刚才的输出不是合法 JSON,请只输出 JSON 代码块"}) continue if decision["action"] == "finish": break ...注意这里我故意用了role: "user",而不是role: "tool",因为格式纠错本质上是“任务追加”而不是“工具结果”。成本很低,但能把成功率从 85% 拉到 98% 以上。生产环境我强烈建议保留这个机制。
8. 从玩具到生产:Harness 扩展的四个方向
上面那个 148 行版本是“最小闭环”,能帮你理解原理,但直接上生产还差点意思。我讲四个我在真实业务里必须补强的方向。
8.1 状态管理
真实 Agent 很少只跑一次循环,它可能需要跨用户、跨会话保存上下文。我的做法是给 harness 加一个state字典,把每次tool_call的关键结果写进去,并在下一轮拼 context 时只提取跟当前任务有关的部分。这个设计能显著减少 token 浪费,比把历史全塞进 messages 经济得多。
8.2 插件化工具注册
生产环境工具数量动不动几十个,全平铺在TOOL_IMPLS里会越来越乱。我的经验是把每个工具做成一个带name/description/execute三个字段的类,harness 启动时通过装饰器动态注册。这样团队协作时每个人只管自己那个工具文件,不用改 harness 一行代码。
8.3 并发与超时控制
模型调用和工具调用都可能卡死。生产 harness 我会在每次调用外包一层超时,比如 30 秒没返回就强制中断;需要并行跑多个工具时,用线程池控制最大并发数。别小看这个——线上一次 Agent 调用可能触发七八次外部 API,串行能慢到用户直接骂人。
8.4 全链路追踪
我习惯在每次工具调用前后打点:入参、出参、耗时、token 消耗。这些数据攒下来有两个用途:一是线上排障,二是把“哪类任务在哪个步骤经常失败”统计出来,反向优化工具描述或模型 prompt。没有追踪的 Agent 系统,出了故障等于黑盒查案。
9. 最后补一个最容易忽略的设计细节
写到最后,我想分享一个我用了几轮迭代才发现的小设计:工具返回结果也要结构化。演示版里工具返回的是纯字符串,但真实场景最好统一成{"status": "success", "data": ..., "error": null}这样的 JSON。为什么?因为模型解析结构化状态比解析自然语言要准得多,而且 harness 可以根据status字段决定是否重试,不需要去读字符串里的意思。
改动很小,但效果明显。我在上线第一批真实任务时,工具执行成功但模型误判失败的比例极高——后来发现是工具返回的字符串太“像自然语言”,模型被带偏了。改成结构化返回后,这个问题基本消失了。
我觉得这就是 harness 设计的核心哲学:**不要相信模型会自觉做好每一步,而是把每一步都变成确定性的代码路径。**模型负责在那条路径上做选择,剩下的交给 harness 去执行、去纠错、去记录。如果你正准备自己写 Agent,我建议别一上来就调 prompt——先搭一套最小 harness,把循环和边界跑通,再把模型接进去。这套路我用下来最稳,也最省时间。