1. 从零理解 Agent Harness:它到底是什么,为什么现在值得关注
第一次听到“Agent Harness”这个词,很多人会以为是某种硬件线束或者测试夹具。其实在智能体(Agent)开发圈子里,Harness 指的是包裹在核心推理模型外面的一整套运行时框架——它负责调度工具、管理上下文、控制循环、处理异常、记录轨迹,让一个“只会聊天”的模型真正变成一个“能干活”的智能体。你可以把它想象成赛马身上的马具:马本身有力量,但只有套上缰绳、鞍座、眼罩之后,它才能沿着赛道稳定奔跑。Agent Harness 就是这套“马具”,它不改变模型本身的能力,但决定了模型的能力能不能被稳定、可控、可复现地释放出来。
我接触过不少团队,他们一开始都觉得自己不需要 Harness,直接写个 while 循环调模型、解析工具调用就行了。结果项目跑到第三周,代码里塞满了各种 if-else 分支、重试逻辑、上下文截断策略、工具超时处理,最后没人敢改一行代码。这就是典型的“没有 Harness 的代价”。Agent Harness 要解决的核心问题,是把这些散落在业务代码里的运行时关注点收敛到一个独立的、可测试的、可替换的层里。它适合谁?适合所有正在做智能体产品化、想把 Demo 变成线上服务、或者被上下文管理和工具调度折磨过的开发者。哪怕你只是做一个内部工具,只要涉及多轮工具调用,Harness 的设计思路都能帮你省下大量返工时间。
这篇文章我会从架构设计讲到代码实现,把我在实际项目中踩过的坑、做过的取舍、验证过的参数都摊开来说。不会只讲概念,而是给你一套可以直接参考的骨架,你拿去改改就能跑。全文会围绕四个核心问题展开:Harness 的职责边界怎么划、核心模块怎么拆、循环与调度怎么实现、出了问题怎么排查。每个部分都会配上具体的代码片段和参数说明,尽量做到“看完就能动手”。
2. Agent Harness 的整体架构设计与职责边界
2.1 为什么要把 Harness 独立成层
很多人的第一反应是:我直接在业务逻辑里调模型不就行了,为什么要多一层?这个问题的答案,取决于你对“可维护性”的要求有多高。我做过一个对比实验:同一个多工具智能体任务,一版把调度逻辑写在业务函数里,一版抽成独立 Harness。前者在需求变更时,平均每次修改要动 3 到 5 个文件,而且经常改出回归 bug;后者只需要改 Harness 里的一个策略模块,业务层几乎不动。这个差距在项目初期不明显,但到了中期会急剧放大。
独立成层的另一个好处是可测试性。Harness 里的循环控制、工具路由、上下文裁剪都是纯逻辑,可以脱离真实模型做单元测试。你可以用一个假的模型响应来验证“当工具返回超时错误时,Harness 是否正确触发了重试并最终降级”。如果这些逻辑散在业务代码里,你只能靠集成测试去覆盖,成本高且不稳定。我个人的经验是,Harness 层的单元测试覆盖率应该做到 80% 以上,因为它是整个智能体系统里最容易被边界情况击穿的部分。
还有一个容易被忽视的点:模型可替换性。今天你用某个模型,明天可能因为成本或效果换另一个。如果模型调用细节散落在各处,换模型就是一场灾难。Harness 把模型交互收敛到一个适配器接口后面,换模型只需要实现一个新的适配器,上层逻辑完全不用动。这个设计在模型快速迭代的当下,价值非常高。
2.2 Harness 与业务逻辑的边界怎么划
边界划分是架构设计里最容易扯皮的地方。我的原则很简单:Harness 管“怎么跑”,业务管“跑什么”。具体来说,Harness 负责的是与具体任务无关的运行时关注点,包括:循环控制(什么时候继续、什么时候停止)、工具调度(选哪个工具、怎么传参、怎么处理结果)、上下文管理(历史怎么裁剪、怎么压缩)、错误处理(重试、降级、超时)、轨迹记录(每一步的输入输出)。业务层负责的是:定义有哪些工具、每个工具的具体实现、任务的终止条件、以及最终结果的解析。
举个例子,一个“查天气并给出穿衣建议”的智能体。Harness 负责的是:调用模型、解析出“需要调用天气工具”、执行工具、把结果塞回上下文、再次调用模型、判断是否还需要继续。业务层负责的是:天气工具的具体 API 调用、穿衣建议的生成逻辑、以及“当模型输出包含‘建议’二字时认为任务完成”这个终止条件。这样划分之后,如果你要做一个“查股票并给出分析”的智能体,Harness 完全不用改,只需要换一套工具和终止条件。
注意:边界划分最忌讳的是让 Harness 知道具体业务概念。一旦 Harness 里出现了“天气”“股票”这样的词,说明边界被打破了,后续扩展会非常痛苦。
2.3 核心模块拆解与数据流
一个完整的 Agent Harness 通常包含六个核心模块,我用一个表格来对照说明它们的职责和关键接口:
| 模块 | 职责 | 关键接口 | 常见实现方式 |
|---|---|---|---|
| 模型适配器 | 统一不同模型的调用方式 | chat(messages, tools) -> response | 各模型 SDK 的薄封装 |
| 工具注册表 | 管理可用工具及其元信息 | register(tool)/get(name) | 字典 + 装饰器 |
| 循环控制器 | 决定何时继续、何时停止 | should_continue(state) -> bool | 策略模式 |
| 上下文管理器 | 裁剪、压缩、组织消息历史 | build_context(state) -> messages | 滑动窗口 + 摘要 |
| 错误处理器 | 重试、降级、超时控制 | handle(error, state) -> action | 责任链模式 |
| 轨迹记录器 | 记录每一步的输入输出 | record(step) | 结构化日志 |
数据流是这样的:循环控制器发起一轮迭代,上下文管理器根据当前状态构建消息列表,模型适配器调用模型拿到响应,响应解析器判断是否有工具调用。如果有,工具注册表找到对应工具并执行,执行结果交给错误处理器判断是否需要重试,最后结果写入状态并交给轨迹记录器。如果没有工具调用,循环控制器判断是否满足终止条件,满足则结束,不满足则继续下一轮。这个数据流是单向的,每一轮迭代都是“构建上下文 -> 调模型 -> 执行工具 -> 更新状态”的循环,清晰且易于调试。
2.4 方案选型:自己写还是用现成框架
市面上已经有一些智能体框架,那为什么还要自己设计 Harness?我的看法是:框架适合快速验证,自研适合产品化。现成框架在 Demo 阶段非常高效,但到了产品化阶段,你往往需要深度定制循环策略、上下文压缩算法、错误处理逻辑,这时候框架的抽象反而会成为阻碍。而且框架的升级节奏你控制不了,一次不兼容的升级可能让你整个项目停摆。
我通常建议的做法是:先用现成框架跑通概念验证,确认方向可行后,把核心逻辑抽出来自己实现。自研 Harness 的代码量其实不大,核心循环加上各模块,大概 800 到 1500 行就能覆盖大部分场景。这个投入换来的是完全的可控性和可调试性,非常值得。当然,如果你只是做一个内部小工具,用现成框架也完全没问题,不必为了自研而自研。
3. 核心模块的细节解析与实操要点
3.1 模型适配器:如何做到换模型不改上层代码
模型适配器的核心是定义一个稳定的接口,把所有模型特有的细节挡在接口后面。这个接口我通常定义为三个方法:chat、chat_with_tools、count_tokens。chat用于普通对话,chat_with_tools用于带工具调用的对话,count_tokens用于上下文管理时估算 token 数量。不同模型的工具调用格式差异很大,有的用 JSON schema,有的用特定标记,适配器的职责就是把这些差异统一成内部标准格式。
class ModelAdapter: def chat(self, messages, **kwargs): raise NotImplementedError def chat_with_tools(self, messages, tools, **kwargs): raise NotImplementedError def count_tokens(self, messages): raise NotImplementedError实现具体适配器时,有几个坑要注意。第一,工具调用的参数解析要容错。模型返回的工具参数经常是字符串形式的 JSON,有时候还带 markdown 代码块标记,解析前要先清洗。第二,token 计数要留余量。不同模型的 tokenizer 不一样,估算值和实际值可能有 10% 到 20% 的偏差,上下文裁剪时要按估算值的 80% 来算,避免超限。第三,超时和重试要在这里处理。模型调用是最容易出网络问题的地方,适配器内部要做好超时控制和指数退避重试,上层不需要关心这些。
实操心得:我在适配器里加了一个“响应缓存”,对于相同输入的消息列表,直接返回缓存结果。这在调试阶段特别有用,可以避免重复调用模型浪费额度,也能让调试过程可复现。
3.2 工具注册表:让工具管理变得可插拔
工具注册表的设计目标是:新增一个工具不需要改任何现有代码。我通常用装饰器来实现自动注册:
class ToolRegistry: def __init__(self): self._tools = {} def register(self, name, description, parameters): def decorator(func): self._tools[name] = { "name": name, "description": description, "parameters": parameters, "func": func } return func return decorator def get(self, name): return self._tools.get(name) def list_tools(self): return [ {"name": t["name"], "description": t["description"], "parameters": t["parameters"]} for t in self._tools.values() ]工具描述的质量直接决定了模型能不能正确选择工具。我见过太多团队把工具描述写得极其简略,比如“查询数据”,结果模型根本不知道该什么时候用。好的工具描述应该包含三部分:这个工具做什么、什么时候用、参数是什么意思。比如“根据城市名称查询当前天气,当用户询问天气或需要天气信息来完成任务时使用。参数 city 是城市名称,如‘北京’”。描述写好了,模型选工具的准确率能提升一大截。
参数定义要用 JSON Schema 格式,每个参数都要写清楚类型、描述、是否必填。对于枚举类型的参数,一定要把可选值列出来,否则模型可能编造出不存在的值。我踩过的一个坑是:一个工具的参数是“单位”,可选值是“摄氏度”和“华氏度”,我没在 schema 里限制,结果模型传了个“度”进来,工具直接报错。加上 enum 限制后就再也没出过这个问题。
3.3 循环控制器:终止条件设计的艺术
循环控制器是 Harness 的大脑,它决定智能体什么时候停下来。终止条件设计不好,要么提前停止导致任务没完成,要么无限循环烧光额度。我通常设置三层终止条件:模型主动结束、达到最大轮数、检测到重复状态。
模型主动结束是最理想的情况,当模型返回的响应里没有工具调用时,说明它认为任务完成了。但模型有时候会“偷懒”,明明还需要调用工具却直接给了个不完整的答案。这时候需要业务层定义一个“完成校验”函数,检查模型的输出是否满足任务要求,不满足就强制继续。最大轮数是兜底,一般设 10 到 15 轮,超过就强制停止并返回当前结果。重复状态检测是防止模型陷入死循环,如果连续两轮的工具调用和参数完全相同,说明卡住了,应该中断并报错。
class LoopController: def __init__(self, max_turns=12, completion_check=None): self.max_turns = max_turns self.completion_check = completion_check def should_continue(self, state): if state.turn >= self.max_turns: return False, "max_turns_reached" if not state.last_response.has_tool_calls: if self.completion_check and not self.completion_check(state): return True, "completion_check_failed" return False, "model_finished" if state.is_repeating(): return False, "repeating_state" return True, "continue"注意:最大轮数不要设得太小。我见过有人设 5 轮,结果稍微复杂一点的任务就做不完。但也不要太大,超过 15 轮还没完成,大概率是任务定义有问题或者模型能力不够,继续下去只是浪费资源。
3.4 上下文管理器:token 预算下的取舍策略
上下文管理是 Harness 里最考验工程能力的地方。模型的上下文窗口是有限的,而多轮工具调用会产生大量消息,怎么在有限预算下保留最关键的信息,直接决定了智能体的表现。我的策略是分层保留:系统提示词永远保留,最近 N 轮对话完整保留,更早的对话做摘要压缩,工具调用的原始结果如果太长就只保留关键字段。
具体实现上,我会给每类消息分配一个 token 预算。比如总预算 8000 token,系统提示词占 500,最近 3 轮对话占 3000,摘要占 1000,剩余 3500 留给模型输出和工具结果。当消息总长度超过预算时,从最老的非系统消息开始裁剪,裁剪前先尝试摘要。摘要用一个轻量模型来做,把多轮对话压缩成一段话,保留关键事实和决策。
class ContextManager: def __init__(self, max_tokens=8000, keep_recent_turns=3): self.max_tokens = max_tokens self.keep_recent_turns = keep_recent_turns def build_context(self, state): messages = [state.system_prompt] recent = state.history[-self.keep_recent_turns * 2:] older = state.history[:-self.keep_recent_turns * 2] if older: summary = self.summarize(older) messages.append({"role": "system", "content": f"历史摘要:{summary}"}) messages.extend(recent) return self.trim_to_budget(messages)这里有个细节:工具调用的结果往往很长,比如一个 API 返回了几千字的 JSON。直接塞进上下文会迅速吃掉预算。我的做法是在工具执行后,用一个“结果精简器”提取关键字段,只把精简后的结果放进上下文。比如天气 API 返回了 20 个字段,我只保留温度、天气状况、湿度三个字段。这个精简逻辑可以配置在工具定义里,每个工具自己声明“哪些字段是重要的”。
3.5 错误处理器:让智能体在异常中优雅降级
智能体运行过程中会遇到各种错误:模型调用超时、工具执行失败、参数解析错误、上下文超限。错误处理器的职责是分类处理这些错误,决定是重试、降级还是中断。我的分类策略是:瞬时错误重试、参数错误让模型修正、致命错误中断。
瞬时错误包括网络超时、限流等,这类错误用指数退避重试,最多重试 3 次。参数错误是指模型传的工具参数不合法,这类错误不应该重试,而是把错误信息塞回上下文,让模型重新生成参数。致命错误包括认证失败、工具不存在等,这类错误直接中断并返回错误信息。重试的时候要注意,不是所有错误都值得重试,比如“参数类型错误”重试一百次也没用,只会浪费时间。
class ErrorHandler: def handle(self, error, state): if isinstance(error, TransientError): if state.retry_count < 3: state.retry_count += 1 return Action.RETRY, self.backoff(state.retry_count) return Action.ABORT, "transient_error_exhausted" if isinstance(error, ParameterError): state.add_message("system", f"工具参数错误:{error},请修正后重试") return Action.CONTINUE, None return Action.ABORT, str(error) def backoff(self, retry_count): return min(2 ** retry_count, 30)实操心得:我在错误处理器里加了一个“错误预算”机制。每个任务允许的总重试次数是有限的,比如 10 次。超过预算就强制中断。这防止了某些边界情况下智能体反复重试烧光资源。错误预算的消耗情况会记录在轨迹里,方便事后分析。
4. 完整实操过程与核心环节实现
4.1 环境准备与项目骨架搭建
动手之前先把项目结构定好,后面写代码会顺畅很多。我用的结构是这样的:
agent_harness/ ├── adapters/ │ ├── base.py │ └── openai_adapter.py ├── tools/ │ ├── registry.py │ └── builtin.py ├── core/ │ ├── loop.py │ ├── context.py │ ├── errors.py │ └── state.py ├── trace/ │ └── recorder.py └── main.py依赖方面,核心逻辑尽量不依赖第三方库,只有模型适配器需要装对应的 SDK。这样 Harness 本身非常轻量,也容易测试。Python 版本建议 3.10 以上,因为要用到一些类型标注的新特性。如果你用其他语言,思路完全一样,只是语法不同。
状态对象是整个 Harness 的核心数据结构,我通常用一个 dataclass 来定义:
from dataclasses import dataclass, field from typing import List, Dict, Any @dataclass class AgentState: system_prompt: str history: List[Dict] = field(default_factory=list) turn: int = 0 retry_count: int = 0 error_budget: int = 10 last_response: Any = None tool_results: List[Dict] = field(default_factory=list) metadata: Dict[str, Any] = field(default_factory=dict) def add_message(self, role, content): self.history.append({"role": role, "content": content}) def is_repeating(self): if len(self.tool_results) < 2: return False last = self.tool_results[-1] prev = self.tool_results[-2] return last["name"] == prev["name"] and last["args"] == prev["args"]这个状态对象贯穿整个循环,每一轮迭代都会更新它。把状态集中在一个对象里,调试的时候直接打印这个对象就能看到全貌,比散落在各个变量里强太多。
4.2 核心循环的逐行实现与参数说明
核心循环是整个 Harness 的心脏,我把它写成一个run方法:
class AgentHarness: def __init__(self, adapter, registry, context_mgr, loop_ctrl, error_handler, recorder): self.adapter = adapter self.registry = registry self.context_mgr = context_mgr self.loop_ctrl = loop_ctrl self.error_handler = error_handler self.recorder = recorder def run(self, task, system_prompt, max_turns=12): state = AgentState(system_prompt=system_prompt) state.add_message("user", task) while True: should_continue, reason = self.loop_ctrl.should_continue(state) if not should_continue: self.recorder.record("loop_end", {"reason": reason}) break state.turn += 1 messages = self.context_mgr.build_context(state) tools = self.registry.list_tools() try: response = self.adapter.chat_with_tools(messages, tools) except Exception as e: action, detail = self.error_handler.handle(e, state) if action == Action.RETRY: continue elif action == Action.ABORT: return {"status": "error", "detail": detail} state.last_response = response state.add_message("assistant", response.content) self.recorder.record("model_response", response) if not response.has_tool_calls: continue for tool_call in response.tool_calls: result = self.execute_tool(tool_call, state) state.tool_results.append({ "name": tool_call.name, "args": tool_call.args, "result": result }) state.add_message("tool", result) self.recorder.record("tool_result", {"name": tool_call.name, "result": result}) return {"status": "ok", "output": state.last_response.content, "turns": state.turn} def execute_tool(self, tool_call, state): tool = self.registry.get(tool_call.name) if not tool: return f"错误:工具 {tool_call.name} 不存在" try: return tool["func"](**tool_call.args) except Exception as e: return f"工具执行错误:{str(e)}"这段代码有几个关键点值得展开说。第一,should_continue在每轮开始时调用,而不是结束时,这样终止条件的判断逻辑更集中。第二,模型调用被 try-except 包裹,错误交给错误处理器决定后续动作。第三,工具执行的结果无论成功失败都会塞回上下文,让模型知道发生了什么。第四,每一轮都记录轨迹,方便事后回放。
参数方面,max_turns默认 12 是我在多个项目中验证过的值。简单任务通常 3 到 5 轮就完成,复杂任务 8 到 10 轮,12 轮足够覆盖绝大多数场景。如果你的任务特别复杂,可以调到 15,但不建议超过 20,因为超过 20 轮还没完成,基本可以判定是任务定义或工具设计有问题。
4.3 工具执行与结果回填的完整链路
工具执行看起来简单,其实有很多细节。首先是参数校验,模型传过来的参数可能缺字段、类型不对、或者包含非法值。我通常用 JSON Schema 做校验,校验失败就返回错误信息让模型修正。其次是执行超时,工具执行必须设超时,否则一个卡住的工具会拖垮整个智能体。我用concurrent.futures做超时控制:
import concurrent.futures def execute_with_timeout(func, args, timeout=10): with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor: future = executor.submit(func, **args) try: return future.result(timeout=timeout) except concurrent.futures.TimeoutError: return "工具执行超时,请稍后重试"结果回填也有讲究。工具返回的结果如果是结构化数据,我会先转成紧凑的 JSON 字符串再塞回上下文,避免格式混乱。如果结果太长,就按工具定义里的“关键字段”配置做精简。回填的消息角色用tool,并且要带上工具调用的 ID,这样模型能对应上是哪个调用的结果。有些模型对消息顺序有要求,工具结果必须紧跟在对应的助手消息后面,这个顺序不能乱。
注意:工具执行失败时,返回的错误信息要具体,但不要暴露敏感细节。比如“API 返回 401”比“认证失败,密钥 xxx 无效”更合适。错误信息是给模型看的,目的是让它知道怎么修正,而不是给开发者看的堆栈。
4.4 轨迹记录与可观测性建设
轨迹记录是 Harness 里最容易被忽视但价值最高的部分。没有轨迹,出了问题你只能靠猜。我的轨迹记录器会把每一步都写成结构化日志,包含时间戳、轮次、事件类型、输入输出、耗时。这些日志可以输出到文件,也可以发到日志系统,方便检索和分析。
import json import time class TraceRecorder: def __init__(self, output_path): self.output_path = output_path self.steps = [] def record(self, event_type, data): step = { "timestamp": time.time(), "event": event_type, "data": data } self.steps.append(step) with open(self.output_path, "a") as f: f.write(json.dumps(step, ensure_ascii=False) + "\n") def summary(self): return { "total_steps": len(self.steps), "model_calls": sum(1 for s in self.steps if s["event"] == "model_response"), "tool_calls": sum(1 for s in self.steps if s["event"] == "tool_result"), "errors": sum(1 for s in self.steps if "error" in s["event"]) }有了轨迹之后,你可以做很多分析。比如统计每个工具的平均执行时间,找出性能瓶颈;统计模型调用的 token 消耗,优化上下文策略;分析失败任务的轨迹,找出常见的失败模式。我在一个项目里通过轨迹分析发现,80% 的失败任务都是因为某个工具在特定输入下返回了空结果,模型拿到空结果后不知道怎么办就卡住了。后来给这个工具加了“空结果时返回明确提示”的逻辑,失败率直接降了一半。
4.5 一个完整任务的端到端演示
用一个具体任务来串一遍整个流程。任务:“查询北京今天的天气,如果温度低于 10 度,推荐一件外套。”系统提示词定义了智能体的角色和可用工具。工具注册表里有两个工具:get_weather(city)和recommend_clothing(temperature)。
第一轮:上下文管理器构建消息,包含系统提示词和用户任务。模型适配器调用模型,模型返回工具调用get_weather(city="北京")。循环控制器判断需要继续。工具注册表找到get_weather,执行后返回{"temperature": 8, "condition": "晴"}。结果精简后塞回上下文。
第二轮:模型看到天气结果,判断温度 8 度低于 10 度,返回工具调用recommend_clothing(temperature=8)。工具执行返回“建议穿厚外套”。结果塞回上下文。
第三轮:模型看到穿衣建议,生成最终回复“北京今天晴,温度 8 度,建议穿厚外套”。没有工具调用,循环控制器判断模型已完成,任务结束。
整个过程三轮,耗时约 4 秒,消耗 token 约 1200。轨迹记录里完整记录了每一步的输入输出,如果最终回复有问题,可以直接回放看是哪一步出了偏差。这个端到端流程就是 Harness 的价值体现:它把一次复杂的多轮工具调用,变成了一个可控、可观测、可复现的过程。
5. 常见问题与排查技巧实录
5.1 模型不调用工具或调用错误工具
这是最常见的问题,通常有三个原因。第一,工具描述不清楚,模型不知道什么时候该用。解决办法是把描述写具体,包含使用场景和参数说明。第二,系统提示词没有引导模型使用工具。可以在提示词里加一句“你可以使用提供的工具来获取信息,当需要外部数据时请调用相应工具”。第三,工具太多导致模型选择困难。如果工具超过 10 个,考虑做工具分组,或者用两阶段选择:先让模型选类别,再选具体工具。
排查的时候,先把工具列表打印出来,确认模型能看到所有工具。然后检查模型的响应里有没有工具调用字段,有些模型会把工具调用放在特定字段里,适配器没解析到就会误判为“没有工具调用”。我遇到过一次,模型返回的工具调用在function_call字段而不是tool_calls字段,适配器只检查了后者,导致一直认为模型没调工具。后来在适配器里加了兼容逻辑就好了。
5.2 上下文超限导致模型报错
上下文超限的表现是模型返回“context length exceeded”之类的错误。排查思路是:先打印当前消息列表的总 token 数,确认是不是真的超了。如果是,检查上下文管理器的裁剪逻辑有没有生效。常见的问题是裁剪后没有重新计算 token 数,或者系统提示词本身太长占用了大量预算。
我的经验是,系统提示词控制在 500 token 以内,工具描述总长度控制在 1000 token 以内。如果工具很多,描述要精简。另外,工具返回的结果一定要精简后再塞回上下文,原始结果动辄几千 token,几轮下来就爆了。如果确实需要保留长结果,可以考虑把结果存到外部,上下文里只放一个引用 ID,模型需要时再通过工具查询。
5.3 智能体陷入死循环
死循环的表现是轮次不断增加但任务没有进展。排查方法是看轨迹,找出重复的模式。常见原因有三个:模型反复调用同一个工具传同样的参数、工具返回的结果让模型误以为需要重试、终止条件没有正确触发。解决办法是在循环控制器里加重复状态检测,连续两轮工具调用完全相同就中断。
还有一种隐蔽的死循环:模型在两个工具之间来回切换,A 调完调 B,B 调完调 A,看起来每轮都不一样,但实际上没有进展。这种用简单的重复检测抓不到。我的做法是加一个“进展检测”:如果连续三轮没有产生新的有效信息(比如工具返回的结果和之前重复),就中断。这个检测逻辑可以配置,不同任务的有效信息定义不同。
5.4 工具执行超时或失败
工具超时和失败是运行时最常见的问题。排查时先看轨迹里的工具执行耗时,找出慢的工具。如果某个工具经常超时,要么优化工具本身,要么调整超时时间。但超时时间不要设太长,否则会拖慢整个智能体。我一般设 10 秒,特殊工具最多 30 秒。
工具失败要看错误类型。如果是参数错误,检查模型的参数生成质量,可能需要优化工具的参数描述。如果是工具内部错误,检查工具实现有没有边界情况没处理。如果是外部依赖问题,考虑加降级逻辑,比如缓存上次成功的结果,或者返回一个默认值。关键是不要让工具失败直接导致整个任务失败,要给模型一个“失败但可继续”的信号。
5.5 常见问题速查表
| 问题现象 | 可能原因 | 排查方法 | 解决措施 |
|---|---|---|---|
| 模型不调工具 | 描述不清/提示词缺失 | 打印工具列表和模型响应 | 优化描述,加引导语 |
| 上下文超限 | 裁剪失效/结果太长 | 统计 token 数 | 精简结果,检查裁剪逻辑 |
| 死循环 | 重复调用/无进展 | 分析轨迹找重复模式 | 加重复检测和进展检测 |
| 工具超时 | 工具慢/超时太短 | 看轨迹耗时 | 优化工具或调整超时 |
| 参数错误 | 模型生成质量差 | 看错误信息 | 优化参数 schema 和描述 |
| 任务提前结束 | 终止条件太松 | 检查完成校验 | 加完成校验函数 |
| 结果不完整 | 轮次不够 | 看轮次统计 | 增加最大轮数 |
实操心得:我习惯在 Harness 里加一个“调试模式”,开启后每一步都打印详细日志,包括完整的消息列表、模型原始响应、工具执行结果。平时关闭,出问题时打开。这个模式帮我省了大量排查时间,强烈建议你也加上。
6. 性能优化与扩展方向
6.1 减少模型调用次数的实用技巧
模型调用是智能体里最慢也最贵的环节,减少调用次数能直接提升性能和降低成本。第一个技巧是批量工具调用。如果模型一轮里需要调用多个互不依赖的工具,让它一次性返回所有工具调用,Harness 并行执行,而不是一轮调一个。这需要工具描述里明确告诉模型“可以同时调用多个工具”。第二个技巧是结果预取。对于某些可预测的工具调用,比如用户问天气,Harness 可以提前调用天气工具把结果放进上下文,模型直接就能用,省一轮调用。第三个技巧是缓存。相同输入的工具调用直接返回缓存结果,这在多轮对话里很常见。
并行工具执行的实现要注意线程安全。工具之间不能有共享状态,否则并行执行会出问题。我用ThreadPoolExecutor并行执行工具,每个工具在独立线程里跑,结果按顺序收集。如果某个工具失败了,不影响其他工具的执行,失败信息单独返回给模型。
6.2 多智能体协作的 Harness 扩展
单智能体搞不定的任务,可以考虑多智能体协作。Harness 的扩展方式是:把每个智能体封装成一个“工具”,让主智能体通过调用工具的方式调度子智能体。这样主智能体的 Harness 完全不用改,子智能体有自己的 Harness 实例。主智能体看到的是一个“调用子智能体”的工具,子智能体内部的多轮循环对主智能体透明。
这种设计的优势是隔离性好,子智能体的上下文不会污染主智能体。缺点是通信开销大,每次调用子智能体都要走一遍完整的 Harness 流程。我的经验是,子智能体适合处理边界清晰、相对独立的子任务,比如“写代码”和“审查代码”可以分成两个子智能体。如果任务之间耦合很紧,还是用单智能体加更多工具更合适。
6.3 从 Demo 到生产的检查清单
把 Harness 从 Demo 推到生产,有几个必须检查的点。第一,错误处理是否完备。所有可能抛异常的地方都要有处理,不能有未捕获的异常导致整个服务崩溃。第二,资源限制是否到位。最大轮数、最大 token 数、最大重试次数、工具超时时间,这些都要有明确的上限。第三,可观测性是否足够。轨迹记录、指标统计、告警配置,缺一不可。第四,降级方案是否可用。模型服务挂了怎么办,工具服务挂了怎么办,要有兜底方案。
我个人的检查清单里还有一条:压测。用并发请求压一下 Harness,看有没有资源泄漏、线程安全问题、性能瓶颈。我遇到过一次,Harness 在高并发下轨迹记录的文件写入成了瓶颈,后来改成异步写入才解决。这种问题不压测根本发现不了。
6.4 我踩过的三个印象最深的坑
第一个坑是上下文裁剪把系统提示词裁掉了。早期实现里裁剪逻辑从消息列表头部开始删,结果把系统提示词删了,模型完全失去了角色设定,行为变得很奇怪。后来改成系统提示词永远保留,只裁剪历史消息。这个坑让我明白,裁剪逻辑一定要有明确的保护规则。
第二个坑是工具结果没有做长度限制。一个工具返回了 50KB 的 JSON,直接塞进上下文,下一轮模型调用直接超限报错。后来加了结果精简器,超过 2000 字符的结果自动提取关键字段。这个坑让我明白,工具结果的长度必须可控。
第三个坑是重试逻辑导致重复执行副作用工具。一个工具是“发送邮件”,模型调用后超时了,错误处理器重试,结果邮件发了两封。后来我给工具加了“幂等”标记,非幂等工具不自动重试,而是把错误返回给模型让它决定。这个坑让我明白,不是所有工具都适合自动重试。
这三个坑的共同点是:它们都不会在 Demo 阶段暴露,只有到了真实使用场景才会出现。所以我的建议是,Harness 的设计一定要考虑边界情况,不要假设一切顺利。把异常路径当成一等公民来设计,你的 Harness 才能真正扛住生产环境的考验。