手写一个 AI Agent:从零实现工具调用与思维链
前言
2026年,AI Agent 已经成为大模型应用最热门的落地形态。GPT-5.5 专为 Agent 时代设计,DeepSeek-V4 百万上下文让 Agent 能处理更复杂的任务——但 Agent 的核心架构其实并不神秘。
Agent = 大模型 + 工具调用 + 记忆管理 + 规划推理
拆开来看,每个模块都不复杂。本文就用纯 Python,从零手写一个可运行的 AI Agent,不依赖 LangChain、AutoGPT 等框架,让你真正理解 Agent 的工作原理。
前置知识: 了解 Python 基础,知道什么是 API 调用即可。
使用模型: 代码兼容 OpenAI 兼容接口(支持 Function Calling 的模型均可),可以用 DeepSeek、通义千问、GLM 等。
1. 核心架构:Agent 到底在做什么?
在动手之前,先建立 Agent 的整体认知。一个标准的 ReAct(Reasoning + Acting)模式 Agent 的工作循环是:
用户提问
↓
Agent 思考(Thought)→ 本次要做什么
↓
Agent 行动(Action)→ 调用某个工具(或直接回答)
↓
收到结果(Observation)→ 工具返回了什么
↓
再思考 → 再行动 → 再观察 → ...
↓
最终回答(Final Answer)
这就是 ReAct 模式,由 Google 在 2022 年提出,是当前大多数 Agent 的基础范式。
我们的实现将围绕这个循环展开:
┌───────────────────────────────────────────┐
│ Agent 主循环 │
│ │
│ Thought → Action → Observation │
│ ↕ (重复直到得出答案) │
│ + │
│ 记忆管理 (控制上下文长度) │
└───────────────────────────────────────────┘
2. 第一步:封装 LLM 调用接口
我们先做一个统一的模型调用层,这样后续切换模型时只需改配置。
import json
import os
from typing import Optional
import requests
class LLMClient:
"""大模型调用封装,兼容 OpenAI API 格式"""
def __init__(self, api_key: str, base_url: str, model: str):
self.api_key = api_key
self.base_url = base_url.rstrip("/")
self.model = model
def chat(
self,
messages: list[dict],
tools: Optional[list[dict]] = None,
temperature: float = 0.7,
max_tokens: int = 4096,
) -> dict:
"""调用对话接口"""
payload = {
"model": self.model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
}
if tools:
payload["tools"] = tools
resp = requests.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
},
json=payload,
)
resp.raise_for_status()
return resp.json()["choices"][0]["message"]
使用示例:
# 配置(以 DeepSeek 为例)
llm = LLMClient(
api_key=os.environ["DASHSCOPE_API_KEY"], # 你的 API Key
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1", # 兼容接口地址
model="qwen-plus", # 模型名称
)
response = llm.chat([
{"role": "user", "content": "你好,请介绍一下自己"}
])
print(response["content"])
这里用 requests 而不是 openai 库,是为了零依赖、让代码更透明。
3. 第二步:工具系统——让 Agent 拥有"双手"
Agent 能做事的关键是工具(Tool)。工具就是一段代码,有名字、描述和参数,模型通过函数调用(Function Calling)来触发它。
3.1 工具定义
我们用一个统一的接口来定义工具:
from typing import Callable, Any
class Tool:
"""工具定义"""
def __init__(
self,
name: str,
description: str,
parameters: dict,
func: Callable[..., Any],
):
self.name = name
self.description = description
self.parameters = parameters # JSON Schema 格式
self.func = func
def to_openai_tool(self) -> dict:
"""转为 OpenAI tools 格式"""
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters,
},
}
def run(self, **kwargs) -> str:
"""执行工具并返回人类可读的结果"""
try:
result = self.func(**kwargs)
return str(result)
except Exception as e:
return f"工具执行错误: {e}"
3.2 自定义几个实用工具
import random
import datetime
def get_weather(city: str) -> str:
"""查询城市天气(模拟)"""
weathers = ["晴 ☀️", "多云 ⛅", "小雨 🌦️", "阴天 ☁️"]
temps = random.randint(15, 35)
return f"{city} 当前天气:{random.choice(weathers)},温度 {temps}°C"
def calculate(expression: str) -> str:
"""执行数学计算"""
# 安全起见,只允许数字和运算符
allowed = set("0123456789+-*/()., ")
if not all(c in allowed for c in expression):
return "表达式包含非法字符"
try:
result = eval(expression, {"__builtins__": {}}, {})
return f"{expression} = {result}"
except Exception as e:
return f"计算错误: {e}"
def get_current_time(format_str: str = "%Y-%m-%d %H:%M:%S") -> str:
"""获取当前时间"""
return datetime.datetime.now().strftime(format_str)
# 注册工具
TOOLS = [
Tool(
name="get_weather",
description="查询指定城市的当前天气",
parameters={
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称,如 北京、上海、深圳",
}
},
"required": ["city"],
},
func=get_weather,
),
Tool(
name="calculate",
description="执行数学表达式计算,支持加减乘除和括号",
parameters={
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "数学表达式,如 (3+5)*2",
}
},
"required": ["expression"],
},
func=calculate,
),
Tool(
name="get_current_time",
description="获取当前日期和时间",
parameters={
"type": "object",
"properties": {},
},
func=get_current_time,
),
]
# 索引工具以便快速查找
TOOL_MAP = {t.name: t for t in TOOLS}
关键点: 工具的描述(description)和参数(parameters)必须写清楚,因为模型靠这些文本来决定调用哪个工具、传什么参数。写得越清楚,模型调用越准确。
4. 第三步:Agent 主循环——思考-行动-观察
这是整个 Agent 最核心的部分。每次循环:
- 把历史 + 新观察发给模型
- 模型返回:要调用工具,还是直接回答
- 如果要调工具 → 执行业务代码 → 把结果放回消息
- 循环直到模型直接回答
4.1 系统提示词
SYSTEM_PROMPT = """你是 AI Agent,你拥有调用工具的权限。
请遵循 ReAct 模式工作:
1. 仔细分析用户的请求
2. 如果需要调用工具,按以下格式输出你的思考过程:
Thought: 分析当前情况,判断下一步需要做什么
Action: 调用合适的工具
(系统将执行工具并返回结果)
3. 如果已有足够信息,直接回答用户
可用工具:
{tools_description}
请开始。"""
4.2 Agent 主类
class Agent:
"""ReAct 模式 AI Agent"""
def __init__(self, llm: LLMClient, tools: list[Tool], max_iterations: int = 10):
self.llm = llm
self.tools = {t.name: t for t in tools}
self.tool_definitions = [t.to_openai_tool() for t in tools]
self.max_iterations = max_iterations
self.messages = []
def _build_tools_description(self) -> str:
"""为系统提示词生成工具描述"""
descs = []
for t in self.tools.values():
params = t.parameters.get("properties", {})
param_str = ", ".join(
f"{name}({info.get('type', '未知')})" for name, info in params.items()
)
descs.append(f"- {t.name}({param_str}): {t.description}")
return "\n".join(descs)
def run(self, user_input: str) -> str:
"""运行 Agent,处理用户输入并返回最终结果"""
# 初始化对话
self.messages = [
{
"role": "system",
"content": SYSTEM_PROMPT.format(
tools_description=self._build_tools_description()
),
},
{"role": "user", "content": user_input},
]
for iteration in range(self.max_iterations):
print(f"\n--- 迭代 {iteration + 1} ---")
# 调用模型
response = self.llm.chat(
messages=self.messages,
tools=self.tool_definitions,
)
# 检查是否有工具调用
if "tool_calls" in response and response["tool_calls"]:
# 模型决定调用工具
self.messages.append(response)
for tool_call in response["tool_calls"]:
tool_name = tool_call["function"]["name"]
try:
arguments = json.loads(tool_call["function"]["arguments"])
except json.JSONDecodeError:
arguments = {}
print(f" → 调用工具: {tool_name}({arguments})")
# 执行工具
if tool_name in self.tools:
result = self.tools[tool_name].run(**arguments)
else:
result = f"错误:未找到工具 {tool_name}"
print(f" ← 工具结果: {result[:100]}...")
# 把工具结果放回对话
self.messages.append({
"role": "tool",
"tool_call_id": tool_call["id"],
"content": result,
})
else:
# 模型直接回答,就是最终结果
final_answer = response.get("content", "")
print(f" ← 最终回答: {final_answer[:100]}...")
return final_answer
return "已达最大迭代次数,无法完成请求。"
4.3 试跑一下
# 初始化 Agent
agent = Agent(llm=llm, tools=TOOLS)
# 测试 1:多工具协同
result = agent.run("北京现在几度?帮我算一下从今天到月底还有几天?")
print(f"\n最终结果:\n{result}")
运行这个例子,Agent 会:
- 调用
get_weather查询北京天气 - 调用
get_current_time获取今天日期 - 调用
calculate计算剩余天数 - 综合结果完整回答
这正是 多步推理 + 多工具协作——Agent 最核心的能力。
5. 第四步:记忆管理
Agent 跑几轮后,messages 会变得很长。对于上下文窗口有限的模型(以及节省 tokens 成本),我们需要记忆管理策略。
这里实现最简单的滑动窗口 + 摘要策略:
class MemoryManager:
"""记忆管理器:控制上下文长度"""
def __init__(self, max_tokens: int = 8000):
self.max_tokens = max_tokens
def trim_messages(
self,
messages: list[dict],
system_prompt: str,
reserve_last: int = 4,
) -> list[dict]:
"""裁剪消息列表,保留最近 N 条"""
# 简化版:保留 system + 最近 reserve_last 条消息
system_msg = next((m for m in messages if m["role"] == "system"), None)
recent = messages[-reserve_last:] if len(messages) > reserve_last else messages
trimmed = []
if system_msg:
trimmed.append(system_msg)
# 如果裁剪后丢失了关键信息,添加一条摘要
if len(messages) > reserve_last + 1:
trimmed.append({
"role": "system",
"content": f"[系统摘要] 前面讨论了 {len(messages) - reserve_last - 1} 轮对话,"
f"已省略。请基于最近的信息继续。"
})
trimmed.extend(recent)
return trimmed
在 Agent run() 方法的每次迭代后加入:
self.messages = MemoryManager(max_tokens=8000).trim_messages(
self.messages, system_prompt
)
实际生产场景中,更复杂的做法包括:
- LLM 摘要:定期让模型对历史对话做摘要,替换掉原始内容
- 向量检索:把关键信息存入向量数据库,需要时召回
- 结构化记忆:把工具调用结果按类型存入不同槽位
6. 完整示例:一个有多轮记忆的对话
agent = Agent(llm=llm, tools=TOOLS)
# 第一轮
print("=== 第一轮 ===")
r1 = agent.run("我的名字叫张三,我是北京的程序员")
print(f"Agent: {r1}\n")
# 第二轮 —— Agent 应该记得之前的信息
print("=== 第二轮 ===")
r2 = agent.run("我叫什么名字?我在哪个城市?")
print(f"Agent: {r2}\n")
# 第三轮 —— 查询本地天气
print("=== 第三轮 ===")
r3 = agent.run("帮我看看我这里的天气怎么样?")
print(f"Agent: {r3}\n")
由于我们把记忆放入了 messages 对话历史中,Agent 会在上下文中看到之前的对话,从而回答:"你叫张三,在北京……“然后调用 `get_weather(city=“北京”)”。
7. 进阶:让 Agent 更聪明
7.1 错误重试
当工具调用失败时,不直接终止,而是让模型自己尝试修复:
def run_with_retry(self, user_input: str, max_retries: int = 3) -> str:
for attempt in range(max_retries):
try:
return self.run(user_input)
except Exception as e:
self.messages.append({
"role": "user",
"content": f"上一步出错了:{e},请换一种方式重试。"
})
return "重试多次仍失败"
7.2 动态工具注册
让 Agent 在运行过程中自己创建新工具(动态生成 Python 函数):
def register_dynamic_tool(self, name: str, code: str, description: str):
"""动态注册一个工具"""
local_scope = {}
exec(code, {"__builtins__": {}}, local_scope)
func = local_scope.get(name)
if not func or not callable(func):
raise ValueError("工具代码必须定义一个同名函数")
tool = Tool(
name=name,
description=description,
parameters={
"type": "object",
"properties": {},
},
func=func,
)
self.tools[name] = tool
self.tool_definitions.append(tool.to_openai_tool())
这样 Agent 就可以先让模型写一段代码,然后注册为工具反复使用——这是一个简单的 Code as Tool 模式。
8. 总结与展望
我们从头实现的 Agent 虽然只有不到 200 行核心代码,但已经包含了现代 Agent 的三大核心机制:
| 模块 | 实现 | 说明 |
|---|---|---|
| 工具调用 | Tool 类 + Function Calling |
模型决定调什么工具、传什么参数 |
| 推理规划 | ReAct 循环 | 思考→行动→观察→再思考 |
| 记忆管理 | 滑动窗口 + 摘要 | 控制上下文长度,保留关键信息 |
这个简陋的 Agent 可以直接跑起来,做天气查询、数学计算、日期计算等任务。如果你想让它更强大:
- 接入 MCP (Model Context Protocol) 协议,统一管理外部工具
- 加入 子 Agent 路由,一个 Agent 搞不定就派子 Agent
- 使用 结构化输出 替代自由文本的 thought 输出,更稳定
最核心的认知是: Agent 不是一个黑盒子,它的本质就是 循环(Loop)+ 工具(Tool)+ 记忆(Memory)。框架让你搭建得更快,但自己手写一遍,才能真正理解每个环节的设计取舍。
如果你想看某个方向的深入实现(比如 MCP 集成、子 Agent 调度、Graph 编排),欢迎评论区留言。
本文配套代码仓库: github.com/your-name/agent-from-scratch(示例地址,欢迎 fork 实战)
全部代码在 Python 3.12+ 下测试通过,使用的依赖仅
requests,零框架依赖。
更多推荐



所有评论(0)