1. 先搞清楚 Harness 到底在解决什么问题很多人第一次听到 AI Agent 里的 Harness 这个词第一反应是“这不就是个壳吗”。我一开始也这么想直到自己从零搭了两三个 Agent 项目踩了一圈坑之后才明白Harness 不是壳它是让模型真正“下地干活”的那套骨架。你可以把大模型想象成一个极其聪明但完全没有手脚的大脑它能思考、能推理、能生成文本但它不知道怎么调用工具、不知道什么时候该停下来、不知道失败了要不要重试、更不知道多个任务之间怎么排队。Harness 就是给这个大脑接上手脚、装上神经、配上调度中心的那一整套工程结构。我见过太多人做 Agent 的做法是写一个 while 循环把用户输入丢给模型模型返回什么就执行什么执行完再丢回去直到模型说“我完成了”。这个做法在 demo 阶段能跑通一旦放到真实场景里问题就全冒出来了——模型陷入死循环怎么办工具调用失败了要不要重试上下文超了怎么截断多个用户同时来请求怎么隔离这些问题恰恰就是 Harness 要解决的核心命题。所以这篇文章我想干一件事把 Harness 拆开看看它到底由哪些子系统组成每个子系统负责什么为什么缺一不可。我会按照我自己搭建 Agent 时的真实思路来写不讲空话直接说每个模块该怎么设计、参数怎么定、坑在哪里。如果你正在做 AI Agent 开发或者准备从零搭建一个能真正跑起来的 Agent 系统这篇内容应该能帮你少走不少弯路。2. Harness 的整体架构与设计思路2.1 为什么需要一个独立的 Harness 层先说一个我踩过的坑。早期我做 Agent 的时候把工具调用逻辑、上下文管理、错误处理全部塞在一个大函数里代码大概长这样接收用户输入拼 prompt调模型解析返回执行工具把结果拼回去再调模型。这个函数写了三百多行跑起来倒也能用但一旦要加一个新工具、换一个模型、或者调整重试策略整个函数就得大改。更麻烦的是我根本没办法单独测试其中任何一个环节。后来我才意识到Agent 的工程复杂度不在于“调模型”这个动作本身而在于模型调用前后那一整套编排逻辑。这套编排逻辑就是 Harness。把它独立出来之后好处非常明显模型可以换、工具可以插拔、上下文策略可以调整、错误处理可以单独配置每个部分互不干扰。这就是为什么现在主流的 Agent 框架不管是 LangGraph 还是其他方案都在强调“编排层”和“执行层”的分离。从设计思路上讲Harness 的核心职责可以归纳成一句话管理 Agent 从接收到任务到完成任务之间的全部状态流转。注意“状态流转”这四个字它意味着 Harness 不只是一个请求转发器而是一个有状态的、能感知上下文的、能做出决策的调度系统。2.2 七个子系统的划分逻辑我把 Harness 拆成七个子系统这个划分不是拍脑袋来的而是按照“一个任务从进入到完成”的生命周期来切的。你可以这样理解用户发来一个请求这个请求要经过接收、理解、规划、执行、观察、修正、输出这几个阶段每个阶段都需要专门的模块来支撑。这七个子系统分别是Agent Loop主循环、LLM Integration模型接入层、Tool Registry工具注册中心、Context Manager上下文管理器、Memory System记忆系统、Error Handler错误处理与重试、Orchestration Scheduler编排调度器。它们之间的关系不是简单的线性串联而是有交叉、有回环的。比如 Agent Loop 在每一轮迭代中都会调用 LLM Integration同时通过 Context Manager 组装上下文通过 Tool Registry 执行工具遇到错误则交给 Error Handler 处理。我之所以强调这个划分是因为很多人在搭建 Agent 时容易把其中几个混在一起。最常见的混淆是把 Context Manager 和 Memory System 当成一回事实际上前者管的是“当前这一轮对话的上下文窗口怎么用”后者管的是“跨会话的信息怎么持久化和检索”。这两个问题的解法完全不同混在一起做最后一定是两头都做不好。3. 七个子系统逐个拆解3.1 Agent Loop整个系统的心脏Agent Loop 是 Harness 最核心的部分它决定了 Agent 的“思考-行动”节奏。最简单的 Agent Loop 就是一个 while 循环调模型看模型要不要调工具要调就调把结果塞回去再调模型直到模型输出最终答案或者达到最大轮次。但真正在生产环境里跑的 Agent Loop需要考虑的东西远不止这些。首先是终止条件的设计。你不能只依赖模型自己说“我完成了”因为模型有时候会一直觉得自己没完成陷入无限循环。我的做法是设置三重终止条件模型明确输出终止信号、达到最大迭代轮次、连续 N 轮没有产生有效工具调用。这三个条件满足任何一个就退出。其次是每轮迭代的状态快照。这个很重要因为 Agent 执行到一半失败了你需要知道它之前做了什么、走到哪一步了。我通常会在每轮迭代结束时把当前的消息列表、已执行的工具调用记录、当前的任务状态序列化存下来。这样即使进程崩溃也能从最近的快照恢复。再就是循环的节奏控制。有些工具调用很快比如查个天气几十毫秒就回来了有些工具调用很慢比如跑一个数据分析脚本可能要几分钟。如果 Agent Loop 不区分这些情况统一用同一个超时策略要么快工具被拖慢要么慢工具被误杀。我的经验是给每个工具单独配置超时时间Agent Loop 根据工具的类型来决定等待策略。# Agent Loop 的核心结构示意 MAX_ITERATIONS 15 NO_PROGRESS_THRESHOLD 3 def agent_loop(task, tools, llm, context_manager): messages context_manager.build_initial_context(task) no_progress_count 0 for i in range(MAX_ITERATIONS): response llm.invoke(messages) if response.is_final_answer: return response.content if response.has_tool_calls: results execute_tools(response.tool_calls, tools) messages context_manager.append_tool_results(messages, results) no_progress_count 0 else: no_progress_count 1 if no_progress_count NO_PROGRESS_THRESHOLD: return Agent 未能取得进展已终止 return 达到最大迭代次数已终止注意最大迭代轮次不要设太大。我见过有人设成 100结果模型在里面绕了 80 多轮才出来token 消耗直接爆炸。根据任务复杂度10 到 20 轮通常够用了。3.2 LLM Integration模型接入层的关键设计LLM Integration 这一层看起来简单不就是调个 API 吗但实际上这一层要处理的问题非常多多模型切换、流式输出、token 计数、速率限制、降级策略、prompt 模板管理。先说多模型切换。在实际项目中我很少只用一个模型。通常的做法是主推理用一个能力强的模型简单的分类或提取任务用一个小模型这样成本和延迟都能优化。LLM Integration 层需要提供一个统一的接口让上层不需要关心底层用的是哪个模型。流式输出是另一个容易被忽视的点。Agent 在执行过程中如果每一步都要等模型完整输出才展示给用户体验会非常差。流式输出让用户可以实时看到 Agent 的思考过程这在交互式场景里几乎是必须的。但流式输出和工具调用放在一起会变得复杂因为工具调用的参数需要完整解析后才能执行不能流式处理。我的做法是文本内容流式输出工具调用参数缓冲后一次性解析。Token 计数和速率限制也是这一层的核心职责。你需要精确知道每次调用消耗了多少 token一方面是为了成本控制另一方面是为了上下文管理提供依据。速率限制则是防止在高并发场景下被模型服务端限流。我通常会在这一层实现一个令牌桶算法控制请求频率。# LLM Integration 层的统一接口设计 class LLMProvider: def __init__(self, model_name, api_config): self.model_name model_name self.rate_limiter TokenBucket( rateapi_config.requests_per_second, capacityapi_config.burst_capacity ) self.token_counter TokenCounter() def invoke(self, messages, toolsNone, streamFalse): self.rate_limiter.acquire() response self._call_api(messages, tools, stream) self.token_counter.record(response.usage) return response def invoke_with_fallback(self, messages, toolsNone): try: return self.invoke(messages, tools) except RateLimitError: return self.fallback_provider.invoke(messages, tools) except ModelOverloadedError: time.sleep(2) return self.invoke(messages, tools)3.3 Tool Registry工具注册与发现机制Tool Registry 是 Agent 的“工具箱”它负责管理所有可用的工具包括工具的定义、参数 schema、执行逻辑、权限控制。这一层的设计好坏直接决定了 Agent 的能力边界和扩展性。我见过很多项目的工具管理是这样的在代码里硬编码一个工具列表每个工具写成一个函数然后在 prompt 里手动描述每个工具的功能。这种做法在工具有五六个的时候还能维护一旦超过十个prompt 就会变得极其臃肿而且模型很容易搞混工具之间的区别。更好的做法是工具自注册机制。每个工具自己声明它的名称、描述、参数 schema 和执行函数Tool Registry 自动收集这些信息生成模型可理解的工具描述。这样新增工具只需要写一个工具类不需要改动其他地方。# 工具自注册示例 class ToolRegistry: def __init__(self): self._tools {} def register(self, tool): self._tools[tool.name] tool def get_schemas(self): return [tool.to_schema() for tool in self._tools.values()] def execute(self, tool_name, params): tool self._tools.get(tool_name) if not tool: raise ToolNotFoundError(f工具 {tool_name} 未注册) return tool.execute(**params) # 注册一个天气查询工具 registry ToolRegistry() registry.register(WeatherTool()) registry.register(SearchTool()) registry.register(CalculatorTool())工具描述的质量非常关键。模型能不能正确选择工具很大程度上取决于工具描述写得好不好。我的经验是描述要具体不要写“查询信息”这种模糊的表述要写“根据城市名称查询当前天气状况返回温度和天气描述”。参数说明也要写清楚格式和取值范围这样模型生成的参数才不容易出错。提示工具数量超过 15 个时建议做工具分组或动态加载。把所有工具一次性塞给模型不仅浪费 token还会降低模型选择工具的准确率。3.4 Context Manager上下文窗口的精打细算Context Manager 要解决的问题是模型的上下文窗口是有限的但 Agent 执行过程中产生的信息是不断增长的怎么在有限的空间里放最有用的信息这个问题的核心是上下文组装策略。我的做法是把上下文分成几个优先级不同的区域系统提示词最高优先级永远保留、当前任务描述高优先级、最近几轮对话中优先级、历史工具调用结果低优先级可压缩、早期对话最低优先级可丢弃或摘要。当上下文接近窗口上限时从最低优先级的区域开始压缩。压缩的方式有两种一种是直接截断把最早的消息丢掉另一种是摘要用一个小模型把一段对话压缩成几句话。截断简单但会丢失信息摘要保留信息但会增加一次模型调用。我的经验是对于工具调用结果这种结构化数据直接截断就好对于对话历史用摘要更合适。class ContextManager: def __init__(self, max_tokens, reserve_for_output2000): self.max_tokens max_tokens self.reserve reserve_for_output self.budget max_tokens - reserve_for_output def assemble(self, system_prompt, task, history, tool_results): context [] used 0 # 系统提示词永远保留 context.append(system_prompt) used count_tokens(system_prompt) # 任务描述 context.append(task) used count_tokens(task) # 按优先级填充历史 remaining self.budget - used for msg in reversed(history): msg_tokens count_tokens(msg) if msg_tokens remaining: context.insert(2, msg) remaining - msg_tokens else: break return context3.5 Memory System跨会话的记忆持久化Memory System 和 Context Manager 的区别我用一个类比来说明Context Manager 像是你工作时的桌面桌面大小有限你只能放当前要用的东西Memory System 像是你的文件柜东西可以一直存着需要的时候去翻。Memory System 通常分三层短期记忆当前会话内的信息其实就是 Context Manager 管的部分、长期记忆跨会话的持久化信息存在数据库或向量库里、工作记忆当前任务相关的临时信息任务结束就清理。长期记忆的实现方式目前主流的是向量数据库加语义检索。把重要的信息 embedding 之后存进向量库需要的时候用当前上下文去检索最相关的几条。但这里有个坑不是所有信息都值得存。我早期做的时候把每一轮对话都存进去结果检索出来的东西又杂又乱反而干扰了模型判断。后来我改成只存三类信息用户明确表达的偏好、任务执行中的关键结论、以及被验证过的有效方案。class MemorySystem: def __init__(self, vector_store, db): self.vector_store vector_store self.db db def store_long_term(self, content, metadata): # 只存储有价值的信息 if not self._is_worth_storing(content): return embedding embed(content) self.vector_store.upsert(embedding, content, metadata) def retrieve_relevant(self, query, top_k5): embedding embed(query) results self.vector_store.search(embedding, top_k) # 过滤掉相似度太低的 return [r for r in results if r.score 0.75] def _is_worth_storing(self, content): # 过滤规则太短的、纯寒暄的、重复的不存 if len(content) 20: return False if is_greeting(content): return False return True3.6 Error Handler让 Agent 有韧性Error Handler 是我认为最被低估的子系统。很多人做 Agent 的时候错误处理就是 try-except 包一下打印个日志就完了。但在生产环境里错误处理做得好不好直接决定了 Agent 是“能用”还是“好用”。Agent 遇到的错误大致分四类模型调用错误超时、限流、返回格式异常、工具执行错误参数错误、外部服务不可用、权限不足、逻辑错误模型陷入循环、生成了无效的工具调用、系统错误内存溢出、网络中断。不同类型的错误处理策略完全不同。模型超时可以重试但重试次数不能太多一般 2 到 3 次就够了而且要用指数退避。工具参数错误不应该重试因为重试同样的参数还是会失败正确的做法是把错误信息返回给模型让模型修正参数后重新调用。逻辑错误需要检测机制比如连续多轮没有有效工具调用就强制终止。class ErrorHandler: def __init__(self, max_retries3, base_delay1.0): self.max_retries max_retries self.base_delay base_delay def handle_model_error(self, error, attempt): if isinstance(error, RateLimitError): delay self.base_delay * (2 ** attempt) time.sleep(delay) return RetryAction(delaydelay) if isinstance(error, TimeoutError): if attempt self.max_retries: return RetryAction(delayself.base_delay) return FallbackAction(fallback_modelsmaller-model) return AbortAction(reasonstr(error)) def handle_tool_error(self, error, tool_call): # 工具错误不重试返回给模型让它修正 return ReturnToModelAction( messagef工具 {tool_call.name} 执行失败{error}。请检查参数并重试。 )3.7 Orchestration Scheduler多任务与并发管理当你的 Agent 需要同时处理多个任务或者一个任务需要拆分成多个子任务并行执行时Orchestration Scheduler 就派上用场了。这个子系统要解决的核心问题是任务怎么排队、怎么分配资源、怎么处理依赖关系。最简单的场景是多个用户同时发来请求你需要一个队列来管理这些请求按优先级或到达顺序处理。复杂一点的场景是一个任务被拆成多个子任务有些子任务之间有依赖关系需要按拓扑顺序执行。我目前用的方案是用一个任务队列管理所有待执行的任务每个任务有优先级和依赖列表。调度器从队列中取出可执行的任务依赖已满足的分配给执行器。执行器可以是线程池也可以是异步协程。对于 I/O 密集型的任务比如调模型、调外部 API用异步协程效率更高对于 CPU 密集型的任务比如数据处理用线程池或进程池更合适。class OrchestrationScheduler: def __init__(self, max_concurrent10): self.task_queue PriorityQueue() self.running {} self.max_concurrent max_concurrent self.semaphore asyncio.Semaphore(max_concurrent) async def submit(self, task): await self.task_queue.put(task) async def run(self): while True: task await self.task_queue.get() if self._dependencies_met(task): asyncio.create_task(self._execute(task)) else: # 依赖未满足放回队列稍后重试 await self.task_queue.put(task) await asyncio.sleep(0.1) async def _execute(self, task): async with self.semaphore: try: result await task.run() self._notify_dependents(task, result) except Exception as e: await self._handle_failure(task, e)4. 子系统之间的协作与数据流4.1 一次完整任务的生命周期把七个子系统串起来看一次完整的任务执行流程是这样的用户请求到达 Orchestration Scheduler调度器创建一个任务并放入队列。任务开始执行时Agent Loop 启动首先通过 Context Manager 组装初始上下文包括系统提示词、任务描述、以及从 Memory System 检索到的相关历史信息。然后 Agent Loop 调用 LLM Integration把组装好的上下文和 Tool Registry 提供的工具描述一起发给模型。模型返回响应后Agent Loop 解析响应。如果是最终答案任务结束结果通过 Orchestration Scheduler 返回给用户同时把关键信息存入 Memory System。如果是工具调用Agent Loop 通过 Tool Registry 执行工具执行结果经过 Error Handler 检查后由 Context Manager 追加到上下文中进入下一轮循环。这个流程看起来是线性的但实际上有很多分支和回环。比如工具执行失败时Error Handler 可能决定重试也可能决定把错误返回给模型上下文超限时Context Manager 可能触发摘要压缩模型调用超时时LLM Integration 可能切换到备用模型。4.2 数据在各子系统之间的流转理解数据流转对于调试和优化 Agent 非常重要。我画不出图这里也不方便画但可以用文字描述清楚消息列表是贯穿整个流程的核心数据结构。它从 Context Manager 生成传给 LLM Integration模型返回的新消息由 Agent Loop 追加进去工具执行结果也追加进去。这个列表在每一轮迭代中不断增长直到任务结束。工具调用记录是另一个重要数据。每次工具调用都会产生一条记录包含工具名称、参数、执行结果、耗时、是否成功。这些记录一方面用于调试和审计另一方面也作为 Memory System 的输入帮助 Agent 在后续任务中避免重复犯错。Token 消耗统计由 LLM Integration 维护每次模型调用后更新。Context Manager 会根据这个统计来决定是否触发压缩。Orchestration Scheduler 也会用这个数据来做成本控制比如当某个任务的 token 消耗超过预算时强制终止。5. 实操中的常见问题与排查技巧5.1 模型不调用工具或调用错误工具这是最常见的问题。模型明明有工具可用但它就是不用或者用了错误的工具。排查思路如下首先检查工具描述是否清晰。我遇到过工具描述写得太抽象模型理解不了的情况。比如把“搜索”工具描述成“获取信息”模型就不知道什么时候该用它。改成“根据关键词在互联网上搜索最新信息返回相关网页摘要”之后调用率明显提升。其次检查系统提示词是否给了模型足够的引导。有时候需要在系统提示词里明确写“当需要获取实时信息时请使用搜索工具”模型才会主动调用。最后检查工具数量是否过多。当工具超过 20 个时模型的选择准确率会明显下降。这时候需要做工具分组或者根据任务类型动态加载相关工具。5.2 上下文超限导致模型输出质量下降上下文快满的时候模型的输出质量会明显下降表现为答非所问、重复之前的内容、或者直接报错。我的处理策略是设置一个预警阈值比如上下文使用率达到 80% 时就触发压缩。压缩时优先丢弃最早的工具调用结果因为这些信息通常已经不需要了。如果压缩后还是超限就对早期对话做摘要。注意摘要操作本身也会消耗 token所以不要频繁触发。我通常设置一个最小间隔比如两次摘要之间至少间隔 5 轮对话。5.3 工具执行超时或返回异常外部工具不可控超时和异常是常态。我的做法是给每个工具设置独立的超时时间并且在工具执行层做熔断。如果一个工具连续失败 3 次就暂时把它从可用工具列表中移除避免 Agent 反复尝试。另外工具返回的结果格式要统一。我见过有的工具返回 JSON有的返回纯文本有的返回 HTML模型处理起来很吃力。统一成结构化的 JSON 格式并且在工具描述里说明返回格式模型的理解准确率会高很多。5.4 并发场景下的资源竞争当多个 Agent 实例同时运行时可能会竞争同一份资源比如同一个数据库连接、同一个文件、同一个外部 API 的配额。我的解决方案是在 Orchestration Scheduler 层做资源隔离每个 Agent 实例有独立的上下文和工具实例。对于共享资源用信号量或分布式锁来控制并发访问。对于外部 API 配额用令牌桶算法做全局限流。问题类型典型表现排查方向解决手段工具不调用模型直接回答不调工具工具描述、系统提示词优化描述增加引导上下文超限输出质量下降、报错token 统计、压缩策略触发压缩、调整阈值工具超时任务卡住、无响应工具超时配置、外部服务状态独立超时、熔断机制并发冲突数据错乱、资源耗尽资源隔离、限流配置信号量、分布式锁死循环反复调用同一工具终止条件、进展检测最大轮次、无进展检测6. 从零搭建 Harness 的实操建议6.1 技术选型与最小可行架构如果你准备从零搭建一个 Harness我的建议是不要一上来就追求大而全。先用最小的架构跑通核心流程然后再逐步补充子系统。最小可行架构只需要三个部分一个 Agent Loop、一个 LLM 调用封装、一个工具注册表。Context Manager 可以先简单实现比如固定保留最近 10 轮对话。Memory System 可以先不做等有跨会话需求时再加。Error Handler 先做基本的重试Orchestration Scheduler 先做单任务串行。技术栈方面Python 生态目前最成熟。异步用 asyncioHTTP 调用用 httpx数据校验用 pydantic向量检索用 faiss 或 chroma。如果你用 LangChain 或 LangGraph它们已经提供了部分子系统的实现但我的建议是核心的 Agent Loop 和 Context Manager 最好自己写因为这两个部分和业务逻辑耦合最深用框架反而受限。6.2 迭代节奏与优先级搭建 Harness 的优先级排序我的经验是这样的第一优先级是 Agent Loop 和 LLM Integration这两个是基础没有它们什么都跑不起来。第二优先级是 Tool Registry 和 Error Handler它们决定了 Agent 能不能稳定执行任务。第三优先级是 Context Manager当任务变复杂、对话变长时它的重要性才体现出来。第四优先级是 Memory System 和 Orchestration Scheduler这两个是进阶需求等单任务跑稳了再考虑。每个子系统的实现也不要追求一步到位。比如 Error Handler第一版只需要做模型调用的重试工具错误直接返回给模型。等遇到具体问题了再针对性补充。6.3 测试与可观测性建设Agent 系统的测试比传统软件难得多因为模型的输出是不确定的。我的做法是分两层测试单元测试测各个子系统的逻辑比如 Context Manager 的压缩策略、Error Handler 的重试逻辑这些是确定性的可以用传统测试方法。集成测试测端到端流程用一组固定的任务检查 Agent 是否能完成但不检查具体路径只检查最终结果。可观测性方面我强烈建议在 Harness 的每个关键节点打日志包括每次模型调用的输入输出和 token 消耗、每次工具调用的参数和结果、每轮迭代的状态变化、每次错误处理的决策。这些日志在排查问题时非常有用。我通常会把日志结构化存储方便后续做分析和统计。# 结构化日志示例 logger.info(agent_iteration, extra{ task_id: task.id, iteration: i, model: llm.model_name, input_tokens: response.usage.input_tokens, output_tokens: response.usage.output_tokens, tool_calls: [tc.name for tc in response.tool_calls], duration_ms: elapsed })7. 一些踩坑之后的个人体会做 Agent 这几年我最大的体会是Harness 的复杂度不在于单个子系统有多难而在于它们之间的交互。一个看似简单的需求比如“让 Agent 支持多轮对话”背后涉及 Context Manager 的上下文组装、Memory System 的持久化、Agent Loop 的状态管理牵一发动全身。另一个体会是不要过度设计。我早期做 Harness 的时候总想着把所有可能的情况都考虑到结果代码写了一堆抽象层实际用到的不到三分之一。后来我改成“遇到问题再解决”的方式先跑通最简单的流程遇到具体问题了再针对性优化效率反而高很多。还有一点模型的能力在快速变化。今天需要 Harness 做的很多补偿性工作比如复杂的 prompt 工程、严格的输出格式校验可能明天模型自己就能做好了。所以 Harness 的设计要保持灵活把和模型能力相关的部分做成可替换的模块这样模型升级时不需要大改。最后分享一个实用技巧在 Agent Loop 里加一个“思考日志”机制让模型在每次工具调用前先输出一段简短的思考说明它为什么要调这个工具、期望得到什么结果。这个日志不参与后续流程只用于调试和审计。我实测下来这个机制不仅方便排查问题还能让模型的工具调用准确率提升不少因为它被迫在调用前想清楚目的。
