PythonAI算法
从零搭建你的第一个 AI Agent
如何使用 Python 从零构建一个基于 ReAct 范式的 AI Agent。内容包括环境配置、LLM 客户端封装、工具系统（计算器、文件读写、网络请求）、对话记忆管理以及 Agent 核心循环逻辑的实现。文章提供了完整的可运行代码示例，并探讨了流式输出、持久化存储及多工具扩展等优化方案，旨在帮助开发者深入理解 AI Agent 的底层机制与开发流程。
如何使用 Python 从零构建一个基于 ReAct 范式的 AI Agent。内容包括环境配置、LLM 客户端封装、工具系统（计算器、文件读写、网络请求）、对话记忆管理以及 Agent 核心循环逻辑的实现。文章提供了完整的可运行代码示例，并探讨了流式输出、持久化存储及多工具扩展等优化方案，旨在帮助开发者深入理解 AI Agent 的底层机制与开发流程。
很多教程上来就贴代码，但如果你不理解 Agent 的核心思想，代码写完了也是一头雾水。
普通 LLM 调用是这样的：
用户输入 → 大模型 → 输出结果
一问一答，模型不会主动做任何事，你问什么它答什么，仅此而已。
AI Agent 是这样的：
用户给目标 → Agent 自主规划步骤 → 调用工具执行 → 观察结果 → 继续规划 → ... → 完成目标
Agent 的核心在于一个循环：思考（Think）→ 行动（Act）→ 观察（Observe），不断迭代，直到任务完成。
这个循环有个专业名字，叫 ReAct（Reasoning + Acting），是目前最主流的 Agent 范式。
一个完整的 Agent 由三部分组成：
| 组件 | 作用 | 类比 |
|---|---|---|
| 大脑（LLM） | 负责推理、规划、决策 | 人的大脑 |
| 工具（Tools） | 执行具体操作（搜索、计算、读写文件等） | 人的双手 |
| 记忆（Memory） | 存储历史对话和中间结果 | 人的记忆 |
理解了这三个组件，你就理解了 Agent 的本质。
本教程将带你从零构建一个任务助手 Agent，它能够：
# 创建虚拟环境（强烈推荐，避免依赖冲突）
python -m venv agent-env
source agent-env/bin/activate # Linux/macOS
# agent-env\Scripts\activate # Windows
# 安装核心依赖
pip install openai python-dotenv requests colorama
# 验证安装
python -c "import openai; print('OpenAI SDK 安装成功')"
💡 没有 OpenAI API Key？ 可以用国内的兼容接口（如 DeepSeek、智谱 GLM），只需修改
base_url即可，后面会说。
mkdir my-agent && cd my-agent
# 创建目录结构
mkdir -p src/{tools,memory,core}
mkdir -p tests
mkdir -p logs
# 创建文件
touch src/__init__.py
touch src/tools/__init__.py
touch src/memory/__init__.py
touch src/core/__init__.py
touch .env
touch main.py
最终目录结构如下：
my-agent/
├── src/
│ ├── core/
│ │ ├── __init__.py
│ │ ├── agent.py # Agent 核心逻辑
│ │ └── llm_client.py # LLM 调用封装
│ ├── tools/
│ │ ├── __init__.py
│ │ ├── calculator.py # 计算器工具
│ │ ├── file_tool.py # 文件读写工具
│ │ └── web_tool.py # 网络请求工具
│ └── memory/
│ ├── __init__.py
│ └── conversation.py # 对话记忆
├── tests/
├── logs/
├── .env # API Key 配置
└── main.py # 入口文件
编辑 .env 文件：
# OpenAI 官方
OPENAI_API_KEY=sk-xxxxxxxxxxxxxxxx
OPENAI_BASE_URL=https://api.openai.com/v1
MODEL_NAME=gpt-4o-mini
# 或者使用 DeepSeek（国内更稳定，价格更低）
# OPENAI_API_KEY=sk-xxxxxxxxxxxxxxxx
# OPENAI_BASE_URL=https://api.deepseek.com/v1
# MODEL_NAME=deepseek-chat
# 或者使用智谱 GLM
# OPENAI_API_KEY=xxxxxxxxxxxxxxxx
# OPENAI_BASE_URL=https://open.bigmodel.cn/api/paas/v4
# MODEL_NAME=glm-4-flash
✅ 推荐新手用 DeepSeek：价格极低（约 OpenAI 的 1/20），API 完全兼容 OpenAI 格式，国内访问稳定。
新建 src/core/llm_client.py：
""" LLM 客户端封装
统一管理大模型调用，支持多种 API 提供商
"""
import os
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()
class LLMClient:
"""大模型调用客户端"""
def __init__(self):
self.client = OpenAI(
api_key=os.getenv("OPENAI_API_KEY"),
base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),
)
self.model = os.getenv("MODEL_NAME", "gpt-4o-mini")
self.total_tokens = 0 # 统计 token 消耗
def chat(self, messages: list, temperature: float = 0.7) -> str:
"""
发送对话请求
Args:
messages: 对话历史，格式为 [{"role": "user/assistant/system", "content": "..."}]
temperature: 温度参数，越高越随机（0~2）
Returns:
模型回复的文本内容
"""
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=temperature,
)
# 统计 token 消耗
self.total_tokens += response.usage.total_tokens
return response.choices[0].message.content
def get_token_usage(self) -> dict:
"""获取 token 使用统计"""
return {"total_tokens": .total_tokens, : .model}
__name__ == :
client = LLMClient()
messages = [
{: , : },
{: , : },
]
reply = client.chat(messages)
()
()
Agent 的能力来自工具。我们先实现三个最常用的工具。
工具一：计算器 src/tools/calculator.py
""" 计算器工具
让 Agent 能够执行数学计算，避免大模型的计算幻觉
"""
import math
class CalculatorTool:
"""安全的数学计算工具"""
name = "calculator"
description = (
"执行数学计算。输入一个数学表达式字符串，返回计算结果。"
"支持：加减乘除、幂运算、开方、三角函数等。"
"示例输入：'2 + 3 * 4'、'sqrt(16)'、'sin(3.14/2)'"
)
# 允许使用的安全函数白名单
SAFE_FUNCTIONS = {
"abs": abs, "round": round, "sqrt": math.sqrt, "pow": math.pow,
"sin": math.sin, "cos": math.cos, "tan": math.tan, "log": math.log,
"log10": math.log10, "log2": math.log2, "pi": math.pi, "e": math.e,
"ceil": math.ceil, "floor": math.floor,
}
def run(self, expression: str) -> str:
"""
执行数学计算
Args:
expression: 数学表达式字符串
Returns:
计算结果字符串，或错误信息
"""
try:
# 安全求值：只允许白名单中的函数
result = eval(expression, {"__builtins__": {}}, self.SAFE_FUNCTIONS)
return f"计算结果：{expression} = {result}"
except ZeroDivisionError:
return "错误：除数不能为零"
except Exception e:
__name__ == :
calc = CalculatorTool()
(calc.run())
(calc.run())
(calc.run())
(calc.run())
工具二：文件读写 src/tools/file_tool.py
""" 文件读写工具
让 Agent 能够读取和写入本地文件
"""
import os
class FileTool:
"""文件操作工具"""
name = "file_tool"
description = (
"读取或写入本地文件。"
"操作类型：'read'（读取文件内容）或 'write'（写入内容到文件）。"
"输入格式：'read:文件路径' 或 'write:文件路径:文件内容'"
)
# 限制可操作的目录（安全沙箱）
ALLOWED_DIR = "./workspace"
def __init__(self):
# 确保工作目录存在
os.makedirs(self.ALLOWED_DIR, exist_ok=True)
def run(self, command: str) -> str:
"""
执行文件操作
Args:
command: 操作命令，格式见 description
Returns:
操作结果字符串
"""
parts = command.split(":", 2)
if len(parts) < 2:
return "错误：命令格式不正确，请使用 'read:路径' 或 'write:路径:内容'"
action = parts[0].strip().lower()
file_path = os.path.join(self.ALLOWED_DIR, parts[1].strip())
# 安全检查：防止路径穿越攻击
if not os.path.abspath(file_path).startswith(os.path.abspath(self.ALLOWED_DIR)):
return "错误：不允许访问工作目录以外的文件"
if action == "read":
return self._read_file(file_path)
elif action == "write":
(parts) < :
content = parts[]
._write_file(file_path, content)
:
() -> :
os.path.exists(path):
:
(path, , encoding=) f:
content = f.read()
Exception e:
() -> :
:
os.makedirs(os.path.dirname(path), exist_ok=)
(path, , encoding=) f:
f.write(content)
Exception e:
__name__ == :
tool = FileTool()
(tool.run())
(tool.run())
工具三：网络请求 src/tools/web_tool.py
""" 网络请求工具
让 Agent 能够获取网页内容（简化版）
"""
import requests
from urllib.parse import urlparse
class WebTool:
"""网络请求工具"""
name = "web_tool"
description = (
"获取指定 URL 的网页文本内容。"
"输入一个完整的 URL（需包含 http:// 或 https://），"
"返回页面的纯文本内容（前 2000 字符）。"
"示例：'https://example.com'"
)
TIMEOUT = 10 # 请求超时时间（秒）
MAX_CONTENT_LENGTH = 2000 # 最大返回内容长度
def run(self, url: str) -> str:
"""
获取网页内容
Args:
url: 目标 URL
Returns:
网页文本内容或错误信息
"""
# 验证 URL 格式
parsed = urlparse(url.strip())
if parsed.scheme not in ("http", "https"):
return "错误：URL 必须以 http:// 或 https:// 开头"
try:
headers = {
"User-Agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 Chrome/120.0.0.0 Safari/537.36"
)
}
response = requests.get(url, headers=headers, timeout=self.TIMEOUT)
response.raise_for_status()
response.encoding = response.apparent_encoding # 简单提取文本（去除 HTML 标签）
text = self._strip_html(response.text)
truncated = text[: self.MAX_CONTENT_LENGTH]
return f"网页内容（前字符）：\n"
requests.exceptions.Timeout:
requests.exceptions.HTTPError e:
Exception e:
() -> :
re
html = re.sub(, , html, flags=re.DOTALL)
html = re.sub(, , html)
html = re.sub(, , html).strip()
html
新建 src/memory/conversation.py：
""" 对话记忆模块
管理 Agent 的上下文历史，支持长度限制和摘要压缩
"""
class ConversationMemory:
"""对话历史管理"""
def __init__(self, max_turns: int = 20, system_prompt: str = ""):
"""
Args:
max_turns: 最大保留的对话轮数（超出后自动裁剪旧记录）
system_prompt: 系统提示词
"""
self.max_turns = max_turns
self.system_prompt = system_prompt
self._history: list[dict] = []
def add_message(self, role: str, content: str):
"""
添加一条消息到历史
Args:
role: 角色，'user' / 'assistant' / 'system'
content: 消息内容
"""
self._history.append({"role": role, "content": content})
# 超出最大轮数时，裁剪最早的记录（保留 system 消息）
non_system = [m for m in self._history if m["role"] != "system"]
if len(non_system) > self.max_turns * 2:
# 删除最早的一轮（user + assistant 各一条）
for i, msg in enumerate(self._history):
if msg["role"] == "user":
self._history.pop(i)
i < (._history) ._history[i][] == :
._history.pop(i)
() -> []:
messages = []
.system_prompt:
messages.append({: , : .system_prompt})
messages.extend(._history)
messages
():
._history.clear()
() -> :
(._history)
() -> :
这是整个教程最关键的部分。
新建 src/core/agent.py：
""" ReAct Agent 核心实现
思路：Thought（思考）→ Action（行动）→ Observation（观察）→ 循环
"""
import re
import json
from colorama import Fore, Style, init
from src.core.llm_client import LLMClient
from src.memory.conversation import ConversationMemory
from src.tools.calculator import CalculatorTool
from src.tools.file_tool import FileTool
from src.tools.web_tool import WebTool
# 初始化彩色输出
init(autoreset=True)
# ── Agent 系统提示词 ──────────────────────────────────────────────────────────
SYSTEM_PROMPT = """你是一个智能任务助手，能够通过调用工具来完成用户交给你的任务。
## 你拥有以下工具：{tool_descriptions}
## 工作流程（严格遵守）：
每次回复必须按照以下格式，直到任务完成：
Thought: [分析当前情况，思考下一步该做什么]
Action: [工具名称]
Action Input: [工具的输入参数]
当工具返回结果后，你会收到：
Observation: [工具返回的结果]
然后继续思考，直到任务完成，最后输出：
Thought: [最终思考，确认任务已完成]
Final Answer: [给用户的最终回答]
## 重要规则：
1. 每次只能调用一个工具
2. 如果不需要工具，直接输出 Final Answer
3. 遇到计算问题，必须使用 calculator 工具，不要自己心算
4. Action 字段只能填写工具名称，不能有其他内容
5. 如果工具返回错误，分析原因并尝试修正后重试
"""
class ReActAgent:
"""
基于 ReAct 范式的 AI Agent
ReAct = Reasoning（推理）+ Acting（行动）
核心循环：Thought → Action → Observation → Thought → ...
"""
MAX_ITERATIONS = 10 # 最大迭代次数，防止死循环
def __init__(self):
# 初始化工具
self.tools = {
"calculator": CalculatorTool(),
"file_tool": FileTool(),
"web_tool": WebTool(),
}
# 构建工具描述（注入到 system prompt）
tool_descriptions = "\n".join(
f"- **{name}**：{tool.description}" name, tool .tools.items()
)
.llm = LLMClient()
.memory = ConversationMemory(
max_turns=,
system_prompt=SYSTEM_PROMPT.(tool_descriptions=tool_descriptions),
)
() -> :
()
()
()
.memory.add_message(, task)
iteration (, .MAX_ITERATIONS + ):
()
response = .llm.chat(.memory.get_messages())
()
parsed = ._parse_response(response)
parsed[] == :
.memory.add_message(, response)
final = parsed[]
()
()
()
()
()
final
parsed[] == :
tool_name = parsed[]
tool_input = parsed[]
()
()
observation = ._execute_tool(tool_name, tool_input)
()
.memory.add_message(, response)
.memory.add_message(
,
)
:
()
.memory.add_message(, response)
.memory.add_message(
, ,
)
() -> :
final_match = re.search(
, response, re.DOTALL
)
final_match:
{: , : final_match.group().strip()}
action_match = re.search(, response)
input_match = re.search(
, response, re.DOTALL,
)
action_match input_match:
{
: ,
: action_match.group().strip(),
: input_match.group().strip(),
}
{: }
() -> :
tool_name .tools:
available = .join(.tools.keys())
:
.tools[tool_name].run(tool_input)
Exception e:
():
.memory.clear()
()
新建 main.py：
""" AI Agent 入口文件 """
from src.core.agent import ReActAgent
def main():
agent = ReActAgent()
print("="*60)
print("🤖 AI Agent 已启动！输入 'quit' 退出，'reset' 重置对话")
print("="*60)
while True:
try:
task = input("\n📝 请输入任务：").strip()
except (KeyboardInterrupt, EOFError):
print("\n\n👋 再见！")
break
if not task:
continue
if task.lower() == "quit":
print("👋 再见！")
break
if task.lower() == "reset":
agent.reset()
continue
# 执行任务
agent.run(task)
if __name__ == "__main__":
main()
python main.py
你会看到类似这样的输出：
============================================================
🤖 AI Agent 已启动！输入 'quit' 退出，'reset' 重置对话
============================================================
📝 请输入任务：计算 (123 + 456) * 789 的结果，然后把结果写入 result.txt 文件
============================================================
🤖 任务开始：计算 (123 + 456) * 789 的结果，然后把结果写入 result.txt 文件
============================================================
── 第 1 轮迭代 ──
LLM 输出：
Thought: 用户需要我先计算数学表达式，再把结果写入文件。先用计算器工具计算。
Action: calculator
Action Input: (123 + 456) * 789
🔧 调用工具：calculator
输入：(123 + 456) * 789
📋 工具返回：计算结果：(123 + 456) * 789 = 466731
── 第 2 轮迭代 ──
LLM 输出：
Thought: 计算结果是 466731，现在把这个结果写入 result.txt 文件。
Action: file_tool
Action Input: write:result.txt:计算结果：(123 + 456) * 789 = 466731
🔧 调用工具：file_tool
输入：write:result.txt:计算结果：(123 + 456) * 789 = 466731
📋 工具返回：成功写入文件：./workspace/result.txt（26 字符）
── 第 3 轮迭代 ──
LLM 输出：
Thought: 两个任务都完成了，可以给出最终答案。
Final Answer: 计算结果为 466731，已成功写入 result.txt 文件。
============================================================
✅ 任务完成！
最终答案：计算结果为 466731，已成功写入 result.txt 文件。
Token 消耗：{'total_tokens': 847, 'model': 'gpt-4o-mini'}
============================================================
工具系统是 Agent 能力的上限。你可以轻松扩展更多工具：
# 示例：添加一个'当前时间'工具
class TimeTool:
name = "get_time"
description = "获取当前的日期和时间。无需输入参数，直接调用即可。"
def run(self, _: str = "") -> str:
from datetime import datetime
now = datetime.now()
return f"当前时间：{now.strftime('%Y年%m月%d日 %H:%M:%S')}"
# 示例：添加一个'天气查询'工具
class WeatherTool:
name = "get_weather"
description = "查询指定城市的天气。输入城市名称（中文），返回天气信息。"
def run(self, city: str) -> str:
import requests
try:
url = f"https://wttr.in/{city}?format=3&lang=zh"
resp = requests.get(url, timeout=5)
return f"天气信息：{resp.text.strip()}"
except Exception as e:
return f"天气查询失败：{str(e)}"
把新工具注册到 Agent 中：
# 在 ReActAgent.__init__ 中添加
self.tools["get_time"] = TimeTool()
self.tools["get_weather"] = WeatherTool()
就这么简单，Agent 立刻获得了新能力。
让 Agent 的思考过程实时显示，体验更好：
def chat_stream(self, messages: list):
"""流式输出版本"""
stream = self.client.chat.completions.create(
model=self.model,
messages=messages,
stream=True, # 开启流式
)
full_response = ""
for chunk in stream:
delta = chunk.choices[0].delta.content or ""
print(delta, end="", flush=True)
full_response += delta
print() # 换行
return full_response
让 Agent 在重启后还能记住之前的对话：
import json
import os
class PersistentMemory(ConversationMemory):
"""支持持久化的对话记忆"""
def __init__(self, save_path: str = "./logs/memory.json", **kwargs):
super().__init__(**kwargs)
self.save_path = save_path
self._load() # 启动时自动加载历史
def add_message(self, role: str, content: str):
super().add_message(role, content)
self._save() # 每次添加消息后自动保存
def _save(self):
os.makedirs(os.path.dirname(self.save_path), exist_ok=True)
with open(self.save_path, "w", encoding="utf-8") as f:
json.dump(self._history, f, ensure_ascii=False, indent=2)
def _load(self):
if os.path.exists(self.save_path):
with open(self.save_path, "r", encoding="utf-8") as f:
self._history = json.load(f)
print()
Q：运行报错 AuthenticationError？
检查 .env 文件中的 API Key 是否正确，注意不要有多余的空格或引号。
Q：Agent 陷入死循环，一直调用同一个工具？
通常是 system prompt 写得不够清晰，或者工具描述有歧义。可以在 system prompt 中加一条： "如果同一个工具连续调用 3 次仍然失败，请直接输出 Final Answer 说明原因。"
Q：LLM 输出格式不对，解析总是失败？
换用更强的模型（如 gpt-4o 替代 gpt-4o-mini），或者在 system prompt 中加入更多格式示例（few-shot）。
Q：工具调用很慢，怎么优化？
Q：想用本地模型（Ollama）怎么接入？
# Ollama 完全兼容 OpenAI API 格式，只需修改 .env：
OPENAI_API_KEY=ollama # 随便填，Ollama 不验证
OPENAI_BASE_URL=http://localhost:11434/v1
MODEL_NAME=qwen2.5:7b # 你本地拉取的模型名
所有代码已整理到以下结构，可直接克隆使用：
my-agent/
├── src/
│ ├── core/
│ │ ├── agent.py ✅ ReAct Agent 核心
│ │ └── llm_client.py ✅ LLM 客户端
│ ├── tools/
│ │ ├── calculator.py ✅ 计算器
│ │ ├── file_tool.py ✅ 文件读写
│ │ └── web_tool.py ✅ 网络请求
│ └── memory/
│ └── conversation.py ✅ 对话记忆
├── .env ✅ API 配置
└── main.py ✅ 入口文件
恭喜你！完成这个教程后，你已经掌握了：
| 知识点 | 掌握程度 |
|---|---|
| AI Agent 的核心概念（LLM + Tools + Memory） | ✅ |
| ReAct 范式（Thought → Action → Observation） | ✅ |
| 如何封装 LLM 调用客户端 | ✅ |
| 如何设计和实现工具系统 | ✅ |
| 如何管理对话上下文记忆 | ✅ |
| 如何解析 LLM 的结构化输出 | ✅ |
| 如何扩展 Agent 能力（添加新工具） | ✅ |
下一步可以探索的方向：
asyncio 实现并发工具调用，大幅提升效率
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online