如果你最近正在做 AI Agent 应用,大概率遇到过这几类情况:模型回话时好时坏,工具调用格式飘忽不定,上下文越跑越长,某个环节死循环烧掉大量 token。很多人第一反应是换更强的模型、写更神奇的 prompt,但问题往往不在模型本身,而在给 Agent 提供执行环境的那个框架上。
这个框架,在业界被称作Agent Harness。很多人也把它和具体产品“Harness Agent”混着叫,其实更准确的词是“Agent 的运行 harness”:Agent 负责出主意,harness 负责兜底、调度、记忆和安全。今天这篇文章,我想用最朴素的工程视角把它讲清楚:Harness 和 Agent 到底有什么区别?Agent 的底层原理是什么?一个最小可运行的 Agent Harness 应该怎么写?以及工程落地时最容易踩的坑在哪里。
文章不会只停留在概念层面。我们会用纯 Python 标准库手写一个最小 Agent Harness,不依赖任何第三方 Agent 框架,演示工具注册、模型抽象、主循环调度、上下文累积和终止控制。跑通这个最小实现之后,你再去看 LangGraph、OpenAI Codex 的 open agent harness,或者任何商业化 Agent 框架,思路会顺畅很多。
1. 明确判断:Agent 是大脑,Harness 是骨架和安全绳
先解决最核心的问题:Harness 和 Agent 到底有什么区别。
Agent是一个能够感知环境、制定计划、调用工具、根据结果继续行动的智能体。它的核心能力是“决策”:下一步做什么,是直接回答用户,还是要调用工具。
Harness看起来没有 Agent 那么“聪明”,但它是承载 Agent 的那套工程系统。它负责让 Agent 能真正跑起来,包括:
- 管理对话历史和上下文窗口;
- 把工具列表暴露给模型;
- 解析模型返回的工具调用指令;
- 执行工具并回传结果;
- 控制循环轮数,防止死循环;
- 记录日志和追踪数据;
- 设置安全边界和权限限制。
打个篮球的比方。Agent 是球场上的明星球员,负责判断局势、选择战术;Harness 是教练团队、场馆安保和比赛规则。球员再厉害,没有比赛规则、没有安保人员、没有战术板,也没办法完成一场正规比赛。
为什么大家总是把这两个词混在一起?因为成熟的 Agent 框架通常自带一套 Harness。你使用 LangChain、LangGraph 或者 OpenAI 的 Agent SDK 时,框架内部已经帮你实现了调度、工具调用、上下文管理。这种“开箱即用”的体验,让人误以为 Agent 天然就应该这么跑。
于是问题出现了:一旦生产环境出现工具调用失败、上下文超长、权限失控、循环不终止,开发者的第一反应往往是“模型不够聪明”。实际上,这些问题的根源大部分在 Harness 层。
| 对比维度 | Agent | Harness |
|---|---|---|
| 角色 | 决策者 | 执行环境和管控层 |
| 核心问题 | 下一步做什么 | 如何安全稳定地执行 |
| 关键能力 | 规划、推理、工具选择 | 调度、记忆、安全、可观测 |
| 工程难点 | prompt、模型选择 | 上下文管理、错误处理、权限控制 |
| 常见失败 | 答非所问 | 死循环、上下文溢出、工具越权 |
一句话总结:Agent 的智能决定了能力的上限,Harness 的工程能力决定了系统能不能交付。
2. 底层原理:Agent 是如何“转圈”的
理解了区别之后,再看底层原理。所有 Agent 应用,无论界面多复杂,本质上都在跑同一个主循环。这个循环通常包括五个步骤:
- 观察(Observe):读取用户输入和已有的对话历史;
- 规划(Plan):模型判断下一步是直接回答,还是需要调用工具;
- 行动(Act):如果需要工具,harness 解析模型返回的工具调用指令,执行对应函数;
- 整合(Integrate):把工具执行结果作为新的上下文,继续交给模型;
- 终止(Terminate):当模型不再返回工具调用,或者达到最大循环次数,结束流程。
这个过程很像人类处理复杂任务的方式:先看问题,查资料,得到结果后继续判断,直到确认信息足够,再给出最终答案。
关键点在于:模型并不知道你的工具是怎么实现的,它只知道工具的描述和参数结构。你提供给模型的是“工具箱说明书”,而不是真正的 Python 函数。真正控制执行过程的是 Harness。
一次典型的工具调用流程如下:
用户问题 ↓ Harness 将历史消息 + 工具描述发送给模型 ↓ 模型返回 assistant 消息,携带 tool_calls ↓ Harness 根据 tool_calls 中的工具名查找注册表 ↓ Harness 执行对应的函数,并把返回值包装成 tool 消息 ↓ tool 消息追加到对话历史,再次发送给模型 ↓ 模型给出最终回答,或继续发起新的工具调用既然主循环看起来这么简单,为什么不能直接写一个for循环?因为真实环境太复杂了:模型返回的 JSON 可能格式错误,工具可能抛异常,上下文可能超过模型窗口限制,某一步可能反复触发同一个工具形成死循环。这些工程问题,都必须在 Harness 层提前设计并处理。
3. Harness 的核心能力拆解
一个能够上生产的 Agent Harness,至少要具备以下六个能力。
| 核心能力 | 解决的问题 | 工程动作 |
|---|---|---|
| 会话管理 | 多轮对话之间如何保存上下文 | 维护消息列表,区分 system / user / assistant / tool |
| 工具注册与调度 | 模型如何知道有哪些工具、如何调用 | 维护工具注册表,动态构建工具 schema |
| 上下文窗口管理 | 上下文超长、Token 成本过高 | 截断、摘要、滑动窗口、持久化 |
| 安全边界 | 工具被恶意调用、越权操作 | 沙箱执行、白名单、参数校验、最小权限 |
| 可观测性 | 循环卡住、工具出错后如何定位 | 日志、Trace 追踪、步骤计数、耗时统计 |
| 终止控制 | 模型反复调用工具导致死循环 | 设置 max_steps、重复检测、超时中断 |
这里特别说明一下“工具注册与调度”。真实项目中,Agent 的工具可能是获取订单信息、查询库存、发送消息、操作数据库。每个工具都应该像一份 API 文档一样被描述清楚,包括:
- 工具名称;
- 功能描述;
- 参数结构;
- 必填参数;
- 返回值约定。
Harness 会把这些描述转换成模型能够理解的 tool schema,随请求发送给模型。模型只需要在返回消息中指明“我要调用哪个工具、传入什么参数”即可,不需要知道背后的实现细节。
安全边界同样重要。一个能够被模型任意调用的工具集,实际上是一个开放给大模型任意操作的系统入口。如果没有权限控制和参数校验,Agent 就可能执行超出预期的操作。这也是为什么很多生产级 Harness 会把工具执行放到沙箱里,并且对每个工具设置单独的授权范围。
4. 环境准备与基础结构
接下来进入实战环节。我们要用纯 Python 标准库实现一个最简 Agent Harness。虽然简单,但结构和生产框架完全一致。
4.1 环境要求
- 操作系统:Windows / macOS / Linux 均可;
- Python 版本:建议 3.10 或更高,代码中使用了
dataclass和类型注解,低版本也可以跑,但 3.10 以上体验最好; - 依赖:不需要安装任何第三方 Agent 框架,只用标准库。
如果你打算把代码里的 MockProvider 替换成真实大模型调用,那么需要另外安装对应的 SDK。本文为了让你无 API Key 也能完整跑通,采用本地模拟模型的方式。
4.2 项目目录结构
建议按下面的结构组织文件:
agent-harness-demo/ ├── models.py # 消息模型、工具描述模型 ├── provider.py # 模型服务抽象,以及本地 Mock 实现 ├── harness.py # Agent Harness 主类 └── main.py # 注册工具并启动示例4.3 关于代码定位
需要提前说明:这份代码是“教学最小实现”,目标是讲清楚 Harness 主循环和工具调度逻辑,不是生产级框架。正式项目还需要考虑异步执行、并行工具调用、持久化存储、鉴权、限流、模型重试、Trace 追踪等一系列能力。建议先跑通最小实现,再逐步补充。
5. 完整代码实现:手写一个最小 Agent Harness
下面我们分四个文件写出完整实现。
5.1 第一步:定义消息和工具描述模型
文件路径:models.py
# 文件路径:models.py from dataclasses import dataclass, field from typing import Any, Callable, Dict, List, Optional @dataclass class Message: role: str # system / user / assistant / tool content: str # 文本内容 tool_calls: Optional[List[Dict[str, Any]]] = None tool_call_id: Optional[str] = None @dataclass class ToolSpec: name: str # 工具名 description: str # 工具描述,给模型看 parameters: Dict[str, Any] # 参数 JSON Schema func: Callable[..., str] # 实际执行的 Python 函数这段代码的核心是两个数据类:
Message统一表示 Agent 对话中的消息。role用来区分消息来源,tool_calls用于承载模型返回的工具调用指令,tool_call_id用于把工具执行结果和对应的调用对应起来。ToolSpec把“工具描述”和“真实函数”绑定在一起。真正给模型调用的是description和parameters,真正执行的是func。
为什么不能只用dict?因为项目一复杂,消息种类会变多,手动维护dict非常容易出错。使用数据类可以明确字段约束,后续也方便扩展。
5.2 第二步:定义模型服务抽象与 Mock Provider
文件路径:provider.py
# 文件路径:provider.py from typing import List from models import Message class ModelProvider: """模型服务抽象:真实项目里替换为 OpenAI / 国产模型 / 本地模型 SDK""" def chat(self, messages: List[Message]) -> Message: raise NotImplementedError class MockProvider(ModelProvider): """本地模拟模型:不需要 API Key 也能演示完整的 Agent 循环。 约定: 1. 如果当前还没有任何工具调用,模型返回两个工具调用指令; 2. 如果已经有工具执行结果,模型读取这些结果并生成最终回答。 """ def chat(self, messages: List[Message]) -> Message: tool_requests = [] for msg in messages: if msg.tool_calls: tool_requests.extend(msg.tool_calls) if len(tool_requests) == 0: return Message( role="assistant", content="我需要先计算订单金额,再获取当前时间。", tool_calls=[ { "id": "call_1", "name": "calculator", "arguments": {"expression": "12*8+100"}, }, { "id": "call_2", "name": "get_now", "arguments": {}, }, ], ) last_tool_results = [ msg.content for msg in messages if msg.role == "tool" ] return Message( role="assistant", content="最终结果:" + ";".join(last_tool_results), )ModelProvider是模型服务抽象层。真实项目中,你会在chat方法里调用大模型 SDK,把messages转换成 SDK 要求的格式,再把模型的响应转换成Message。
MockProvider的作用是在没有真实模型的情况下模拟模型行为。它的逻辑很简单:
- 第一次调用时,返回两个工具调用指令,模拟模型说“我要算数并查询时间”;
- 第二次调用时,读取之前所有
role == "tool"的消息内容,拼成最终回答。
这样就能在完全不依赖外部 API 的情况下,演示“模型决定调用工具 → 工具执行 → 模型整合结果”的完整闭环。
5.3 第三步:实现 Harness 主类
文件路径:harness.py
# 文件路径:harness.py import json import logging from typing import Dict, List from models import Message, ToolSpec from provider import ModelProvider logger = logging.getLogger("agent_harness") class AgentHarness: def __init__( self, provider: ModelProvider, system_prompt: str, max_steps: int = 5, ): self.provider = provider self.system_prompt = system_prompt self.tools: Dict[str, ToolSpec] = {} self.messages: List[Message] = [ Message(role="system", content=system_prompt) ] self.max_steps = max_steps def register_tool(self, spec: ToolSpec) -> None: """注册工具:只有注册过的工具才能被模型调用""" self.tools[spec.name] = spec logger.info("registered tool: %s", spec.name) def build_tool_schema(self) -> List[Dict]: """把工具描述转换成模型需要的工具 schema 格式""" schema = [] for spec in self.tools.values(): schema.append( { "type": "function", "function": { "name": spec.name, "description": spec.description, "parameters": spec.parameters, }, } ) return schema def run(self, user_input: str) -> str: """Agent 主循环入口""" self.messages.append(Message(role="user", content=user_input)) for step in range(1, self.max_steps + 1): logger.info("step %d start, messages=%d", step, len(self.messages)) assistant_msg = self.provider.chat(self.messages) self.messages.append(assistant_msg) if not assistant_msg.tool_calls: return assistant_msg.content for call in assistant_msg.tool_calls: tool_result = self._execute_tool(call) self.messages.append( Message( role="tool", content=tool_result, tool_call_id=call["id"], ) ) raise RuntimeError( f"exceed max_steps={self.max_steps}, possible infinite loop" ) def _execute_tool(self, call: Dict) -> str: """执行单个工具调用,并把结果统一转换成字符串""" name = call.get("name") arguments = call.get("arguments", {}) spec = self.tools.get(name) if spec is None: return json.dumps( {"error": f"tool {name} not found"}, ensure_ascii=False ) try: logger.info("execute tool %s, args=%s", name, arguments) return str(spec.func(**arguments)) except Exception as exc: logger.exception("tool %s execute failed", name) return json.dumps( {"error": str(exc)}, ensure_ascii=False )这是整个 Harness 的核心。重点看run方法里的循环:
- 把用户输入追加到消息列表;
- 调用模型的
chat方法; - 把模型返回的 assistant 消息追加到历史;
- 如果模型没有返回
tool_calls,说明任务完成,直接返回; - 如果有
tool_calls,逐个执行工具,并把工具结果以tool消息追加到历史; - 进入下一轮循环。
max_steps是终止控制的核心。无论模型多聪明,循环次数必须有上限。生产环境中,这里还要加入“重复工具调用检测”:如果同一参数反复调用同一个工具,应尽早中断。
_execute_tool方法做了一件很重要的事情:把工具执行结果统一转换成字符串。为什么?因为模型只认文本。你在工具函数里返回dict、list、float,最终都要序列化成字符串才能作为上下文继续传给模型。
5.4 第四步:注册真实工具并启动
文件路径:main.py
# 文件路径:main.py import datetime import logging from harness import AgentHarness from models import ToolSpec from provider import MockProvider logging.basicConfig( level=logging.INFO, format="%(asctime)s %(name)s %(levelname)s %(message)s", ) def calculator(expression: str) -> str: """计算普通四则运算表达式,例如 '12*8+100'。 注意:这里使用 eval 仅用于教学演示。 生产环境请使用 ast.literal_eval 或自定义安全解释器。 """ allowed = set("0123456789+-*/(). ") if any(ch not in allowed for ch in expression): return "非法表达式,只支持数字和四则运算符" return str(eval(expression, {"__builtins__": {}}, {})) def get_now() -> str: """获取当前时间""" return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") def main(): harness = AgentHarness( provider=MockProvider(), system_prompt="你是一个订单助手,可以访问计算器工具和当前时间工具。", max_steps=5, ) harness.register_tool( ToolSpec( name="calculator", description="计算四则运算表达式", parameters={ "type": "object", "properties": { "expression": {"type": "string"} }, "required": ["expression"], }, func=calculator, ) ) harness.register_tool( ToolSpec( name="get_now", description="获取当前时间", parameters={ "type": "object", "properties": {}, }, func=get_now, ) ) result = harness.run( "帮我计算 12*8+100 的结果,顺便告诉我现在时间" ) print("FINAL:", result) if __name__ == "__main__": main()这段代码注册了两个工具:
calculator:计算四则运算表达式,描述、参数结构、真实函数都在一个ToolSpec里;get_now:获取当前时间,不需要参数。
calculator中使用了白名单字符校验。虽然这是教学示例,但已经体现了“参数校验”的思路:不要盲目信任模型生成的内容,也不能让工具执行任意代码。
5.5 关键设计逻辑说明
上面四个文件合起来,就是一套完整的 Agent 循环。它的设计哲学可以总结为:
- 模型不直接接触工具函数:模型只拿到工具 schema,真正执行函数的是 Harness;
- 所有状态都走消息列表:系统提示词、用户输入、助手返回、工具结果,全部按顺序追加到
messages; - 工具结果必须回填:工具执行后必须把结果追加到历史,否则模型无法“知道”工具已经执行过;
- 主循环必须有上限:
max_steps防止死循环消耗大量资源; - 抽象 Provider 接口:以后换真实模型,只需要替换
MockProvider。
6. 运行结果与效果验证
在项目目录下执行:
cd agent-harness-demo python main.py预期输出大致如下:
2026-06-15 10:31:22,001 agent_harness INFO step 1 start, messages=2 2026-06-15 10:31:22,001 agent_harness INFO execute tool calculator, args={'expression': '12*8+100'} 2026-06-15 10:31:22,002 agent_harness INFO execute tool get_now, args={} 2026-06-15 10:31:22,002 agent_harness INFO step 2 start, messages=4 2026-06-15 10:31:22,002 agent_harness INFO step 2 end, assistant returns final content FINAL: 最终结果:196;2026-06-15 10:31:22如何判断运行成功?
- 日志中出现了
execute tool calculator和execute tool get_now,说明工具调度正常; step 2 start时消息数量从 2 变成 4,说明工具结果已经追加到上下文;- 最终打印
FINAL: 最终结果:196;...,说明模型拿到了工具结果并生成了最终回答。
如果程序没有任何输出,第一步先确认logging.basicConfig是否设置成功。Windows 终端下如果日志没有刷出来,可以尝试在main.py顶部增加:
logging.basicConfig(level=logging.INFO, force=True)如果FINAL没有打印,而是抛出了RuntimeError: exceed max_steps=5,说明主循环一直没有等到模型返回最终结果。这是最常见的问题,下一节专门讲排查思路。
7. 常见问题与排查思路
自己实现 Agent Harness 时,下面几个问题几乎一定会遇到。
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
报错exceed max_steps | 模型反复发起工具调用,主循环不终止 | 打印每轮 assistant 消息和工具调用参数 | 提高循环上限;加入重复调用检测;检查工具是否总是返回导致模型再次调用的结果 |
工具返回tool xxx not found | 模型调用了未注册的工具名 | 打印self.tools.keys()与模型返回的工具名 | 检查工具 schema 与注册名称是否一致;考虑增加模糊匹配 |
| 工具参数缺失 | 模型返回的 arguments 缺少必填字段 | 打印原始tool_callsJSON | 在_execute_tool中补充默认值;在 schema 中声明required;增加参数校验 |
| 上下文越来越长,Token 成本飙升 | 每轮工具结果都追加到消息列表,没有压缩 | 观察len(self.messages)和模型 token 计费 | 引入摘要压缩、滑动窗口、过期消息清理 |
| 模型返回 JSON 格式不稳定 | 工具调用结构经常解析失败 | 把模型原始输出原样打到日志中 | 增加重试机制;使用更强的 JSON 约束提示;必要时做本地 schema 校验 |
| 生产环境工具执行危险操作 | 模型被诱导调用越权工具 | 审计工具注册表 | 权限模型、沙箱执行、人工审批、操作审计 |
这里特别强调“重复调用检测”。真实项目里最可怕的不是模型不会调用工具,而是模型疯狂重复调用同一个工具。比如查询库存接口被连续调用 20 次,而第一次的结果已经明确告诉它“库存不足”。为了应对这种情况,生产 Harness 至少要记录:
- 每个工具执行的次数;
- 相同参数是否重复执行;
- 单次任务的总工具调用数。
一旦超过阈值,立即中断并让模型基于已有结果直接回答。
8. 最佳实践与工程建议
如果你的目标不是做一个玩具 Demo,而是把 Agent 引入真实业务系统,下面这些建议建议收藏。
8.1 把工具当 API 设计,而不是当函数随机写
工具 schema 是模型与真实世界的接口契约。每个工具都应该有清晰的名称、描述、参数说明、返回值约定。描述要写清楚“什么时候该用这个工具”,否则模型会在不合适的场景调用它。至少做到:
- 工具名称全局唯一;
- 描述包含使用前提和边界;
- 参数必须有类型约束;
- 返回值固定为 JSON 或字符串,并注明错误格式。
8.2 所有工具执行都要考虑幂等和异常
Agent 工具不是普通函数调用,它可能因为网络超时、模型重试、Harness 重启而重复执行。工具函数必须能够安全地多次执行。涉及数据库写入、消息发送、支付扣款时,特别要设计幂等键和事务保护。
8.3 安全边界必须独立于模型
不要以为“用户不会输入恶意内容,模型会自己判断”。安全应该放在 Harness 层,而不是依赖模型自觉。建议:
- 工具执行前做参数白名单校验;
- 敏感操作用独立身份认证;
- 高权限工具要求人工审批;
- 外部命令执行放到沙箱环境;
- 所有工具调用记录完整审计日志。
8.4 从一开始就引入 Trace
Agent 应用调试非常困难,因为一次回答可能包含多次模型调用和工具调用。建议在 Harness 设计之初就给每次任务分配一个trace_id,记录:
- 每一步的输入输出;
- 模型调用的 token 消耗;
- 工具执行耗时;
- 错误信息。
没有 Trace 的 Agent 项目,线上出了问题基本只能靠猜。
8.5 上下文管理要提前设计
很多人第一版 Harness 都是“把所有消息一股脑发给模型”,直到账单爆炸才想起来做上下文管理。核心策略包括:
- 只保留最近的 N 轮消息;
- 对早期对话做摘要压缩;
- 工具执行结果只保留结构化关键字段;
- 长文档先检索再拼装。
8.6 用模拟模型跑通集成测试
在本文中我用 MockProvider 只是为了演示。但真正的工程价值在于:你可以把 MockProvider 当作测试替身,用它验证 Harness 的循环逻辑而不消耗真实模型费用。生产项目中,建议设计一套“固定剧本”的 FakeProvider,用于 CI 回归测试。
9. 总结与后续学习方向
回到开头那句话:真正决定 Agent 应用能不能落地的,不是模型选得多强,而是 Harness 设计得是否扎实。从本文不难看出,一个 Agent 能跑起来并不难,难的是让它稳定、安全、可观测地在生产环境运行。
你已经学会了:
- 区分 Agent 与 Harness;
- 理解 Agent 主循环的五个步骤;
- 掌握工具注册、消息回填、循环控制、终止判断;
- 用纯 Python 写出一份可运行的最小 Agent Harness;
- 知道常见异常和排查思路。
下一步可以做什么?
如果你对框架内部机制感兴趣,可以去看 LangGraph 的执行图设计,或者研究 OpenAI Codex 的 open agent harness 源码,看它如何处理工具注册、沙箱执行和上下文管理。如果你想深入工程化,建议先做三件事:把 MockProvider 替换成真实模型、给 Harness 加上 Trace 日志、为工具调用接入参数校验和幂等控制。
等你把这三件事做完,再回头看市面上的 Agent 框架,就不会觉得它们“神奇”了。它们的底层,不过是一个考虑得更加周密的 Harness 而已。