尧图网站设计 尧图网站设计YAOTU DESIGN
ARTICLE DETAIL

资讯详情

深耕网站设计与一线实操的经验洞察。

手写最小Agent Harness:掌握大模型工具调用与循环控制

手写最小Agent Harness:掌握大模型工具调用与循环控制 如果你最近正在做 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 层。对比维度AgentHarness角色决策者执行环境和管控层核心问题下一步做什么如何安全稳定地执行关键能力规划、推理、工具选择调度、记忆、安全、可观测工程难点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( roleassistant, content我需要先计算订单金额再获取当前时间。, tool_calls[ { id: call_1, name: calculator, arguments: {expression: 12*8100}, }, { id: call_2, name: get_now, arguments: {}, }, ], ) last_tool_results [ msg.content for msg in messages if msg.role tool ] return Message( roleassistant, 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(rolesystem, contentsystem_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(roleuser, contentuser_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( roletool, contenttool_result, tool_call_idcall[id], ) ) raise RuntimeError( fexceed 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: ftool {name} not found}, ensure_asciiFalse ) 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_asciiFalse )这是整个 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( levellogging.INFO, format%(asctime)s %(name)s %(levelname)s %(message)s, ) def calculator(expression: str) - str: 计算普通四则运算表达式例如 12*8100。 注意这里使用 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( providerMockProvider(), system_prompt你是一个订单助手可以访问计算器工具和当前时间工具。, max_steps5, ) harness.register_tool( ToolSpec( namecalculator, description计算四则运算表达式, parameters{ type: object, properties: { expression: {type: string} }, required: [expression], }, funccalculator, ) ) harness.register_tool( ToolSpec( nameget_now, description获取当前时间, parameters{ type: object, properties: {}, }, funcget_now, ) ) result harness.run( 帮我计算 12*8100 的结果顺便告诉我现在时间 ) 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, messages2 2026-06-15 10:31:22,001 agent_harness INFO execute tool calculator, args{expression: 12*8100} 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, messages4 2026-06-15 10:31:22,002 agent_harness INFO step 2 end, assistant returns final content FINAL: 最终结果1962026-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(levellogging.INFO, forceTrue)如果FINAL没有打印而是抛出了RuntimeError: exceed max_steps5说明主循环一直没有等到模型返回最终结果。这是最常见的问题下一节专门讲排查思路。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 从一开始就引入 TraceAgent 应用调试非常困难因为一次回答可能包含多次模型调用和工具调用。建议在 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 而已。
返回列表