这次我们来看一个很多人听过但还没真正搞懂的概念Harness Agent。它不是某个新的对话模型也不是提示词技巧而是让 Agent 真正稳定跑起来的那层“运行框架”。如果你写过 Agent 应用发现它总是跑着跑着就断了、工具调用乱了、任务稍长一点就失控那问题大概率不在模型而在你的代码里缺了一个 Harness。这篇文章不绕弯子直接讲三件事Harness 和 Agent 到底什么区别、底层是怎么完成“循环推理—调用工具—观察结果—再决策”的、以及怎么用代码在本地搭一个最小可用的 Harness。内容会覆盖核心能力拆解、环境准备、代码实战、接口封装、批量任务、资源占用观察和常见问题排查。适合刚开始接触 Agent 开发、想理解 Agent 底层原理或者手头有 Agent 任务需要工程化落地的读者。1. 核心能力速览先把 Harness 的能力边界和定位说清楚。下面这张表是 Harness Agent 的整体能力画像具体实现会随语言和框架变化但核心职责是通用的。能力项说明项目类型Agent 运行框架 / 轻量级 Agent 运行时生态来源与 OpenAI Codex 的 Open Agent Harness 方向类似目标是让 Agent 逻辑可复用、可嵌入核心职责管理 Agent 的完整运行周期任务接收、工具注册、循环决策、工具执行、结果反馈、终止判断与 Agent 的关系Agent 是决策模块Harness 是承载 Agent 的运行环境与执行引擎主要功能工具注册与调度、上下文管理、多轮调用、错误恢复、审计日志、批处理入口推荐运行环境本地 Python 3.9 macOS / Linux / Windows需要联网调用大模型 API显存占用本地不跑推理时几乎可以忽略如果本地跑小型开源模型按模型大小占用显存需实际测试支持平台跨平台纯 Python 实现依赖具体模型 API 时以模型服务可用性为准启动方式命令行脚本、HTTP 服务、任务队列消费进程是否支持 API可以封装为 HTTP API支持同步和异步任务提交是否支持批量任务支持通过任务队列或循环调用实现适合场景个人 Agent 项目、企业内部自动化任务、需要审计和可控性的生产环境需要说明一点Harness 本身不负责模型推理它的价值在于把“模型怎么想”和“工具怎么做”连接起来。模型输出一个“调用某工具”的意图Harness 负责找到工具、执行它、把结果带回给模型再让模型决定下一步。这个循环是整个 Agent 系统能够完成复杂任务的基础。2. Harness 与 Agent 的区别很多人会把“Agent”和“Harness”混为一谈或者认为 Harness 是 Agent 的某个高级版本。实际上两者讨论的是不同层面的东西。Agent 通常指那个能做决策的模块它接收用户任务通过大模型推理决定是给出最终答案还是需要调用某个工具。它像一个“大脑”负责思考。但思考本身不能操作外部系统它需要有人帮它去执行工具、读取返回结果、把结果塞回上下文、判断下一步该干嘛甚至处理工具抛出的异常。这套“帮大脑干活”的外围系统就是 Harness。更直接地说Agent 描述的是能力Harness 描述的是运行机制。同一个 Agent 模型放在不同的 Harness 里最终表现可能差别很大。一个好的 Harness 能让模型稳定地完成工具调用不会因为一次返回格式错误就中断也不会因为上下文过长就丢失关键信息。对比项AgentHarness本质决策模块运行框架核心任务推理、规划、决定下一步动作执行循环、工具调度、状态管理是否调模型是通常负责调用模型但不做业务决策是否管工具只声明需要工具负责找到、调用并处理工具结果是否管上下文只消费上下文负责上下文组装、截断和持久化是否处理异常不感知负责捕获、记录、恢复典型类比大脑身体和神经系统在真实工程中你会把 Agent 的理解放在模型提示词和工具描述里而把稳定性、可观测性、批量执行这些工程问题交给 Harness 解决。掌握 Harness等于掌握 Agent 项目从“能跑”到“能稳定跑”的关键。3. Harness 的底层运行原理Harness 的工作方式可以概括成一个循环循环推理、工具调用、结果反馈、继续推理直到完成任务或达到终止条件。3.1 主执行循环绝大多数 Agent Harness 的核心是下面这个循环接收用户任务。将系统提示词、历史记录、可调用工具定义、当前任务组合为模型输入。调用大模型得到结构化输出。判断输出类型如果是最终回答则结束并返回结果如果是工具调用请求则执行对应工具如果是错误或拒绝运行则记录状态。将工具调用结果写回上下文。回到第 2 步直到达到最大轮数或任务完成。这里的“结构化输出”是工程难点。模型输出必须能被程序可靠解析所以大多数 Harness 会要求模型输出 JSON 或者使用特殊分隔符。解析失败时 Harness 不能直接崩溃而是要把错误信息重新喂给模型让它修正输出。3.2 工具注册与参数约束工具注册的本质是把一段可执行代码的描述暴露给模型。Harness 维护一张工具表每个工具包含名称、描述、参数 JSON Schema、执行函数。模型看到的不是 Python 源码而是被序列化成 JSON 的描述信息。这种设计的核心价值在于模型只需要理解“有什么工具、接受什么参数”不需要理解代码实现。参数约束是最容易忽略的部分。一个工具的参数如果定义得太随意模型就可能传错类型或缺字段。因此Harness 通常会在执行前先做参数校验。校验失败时Harness 返回一个错误信息给模型让它重新生成参数而不是直接抛异常终止。3.3 上下文管理与截断策略Agent 每执行一轮对话历史就会变长。模型上下文窗口有限给工具描述、系统提示词、用户任务预留空间之后能放历史记录的余量更加有限。Harness 必须有一套上下文截断策略。常见策略有三种只保留最近的 N 轮对话将早期对话压缩成摘要把大段工具返回结果截断到几百字符并提示模型“内容已截断如需完整数据请使用指定参数获取”。生产级别的 Harness 通常组合使用这三种策略。3.4 错误恢复与重试机制模型调用可能因为网络超时、限流、返回格式异常等原因失败。工具执行也可能抛异常。Harness 的设计目标是“尽量不中断”。遇到模型调用失败可以指数退避重试遇到工具报错把异常信息作为工具结果返回给模型让它自己调整策略。这一层像保险丝能明显提升长时间任务的完成率。4. 核心能力拆解Harness 不是一个大而全的平台它是由一系列可插拔能力组成的。理解这些能力你才能按需选择要自己实现哪些部分。4.1 工具扩展与热插拔好的 Harness 允许你像注册插件一样添加工具。新增一个工具只需要写一个普通函数加一段描述和参数 Schema然后注册进 Harness。业务代码不需要和模型逻辑耦合。这种设计让团队可以并行开发一个人管模型行为另一个人只管实现工具函数。4.2 日志与可观测性Agent 应用出问题最大的难点在于难以复现。模型输出是随机的同样任务每次执行路径可能不一样。Harness 必须把每次模型响应、每次工具调用、参数、结果、耗时、token 消耗完整记录下来。日志不只是为了排错也是为了审计特别是在工具能操作数据库或发送消息的场景下完整的审计日志是安全基线。4.3 并发与批量任务当你有一个任务列表需要处理时逐个循环执行效率太低。Harness 可以支持批量任务将每个任务作为独立“运行”提交。可以在进程内做并发也可以把任务丢进消息队列由多个工作进程消费。并发控制的关键是限流既要防止模型 API 被限流也要防止工具操作外部系统时造成压力。4.4 可配置的安全边界Agent 能调用的工具越多风险边界就越大。Harness 需要提供安全控制点哪些工具允许调用哪些工具需要人工确认哪些操作在特定条件下禁止。生产环境中比如删除数据、发送付款、修改线上配置这类高危操作都应设置为“需要人工审批”。很多团队把审批做成 Hook工具执行前检查审批状态未获批则挂起等待。5. 环境准备与最小运行骨架在写完整代码之前先把环境准备好。Harness 本身是纯 Python 的逻辑不依赖特定的深度学习框架核心依赖只有模型 API 客户端和数据校验库。5.1 环境与依赖建议使用 Python 3.9 以上的版本创建独立的虚拟环境。如果你只是跑最小示例可以先用标准库实现不装任何额外依赖。正式项目中建议安装Python 3.9OpenAI SDK 或其他模型服务 SDKpydantic 用于参数校验FastAPI 用于将 Harness 封装成 HTTP 服务Redis 或消息队列用于批量任务分发5.2 项目目录建议按下面的目录组织项目harness-demo/ ├── main.py # 命令行入口 ├── harness/ │ ├── __init__.py │ ├── core.py # Harness 主循环 │ ├── tools.py # 工具定义与注册 │ └── llm.py # 模型调用封装 ├── tests/ │ └── test_harness.py └── examples/ └── simple_task.py目录结构不复杂但把“框架代码”和“业务工具”分开后续加工具时不需要改动主循环。6. 代码实战从命令行 Agent 到完整 Harness下面直接从零搭一个最小可用的 Harness。这里不绑定具体模型 SDK用抽象接口演示核心逻辑方便你理解原理。6.1 定义工具基类和注册表# harness/core.py from __future__ import annotations import json import time from dataclasses import dataclass, field from typing import Any, Callable, Optional dataclass class Tool: name: str description: str func: Callable[..., Any] parameters: dict field(default_factorydict) def to_schema(self) - dict: 把工具描述序列化为模型可见的 JSON Schema return { type: function, function: { name: self.name, description: self.description, parameters: self.parameters, }, } class ToolRegistry: 工具注册表负责维护名称到工具的映射 def __init__(self): self._tools: dict[str, Tool] {} def register(self, tool: Tool) - None: if tool.name in self._tools: raise ValueError(ftool {tool.name} already registered) self._tools[tool.name] tool def get(self, name: str) - Optional[Tool]: return self._tools.get(name) def list_schemas(self) - list[dict]: return [t.to_schema() for t in self._tools.values()] def call(self, name: str, arguments: dict) - str: tool self.get(name) if tool is None: return f错误工具 {name} 不存在 try: result tool.func(**arguments) return json.dumps(result, ensure_asciiFalse, defaultstr) except Exception as exc: return f工具执行失败: {exc}这里的关键是call方法。它接收工具名和参数字典执行后把结果序列化成字符串返回。字符串形式的结果可以直接写回上下文交给模型继续推理。无论工具成功还是失败返回的都是可读文本不会让异常直接打断主循环。6.2 实现一个工具# harness/tools.py import json from harness.core import Tool # 示例工具模拟查询本地文件信息业务中可替换为真实工具 def get_file_size(file_path: str) - dict: import os if not os.path.exists(file_path): return {error: f文件不存在: {file_path}} size os.path.getsize(file_path) return {file: file_path, size_bytes: size} def add_two_numbers(a: int, b: int) - dict: return {result: a b} def register_sample_tools(registry) - None: registry.register( Tool( nameget_file_size, description获取本地文件的大小参数为文件路径, funcget_file_size, parameters{ type: object, properties: { file_path: {type: string, description: 文件绝对路径} }, required: [file_path], }, ) ) registry.register( Tool( nameadd_two_numbers, description计算两个数字之和, funcadd_two_numbers, parameters{ type: object, properties: { a: {type: integer}, b: {type: integer}, }, required: [a, b], }, ) )工具函数返回 dict外层call会统一序列化。这样做的好处是对齐 JSON 往返格式避免出现不可序列化的对象。6.3 实现 Harness 主循环主循环是 Harness 的心脏。我们用一个抽象方法_think代表模型调用实际项目中会替换为真实 SDK 请求。# harness/core.py class Harness: def __init__( self, registry: ToolRegistry, system_prompt: str , max_turns: int 8, verbose: bool True, ): self.registry registry self.system_prompt system_prompt self.max_turns max_turns self.verbose verbose self.history: list[dict[str, Any]] [] self.turn_count 0 def _think(self, task: str) - dict[str, Any]: 真正的模型推理逻辑需要接入大模型 API。 返回格式约定 - {type: final, content: 最终回答} - {type: tool_call, name: 工具名, arguments: {...}} raise NotImplementedError(请接入真实模型调用) def _build_messages(self, task: str) - list[dict[str, Any]]: tool_schemas self.registry.list_schemas() system self.system_prompt if tool_schemas: system \n\n### 可用工具\n json.dumps(tool_schemas, ensure_asciiFalse) messages [{role: system, content: system}] messages.extend(self.history) messages.append({role: user, content: task}) return messages def run(self, task: str) - str: self.history [] self.turn_count 0 for _ in range(self.max_turns): self.turn_count 1 if self.verbose: print(f\n--- Turn {self.turn_count} ---) decision self._think(task) if decision[type] final: content decision.get(content, ) self.history.append({role: assistant, content: content}) return content if decision[type] tool_call: tool_name decision[name] arguments decision.get(arguments, {}) if self.verbose: print(f调用工具: {tool_name} 参数: {arguments}) result self.registry.call(tool_name, arguments) if self.verbose: print(f工具结果: {result[:200]}) self.history.append({ role: assistant, content: f调用工具 {tool_name} 参数 {json.dumps(arguments, ensure_asciiFalse)}, }) self.history.append({ role: tool, content: result, }) continue if decision[type] error: error_msg decision.get(message, 未知错误) if self.verbose: print(f模型返回错误: {error_msg}) self.history.append({role: assistant, content: f错误: {error_msg}}) continue return 达到最大轮数任务未完成上面把主循环的逻辑写得很明确。每一步模型输出都不是“黑盒”而是可以被程序解析的结构。模型没有“自己执行工具”的能力它只会输出调用意图真正的执行权在 Harness 手里的registry。6.4 接入真实模型调用把_think换成真实模型调用是实际项目中最核心的改写。这里给一个使用通用 OpenAI 兼容接口的示例方便你替换为自己的模型服务地址和密钥# llm.py import json import os import openai client openai.OpenAI( api_keyos.getenv(LLM_API_KEY), base_urlos.getenv(LLM_BASE_URL, https://api.openai.com/v1), ) def create_harness_with_llm(harness: Harness, model: str): original_think harness._think def think(task: str) - dict: messages harness._build_messages(task) response client.chat.completions.create( modelmodel, messagesmessages, toolsharness.registry.list_schemas(), tool_choiceauto, temperature0.2, ) choice response.choices[0] msg choice.message if msg.tool_calls: tool_call msg.tool_calls[0] func tool_call.function return { type: tool_call, name: func.name, arguments: json.loads(func.arguments or {}), } return {type: final, content: msg.content or } harness._think think return harness这样 Harness 的框架逻辑完全不变只是把策略部分替换成真实模型。你还可以在这里做更多事比如记录 token 消耗、加入重试机制、对工具调用参数做二次校验这些都属于 Harness 的能力扩展。6.5 运行测试写一个简单的入口脚本跑通整个流程# main.py from harness.core import Harness, ToolRegistry from harness.tools import register_sample_tools from llm import create_harness_with_llm SYSTEM_PROMPT 你是一个执行助手。需要调用工具时使用可用工具完成如果任务信息不足明确说明缺少什么。 def main(): registry ToolRegistry() register_sample_tools(registry) harness Harness( registryregistry, system_promptSYSTEM_PROMPT, max_turns5, verboseTrue, ) harness create_harness_with_llm(harness, modelos.getenv(LLM_MODEL, gpt-4o-mini)) result harness.run(请计算 12345 和 6789 的和) print(\n最终结果:, result) if __name__ __main__: main()运行命令export LLM_API_KEY你的密钥 export LLM_BASE_URL你的模型服务地址 python main.py预期输出类似--- Turn 1 --- 调用工具: add_two_numbers 参数: {a: 12345, b: 6789} 工具结果: {result: 19134} --- Turn 2 --- 最终结果: 12345 和 6789 的和是 19134这个流程虽然简单但已经具备 Harness 的核心要素工具注册、模型决策、工具调用、结果回填、最终回答。后续所有复杂能力都是在这个循环上迭代出来的。7. 接口 API 与批量任务本地跑通命令行版之后下一步是把 Harness 封装成服务让它可以被外部系统调用。7.1 HTTP 服务封装使用 FastAPI 可以快速把 Harness 包成一个 HTTP API。建议区分同步接口和异步接口。同步接口适合短任务异步接口适合长任务。# api.py import asyncio import uuid from fastapi import FastAPI from pydantic import BaseModel, Field from harness.core import Harness, ToolRegistry from llm import create_harness_with_llm app FastAPI() # 全局 Harness 实例生产环境可按需创建多个实例 registry ToolRegistry() register_sample_tools(registry) harness create_harness_with_llm( Harness(registryregistry, system_promptSYSTEM_PROMPT), modelos.getenv(LLM_MODEL, gpt-4o-mini), ) TASKS: dict[str, dict] {} class TaskRequest(BaseModel): task: str Field(..., description用户任务描述) class TaskResponse(BaseModel): task_id: str status: str app.post(/api/task, response_modelTaskResponse) async def submit_task(req: TaskRequest): task_id uuid.uuid4().hex TASKS[task_id] {status: pending, result: None} loop asyncio.get_event_loop() def run_in_thread(): result harness.run(req.task) TASKS[task_id][status] completed TASKS[task_id][result] result asyncio.create_task(loop.run_in_executor(None, run_in_thread)) return TaskResponse(task_idtask_id, statuspending) app.get(/api/task/{task_id}) async def get_task(task_id: str): task TASKS.get(task_id) if not task: return {error: task not found} return task这里先提交任务再轮询结果。优点是不会因为长任务阻塞 HTTP 连接前端或调用方可以展示“排队中”状态。启动服务uvicorn api:app --host 127.0.0.1 --port 8000调用示例curl -X POST http://127.0.0.1:8000/api/task \ -H Content-Type: application/json \ -d {task: 计算 100 和 200 的和}返回{task_id: a1b2c3d4..., status: pending}轮询结果curl http://127.0.0.1:8000/api/task/a1b2c3d4...这种方式非常适合接入现有后端系统例如工单处理、自动运维、内容生成管道。7.2 批量任务队列当你有大量任务要执行时HTTP 接口一次一个不够高效。可以用一个简单的队列解决# batch.py import json import time from concurrent.futures import ThreadPoolExecutor from harness.core import Harness, ToolRegistry from harness.tools import register_sample_tools from llm import create_harness_with_llm def build_harness(): registry ToolRegistry() register_sample_tools(registry) return create_harness_with_llm( Harness(registryregistry, system_promptSYSTEM_PROMPT), modelos.getenv(LLM_MODEL, gpt-4o-mini), ) def process_one(task: str): harness build_harness() return harness.run(task) def run_batch(tasks: list[str], max_workers: int 3): results [] with ThreadPoolExecutor(max_workersmax_workers) as pool: futures [pool.submit(process_one, t) for t in tasks] for fut in futures: results.append(fut.result()) return results if __name__ __main__: tasks [ 计算 1 和 2 的和, 计算 100 和 200 的和, 计算 -5 和 8 的和, ] results run_batch(tasks, max_workers2) for task, result in zip(tasks, results): print(f任务: {task}\n结果: {result}\n)并发数需要控制。max_workers2是最保守的设置。并发太高模型 API 容易限流外部工具也容易撑不住。更稳妥的做法是把任务投递到 Redis 队列由单独的工作进程消费这样即使某个进程崩溃任务也不会丢失。8. 性能观察与资源占用Harness 本身几乎不消耗显存因为推理发生在模型服务端。但如果你在本地跑开源模型就需要关注资源占用。8.1 什么在消耗资源Harness 的资源消耗主要在三个方面token 消耗每轮模型调用都会消耗 token工具返回结果越长消耗越多。上下文长度任务轮数越多历史记录越长模型输入变长响应时间和费用都会上升。并发执行同时运行多个 Harness 实例时模型服务的吞吐能力会成为瓶颈。8.2 如何观察在 Harness 代码里加入 token 计数和耗时统计是最直接的做法# 在 _think 方法中记录模型返回的 usage 字段 usage response.usage print(f本轮 tokens: prompt{usage.prompt_tokens}, completion{usage.completion_tokens}, total{usage.total_tokens}) print(f本轮耗时: {elapsed:.2f}s)执行完成后统计总轮数、总 token 数、总耗时就能判断当前任务是否“过重”。如果任务很简单却消耗了几万 token多半是上下文管理没做好。8.3 降低资源消耗的策略控制 max_turns避免模型陷入死循环。截断过长的工具返回结果只保留关键信息。对历史对话做摘要而不是无限保留原始记录。批量任务限流避免拥挤时段产生超时重试。能一次调用工具拿到的数据就不要让模型拆成多次调用来拿。9. 常见问题与排查方法Harness 跑起来之后你大概率会遇到下面几类问题。这里整理成排查表。问题现象可能原因排查方式解决方案模型不调用工具直接给最终回答工具描述不清或模型不支持工具调用检查工具 schema 和模型版本优化工具描述在 system prompt 中明确要求必须调用工具模型输出非法 JSON模型返回格式不稳定查看原始响应日志增加重试逻辑将解析错误信息反馈给模型重生成工具参数频繁缺失参数 Schema 不够严格检查模型调用参数日志使用 pydantic 做参数校验缺失时自动填充默认值或返回错误任务跑几轮后上下文溢出历史记录无限累积打印每轮 token 用量实现摘要压缩或早期截断策略批量任务大量超时并发过高触发 API 限流查看模型 API 错误码降低并发数增加指数退避重试工具执行成功但模型不认结果返回格式嵌套过深查看传给模型的工具返回内容将工具返回统一转为简洁 JSON并给出必要的自然语言摘要API 服务重启后任务丢失任务状态只存内存查看服务日志接入 Redis 或数据库持久化任务状态同一个任务每次结果不一致模型温度设置过高对比多次运行结果降低 temperature固定 seed日志杂乱无法定位问题缺少请求 ID 关联检查日志结构为每个任务生成 trace_id贯穿所有日志10. 最佳实践与使用建议Harness 解决的是稳定性和工程化问题。要让它在生产环境可靠运行下面这些建议值得落实。10.1 从最小任务开始验证第一次搭 Harness 时不要直接上复杂工作流。先用一个“单工具、单轮调用、立刻返回”的任务验证链路是否通。链路通了再逐步增加工具数量和任务复杂度。这样一旦出问题定位范围很小。10.2 日志必须有 trace_id给每个任务生成一个唯一 ID贯穿模型调用、工具执行、最终结果所有日志。这样排查问题时可以把一次任务的所有操作串起来。没有 trace_id多任务并发时日志会互相穿插很难排查。10.3 工具权限最小化Harness 能调用的工具必须按最小权限原则配置。一个只读工具就不要给它写权限一个只查单条数据的工具就不要让它接受“删除全部”这类参数。对于高危操作比如删除、写入、发送消息建议加入人工审批钩子默认拒绝。10.4 注意数据隐私与授权Harness 在运行过程中可能把用户输入、文档内容、数据库查询结果发送到模型服务。在接入真实业务数据前必须确认模型服务的数据合规边界。涉及他人声音、人脸、版权素材的生成或处理任务必须确保已经获得明确授权。生产系统中敏感字段在进入模型上下文之前应先做脱敏处理。10.5 保留最小可运行配置把一份最小可运行的 Harness 代码、模型配置、依赖清单单独保存作为以后新项目的基础模板。模型服务切换、参数调整都在模板上迭代不要每次从空文件开始写。11. 总结与下一步Harness 真正的价值是把 Agent 从一个“能回答问题的模型”变成一个“可执行的系统”。它接手了工具调度、状态管理、异常恢复、审计这些工程问题让上层业务只需要关注“模型该做什么决策”和“工具能提供什么能力”。如果你是第一次接触这个概念建议最先验证三件事第一Harness 主循环能不能把一次工具调用完整跑通第二模拟一次工具返回异常观察 Harness 是否能把错误信息回传给模型并继续执行第三用一个多工具任务测试上下文轮转是否符合预期。这三件事跑通你就已经理解 Harness 的核心机制了。最常见的坑其实不是代码写错而是没有给工具定义清晰的参数约束以及没有对模型返回做容错。这两个问题会在任务复杂度上升后集中爆发建议在项目一开始就处理。下一步可以继续扩展的方向很多接入更多真实业务工具把 HTTP 服务换成更稳定的异步任务队列加入 pydantic 严格参数校验实现上下文自动摘要或者把 Harness 嵌入现有的自动化平台。底层原理不变变的只是接入方式。把这一层理解透后面所有 Agent 项目都会顺畅很多。建议收藏这篇文章动手搭的时候按步骤来能少走很多弯路。