从零手搓一个Agent：核心循环、工具调用与状态管理实战

1. 为什么我要自己动手写一个 Agent

市面上关于 Agent 的文章，十篇里有八篇在讲概念——什么自主规划、工具调用、记忆机制，翻来覆去就那几个词。但真正落到代码层面，很多人就卡住了：Agent 的循环到底怎么写？工具调用的结果怎么塞回上下文？多轮对话的状态存在哪里？这些问题，光看架构图是看不出来的。

我最初也是被这些问题困扰，后来看到一个叫 pi-agent 的轻量实现，代码量不大，但把 Agent 的核心骨架搭得很清楚。于是我决定参照它的思路，从零手搓一个小 Agent，不依赖任何重型框架，纯靠基础库把整个链路跑通。这篇文章就是整个过程的完整记录——包括设计思路、关键代码、踩过的坑，以及我对每一步“为什么这么做”的理解。

这篇文章适合两类人：一是已经用过某些 Agent 框架但想搞清楚底层原理的开发者；二是想自己写一个 Agent 但不知道从哪下手的新手。我会尽量把每个环节讲透，代码可以直接抄作业，也可以根据自己需求改。

2. 整体设计：一个 Agent 到底需要什么

2.1 核心循环：Agent 的心跳

Agent 和普通的聊天机器人最大的区别在于：它不是“你问一句我答一句”就结束了，而是会根据你的目标，自己决定下一步做什么，调用工具，拿到结果，再决定下一步，直到任务完成或者达到终止条件。

这个“决定-执行-观察-再决定”的过程，就是 Agent 的核心循环。用伪代码表示大概是这样：

while not done: response = llm.chat(messages) if response.has_tool_call: result = execute_tool(response.tool_call) messages.append(result) else: done = True

看起来很简单对吧？但魔鬼在细节里。比如：怎么判断任务完成了？工具调用失败了怎么办？循环次数太多怎么兜底？这些后面会逐一展开。

2.2 模块拆解：四个核心组件

参照 pi-agent 的设计，我把整个 Agent 拆成了四个模块：

• LLM 接口层：负责和模型通信，发送消息、接收回复。这一层要做的是把不同模型的 API 差异抹平，上层不需要关心用的是哪家模型。
• 工具注册与调度：定义工具的描述格式，解析模型的工具调用请求，执行对应的函数，把结果格式化后返回。
• 对话状态管理：维护消息历史，包括用户输入、模型回复、工具调用记录、工具返回结果。这是 Agent 的“记忆”。
• 主循环控制器：串联以上所有模块，控制循环的启动、终止和异常处理。

为什么这么拆？因为这样每个模块都可以独立替换。比如你想换个模型，只改 LLM 接口层就行；想加新工具，只改工具注册表就行。这种解耦在实际开发中非常关键，否则代码会很快变成一团乱麻。

2.3 技术选型：为什么不用现成框架

你可能会问：LangChain、AutoGPT 这些框架不香吗？为什么要自己写？

我的理由有三个。第一，学习目的——用框架你只能学到怎么调 API，自己写才能理解每一步在干什么。第二，可控性——框架的抽象层有时候会挡住关键细节，出了问题很难排查。第三，轻量需求——很多场景下你只需要一个简单的 Agent，引入重型框架反而增加了复杂度和不确定性。

当然，如果你是要做生产级的产品，该用框架还是得用。但作为学习和理解原理的手段，自己动手写一遍是最有效的。

3. 核心细节：每个模块怎么实现

3.1 LLM 接口层：统一通信协议

这一层的核心任务是：给定一组消息，返回模型的回复。回复可能是纯文本，也可能包含工具调用请求。

我选择用 OpenAI 兼容的接口格式作为标准，因为目前大部分模型服务都支持这个格式。核心代码大概长这样：

import httpx import json class LLMClient: def __init__(self, base_url, api_key, model): self.base_url = base_url self.api_key = api_key self.model = model def chat(self, messages, tools=None): payload = { "model": self.model, "messages": messages, } if tools: payload["tools"] = tools response = httpx.post( f"{self.base_url}/chat/completions", headers={"Authorization": f"Bearer {self.api_key}"}, json=payload, timeout=60, ) return response.json()["choices"][0]["message"]

这里有几个关键点需要注意。

超时设置。Agent 场景下，模型可能需要较长时间来推理，尤其是带工具调用的复杂任务。我一开始设了 30 秒，结果经常超时。后来改成 60 秒，基本够用。如果你的场景更复杂，可以设到 120 秒。

错误处理。网络请求失败、API 返回错误码、返回格式不符合预期——这些都要处理。我的做法是封装一个重试机制，最多重试 3 次，每次间隔递增。如果连续失败，就抛出一个明确的异常，让主循环决定怎么处理。

消息格式。OpenAI 的消息格式里，每条消息有role和content两个核心字段。role可以是system、user、assistant、tool。工具调用的结果需要用tool角色返回，并且带上tool_call_id来对应是哪次调用。这个细节很容易搞错，搞错了模型就不知道工具返回的结果对应哪个请求。

3.2 工具注册：让模型知道它能做什么

工具是 Agent 能力的延伸。没有工具，Agent 就只是一个聊天机器人；有了工具，它才能查数据、算数学、读写文件、调用外部服务。

工具的定义遵循 OpenAI 的 function calling 格式：

tools = [ { "type": "function", "function": { "name": "calculator", "description": "执行数学计算，支持加减乘除和幂运算", "parameters": { "type": "object", "properties": { "expression": { "type": "string", "description": "数学表达式，如 '2 + 3 * 4'" } }, "required": ["expression"] } } } ]

然后在本地维护一个名称到函数的映射：

TOOL_REGISTRY = {} def register_tool(name, func): TOOL_REGISTRY[name] = func def execute_tool(name, arguments): if name not in TOOL_REGISTRY: return f"错误：未找到工具 {name}" try: result = TOOL_REGISTRY[name](**arguments) return str(result) except Exception as e: return f"工具执行失败：{e}"

这里有个经验：工具的 description 写得越清楚，模型调用得越准确。我一开始写得很简略，结果模型经常传错参数或者在不该调用的时候调用。后来我把每个工具的描述都写成了“什么场景下用、参数是什么意思、返回什么格式”，准确率明显提升。

另外，参数校验不能省。模型有时候会传一些奇怪的参数，比如该传数字的传了字符串，该传数组的传了单个值。在工具函数内部做好类型检查和转换，能避免很多莫名其妙的报错。

3.3 对话状态管理：Agent 的记忆

消息历史是 Agent 的短期记忆。每次循环，我们都要把完整的消息历史发给模型，模型才能知道之前发生了什么。

class Conversation: def __init__(self, system_prompt): self.messages = [ {"role": "system", "content": system_prompt} ] def add_user_message(self, content): self.messages.append({"role": "user", "content": content}) def add_assistant_message(self, message): self.messages.append(message) def add_tool_result(self, tool_call_id, content): self.messages.append({ "role": "tool", "tool_call_id": tool_call_id, "content": content })

看起来简单，但有两个坑我踩过。

第一个坑：消息太长。Agent 循环几轮之后，消息历史会变得很长，可能超出模型的上下文窗口。解决办法有两种：一是截断最早的消息（但要保留 system prompt），二是对历史消息做摘要压缩。我目前用的是简单截断，保留最近 N 条消息。如果任务复杂，建议上摘要方案。

第二个坑：工具调用消息的完整性。当模型返回一个工具调用请求时，assistant 消息里会包含tool_calls字段。这条消息必须原样保存，不能只保存文本内容。否则下一轮发给模型时，模型会发现工具调用的结果没有对应的请求，直接报错。

3.4 主循环：把所有东西串起来

主循环是整个 Agent 的引擎。它的逻辑是：

1. 把当前消息历史发给模型
2. 检查模型回复中是否有工具调用
3. 如果有，执行工具，把结果加入消息历史，回到步骤 1
4. 如果没有，输出模型的文本回复，结束

def run_agent(conversation, llm_client, tools, max_iterations=10): for i in range(max_iterations): response = llm_client.chat(conversation.messages, tools=tools) conversation.add_assistant_message(response) if not response.get("tool_calls"): return response.get("content", "") for tool_call in response["tool_calls"]: name = tool_call["function"]["name"] args = json.loads(tool_call["function"]["arguments"]) result = execute_tool(name, args) conversation.add_tool_result(tool_call["id"], result) return "达到最大循环次数，任务未完成"

max_iterations是一个安全阀。没有它，模型可能会陷入无限循环——比如反复调用同一个工具但每次都得不到想要的结果。我一般设 10 次，大部分任务 3-5 次循环就能完成。

4. 实操过程：从零跑通一个完整 Agent

4.1 环境准备与依赖安装

整个项目只需要三个依赖：

pip install httpx json5

httpx用来发 HTTP 请求，json5用来解析模型返回的 JSON（有时候模型会返回带注释或尾逗号的 JSON，标准库解析不了）。

如果你用的模型服务需要额外的 SDK，按需安装即可。但我建议尽量用 HTTP 接口直接调用，少一层封装少一层坑。

4.2 定义工具函数

我先定义两个最基础的工具：计算器和当前时间查询。

import math from datetime import datetime def calculator(expression: str) -> str: allowed = set("0123456789+-*/.() ") if not all(c in allowed for c in expression): return "错误：表达式包含不允许的字符" try: result = eval(expression, {"__builtins__": {}}, {"math": math}) return str(result) except Exception as e: return f"计算错误：{e}" def get_current_time(timezone: str = "local") -> str: now = datetime.now() return now.strftime("%Y-%m-%d %H:%M:%S")

calculator里我做了字符白名单校验，只允许数字和基本运算符。这是为了防止模型生成恶意表达式。虽然在自己本地跑风险不大，但养成好习惯没坏处。

4.3 组装并运行

把前面所有模块拼起来：

SYSTEM_PROMPT = """你是一个智能助手，可以使用工具来帮助用户解决问题。 当需要计算时，使用 calculator 工具。 当需要知道当前时间时，使用 get_current_time 工具。 请一步一步思考，先理解用户需求，再决定是否需要调用工具。""" def main(): llm = LLMClient( base_url="你的模型服务地址", api_key="你的密钥", model="你的模型名称" ) register_tool("calculator", calculator) register_tool("get_current_time", get_current_time) conversation = Conversation(SYSTEM_PROMPT) conversation.add_user_message("帮我算一下 (15 + 27) * 3 等于多少，然后告诉我现在几点") result = run_agent(conversation, llm, tools) print(result) if __name__ == "__main__": main()

运行之后，你会看到 Agent 先调用 calculator 算出 126，再调用 get_current_time 获取时间，最后把两个结果整合成一段完整的回复。整个过程不需要你手动干预，Agent 自己决定调用哪些工具、按什么顺序调用。

4.4 参数选择与调优记录

在实际跑的过程中，我调整了几个参数，记录一下：

这些值不是固定的，你需要根据自己的场景调整。比如如果你的任务很简单，最大循环次数设 5 就够了；如果任务特别复杂，可能要设到 20。

5. 常见问题与排查技巧

5.1 模型不调用工具怎么办

这是最常见的问题。模型要么直接用自己的知识回答，要么说“我无法完成这个任务”。

排查思路：

• 检查工具描述是否清晰。模型是根据 description 来判断什么时候该用工具的。如果描述太模糊，模型就不知道什么时候该调用。
• 检查 system prompt。在 system prompt 里明确告诉模型“你有以下工具可用，遇到 X 场景请使用 Y 工具”。
• 检查模型是否支持 function calling。不是所有模型都支持工具调用。如果不支持，你需要在 prompt 里让模型输出特定格式的文本，然后自己解析。

5.2 工具调用参数错误

模型传的参数类型不对、缺少必填参数、参数名拼错——这些都会导致工具执行失败。

我的做法是在工具函数入口做严格的参数校验，返回明确的错误信息。这样模型在下一轮看到错误信息后，有机会自我修正。比如模型传了{"expr": "1+1"}但参数名应该是expression，工具返回“缺少必填参数 expression”，模型下一轮就会传对。

5.3 循环停不下来

有时候模型会反复调用同一个工具，每次都得到相同的结果，但就是不结束。

解决办法有两个：一是设max_iterations硬性截断；二是在工具返回结果里加入提示，比如“如果这个结果已经满足需求，请直接回复用户，不要再调用工具”。

5.4 常见问题速查表

5.5 独家避坑技巧

技巧一：给工具返回值加前缀。比如计算器返回"计算结果：126"而不是直接返回"126"。这样模型更容易理解这个结果的含义，减少误用。

技巧二：在 system prompt 里加 few-shot 示例。给模型展示一两个完整的“用户提问-工具调用-结果整合”的示例，能显著提升工具调用的准确率。

技巧三：日志要打全。每次请求和响应都完整记录下来，包括消息历史、工具调用参数、工具返回结果。出问题的时候，翻日志比猜原因快得多。

技巧四：工具粒度要适中。太粗的工具（比如一个工具能做十件事）模型不知道怎么传参；太细的工具（比如每个数学运算一个工具）模型要调用很多次。我一般按“一个工具做一类事”的原则来划分。

6. 后续可以怎么扩展

这个最小 Agent 跑通之后，你可以根据需求往几个方向扩展。

加记忆。目前的对话历史只存在内存里，程序退出就没了。你可以加一个持久化层，把消息存到文件或数据库，下次启动时加载回来。

加更多工具。比如网页搜索、文件读写、代码执行、API 调用。每加一个工具，Agent 的能力边界就扩大一圈。

加多 Agent 协作。一个 Agent 负责规划，一个负责执行，一个负责审核。这种模式在复杂任务上效果更好，但实现复杂度也更高。

加流式输出。目前是等模型完整回复后才处理，用户体验不够好。改成流式输出，模型边生成边显示，体验会好很多。

加人工确认。对于危险操作（比如删除文件、发送请求），在工具执行前弹一个确认框，让用户决定是否继续。这在生产环境里是必须的。

我自己在实际使用中发现，最小 Agent 最大的价值不是它能做多少事，而是它让你真正理解了 Agent 的工作原理。一旦你理解了循环、工具调用、状态管理这三个核心概念，再去用任何框架都会觉得理所当然。反过来，如果你直接上框架，遇到问题可能连从哪里排查都不知道。

最后分享一个小技巧：调试 Agent 的时候，把每一轮的消息历史打印出来，你会直观地看到模型是怎么“思考”的。有时候模型的决策逻辑会让你意外，但正是这些意外让你更懂它。