从零手写一个Agent框架-理解Agent的本质
本章不使用任何框架，从零编写一个能实际运行的 Agent——大约 120 行 Python，包含工具注册、LLM 调用、循环逻辑、错误处理。通过手写实现，可以清晰理解 LangChain（一个构建 LLM 应用的开源 Python 框架，封装了 Agent 循环、工具调用、链式执行等复杂逻辑）那个循环到底在干什么，以及框架封装了哪些工程问题。
从零手写一个 Agent 框架：理解 Agent 的本质
本章不使用任何框架，从零编写一个能实际运行的 Agent——大约 120 行 Python，包含工具注册、LLM 调用、循环逻辑、错误处理。通过手写实现，可以清晰理解 LangChain（一个构建 LLM 应用的开源 Python 框架，封装了 Agent 循环、工具调用、链式执行等复杂逻辑）那个循环到底在干什么，以及框架封装了哪些工程问题。
1.1 Agent 的本质是什么
极简 Agent 框架的四大组成——LLM 核心、工具注册表、执行循环、记忆模块
先把本质说清楚，再写代码。
一个 LLM-based Agent，本质上是三件事的组合：
一个循环：持续运行，直到任务完成或达到上限。每次迭代，模型要么输出最终答案（退出循环），要么输出工具调用请求（继续循环）。
工具调用：模型通过 function calling 告诉外部系统"我要调用某个工具，参数是这些"。外部系统执行工具，把结果返回给模型。
状态管理：所有的上下文——用户输入、模型输出、工具调用请求、工具执行结果——都存在一个叫 messages 的列表里。每次循环都把新信息追加进去，模型每次都能看到完整历史。
就这三件事。LangChain 的 AgentExecutor，LangGraph（LangChain 旗下专门用于构建有状态 Agent 工作流的框架）的 StateGraph，CrewAI（一个专注于多 Agent 协作的 Python 框架，让多个 Agent 像团队成员一样分工合作）的 Crew，底层都在做这三件事，只是在上面加了更多抽象层。
1.2 手写 Agent：完整代码
# agent_from_scratch.py
# 依赖：pip install openai
# 需要设置环境变量：OPENAI_API_KEY
import json
import inspect
import os
from typing import Any, Callable
from openai import OpenAI
client = OpenAI()
# ─────────────────────────────────────────
# 第一部分：工具注册系统
# ─────────────────────────────────────────
# 全局工具注册表
_tool_registry: dict[str, dict] = {}
def tool(func: Callable) -> Callable:
"""
工具注册装饰器。
读取函数签名和 docstring，自动生成 OpenAI function calling 所需的 schema。
"""
sig = inspect.signature(func)
properties = {}
required = []
for param_name, param in sig.parameters.items():
# 从类型注解提取类型
annotation = param.annotation
if annotation == inspect.Parameter.empty:
param_type = "string"
elif annotation == int:
param_type = "integer"
elif annotation == float:
param_type = "number"
elif annotation == bool:
param_type = "boolean"
else:
param_type = "string"
properties[param_name] = {"type": param_type}
# 没有默认值的参数是必填项
if param.default == inspect.Parameter.empty:
required.append(param_name)
# 把函数信息存入注册表
_tool_registry[func.__name__] = {
"function": func,
"schema": {
"type": "function",
"function": {
"name": func.__name__,
"description": func.__doc__ or "",
"parameters": {
"type": "object",
"properties": properties,
"required": required,
}
}
}
}
return func
def get_tools_schema() -> list[dict]:
"""返回所有注册工具的 schema，用于传给 OpenAI API"""
return [info["schema"] for info in _tool_registry.values()]
def execute_tool(name: str, arguments: dict) -> Any:
"""根据工具名和参数执行工具，返回结果"""
if name not in _tool_registry:
raise ValueError(f"工具 '{name}' 未注册")
func = _tool_registry[name]["function"]
return func(**arguments)
# ─────────────────────────────────────────
# 第二部分：工具定义
# ─────────────────────────────────────────
@tool
def get_weather(city: str) -> str:
"""查询指定城市的当前天气和温度。"""
# 实际项目中这里调用真实天气 API
weather_data = {
"北京": "晴，22°C，东南风 3 级，空气质量良",
"上海": "多云，25°C，东风 2 级，空气质量优",
"广州": "小雨，28°C，南风 4 级，空气质量中",
"深圳": "晴，27°C，西南风 3 级，空气质量良",
}
return weather_data.get(city, f"抱歉，暂无 {city} 的天气数据")
@tool
def calculate(expression: str) -> str:
"""
计算数学表达式。支持加减乘除、幂运算。
示例：'2 + 3 * 4'、'(10 + 5) / 3'、'2 ** 10'
"""
try:
# 只允许安全的数学运算
allowed = set("0123456789+-*/().**% ")
if not all(c in allowed for c in expression):
return "错误：不支持的字符，只允许数字和基本运算符"
result = eval(expression, {"__builtins__": {}})
return f"{expression} = {result}"
except ZeroDivisionError:
return "错误：除数不能为零"
except Exception as e:
return f"计算出错：{str(e)}"
@tool
def search_knowledge(query: str) -> str:
"""搜索知识库，查找相关信息。"""
# 模拟知识库查询
knowledge = {
"python": "Python 是一种高级编程语言，以简洁易读著称，广泛用于 AI/ML、Web 开发、自动化脚本等领域。",
"agent": "AI Agent 是一种能感知环境、做决策、执行行动的自主系统，通过 LLM 提供推理能力。",
"langchain": "LangChain 是一个构建 LLM 应用的开源框架，提供工具链、记忆、Agent 等抽象。",
}
query_lower = query.lower()
for key, value in knowledge.items():
if key in query_lower:
return value
return f"知识库中未找到关于 '{query}' 的相关信息"
# ─────────────────────────────────────────
# 第三部分：Agent 核心循环
# ─────────────────────────────────────────
class AgentFromScratch:
def __init__(
self,
model: str = "gpt-4o",
max_iterations: int = 10,
system_prompt: str = None,
):
self.model = model
self.max_iterations = max_iterations
self.system_prompt = system_prompt or (
"你是一个有帮助的 AI 助手。当需要时，使用提供的工具来完成任务。"
"工具调用完成后，根据工具返回的结果给出最终答案。"
)
def run(self, user_input: str, verbose: bool = True) -> str:
"""
运行 Agent，返回最终答案。
verbose=True 时打印每步执行过程。
"""
# 初始化 messages：系统 prompt + 用户输入
messages = [
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": user_input},
]
if verbose:
print(f"\n用户输入：{user_input}")
print("─" * 50)
# Agent 循环
for iteration in range(self.max_iterations):
if verbose:
print(f"\n[迭代 {iteration + 1}] 调用 LLM...")
# 调用 LLM
response = client.chat.completions.create(
model=self.model,
messages=messages,
tools=get_tools_schema(),
tool_choice="auto", # 让模型自己决定要不要调工具
)
message = response.choices[0].message
# 情况一：模型决定直接回答，不调工具——退出循环
if not message.tool_calls:
if verbose:
print(f"\n最终回答：{message.content}")
return message.content
# 情况二：模型决定调用工具
# 先把模型的回应（含工具调用请求）加入 messages
messages.append(message)
if verbose:
print(f"模型请求调用 {len(message.tool_calls)} 个工具")
# 执行每一个工具调用
for tool_call in message.tool_calls:
tool_name = tool_call.function.name
tool_args = json.loads(tool_call.function.arguments)
if verbose:
print(f" -> 调用工具：{tool_name}({tool_args})")
# 执行工具，处理错误
try:
tool_result = execute_tool(tool_name, tool_args)
result_str = str(tool_result)
except Exception as e:
result_str = f"工具执行失败：{str(e)}"
if verbose:
print(f" 结果：{result_str}")
# 把工具执行结果加入 messages
# 注意：tool_call_id 必须和请求里的 id 对应
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result_str,
})
# 工具执行完毕，继续循环，让模型基于结果决定下一步
# 达到最大迭代次数，强制返回
if verbose:
print(f"\n达到最大迭代次数 ({self.max_iterations})，强制返回")
return "抱歉，任务处理超出最大步骤限制，未能完成。"
# ─────────────────────────────────────────
# 第四部分：运行示例
# ─────────────────────────────────────────
if __name__ == "__main__":
agent = AgentFromScratch(model="gpt-4o", max_iterations=10)
# 测试 1：单工具调用
agent.run("北京今天天气怎么样？需要带雨伞吗？")
print("\n" + "=" * 60 + "\n")
# 测试 2：数学计算
agent.run("帮我计算 (2 ** 8 + 100) / 3 的结果，保留两位小数")
print("\n" + "=" * 60 + "\n")
# 测试 3：多工具调用
agent.run("北京和上海今天的天气分别是什么？哪个城市更适合户外运动？")
print("\n" + "=" * 60 + "\n")
# 测试 4：不需要调工具的问题
agent.run("你好，能简单介绍一下你自己吗？")1.3 执行流程
1.4 理解 messages 的结构
messages 列表是整个 Agent 的"工作记忆"，理解它的结构就理解了一半 Agent 的工作原理。
一次完整的工具调用交互，messages 里会出现四种角色：
messages = [
# 系统指令
{"role": "system", "content": "你是一个助手..."},
# 用户问题
{"role": "user", "content": "北京今天天气怎样？"},
# 模型决定调工具（注意：tool_calls 字段，content 可以为 None）
{
"role": "assistant",
"content": None,
"tool_calls": [{
"id": "call_abc123",
"type": "function",
"function": {
"name": "get_weather",
"arguments": '{"city": "北京"}'
}
}]
},
# 工具执行结果（tool_call_id 必须和上面的 id 对应）
{
"role": "tool",
"tool_call_id": "call_abc123",
"content": "晴，22°C，东南风 3 级"
},
# 模型基于工具结果给出最终回答
{
"role": "assistant",
"content": "北京今天天气晴朗，气温 22°C，东南风，不需要带雨伞。"
}
]每次循环，都在往这个列表里追加新内容。模型每次调用都能看到完整历史，这就是 Agent 能"记住"之前做了什么的原因。
1.5 手写版的局限，以及为什么需要框架
这段代码能跑，但和生产级的框架比，缺了很多东西。
没有流式输出。现在用户要等 Agent 全部执行完才能看到结果。LangChain/LangGraph 支持流式推送，每生成一个 token 就推给前端，用户体验好得多。
没有可观测性。这段代码用 print 打印执行过程，生产环境里需要结构化的日志、追踪每一次 LLM 调用的延迟和 token 消耗、在 Langfuse 或 LangSmith 上可视化整个执行链路。
没有持久化。每次调用 agent.run() 都是全新的，没有多轮对话记忆，没有把执行状态保存到数据库。LangGraph 的 Checkpoint 机制就是解决这个问题的。
没有并发控制。多工具调用时，串行执行每个工具。如果同时查北京和上海的天气，完全可以并行调用，节省时间。
没有 Human-in-the-loop。某些关键操作（比如发送邮件、提交订单）执行前应该等人工确认。LangGraph 的 interrupt 机制就是做这个的。
这就是为什么框架存在——不是为了藏起复杂性，而是把这些通用的工程问题统一解决好，让开发者专注在业务逻辑上。
但如果不理解底层的循环是怎么工作的，就没法调试框架出的问题。Agent 在生产环境里出现奇怪行为时——比如工具调用一直循环不退出、或者模型明明有工具却不调——不知道该去哪里找原因。
知道 AgentExecutor 底层是在维护一个 messages 列表、每次循环追加内容，就知道去检查 messages 的内容。知道模型的退出条件是"不再输出 tool_calls"，就知道去检查 system prompt 里有没有说清楚什么时候应该直接回答。
框架用起来，原理懂了，出了问题才知道去哪里找。