从零构建AI Agent核心：Agent Loop与工具调用全链路解析

1. 从零开始理解 Agent 的核心骨架最近和不少刚接触 Agent 领域的朋友交流发现大家普遍有个困惑看了很多关于“智能体”、“自主智能”的宏大叙事和框架介绍但回到代码层面一个最基础、能跑起来的 Agent 到底长什么样它的心脏——“Agent Loop”是如何跳动的又是如何调用工具完成任务的这些看似基础的问题恰恰是理解一切复杂 Agent 系统的基石。今天我们就抛开那些眼花缭乱的框架和营销术语亲手拆解并构建一个“最小可行 Agent”。这个 Agent 的目标极其单纯接收一个用户指令通过思考决定是否需要调用工具如果需要就调用合适的工具并处理结果最终给出回答。我们将聚焦于两个最核心的机制Agent Loop智能体循环与工具使用Tool Use的全链路。通过这个过程你不仅能看清 Agent 内在的工作流更能掌握其设计精髓为后续学习更复杂的多智能体协作、记忆、规划等高级能力打下坚实基础。无论你是想入门 Agent 开发的工程师还是希望理解其原理的产品经理或研究者这篇从第一性原理出发的拆解都会让你有豁然开朗的感觉。我们用的“语言”是 Python但更重要的是其背后通用的设计模式与思想。2. 项目整体设计与核心思路拆解在动手写代码之前我们必须先想清楚设计。一个能“跑起来”的最小 Agent绝不是简单地把大语言模型LLM的 API 包装一下。它的核心在于引入了“循环”和“工具”这两个关键概念使得系统从单纯的“问答机”变成了可以自主执行多步骤任务的“智能体”。2.1 什么是 Agent Loop你可以把 Agent Loop 想象成一个智能体的“思考-行动”循环。它不同于一次性的 API 调用而是一个持续的过程观察Observation接收当前的状态信息通常是用户的输入或上一步工具执行的结果。思考Thinking基于观察决定下一步要做什么。是直接给出最终答案还是需要调用某个工具来获取更多信息行动Action如果决定调用工具就执行对应的工具并获取工具返回的结果。回到观察将工具返回的结果作为新的“观察”输入给下一步的“思考”。这个循环会一直进行直到智能体认为它已经收集到足够的信息可以给出最终答案为止。这个循环机制是 Agent 能够处理复杂、多步骤任务的根本。2.2 工具使用的全链路解析工具Tool是 Agent 延伸自己能力的“手脚”。一个工具本质上是一个函数它有着明确的名称、描述和参数。全链路指的是从“决定使用工具”到“消化工具结果”的完整过程工具描述与注册如何让 LLM 知道有哪些工具可用我们需要以结构化的方式通常是 JSON Schema向 LLM 描述工具的功能和输入参数。工具调用决策LLM 在思考阶段根据当前对话上下文和任务判断是否需要调用工具以及调用哪一个。调用格式解析LLM 的输出需要被规范化为一个标准的工具调用请求如{“name”: “calculator”, “arguments”: {“a”: 5, “b”: 3}}。工具执行系统根据解析出的工具名和参数找到对应的本地函数并执行。结果整合将工具执行的原始结果如8再次反馈给 LLM让 LLM 基于这个新信息进行下一轮的思考或生成最终回答。2.3 最小系统的技术选型为了最清晰地展示原理我们做如下技术选型LLM 接口使用 OpenAI 格式的 API如 OpenAI GPT、DeepSeek、智谱等国内可访问的模型。这几乎是所有 Agent 框架的通用接口标准兼容性最好。编排框架不直接使用 LangChain 或 LlamaIndex 等高级框架。我们将从零开始用纯 Python 和requests库实现核心循环。这能让你彻底摆脱框架的“黑箱”理解每一行代码在做什么。学会这个你再去看任何框架都会一目了然。工具示例实现两个最简单的工具一个计算器calculator和一个查询当前时间的工具get_current_time。这个设计思路的优势在于极致的透明度和教育性。你会亲眼看到消息列表message history如何构建、函数描述如何注入、模型响应如何被解析、状态如何流转。理解了这些任何封装过的框架对你来说都将不再是魔法。3. 核心模块拆解与实现要点现在我们进入实战环节一步步搭建这个最小 Agent。我们将系统分为几个核心模块每个模块都承担明确的责任。3.1 模块一工具系统的定义与注册工具是 Agent 能力的扩展。首先我们需要一种方式来定义和存储工具。import json from datetime import datetime from typing import Dict, Any, Callable, List # 工具定义一个工具包含名称、描述、参数schema和执行函数 class Tool: def __init__(self, name: str, description: str, parameters: Dict[str, Any], func: Callable): self.name name self.description description self.parameters parameters # 符合JSON Schema格式 self.func func def execute(self, **kwargs) - str: 执行工具并返回字符串结果 try: result self.func(**kwargs) return str(result) except Exception as e: return fError executing tool {self.name}: {str(e)} # 具体的工具实现 def calculator(a: float, b: float, operation: str) - float: 一个简单的计算器工具。 if operation add: return a b elif operation subtract: return a - b elif operation multiply: return a * b elif operation divide: if b 0: return Error: Division by zero return a / b else: return fError: Unsupported operation {operation} def get_current_time(format: str %Y-%m-%d %H:%M:%S) - str: 获取当前时间。 now datetime.now() return now.strftime(format) # 工具注册表 class ToolRegistry: def __init__(self): self._tools: Dict[str, Tool] {} def register(self, tool: Tool): self._tools[tool.name] tool def get_tool(self, name: str) - Tool: return self._tools.get(name) def get_tool_list_for_llm(self) - List[Dict]: 生成给LLM看的工具描述列表 tools_for_llm [] for tool in self._tools.values(): tools_for_llm.append({ type: function, function: { name: tool.name, description: tool.description, parameters: tool.parameters } }) return tools_for_llm # 初始化工具注册表并注册工具 registry ToolRegistry() registry.register(Tool( namecalculator, descriptionPerform a basic arithmetic operation on two numbers., parameters{ type: object, properties: { a: {type: number, description: The first number}, b: {type: number, description: The second number}, operation: { type: string, enum: [add, subtract, multiply, divide], description: The arithmetic operation to perform } }, required: [a, b, operation] }, funccalculator )) registry.register(Tool( nameget_current_time, descriptionGet the current date and time., parameters{ type: object, properties: { format: { type: string, description: Python strftime format string. Default is %Y-%m-%d %H:%M:%S, default: %Y-%m-%d %H:%M:%S } }, required: [] }, funcget_current_time ))实现要点与注意事项参数 Schema 是关键parameters字段必须严格按照 JSON Schema 格式定义。这是 LLM 能正确理解如何调用工具的前提。描述要清晰枚举类型enum和必填字段required要明确。工具执行结果必须为字符串LLM 处理的是文本。即使工具内部计算返回数字最终也要转换成字符串。这统一了数据接口。错误处理在Tool.execute()方法中包裹了异常捕获确保任何工具错误都不会导致整个 Agent 崩溃而是以错误信息的形式返回让 LLM 知晓。注册表的中心化ToolRegistry提供了统一的管理和查询入口并为后续生成 LLM 可识别的工具列表提供了方法。3.2 模块二与大语言模型LLM的通信这是 Agent 的“大脑”。我们需要一个稳定的方式来发送请求并解析响应。import requests import os class LLMClient: def __init__(self, base_url: str, api_key: str, model: str): 初始化LLM客户端。 以 OpenAI 兼容 API 为例如 DeepSeek, Qwen 等。 self.base_url base_url.rstrip(/) self.api_key api_key self.model model self.headers { Authorization: fBearer {self.api_key}, Content-Type: application/json } def chat_completion(self, messages: List[Dict], tools: List[Dict] None) - Dict[str, Any]: 发送聊天补全请求支持工具调用。 payload { model: self.model, messages: messages, stream: False # 为简化关闭流式输出 } if tools: payload[tools] tools # 关键参数让模型在需要时主动选择调用工具 payload[tool_choice] auto try: # 注意端点路径通常是 /v1/chat/completions response requests.post( f{self.base_url}/v1/chat/completions, headersself.headers, jsonpayload, timeout30 ) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: print(fLLM API请求失败: {e}) # 在实际项目中这里应有更完善的降级或重试逻辑 return {choices: [{message: {content: 抱歉我暂时无法思考。, role: assistant}}]} # 示例配置请替换为你的实际配置 # 例如使用 DeepSeek 的 API llm_client LLMClient( base_urlhttps://api.deepseek.com, api_keyos.getenv(DEEPSEEK_API_KEY, your-api-key-here), modeldeepseek-chat )实操心得与避坑指南API 兼容性是第一道坎虽然都声称兼容 OpenAI但不同服务商在端点路径、参数支持如tool_choice上可能有细微差别。首次对接时务必先用一个最简单的纯文本对话请求测试通路。tool_choice参数设置为auto是精髓。它告诉模型“你自己判断是否需要调用工具”。如果设为none模型即使知道有工具也不会调用如果指定某个工具名则会强制调用这适用于特定场景。超时与重试网络请求是不稳定的。在生产环境中必须设置合理的超时如30秒并实现重试机制例如对5xx错误重试2次。我们示例中仅做了简单异常处理实际项目需要加强。API Key 管理永远不要将 API Key 硬编码在代码中。使用环境变量os.getenv或安全的配置管理服务。3.3 模块三Agent Loop 引擎的实现这是整个系统最核心的部分它将上述模块串联起来实现思考-行动的循环。class SimpleAgent: def __init__(self, llm_client: LLMClient, tool_registry: ToolRegistry, max_iterations: int 10): self.llm llm_client self.tools tool_registry self.max_iterations max_iterations # 防止无限循环 self.conversation_history: List[Dict] [] def _parse_tool_call(self, message: Dict) - Dict: 解析LLM响应中的工具调用请求。 # 响应结构可能因不同API提供商略有差异这里处理通用情况 tool_calls message.get(tool_calls) if not tool_calls: return None # 通常一次只处理一个工具调用简化场景 first_call tool_calls[0] return { id: first_call.get(id), name: first_call[function][name], arguments: json.loads(first_call[function][arguments]) } def run(self, user_input: str) - str: 运行Agent处理用户输入返回最终结果。 print(f\n[用户] {user_input}) # 1. 初始化对话历史加入用户消息 self.conversation_history [{role: user, content: user_input}] # 2. 开始循环 for iteration in range(self.max_iterations): print(f\n--- 循环迭代 {iteration 1} ---) # 2.1 准备工具列表仅在需要时发送减少token消耗 tools_for_llm self.tools.get_tool_list_for_llm() # 注意并非每次循环都必须发送工具列表。一种优化策略是只在历史记录中首次出现工具调用需求时发送。 # 这里为简单起见每次循环都发送。 # 2.2 调用LLM进行“思考” print(f 思考中... (历史消息数: {len(self.conversation_history)})) response self.llm.chat_completion(self.conversation_history, tools_for_llm) assistant_message response[choices][0][message] self.conversation_history.append(assistant_message) # 记录助手的原始响应 # 2.3 检查是否需要“行动”调用工具 tool_call self._parse_tool_call(assistant_message) if tool_call: # 需要调用工具 tool_name tool_call[name] tool_args tool_call[arguments] print(f 决定调用工具: {tool_name}, 参数: {tool_args}) # 查找并执行工具 tool self.tools.get_tool(tool_name) if tool: tool_result tool.execute(**tool_args) print(f 工具执行结果: {tool_result}) # 2.4 将工具执行结果作为新的“观察”加入历史 # 格式非常重要role必须是“tool”并包含对应的tool_call_id self.conversation_history.append({ role: tool, tool_call_id: tool_call[id], name: tool_name, content: tool_result }) # 循环继续进入下一轮“思考” continue else: print(f 错误未找到工具 {tool_name}) # 可以将错误信息反馈给LLM这里简单处理为终止 return f系统错误请求的工具 {tool_name} 不存在。 # 2.5 如果不需要调用工具则LLM给出了最终答案 final_answer assistant_message.get(content, ) print(f 生成最终答案。) return final_answer # 3. 循环超过最大次数强制终止 return f经过 {self.max_iterations} 轮尝试仍未完成可能任务过于复杂或陷入循环。 # 组装并运行Agent agent SimpleAgent(llm_client, registry, max_iterations5)核心机制深度解析消息历史的维护conversation_history是这个循环的“状态存储器”。它严格按照[user, assistant, tool, user, assistant...]这样的顺序记录每一次交互。LLM 正是基于这个完整的历史上下文来决定下一步行动。循环的驱动力for循环和continue语句是实现循环的关键。当检测到工具调用时我们添加工具结果后用continue直接跳转到下一轮循环再次调用 LLM。只有当 LLM 的响应中不包含工具调用时循环才结束返回最终内容。工具结果的格式将工具结果加入历史时role必须设为tool并且tool_call_id必须与之前 LLM 请求中的id对应。这是 OpenAI 制定的规范用于将工具执行结果与特定的调用请求关联起来确保上下文连贯。安全护栏max_iterations参数至关重要。它防止了因逻辑错误或模型“鬼打墙”导致的无限循环是生产系统必备的防护措施。4. 全链路实操与效果演示让我们用几个具体的例子让这个 Agent 真正跑起来观察其内部状态的全链路流转。4.1 场景一简单计算任务我们问“123乘以456等于多少”result agent.run(123乘以456等于多少) print(f\n[最终答案] {result})预期的内部日志与状态流转[用户] 123乘以456等于多少 --- 循环迭代 1 --- 思考中... (历史消息数: 1) 决定调用工具: calculator, 参数: {a: 123, b: 456, operation: multiply} 工具执行结果: 56088 --- 循环迭代 2 --- 思考中... (历史消息数: 3) 生成最终答案。 [最终答案] 123乘以456等于56088。链路拆解迭代1LLM 看到用户问题结合可用的工具描述判断需要计算。它生成一个格式正确的tool_calls请求指定使用calculator工具并正确解析出了数字和“乘”对应的操作multiply。状态更新历史记录变为[用户问题 assistant的tool_call请求 tool的执行结果]。迭代2LLM 看到工具返回的结果56088判断信息已完备无需再调用工具于是生成最终的自然语言答案循环结束。4.2 场景二多步骤混合任务我们问“现在几点了如果是下午就告诉我123加456的结果。”result agent.run(现在几点了如果是下午就告诉我123加456的结果。) print(f\n[最终答案] {result})预期的内部日志与状态流转[用户] 现在几点了如果是下午就告诉我123加456的结果。 --- 循环迭代 1 --- 思考中... (历史消息数: 1) 决定调用工具: get_current_time, 参数: {format: %Y-%m-%d %H:%M:%S} 工具执行结果: 2024-05-27 14:30:15 --- 循环迭代 2 --- 思考中... (历史消息数: 3) 决定调用工具: calculator, 参数: {a: 123, b: 456, operation: add} 工具执行结果: 579 --- 循环迭代 3 --- 思考中... (历史消息数: 5) 生成最终答案。 [最终答案] 现在是2024-05-27 14:30:15是下午。123加456的结果是579。链路拆解迭代1LLM 理解任务的第一步是获取时间调用get_current_time。迭代2LLM 收到时间结果“14:30:15”根据上下文判断是下午于是执行下一步指令调用calculator进行加法运算。迭代3LLM 收到计算结果579结合之前的时间信息组织成一句连贯的回复任务完成。这个例子完美展示了 Agent Loop 如何让 LLM 像“程序员”一样逐步分解并执行一个条件性任务。4.3 场景三无需工具的直接回答我们问“你好请介绍一下你自己。”result agent.run(你好请介绍一下你自己。) print(f\n[最终答案] {result})预期的内部日志与状态流转[用户] 你好请介绍一下你自己。 --- 循环迭代 1 --- 思考中... (历史消息数: 1) 生成最终答案。 [最终答案] 你好我是一个AI智能助手能够通过调用工具来帮助你完成计算、查询时间等任务。我的核心是一个大型语言模型通过循环思考的过程来理解你的需求并采取行动。有什么可以帮你的吗链路拆解LLM 分析问题发现这是一个无需调用工具的纯对话问题因此直接在第一次迭代中就生成最终答案循环仅执行一次。这体现了 Agent 的灵活性只在必要时才动用工具。5. 常见问题、调试技巧与进阶思考即使在这个最小系统中你也会遇到各种问题。下面是我在实践和教学中总结的常见坑点与解决方案。5.1 问题一LLM 不调用工具直接猜测答案现象问“123*456等于多少”LLM 直接输出一个猜测的数字可能对也可能错而不是调用计算器。排查思路检查工具描述工具的描述description是否清晰参数schema定义是否准确模糊的描述会导致 LLM 不理解工具的用途。尝试将描述写得更具体如“用于对两个数字进行精确的算术运算支持加、减、乘、除”。检查系统提示词System Prompt我们目前的实现没有显式添加系统消息。一个强大的技巧是在conversation_history的开头插入一条系统消息明确指示模型“你是一个可以调用工具的助手。当用户的问题涉及计算、查询实时信息等需要精确结果或外部能力的任务时你必须调用相应的工具来完成不要自行猜测。” 这能极大提高工具调用的主动性。检查 API 参数确认调用chat_completion时tools参数正确传入了工具列表并且tool_choice设置为auto。模型能力某些较小的或未经专门微调的模型工具调用能力可能较弱。可以尝试更换更主流的模型如 GPT-4, DeepSeek-V2, Qwen-Max。5.2 问题二工具调用参数解析错误现象LLM 决定调用工具但生成的参数 JSON 格式错误或者参数值类型不对例如把数字传成了字符串。排查思路强化 Schema 定义在参数的description里明确指定类型例如“type”: “number”。对于枚举值使用enum字段。输出调试在_parse_tool_call函数中将解析前的assistant_message和解析后的tool_call都打印出来检查原始输出。有时 LLM 会在 JSON 外包裹 Markdown 代码块或多余的解释文字需要更健壮的解析逻辑来处理。后处理清洗在解析 JSON 前可以尝试用正则表达式提取arguments字符串中的 JSON 部分或者使用json.loads()的异常处理失败时尝试一些简单的字符串修复。5.3 问题三陷入无限循环或无效循环现象Agent 在两个工具间来回调用或者反复调用同一个工具但无法推进任务。排查思路检查最大迭代次数确保设置了max_iterations如10这是最后的安全网。分析对话历史将每一轮迭代后的conversation_history完整打印出来。观察 LLM 在收到工具结果后是否做出了合理的下一步决策。问题可能出在工具结果的信息量不足或者 LLM 对任务的理解有偏差。引入思维链Chain-of-Thought在系统提示词中鼓励模型“逐步思考”。例如在调用工具前让模型在content字段中先输出它的推理过程虽然这部分内容不会给用户看这有时能提高决策质量。工具设计检查工具是否具有“幂等性”调用是否产生了副作用工具返回的结果是否清晰、无歧义5.4 进阶思考从这个最小系统出发当你彻底理解了这个最小系统你就拥有了理解和构建更复杂 Agent 的钥匙规划Planning当前的 Agent 是“反应式”的走一步看一步。高级 Agent 会在循环开始前或过程中先制定一个计划Plan例如“第一步获取时间第二步判断上下午第三步如果是下午则计算。” 这能提高复杂任务的成功率。记忆Memory我们的conversation_history就是一种简单的短期记忆。可以引入向量数据库来实现长期记忆让 Agent 记住跨对话的关键信息。工具检索Tool Retrieval当工具有成百上千个时不能每次都把所有工具描述发给 LLMToken 消耗大且干扰模型。需要根据当前对话动态检索最相关的几个工具。多智能体协作多个这样的 SimpleAgent 实例可以组合起来每个负责特定领域通过一个协调器Orchestrator来分配任务和整合结果解决更宏大的问题。这个“最小 Agent”就像乐高积木的基础颗粒。所有令人眼花缭乱的 Agent 框架和应用本质上都是在这些核心机制之上增加了更便捷的抽象、更强大的组件和更优的工程实践。理解了这个循环和链路你就掌握了智能体最本质的脉搏。