什么是 Agent Harness：从概念到代码实战

1. 引言为什么需要 Agent Harness随着大语言模型LLM能力的增强越来越多的应用开始从「单次问答」走向「多步骤、多工具、多智能体协作」的复杂场景。Agent智能体需要规划任务、调用工具、读取结果、反思纠错甚至与其他 Agent 协作。然而把这样一套流程从零搭建起来往往要处理大量与业务无关的重复工作上下文管理、工具注册、循环控制、错误重试、日志追踪、安全护栏等。Agent Harness智能体框架/运行容器正是为了解决这些问题而出现的。它是一套把 Agent 的「思考—行动—观察」循环ReAct 模式封装起来的运行环境让开发者可以专注于定义 Agent 的行为和工具而不必重复实现底层编排逻辑。简单来说Agent Harness 承担了以下职责循环控制驱动 Agent 在「推理 → 调用工具 → 观察结果 → 再次推理」之间循环直到任务完成或达到上限。上下文管理维护对话历史、工具调用记录和中间结果控制发送给 LLM 的 Token 长度。工具注册与调度统一管理 Agent 可用的工具负责参数校验、调用和结果回传。错误处理与重试捕获工具异常、解析失败等错误决定是重试、跳过还是终止。可观测性记录每一步的输入输出便于调试、审计和评估。安全与护栏限制工具权限、敏感操作确认、输出内容过滤等。2. Agent Harness 的核心概念要深入理解 Agent Harness需要先掌握几个关键概念。2.1 Agent智能体Agent 是具备自主决策能力的执行单元。它接收用户目标通过 LLM 进行推理决定下一步是直接回答还是调用某个工具。Agent 通常包含一个「系统提示词System Prompt」用来定义它的角色、能力和行为边界。2.2 Tool工具工具是 Agent 与外部世界交互的接口可以是函数、API、数据库查询或代码解释器。每个工具都有名称、描述和参数 SchemaLLM 根据这些信息决定何时调用、传什么参数。2.3 Harness运行容器Harness 是承载 Agent 运行的外壳。它不关心 Agent 的具体业务逻辑而是提供通用的执行框架接收输入、调用 LLM、解析输出、执行工具、维护状态、处理终止条件。Harness 与 Agent 的关系类似于「操作系统」与「应用程序」的关系。2.4 ReAct 循环ReActReasoning Acting是 Agent 最经典的运行模式。每一步循环包含Thought思考LLM 分析当前状态决定下一步行动。Action行动选择一个工具并给出参数。Observation观察执行工具把结果返回给 LLM。Harness 负责把这个循环自动化直到 LLM 输出最终答案Final Answer或达到最大迭代次数。3. 一个最小可运行的 Agent Harness 实战下面我们用 Python 从零实现一个极简但完整的 Agent Harness。它不依赖任何第三方 Agent 框架只使用 OpenAI SDK 作为 LLM 后端帮助你理解底层原理。3.1 环境准备首先安装依赖pip install openai3.2 定义工具我们先定义两个简单的工具一个计算器和一个天气查询模拟。import json import datetime def calculator(expression: str) - str: 计算数学表达式例如 1 2 * 3。 # 注意生产环境应使用安全解析库这里仅作演示 try: result eval(expression, {builtins: {}}, {}) return str(result) except Exception as e: return f计算错误: {e} def get_weather(city: str) - str: 查询指定城市的天气模拟数据。 weather_map { 北京: 晴25°C, 上海: 多云28°C, 广州: 阵雨30°C, } return weather_map.get(city, f暂无 {city} 的天气数据) TOOLS { calculator: { description: 计算数学表达式输入如 1 2 * 3, parameters: {expression: string}, function: calculator, }, get_weather: { description: 查询指定城市的天气, parameters: {city: string}, function: get_weather, }, }3.3 构建工具描述供 LLM 识别为了让 LLM 知道有哪些工具可用我们需要把工具信息转换成 JSON Schema 格式并放入系统提示词中。def build_tool_schema(): schema [] for name, meta in TOOLS.items(): props {p: {type: string} for p in meta[parameters]} schema.append({ type: function, function: { name: name, description: meta[description], parameters: { type: object, properties: props, required: list(meta[parameters].keys()), }, }, }) return schema3.4 实现 Harness 核心循环这是最关键的部分。Harness 维护一个消息列表循环调用 LLM解析工具调用并执行直到得到最终答案。from openai import OpenAI class AgentHarness: def init(self, modelgpt-4o, max_iterations10): self.client OpenAI() self.model model self.max_iterations max_iterations self.messages [] self.tool_schema build_tool_schema() def run(self, user_input: str) -gt; str: 启动 Agent返回最终答案。 self.messages [ {role: system, content: 你是一个智能助手可以调用工具来回答问题。 如果需要工具请调用否则直接回答。}, {role: user, content: user_input}, ] for step in range(self.max_iterations): print(f\n 第 {step 1} 轮循环 ) response self.client.chat.completions.create( modelself.model, messagesself.messages, toolsself.tool_schema, tool_choiceauto, ) message response.choices[0].message # 1. 没有工具调用说明 LLM 给出了最终答案 if not message.tool_calls: print(Agent 直接回答无需调用工具。) return message.content 2. 有工具调用执行工具并回传结果 self.messages.append(message) for tool_call in message.tool_calls: tool_name tool_call.function.name args json.loads(tool_call.function.arguments) print(f调用工具: {tool_name}({args})) # 执行工具 tool_meta TOOLS.get(tool_name) if tool_meta is None: result f错误: 未知工具 {tool_name} else: try: result tool_meta[function](**args) except Exception as e: result f工具执行异常: {e} # 把工具结果作为 tool 角色的消息回传 self.messages.append({ role: tool, tool_call_id: tool_call.id, content: result, }) print(f工具返回: {result}) return 已达到最大迭代次数任务未完成。 使用示例 if name main: harness AgentHarness() answer harness.run(北京天气怎么样顺便算一下 25 * 4 10 等于多少。) print(f\n最终答案: {answer})3.5 运行效果运行上述代码你会看到类似下面的输出 第 1 轮循环 调用工具: get_weather({city: 北京}) 工具返回: 晴25°C 调用工具: calculator({expression: 25 * 4 10}) 工具返回: 110 第 2 轮循环 Agent 直接回答无需调用工具。 最终答案: 北京今天天气晴朗气温 25°C。另外25 * 4 10 的计算结果是 110。可以看到Harness 自动完成了「识别工具 → 执行 → 汇总答案」的完整流程。这就是 Agent Harness 最核心的价值把多步工具调用编排成一次自然语言交互。4. 进阶加入记忆与状态管理上面的最小实现已经能跑通基本流程但真实场景往往需要更复杂的状态管理。下面我们扩展 Harness加入「短期记忆窗口」和「最大 Token 控制」。class AdvancedHarness(AgentHarness): def __init__(self, modelgpt-4o, max_iterations10, max_history20): super().__init__(model, max_iterations) self.max_history max_history # 最多保留多少条历史消息 def _trim_history(self): 裁剪历史消息只保留最近的 N 条避免超出上下文窗口。 if len(self.messages) gt; self.max_history: # 保留 system 消息 最近的 max_history - 1 条 system_msg self.messages[0] recent self.messages[-(self.max_history - 1):] self.messages [system_msg] recent print(f[Harness] 已裁剪历史当前消息数: {len(self.messages)}) def run(self, user_input: str) -gt; str: self.messages [ {role: system, content: 你是一个智能助手可以调用工具来回答问题。}, {role: user, content: user_input}, ] for step in range(self.max_iterations): response self.client.chat.completions.create( modelself.model, messagesself.messages, toolsself.tool_schema, tool_choiceauto, ) message response.choices[0].message if not message.tool_calls: return message.content self.messages.append(message) for tool_call in message.tool_calls: tool_name tool_call.function.name args json.loads(tool_call.function.arguments) tool_meta TOOLS.get(tool_name) result tool_meta[function](**args) if tool_meta else f未知工具: {tool_name} self.messages.append({ role: tool, tool_call_id: tool_call.id, content: result, }) # 每轮结束后裁剪历史 self._trim_history() return 已达到最大迭代次数。lt;/codegt;lt;/pregt; 这个进阶版本在每轮循环后检查消息数量超过阈值就裁剪掉最早的历史保留系统提示词从而防止长对话撑爆上下文窗口。 5. 主流 Agent Harness 框架对比 理解了底层原理后再看主流框架就会轻松很多。它们本质上是把上面这些逻辑做成了开箱即用的产品。 框架 语言 核心特点 适用场景 LangChain / LangGraph Python / JS 生态丰富支持图状编排、记忆、多 Agent 快速原型、复杂工作流 AutoGen Python 多 Agent 对话协作支持人机混合 多角色协作、研究实验 CrewAI Python 角色化团队协作声明式配置 任务分工明确的团队场景 OpenAI Assistants API 任意 托管式 Harness内置检索、代码解释器 快速接入 OpenAI 生态 自研 Harness 任意 完全可控无框架依赖 对性能、安全、定制要求高的场景 选择建议如果追求开发速度和生态选 LangChain 或 CrewAI如果需要深度定制和完全掌控自研 Harness 是更好的选择——毕竟核心循环并不复杂正如我们上面实现的。 6. 实战用 LangGraph 实现带记忆的 Agent 最后我们看一个使用 LangGraph 构建 Agent 的实战示例感受主流框架的写法。 pip install langgraph langchain-openai from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.tools import tool from langchain_core.messages import HumanMessage, AIMessage 定义工具 tool def multiply(a: int, b: int) - int: 计算两个整数的乘积。 return a * b tool def add(a: int, b: int) - int: 计算两个整数的和。 return a b tools [multiply, add] llm ChatOpenAI(modelgpt-4o).bind_tools(tools) 定义状态 class AgentState(TypedDict): messages: Annotated[list, lambda x, y: x y] # 消息累积 定义节点 def agent_node(state: AgentState): Agent 节点调用 LLM 决定下一步。 result llm.invoke(state[messages]) return {messages: [result]} def tools_node(state: AgentState): 工具节点执行 LLM 请求的工具调用。 last_message state[messages][-1] outputs [] for tool_call in last_message.tool_calls: tool_name tool_call[name] tool_args tool_call[args] tool_map {multiply: multiply, add: add} result tool_map[tool_name].invoke(tool_args) outputs.append( AIMessage( contentstr(result), nametool_name, tool_call_idtool_call[id], ) ) return {messages: outputs} 构建图 def should_continue(state: AgentState) - str: 判断是否继续循环有工具调用则去工具节点否则结束。 last_message state[messages][-1] return tools if last_message.tool_calls else END graph StateGraph(AgentState) graph.add_node(agent, agent_node) graph.add_node(tools, tools_node) graph.set_entry_point(agent) graph.add_conditional_edges(agent, should_continue, {tools: tools, END: END}) graph.add_edge(tools, agent) app graph.compile() 运行 result app.invoke({ messages: [HumanMessage(content计算 3 乘以 4然后加上 5 的结果。)] }) print(result[messages][-1].content) LangGraph 的核心思想是把 Agent 流程建模成一张「状态图」Agent 节点负责推理工具节点负责执行条件边决定流程走向。相比手写循环它提供了更清晰的可视化、可中断、可恢复的能力适合构建复杂的生产级 Agent。 总结 Agent Harness 是构建智能体应用的基础设施它把「推理—行动—观察」的循环、工具调度、上下文管理和错误处理等通用能力封装起来让开发者专注于业务逻辑。通过本文的手写实现你应该已经理解了它的底层原理而通过 LangGraph 示例你也看到了主流框架如何把这些能力产品化。 在实际项目中建议从简单的手写 Harness 开始验证思路再根据复杂度逐步引入框架。无论选择哪条路理解核心循环和状态管理都是掌握 Agent 开发的关键。