ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

手写Agent原生实现:工具调用与模型输出解析实战

手写Agent原生实现:工具调用与模型输出解析实战 这几年 AI 应用开发最热的方向非 Agent 莫属。尤其是 Agent 的工具调用和模型输出解析几乎成了 LLM 应用岗位面试的必考点。我和不少读者聊过很多人框架用得很溜LangChain、CrewAI 都写过 demo但一问“不用框架原生手写一个带工具调用的 Agent”就卡住了。原因很简单框架帮你把底层逻辑封装好了你只看到了 API没看到流程。而面试官真正想考察的恰恰是你对 Agent 底层运行机制的理解——模型输出怎么解析、工具如何定义和注册、执行结果如何回填到对话里、循环什么时候终止。这套东西不亲手写一遍光靠背文档是答不好“为什么”的。这篇内容我会完整拆解一个简易 Agent 的原生实现重点放在两个核心环节模型输出解析和工具调用实现。全程不依赖 Agent 框架只用一个 Python 文件和 HTTP 请求代码可以直接抄去跑通面试也能当成项目经历来讲。1. Agent 核心机制为什么工具调用与输出解析绕不开1.1 工具调用是 Agent 区别于普通对话的分水岭普通聊天机器人拿到一个问题只会凭训练记忆生成一段文字。Agent 不一样它能在生成过程中主动请求调用外部工具——查数据库、调 API、执行计算、访问文件系统。这个“主动请求”就是 Agent 的灵魂。为什么工具调用这么关键因为它是 Agent 从“语言模型”变成“行动模型”的转折点。没有工具模型只是嘴炮只能基于自己的知识给你一个没有时效性、没有事实依据的答案有了工具模型才能真正去查、去算、去操作进而闭环完成一个任务。举个例子。你问“北京现在天气怎么样”普通对话模型只能根据训练数据里的历史信息胡编一个气温而 Agent 会意识到自己不知道实时天气于是输出一个工具调用请求{name: get_weather, arguments: {city: 北京}}程序收到这个请求后去调用天气 API拿到真实数据再把结果回传给模型模型基于真实数据给出最终答复。这个“请求工具 - 执行工具 - 回填结果 - 继续推理”的循环就是 Agent 最基本的工作方式。面试官问工具调用其实就是在考察你是否理解这个闭环。很多人只会调框架 API说不清楚里面的数据流转那这道题基本就废了。1.2 原生手写 vs 框架封装面试到底在考什么框架封装得很好的时候你可能只需要几行代码就能跑一个 Agentfrom langchain.agents import create_react_agent但手写一次之后你会发现框架帮你隐藏了太多细节而这些细节恰恰是面试场上真正拉分的地方。我列一下最典型的几个模型输出到底以什么格式返回工具调用请求是自然语言里的 JSON还是结构化字段工具参数从 JSON 字符串里怎么可靠地解析出来解析失败怎么办工具执行报错时错误信息要不要回传给模型用什么格式回传多轮工具调用之间怎么衔接什么时候停止循环工具列表怎么维护新增一个工具需要改动多少代码这些问题的答案只有自己从零写一遍才能真正讲清楚。面试时你说“我理解底层流程”和你说“我用过 LangChain”可信度完全不一样。前者意味着你遇到问题能自己排查后者很可能只是调包侠。原生手写还有一个实际好处不依赖特定平台。很多大模型服务商对 function calling 的支持不一致有的返回格式不同有的费用更高有的干脆不支持。你必须有“不靠平台特性也能实现工具调用”的能力这套能力就是最朴素的“模型输出解析 工具分发执行”。2. 模型输出解析从自由文本到结构化指令2.1 模型输出的真实形态调用大模型接口时如果走普通的 chat completion返回的 content 是一段自然语言。假设你让模型帮忙查询天气它可能输出这样一段话我来帮你查询北京的天气需要调用 get_weather 工具参数是{city: 北京}这段文字本身不是可以直接执行的指令。你需要从里面识别出“工具名”和“参数”这就引出了模型输出解析。在真实场景里模型输出的形态五花八门我总结为三种典型情况输出形态示例解析难度纯自然语言描述“我调用 get_weather 查一下北京天气”高需要靠语义理解带 Markdown 标记的伪 JSON代码块中夹着 JSON前后还有说明文字中清洗后解析结构化字段tool_callsAPI 直接返回 function.call 结构低但 arguments 仍需解析手写 Agent 时最常见的指令格式是让模型输出纯 JSON。虽然现在很多平台提供了原生的 function calling 接口但理解“从自由文本中提取指令”这个能力永远是你的基本功和兜底方案。2.2 JSON 解析与兜底策略最简单的思路是在 System Prompt 里要求模型输出纯 JSON当我需要调用工具时请严格输出以下格式不要包含任何其他内容 {name: 工具名, arguments: {参数名: 参数值}}但模型的服从性并不总是那么可靠。它仍然可能输出好的我马上为你查询 json {name: get_weather, arguments: {city: 北京}}如果你直接 json.loads()必定抛异常。这时候就需要一个健壮的解析函数按清洗优先级处理 1. 先剔除首尾空白和多余说明文字 2. 提取 markdown 代码块中包裹的 JSON 3. 遍历所有疑似 JSON 对象的片段找到第一个能成功解析的对象 4. 解析失败时回退到“截取第一个大括号到最后一个大括号”的策略。 我写了一个比较通用的提取函数实测覆盖了大部分真实场景 python import json import re def extract_json(text: str): if not isinstance(text, str) or not text.strip(): return None # 第一步直接尝试完整解析 try: return json.loads(text.strip()) except json.JSONDecodeError: pass # 第二步提取 markdown 代码块中的内容 code_block re.search(r(?:json)?\s*([\s\S]*?), text) candidate code_block.group(1) if code_block else text # 第三步用正则寻找所有疑似 JSON 对象逐个尝试解析 for match in re.finditer(r\{[\s\S]*?\}, candidate): try: return json.loads(match.group(0)) except json.JSONDecodeError: continue return None这个函数不是万能的嵌套 JSON 或数组型的复杂结构会有点力不从心但对付 Agent 工具调用返回的参数对象已经足够了。为什么先用正则匹配大括号因为模型经常在 JSON 前后夹带“好的”“稍等”这类口语逐段定位比直接强解析要稳得多。2.3 更稳的方案结构化输出与 function calling正则和 JSON 解析本质上是“从自由文本里捞指令”总会存在失败概率。所以现在的主流做法是让模型直接输出结构化的字段。以 OpenAI 兼容接口为例请求时传入tools参数模型返回的消息里会多出一个tool_calls字段{ tool_calls: [ { id: call_xxx, type: function, function: { name: get_weather, arguments: {\city\: \北京\} } } ] }这里有个容易被忽视的细节arguments字段依然是字符串不是对象。你还是需要json.loads()解析一次并且要做参数校验。所以面试时经常有连环问tool_calls 里的 arguments 是字符串你怎么处理答案很简单——解析、校验、出错就反馈给模型重试。结构化输出最大的优势是把“工具意图识别”从自然语言理解问题变成了“判断题 JSON 解析问题”稳定性和可验证性都大幅提升。但它也有依赖服务商必须支持 tool_calls 接口或者你得自己部署支持 agentic 的模型服务。因此我的建议很明确面试和实际项目中结构化输出作为首选JSON 清洗解析作为兜底。两套都掌握你才能在任何环境下做 Agent。3. 手写 Agent 工具实现定义、注册与执行3.1 工具定义schema 是模型的“操作说明书”工具定义不仅是给模型看的说明书还是程序侧参数校验的基础。一个工具描述得不好模型会频繁传错参数甚至完全不知道该调什么。工具 schema 的三要素是name工具名、description用途描述、parameters参数清单及其约束。其中description的质量对模型调用准确率的影响我这几年体会很深。给模型看的两件事最重要这个工具什么时候用参数怎么填。举个例子{ name: get_weather, description: 获取指定城市的实时天气信息。当用户询问天气、气温、降雨等情况时使用。城市名使用中文例如北京。, parameters: { type: object, properties: { city: { type: string, description: 城市名称如北京、上海、广州 } }, required: [city] } }我踩过最典型的坑是description 只写“查询天气”四个字结果模型经常把一整句话塞进参数里比如city: 请问北京今天天气怎么样。后来改成了“当用户询问天气时使用 参数格式示例 限制说明”三层描述调用准确率明显提升。3.2 工具注册表与动态分发工具定义好之后不能只靠硬编码 if/else 去分发。Agent 在循环里要根据模型输出的name去查找对应工具所以需要一个注册表结构把工具名、描述、参数 schema、可执行函数统一管理起来。我用的是装饰器注册和 Flask 路由注册的玩法类似非常简洁。每个工具只需要定义一次Agent 侧集中获取所有工具列表TOOL_REGISTRY {} def register_tool(nameNone, descriptionNone, parametersNone): def decorator(func): tool_name name or func.__name__ TOOL_REGISTRY[tool_name] { name: tool_name, description: description or func.__doc__ or , parameters: parameters or {type: object, properties: {}}, func: func, } return func return decorator register_tool( description获取指定城市的实时天气信息。当用户询问天气、气温等情况时使用。, parameters{ type: object, properties: { city: {type: string, description: 城市名称如北京} }, required: [city] } ) def get_weather(city: str) - str: # 这里调用真实天气 API import json return json.dumps({city: city, temperature: 18, condition: 晴}, ensure_asciiFalse)注册表的价值在于新增工具只需要加一个函数、一个装饰器不用改 Agent 主逻辑工具 schema 列表可以一键提取直接传给模型做声明执行时通过name查注册表拿到func再调用天然支持动态分发。3.3 主循环Agent 的 ReAct 核心Agent 主循环本质上是 ReAct 模式Reasoning推理 Acting执行。流程可以用几句话讲清楚把用户请求和工具描述拼进消息列表调用模型获取回复从回复中解析工具调用请求可能没有如果有工具调用执行工具把结果作为新消息回填到对话历史回到第 2 步如果没有工具调用把模型当前回复作为最终答案返回用户。这个循环是 Agent 的“心脏”。我见过很多新手写着写着就懵了问题多半出在第 4 步工具执行结果到底应该以什么角色、什么格式放回对话主流做法是走 OpenAI 兼容的对话规范用roletool的消息并带上对应的tool_call_id。手写时必须注意消息角色或字段对应不上模型很可能会直接报错或者忽略工具结果。这部分直接放到下面的完整代码里演示。4. 完整代码示例一个能跑的简易 Agent4.1 环境准备为了把“原生手写”贯彻到底这个示例不用任何 Agent 框架和 OpenAI SDK只用 Python 标准库加一个requests。它对接的是兼容 OpenAI 风格的模型服务接口你本地部署一个模型服务就能跑通。需要准备的信息LLM_BASE_URL模型服务地址如http://localhost:8000/v1LLM_API_KEY密钥本地服务通常随便填LLM_MODEL模型名称通过环境变量传入避免把密钥写死在代码里。4.2 核心实现从工具定义到主循环下面这段代码就是完整的一体版删掉了多余注释保留关键说明。它能跑通一次带工具调用的 Agent 对话。import json import os import re import requests BASE_URL os.environ.get(LLM_BASE_URL, http://localhost:8000/v1) API_KEY os.environ.get(LLM_API_KEY, sk-local) MODEL os.environ.get(LLM_MODEL, local-model) TOOL_REGISTRY {} def register_tool(nameNone, descriptionNone, parametersNone): def decorator(func): tool_name name or func.__name__ TOOL_REGISTRY[tool_name] { name: tool_name, description: description or func.__doc__ or , parameters: parameters or {type: object, properties: {}}, func: func, } return func return decorator register_tool( description获取指定城市的实时天气信息。当用户询问天气、气温、降雨等情况时使用。, parameters{ type: object, properties: { city: {type: string, description: 城市名称如北京、上海} }, required: [city] } ) def get_weather(city: str) - str: # 实际项目中替换成真实天气 API 调用 return json.dumps({city: city, temperature: 18, condition: 晴}, ensure_asciiFalse) def get_tool_schemas(): return [ { type: function, function: { name: tool[name], description: tool[description], parameters: tool[parameters], } } for tool in TOOL_REGISTRY.values() ] def chat_completion(messages, toolsNone): payload {model: MODEL, messages: messages} if tools: payload[tools] tools payload[tool_choice] auto resp requests.post( f{BASE_URL}/chat/completions, headers{Authorization: fBearer {API_KEY}}, jsonpayload, timeout30, ) resp.raise_for_status() return resp.json()[choices][0][message] def parse_tool_call(message): 优先解析结构化 tool_calls其次尝试从文本中提取 JSON 指令。 if message.get(tool_calls): tc message[tool_calls][0][function] try: args json.loads(tc[arguments]) except json.JSONDecodeError: return None, None, None return message[tool_calls][0][id], tc[name], args content message.get(content) or parsed extract_json(content) if parsed and parsed.get(name): return None, parsed[name], parsed.get(arguments, {}) return None, None, None def extract_json(text: str): # 前面写过的清洗逻辑 try: return json.loads(text.strip()) except json.JSONDecodeError: pass code_block re.search(r(?:json)?\s*([\s\S]*?), text) candidate code_block.group(1) if code_block else text for match in re.finditer(r\{[\s\S]*?\}, candidate): try: return json.loads(match.group(0)) except json.JSONDecodeError: continue return None def execute_tool(name, arguments): tool TOOL_REGISTRY.get(name) if not tool: return f错误工具 {name} 不存在 try: return tool[func](**arguments) except TypeError as e: return f错误参数不匹配{e} except Exception as e: return f错误执行工具时发生异常{e} def agent_run(user_input, max_steps5): messages [{role: user, content: user_input}] tools get_tool_schemas() for step in range(max_steps): message chat_completion(messages, tools) tool_call_id, name, arguments parse_tool_call(message) if not name: return message.get(content) or # 执行工具 result execute_tool(name, arguments) print(f[step {step}] 调用工具: {name}{arguments} - {result}) # 回填工具结果 messages.append({ role: assistant, content: message.get(content) or , tool_calls: message[tool_calls], }) messages.append({ role: tool, tool_call_id: tool_call_id, content: result, }) return 已达到最大工具调用轮次任务未能完成。 if __name__ __main__: print(agent_run(北京天气怎么样))4.3 代码里的几个细节为什么这样写我把这段代码里的几个关键点单独拿出来讲面试时你能答出这些细节印象分会高不少。第一parse_tool_call优先处理结构化tool_calls只有在没有结构化字段时才走 JSON 提取这符合“稳定方案优先兜底方案救命”的原则。第二工具执行的异常处理分了两层TypeError单独捕获说明参数不对通用Exception兜底保证任何工具挂掉都不会让整个 Agent 卡死。错误消息照样回填给模型模型可能会自己修正参数重新调用。第三回填消息时我把 assistant 消息含 tool_calls和 tool 消息都推进了 messages。很多刚入门的同学容易少补 assistant 那条消息导致对话历史和模型请求格式对不上。5. 面试核心高频追问与避坑经验5.1 五个必会的高频追问面试官不会只问“你写过 Agent 吗”他们会顺着你的回答往下钻。我把最常见的五个追问整理成了速查表你直接按这个思路答就行。追问回答要点模型返回的 JSON 非法怎么办设计重试机制把解析失败的错误信息回传给模型让它重新生成设置最大重试次数超限后降级为纯文本回答工具执行报错怎么处理捕获异常把错误摘要作为 tool 消息回填让模型判断是继续尝试、换参数还是换方案多个工具调用能并行吗tool_calls 本身可以包含多个调用程序侧可以并发执行但要处理参数校验、线程安全和 API 速率限制如何防止模型无限循环调用设置 max_steps 上限在 prompt 里提示“不要重复调用同一个工具”记录调用历史并加入上下文参数校验失败怎么办用 jsonschema 做参数校验失败时把具体错误字段回传给模型并提示可能的格式示例别只背答案。每个追问都对应你手写代码里的一个具体决策面试官就是想听到你做过的判断哪怕它不够完美。5.2 真实业务中的避坑心得这里分享几个我实际踩过的坑都是在文档里找不到的细节经验。第一个坑是工具描述太笼统。最开始我写的 description 只有一句话像“查询天气”。结果模型经常把用户原话直接当参数传进来查出来的东西驴头不对马嘴。后来我总结了一个固定句式触发场景 参数格式示例 限制条件。比如“当用户询问天气、气温、降雨等情况时使用。城市名使用中文例如北京。不要传入完整的问句。”这样模型调用准确率从六七成直接干到了九成以上。第二个坑是工具结果超长。有些工具返回的内容很大比如数据库查询结果几百 KB全部塞进上下文会给模型服务带来压力也浪费 token。我的做法是超过一定长度就截断只保留前 N 个字符并在开头标注“结果已截断”。对模型来说关键信息保留住就够了。第三个坑是 Agent 死循环。典型场景是模型反复调用同一个工具明明参数都错了还一直重试。如果不加 max_steps这个循环会一直消耗你的 API 额度。所以主循环里的次数上限不是可选优化是必选项。第四个坑是日志缺失。工具调用链路一旦出错没有日志你根本不知道模型输出了什么、工具返回了什么。我建议至少把每一步的 prompt、模型输出、解析结果、工具返回都写到结构化日志里出问题 5 分钟内能定位。5.3 面试话术怎么说才像真懂很多人的技术点其实会但表达太散。面试官问“介绍一下你的 Agent 项目”如果你说“我用了 LangChain里面有个 Agent配合工具能回答问题”完了完全没突出你的能力。用下面这个话术框架能把项目讲得又清楚又有含金量我用原生方式手写过一个简易 Agent核心是三步模型输出解析、工具注册执行、结果回填。解析层我写了 JSON 清洗函数能从 markdown 代码块里提取工具指令也兼容了 service 商的结构化 tool_calls 返回。工具层用注册表维护 schema 和执行函数新增工具只需要加一个装饰器。执行结果以 roletool 的消息回填进对话主循环根据模型是否请求工具来决定继续还是终止。这样讲有几个好处有技术深度原生手写、有细节JSON 清洗、注册表、有工程意识结果回填、循环终止。面试官顺着任何一个点往下问你都能展开因为你真的写过。最后的体会我个人在带新人时发现凡是用心手写过 Agent 的同学后面学任何框架都很快反过来一上来就套框架的遇到问题往往无从下手。工具调用和模型输出解析这两件事是整个 Agent 系统里最容易被框架隐藏、但又最值得搞懂的部分。如果你想给自己留一个进阶方向可以在现在这个简易版本上继续加内容给 Agent 加短期记忆和长期记忆、实现多工具并行调度、把注册表改成支持异步、或者让多个 Agent 围绕同一个任务协作。但无论加多少功能底层那一套“解析指令 - 查表执行 - 回填结果 - 循环判断”的逻辑永远不会变。把这篇里的代码跑通再对着面试真题练几遍表达这个考点你就能真正吃透了。
返回列表