首页 / 资讯中心 / 文章详情

手写最小可运行Agent:核心循环与模型输出解析实践

手写最小可运行Agent:核心循环与模型输出解析实践 ★ FEATURED ARTICLE
面试官如果只让你回答一个问题“写一个最小可运行的 Agent你的代码长什么样”大部分人第一反应是掏 LangChain但我的建议是不要在面试里先掏框架。你真正该展示的是你能手写 tool calling 循环、手写模型输出解析并且在边界问题上说清楚“为什么这么设计”。这篇文章我就把自己在面试和工作中反复写过的“原生手写 Agent 简易版”拆给你看代码量不大但会把 Agent 的原理、解析细节、踩坑点一次讲透。这篇文章适合两类人一类是准备 Agent 相关岗位面试的开发者另一类是已经用框架写过 Demo但觉得底层逻辑很模糊想回到原点把它彻底打通的人。如果你只是调包遇到“为什么模型输出解析失败”“怎么防止死循环”“没有 function calling 接口时怎么办”这类问题很容易卡住。手写一遍这些问题就不再是黑盒了。1. 为什么“原生手写”成了面试核心1.1 框架会过时核心循环不会Agent 这个词现在被玩得很花什么规划、记忆、工具调用、多 Agent 协作听着都很高大上。但如果你把热闹的包装剥掉一个最基础的 Agent 其实就是一个循环把用户的诉求交给模型模型决定是直接回答还是调用某个工具如果是调用工具Agent 执行完再把结果回填给模型继续循环直到模型给出最终答案。这个循环几乎是所有 Agent 框架的底座。LangChain 的 AgentExecutor、LlamaIndex 的 AgentRunner、各类开源 AutoGPT 变种本质上都是在围绕这个循环做增强。框架的好处是省事但坏处是容易让人产生“我会 Agent”的错觉。面试官见多了背源码的人所以越来越喜欢问你不用框架能不能自己写一个最小实现这个问题的潜台词是你到底是理解了原理还是只记住了 API。我自己带过几个新人用 LangChain 写 Agent 很快但一问“模型返回的 tool_calls 字段你理解成什么”就说不清楚了。这不怪他们框架把太多细节藏起来了。手写一遍你会被迫面对模型输出、消息格式、工具注册、循环终止这些问题而这些恰恰是真正决定 Agent 能不能稳定的关键。1.2 面试官真正想考察的三层能力第一层是会抽象。你能不能把 Agent 讲成一句话模型负责推理工具负责执行Agent 负责把两者连接起来循环。这一层决定你有没有全局认知。第二层是能处理不确定性。模型输出不是稳定的 JSON它会多说话、会漏括号、会输出 markdown 代码块、甚至会在 JSON 前后加一堆解释。你的解析层能不能兜住这些情况这一层最见功力也是我这次想重点展开的地方。第三层是有边界意识。Agent 会无限循环吗工具返回结果太大导致上下文爆掉怎么办工具本身的代码安全吗面试官如果追问这些说明他在乎的是“生产可用”不是“跑通 Demo”。手写实现的时候把这些考虑放进去回答会自然很多。2. Agent 最小可行架构三条核心链路2.1 主循环Agent Loop主循环是 Agent 的心脏。现在还没到写代码的阶段先用伪代码理解一遍1. 初始化 messages把系统提示词和用户问题放进去 2. 调用语言模型拿到响应 3. 判断响应的类型 - 如果是最终答案退出循环返回结果 - 如果是工具调用继续第 4 步 4. 根据模型给出的工具名和参数执行对应工具 5. 把工具返回结果作为一条消息追加进 messages 6. 回到第 2 步这就是最朴素的 Agent 主循环。面试的时候可以先把这个循环画出来再讲实现细节对方会觉得你的思路很清晰。比一上来就贴代码好得多。关键点在于messages 是整个循环的“共享记忆”。每一轮模型输出和工具观察结果都必须回到消息列表里模型才能根据最新状态继续推理。有人会把工具结果放在一个独立变量里只在下一轮拼到 prompt 中这样也能跑但一旦要多轮工具调用消息列表的一致性就很容易出问题。所以我的建议是从第一版开始就统一用 messages 列表让模型看到完整对话上下文。2.2 工具注册中心Tool Registry工具注册中心听起来高级其实就是一张“工具清单”。Agent 需要知道有哪些工具、每个工具叫什么、是干什么的、参数长什么样才能在合适的时机调用合适的工具。每个工具定义至少包含四部分名字、描述、参数结构、执行函数。描述尤其重要因为模型是根据描述来决定调不调用的。你写“根据表达式计算数学题的结果”它就更容易在需要计算的时候选中这个工具你只写“calculator”模型也能猜个大概但效果会差不少。面试时我建议你强调一点工具注册中心让 Agent 具备“可扩展性”。你不需要改 Agent 主循环只要往注册表里加一个新工具Agent 就获得了一项新能力。这个设计是 Agent 能否规模化的关键也是框架里大量使用装饰器注册工具的原因。2.3 模型输出解析层Output Parser这是所有 Agent 实现里最核心、也最容易出问题的地方。模型是概率输出它不会天然遵守“你规定的格式”。哪怕你用了 function calling 接口也要处理工具参数里的 JSON 字符串如果不依赖 function calling完全靠 Prompt 让模型输出一段可解析文本那解析层的复杂度会再上一个台阶。输出解析层要做的事情是把模型的原始输出转换成 Agent 能执行的结构化指令也就是“是最终回答”还是“工具调用工具名是什么、参数是什么”。解析失败时不能直接崩掉最好能把错误信息反馈给模型让它自己修复。很多手写实现会“死”在解析层。因为模型的输出花样太多永远有人猜不到。我的策略是双保险优先使用结构化 function calling 接口同时写好一个通用的文本解析函数兜底。下面第三章专门讲这里的细节。3. 手写实现从模型输出解析开始3.1 模型输出解析为什么最难说清楚因为这一层面对的“输入”是自然语言而模型输出的格式天然不稳定。你让模型输出 JSON它可能给你一个被 json 代码块包裹的 JSON你让模型只输出 JSON它可能在 JSON 前写一句“好的我来计算一下”你让模型不要在 JSON 里有注释它偶尔还是会用//写注释。每一个问题单看都不难但叠在一起解析代码就会变得很“脏”。我见过一个很典型的例子好的我来查询天气。结果如下 json { // 城市名称 city: 北京, // 天数 days: 3 }这段内容如果直接 json.loads会报 Invalid control character 或找不到花括号。所以解析之前必须先做一层“净化”把代码块标记剥掉只保留花括号内部的 JSON如果有多余的注释或者尾逗号也要做简单的兼容处理。 下面给出一个实用的提取函数 python import json import re class OutputParseError(Exception): 模型输出解析失败时抛出的异常 def extract_json_block(text: str) - str: # 先尝试从 markdown 代码块中提取 pattern r(?:json)?\s*(.*?)\s* m re.search(pattern, text, re.S) if m: return m.group(1).strip() # 再尝试从第一个 { 到最后一个 } 提取 start text.find({) end text.rfind(}) if start ! -1 and end ! -1 and end start: return text[start:end 1] raise OutputParseError(f无法从输出中找到 JSON 片段: {text[:120]!r}) def safe_json_loads(text: str) - dict: raw extract_json_block(text) # 做简单清理把注释行和尾逗号去掉这只是一个保守版本 raw re.sub(r(?m)//[^\n]*$, , raw) # 删除 // 注释 raw re.sub(r,\s*([}\]]), r\1, raw) # 删除尾逗号 try: return json.loads(raw) except json.JSONDecodeError as exc: raise OutputParseError(fJSON 解析失败: {exc}内容: {raw[:120]!r}) from exc这个函数不算完美但已经能覆盖大部分模型输出问题。面试的时候你可以直接说解析层要解决的是“格式不确定性”所以第一步是从文本里可靠地提取 JSON第二步是兼容常见错误第三步是把失败信息回传给模型。3.2 解析方案一function calling 格式现在主流模型普遍支持 function calling也叫 tool use。它的好处是模型返回的不是让你再解析的自然语言而是结构化字段。OpenAI 风格的响应里会有一个tool_calls数组每个元素包含id、type、function其中function里又有name和argumentsarguments 是字符串需要再解析一次。先把最简单的 LLM 客户端抽象出来后面所有示例都基于它import os from openai import OpenAI class LLMClient: def __init__(self, model: str gpt-4o-mini): self.model model self.client OpenAI( api_keyos.getenv(LLM_API_KEY), base_urlos.getenv(LLM_API_BASE, https://api.llm.example.com/v1), ) def chat(self, messages: list[dict], **kwargs) - dict: resp self.client.chat.completions.create( modelself.model, messagesmessages, **kwargs, ) return resp.choices[0].message.model_dump()注意这个base_url只是一个占位示例。你自己使用时换成实际可用的模型网关地址就行原理完全一致。有了这个客户端之后function calling 解析逻辑可以写成def parse_tool_calls(message: dict) - list[dict]: 从模型返回的消息里解析出工具调用列表 result [] for call in message.get(tool_calls) or []: fn call.get(function, {}) name fn.get(name, ) arguments_raw fn.get(arguments) or {} try: arguments safe_json_loads(arguments_raw) except OutputParseError: arguments {} result.append({ id: call.get(id), name: name, arguments: arguments, }) return result这段代码的核心是把tool_calls里的arguments字符串用安全的 JSON 加载器解析成字典同时处理好解析失败的情况。我见过有人在这里直接json.loads一旦模型返回空字符串或者格式错误整个 Agent 就会崩溃。所以哪怕是 function calling 的 arguments也一定要经过自己的解析层不要赌模型一定给一个合法的 JSON。3.3 解析方案二通用文本协议 最少正则如果你使用的模型不支持 function calling或者面试官故意问“没 function calling 怎么办”就需要让它输出一个约定好的文本协议。ReAct 风格的Action / Action Input / Observation / Final Answer是最经典的方案但用一个更简单的tool_call标签会让解析代码更好写。在系统提示词里明确告诉模型你需要调用工具时请严格按下面格式输出不要有多余解释 tool_call{name: 工具名, arguments: {参数名: 参数值}}/tool_call 当你看到工具返回的 Observation 后继续推理。如果可以回答用户直接输出最终答案。然后解析层只要做两件事找出所有tool_call标签把标签里的 JSON 解析出来如果没有标签就认为这是最终回答。这样比 ReAct 的正则简单很多也更容易在面试现场写清楚。TOOL_CALL_PATTERN re.compile(rtool_call(.*?)/tool_call, re.S) def parse_model_output(content: str): 把模型输出解析成 (类型, 数据) matches TOOL_CALL_PATTERN.findall(content) if not matches: return {type: final, content: content} calls [] for raw in matches: try: parsed safe_json_loads(raw) calls.append(parsed) except OutputParseError: # 解析失败时不要立刻终止把失败信息返回给模型修复 raise OutputParseError(f工具调用格式错误: {raw[:120]!r}) return {type: tool_call, calls: calls}这里的findall而不是search很重要因为模型可能在一轮输出里连续调用多个工具所有标签都应该被解析出来。3.4 解析失败时的兜底策略解析失败是必然会发生的事所以兜底策略要提前写好。我最常用的策略是“反馈式修复”把解析异常作为一条工具结果回传给模型让模型重新审视自己的输出。而不是直接抛给用户一个失败。举个例子模型输出了tool_call{name: calculator, arguments: {expression: 12,}}/tool_call尾逗号导致解析失败。我们捕获OutputParseError后把下面这条消息塞回对话你上一条输出的工具调用格式无法解析JSON 解析失败: Expecting property name enclosed in double quotes... 请重新输出正确的 tool_call 格式。模型看到之后通常会重新生成一个合法的格式。这个“错误信息也是上下文”的思路在手写 Agent 里非常重要。它跟人类的纠错方式很像你写错了我指出具体哪里错你重写。这个策略还可以推广到工具执行失败工具抛了异常不要直接中断而是把异常消息作为 Observation 回传模型可能会换一种方式再试。4. 工具层实现与 Agent 循环组装4.1 定义工具的协议工具定义直接决定 Agent 能不能正确调用。参考 OpenAI 的 tools 结构我会这样定义tools {} def register_tool(name: str, description: str, parameters: dict): def wrapper(func): tools[name] { name: name, description: description, parameters: parameters, handler: func, } return func return wrapperparameters需要符合 JSON Schema 的子集一般写法是{ type: object, properties: { expression: { type: string, description: 要计算结果的数学表达式 } }, required: [expression] }为什么参数要描述得这么细因为模型没有“读代码”的能力它只能看到你的 schema 和 description然后猜测这个工具怎么用。description 写得模糊它就会传错参数。这一点在产品环境里吃过很多亏我后来养成的习惯是每个参数都假设“模型是个完全不了解系统的外人”把边界和格式写清楚。4.2 用装饰器做工具注册表注册两个演示工具一个计算数学表达式一个查天气。import ast register_tool( namecalculator, description计算数学表达式的结果例如 1 2 * 3, parameters{ type: object, properties: { expression: {type: string, description: 合法的数学表达式} }, required: [expression] } ) def calculator(expression: str) - str: # 注意生产环境不要直接 eval这里仅用于演示 try: result eval(expression[:200]) return str(result) except Exception as exc: return f计算失败: {exc} register_tool( nameget_weather, description查询指定城市的实时天气, parameters{ type: object, properties: { city: {type: string, description: 城市名称例如 北京} }, required: [city] } ) def get_weather(city: str) - str: # 这里应该接真实天气 API演示时返回模拟数据 return f{city}天气晴气温 25 摄氏度你看注册表的作用就是把函数变成 Agent 能够感知和调用的“工具”。后面每增加一个工具只需要再加一个装饰器函数主循环代码一行都不用改。工具执行函数有一个容易被忽略的原则永远不要直接抛异常到外面。更好的做法是把异常捕获并转成字符串返回这样异常信息会成为模型的观察结果帮助它继续推理。这个原则在下面的执行函数里体现得很明显。def execute_tool(name: str, arguments: dict) - str: if name not in tools: return f错误工具 {name} 不存在可用工具{list(tools.keys())} func tools[name][handler] try: result func(**arguments) return str(result) except TypeError as exc: return f错误工具参数不匹配异常信息{exc} except Exception as exc: return f错误工具执行失败异常信息{exc}4.3 Agent.run 完整循环现在把解析、工具注册、消息列表串起来。这个 Agent 同时支持 function calling也预留了文本协议的路由。class MinimalAgent: def __init__(self, llm: LLMClient, max_iterations: int 8): self.llm llm self.max_iterations max_iterations def build_tools_schema(self): return [ {k: v for k, v in tool.items() if k ! handler} for tool in tools.values() ] def run(self, user_input: str) - str: messages [{role: user, content: user_input}] for step in range(self.max_iterations): resp self.llm.chat( messages, toolsself.build_tools_schema(), temperature0, ) message resp messages.append(message) tool_calls parse_tool_calls(message) if not tool_calls: # 没有工具调用说明模型给出了最终回答 return message.get(content, ) for call in tool_calls: observation execute_tool(call[name], call[arguments]) messages.append({ role: tool, tool_call_id: call[id], content: observation, }) raise TimeoutError(fAgent 超过 {self.max_iterations} 轮未得出最终答案)这套代码是一个非常经典的最小实现。面试手写时把这个类写在白板上再补几句边界处理基本就过了。有几个细节必须说明白。第一条messages.append(message)是把模型的完整输出保留住包括tool_calls字段。如果你只把content放回去或者擅自改成role: user到下一轮调用时模型服务端会因为找不到对应的tool_call_id报错。这是用 function calling 最容易踩的坑。第二条parse_tool_calls(message)拿到一个空列表说明模型没有请求工具那就是最终答案。这里隐含了一个假设模型要么回复最终答案要么调用工具。实际场景中模型可能既不调用工具又只说一句“我需要调用工具”这种无效输出在第一次迭代里偶尔会出现。兜底做法是如果content里包含tool_call标签就尝试走文本解析否则把它当最终答案返回必要时再做一次“答案质量校验”。4.4 完整演示让 Agent 使用计算器与天气查询假设用户输入是“北京今天气温多少度如果比 20 度高帮我算一下高了百分之几。”运行MinimalAgent.run后大概流程是这样的第一轮模型看到用户问题决定先查天气所以返回一个tool_calls调用get_weather参数是{city: 北京}。Agent 执行工具把“北京天气晴气温 25 摄氏度”追加到消息列表。第二轮模型看到天气结果发现 25 度比 20 度高于是调用calculator参数是{expression: (25-20)/20*100}。Agent 执行工具把“25.0”追加到消息列表。第三轮模型看到计算结果给出最终回答“北京当前气温 25 摄氏度比 20 度高 25%。” 循环退出。这就是一次完整的多工具协作 Agent 调用。整个过程里模型只负责“决定用什么工具、传什么参数、解释结果”真正干活的是你注册的 Python 函数。这个边界感很重要面试时最好主动讲出来。5. 面试高频追问边界条件与优化5.1 上下文窗口超限怎么办Agent 每次循环都会往 messages 里追加内容工具返回结果可能很大多轮下来上下文很容易超限。最简单的方案是在每次追加前统计消息总长度如果超过预设阈值就丢最老的对话轮次。注意不要丢系统提示词和最新的用户问题优先丢那些已经被工具结果覆盖的中间轮次。还可以做“观察压缩”。工具返回一个几百行的 JSON 时Agent 通常只需要关键字段。可以考虑把大结果先交给模型做一次摘要再把摘要放回对话。代价是多一次模型调用但能显著延长 Agent 的有效工作长度。面试时说出这个思路会让对方觉得你不是只会写循环的初学者。5.2 死循环和小作文问题没有迭代上限的 Agent 是不敢上生产的。max_iterations是最基础的防线。但这还不够因为模型可能会在倒数第二轮输出一个工具调用执行完后继续调用同一个工具直到撞上上限。所以更稳的策略是记录每个工具的调用次数如果某个工具连续被调用多次且参数几乎一样就主动打断并告诉模型“你已经重复调用相同工具请换一种思路或直接回答”。这种“重复检测”很实用。我遇到过模型因为天气 API 返回“稍后重试”就反复用相同参数查询十几次的情况。有了重复检测既节省 token也避免了用户等待。5.3 工具返回内容太大怎么办工具返回的内容如果直接完整塞进 messages很可能撑爆上下文。处理方式很简单截断。截断长度一般取 1000 到 2000 个字符超出部分用...(内容过长已截断)代替。但要注意截断后的 Observation 会丢失信息模型可能因此给出错误回答所以要尽量在工具端做字段筛选。比如查询数据库的工具SQL 只取需要的字段不要SELECT *调用 HTTP API也不要返回整包 Response。工具返回给 Agent 的内容不是给你自己看的是给模型看的越精简越容易被模型利用。这个经验对线上排查尤其重要。5.4 多工具并行与工具冲突OpenAI 风格的 function calling 支持一次返回多个tool_calls。上面的示例里我用for循环串行执行这没问题而且能保证顺序稳定。如果工具之间没有依赖也可以用ThreadPoolExecutor并行执行能加快整体速度。但并行会带来两个问题一是工具执行的顺序不确定某些有副作用的工具可能互相影响二是工具返回结果追加进消息的顺序必须和tool_calls的顺序一致否则模型会对应错。我的建议是第一版老老实实串行。面试时能主动提“并行会导致顺序问题所以默认串行”这比盲目上并发更让人放心。6. 常见问题与排查技巧实录6.1 模型输出解析失败的三个典型报错Expecting property name enclosed in double quotes模型用了单引号或裸字段名。原因是最常见的。我们的safe_json_loads只能做保守清理最好的办法是让模型重新生成并把错误信息反馈给它。Invalid control characterJSON 里出现了换行符或特殊控制字符。处理方式是把原始文本里的非法控制字符替换掉或者干脆提示模型不要输出转义序列。Extra data模型在一个 JSON 后面又输出了另一个 JSON 或文字。说明你的提取逻辑只取了第一个{到最后一个}把多余内容一起包进来了。可以改成找配对的 JSON 块或者用tool_call标签做精确切分。我整理过一个简单的排查原则先看原始输出再决定是修解析代码还是让模型重试。很多人一上来就加正则结果模型换一种输出方式正则又废了。稳一点的做法是“提取优先修复兜底重试保底”。6.2 工具执行阶段的坑最坑的工具是计算器。用eval虽然简单但安全风险极大。如果面试官问“你写的 eval 会不会有问题”你一定要能答上来生产环境换成受控的表达式求值库比如sympy或者用 AST 解析白名单节点。这个安全意识是加分项。另一个问题是参数不匹配。模型生成的参数名可能和工具定义不完全一致比如多传了一个没有定义的参数导致TypeError。上面的execute_tool里把 TypeError 捕获并转成了字符串返回而不是直接崩掉这是一个很关键的细节。还有一条建议工具结果尽量返回字符串不要返回对象。因为最终都要拼接进消息列表字符串更可控也不会出现“不能 JSON 序列化”的问题。如果你的工具本身返回结构体可以加一个序列化步骤。6.3 不能忽略的 temperature 与 few-shot 参数Agent 场景里temperature一定要低我通常设为 0。温度高了模型会在格式和参数上“自由发挥”解析失败率会肉眼可见地上升。这不是让它“有创造力”的地方工作流稳定比什么都重要。除了 temperature还可以在系统提示词里放一个工具调用的 few-shot 示例。比如示例里get_weather的用途是“中文城市名查询”模型看到这种具体例子后生成参数会更准确。很多人忘了 few-shot 对结构化输出的影响其实它比任何解析代码都好用因为它是从源头降低错误率。6.4 如何准备一份可展示的手写测试用例面试手写 Agent主动补几个测试会非常加分。核心是先 mock 掉 LLM不要真的每次调用 API。你只要把模型的返回体控制成固定的tool_calls和“最终答案”就能验证 Agent 是否把工具结果正确回填。一个粗略的测试思路def test_agent_should_call_calculator_then_return_final(): llm FakeLLM() # 第一次返回 calculator 调用第二次返回最终答案 agent MinimalAgent(llm) answer agent.run(11等于几) assert answer 112 assert llm.messages_contains_tool_result() # 断言工具结果进入了下一轮不用写多完整有一个验证思路就够了。面试官看到你能为手写代码补测试说明你的工程素养不是临时背出来的。7. 从手写 Agent 到理解框架源码7.1 开源框架到底帮你做了什么当你手写过一遍之后再回头看 LangChain 的AgentExecutor或者各类 Agent 框架会有一种“原来如此”的感觉。框架在核心循环之上替你完成了大量工程化工作提示词模板、工具加载、历史记忆的持久化、token 成本统计、重试机制、日志追踪、多 Agent 编排等等。这些能力都很重要但如果你不理解底层循环出问题时依然不知道怎么排查。比如 Agent 不断重复调用同一个工具很多人会去翻框架配置其实手写版里加一个“重复调用检测”就解决了。框架给你的是便利手写给你的是诊断能力。7.2 下一步扩展记忆、重试与可观测手写版本目前的 messages 都在内存里用户对话一结束就没了。要扩展成 product 级别第一步是加记忆层把之前的对话摘要或者关键信息存成长期记忆下次进入 Agent 时加载到系统提示词里。最简单的方式是用另一个 LLM 调用把历史 messages 压缩成一段摘要存入数据库或 Redis。第二步是重试机制。模型服务偶尔会超时或返回 5xx这时 Agent 不能直接把异常抛给用户应该做几次指数退避的重试。这个在手写框架里就是一个外层while的事但很多人从来没考虑过。第三步是可观测。手写 Agent 至少要打三种日志模型输入消息、模型输出解析结果、工具执行结果。面试时如果主动说“我会给 Agent 加日志方便排查哪一轮出了问题”这是很实际的工程思维。我最深的体会是面试手写 Agent考察的不是你背下多少代码而是你能不能把“模型、工具、循环”这三个词的关系讲清楚。掌握这个主干任何框架在你眼里都只是一层封装你才有能力在出现问题的时候定位到根因而不是重新安装一个插件碰运气。
阅读完成 · 觉得有帮助?
咨询建站