从零搭建你的第一个 AI Agent
如何从零开始使用 Python 构建一个基于 ReAct 范式的 AI Agent。内容涵盖环境配置、核心组件封装(LLM 客户端、工具系统、对话记忆)、Agent 逻辑实现及运行测试。通过计算器、文件读写和网络请求等工具示例,展示了 Agent 自主规划、调用工具并完成任务的能力。文章提供了完整的代码结构和扩展建议,适合希望深入理解 Agent 底层逻辑的开发者。

如何从零开始使用 Python 构建一个基于 ReAct 范式的 AI Agent。内容涵盖环境配置、核心组件封装(LLM 客户端、工具系统、对话记忆)、Agent 逻辑实现及运行测试。通过计算器、文件读写和网络请求等工具示例,展示了 Agent 自主规划、调用工具并完成任务的能力。文章提供了完整的代码结构和扩展建议,适合希望深入理解 Agent 底层逻辑的开发者。

如果说 2024 年是大模型的元年,那 AI Agent 正在成为新的爆发点。
现在的 AI 已经不只是"聊天机器人"了——它开始接管我们的 IDE、终端、浏览器,甚至整个工作流。GitHub 上 AI 相关仓库突破 430 万,其中 Agent 框架类项目增速最猛。
但很多人对 Agent 的理解还停留在"就是 ChatGPT 加了几个插件"的阶段。
这篇文章要做的,就是把 Agent 的底层逻辑彻底讲清楚,并带你从零写出一个真正能用的 Agent。
不是调用现成框架的那种"Hello World",而是理解每一行代码在做什么,搞清楚 Agent 的核心机制,以后遇到任何 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": self.total_tokens, "model": self.model}
# 使用示例
if __name__ == "__main__":
client = LLMClient()
messages = [
{"role": "system", "content": "你是一个有帮助的助手。"},
{"role": "user", "content": "你好,请用一句话介绍自己。"},
]
reply = client.chat(messages)
print(f"模型回复:{reply}")
print(f"Token 消耗:{client.get_token_usage()}")
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 as e:
return f"计算错误:{str(e)},请检查表达式格式"
# 测试
if __name__ == "__main__":
calc = CalculatorTool()
print(calc.run("2 + 3 * 4")) # 14
print(calc.run("sqrt(144)")) # 12.0
print(calc.run("sin(pi/2)")) # 1.0
print(calc.run("log(e)")) # 1.0
工具二:文件读写 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":
if len(parts) < 3:
return "错误:写入操作需要提供文件内容"
content = parts[2]
return self._write_file(file_path, content)
else:
return f"错误:不支持的操作类型 '{action}',请使用 'read' 或 'write'"
def _read_file(self, path: str) -> str:
"""读取文件"""
if not os.path.exists(path):
return f"错误:文件 '{path}' 不存在"
try:
with open(path, "r", encoding="utf-8") as f:
content = f.read()
return f"文件内容:\n{content}"
except Exception as e:
return f"读取失败:{str(e)}"
def _write_file(self, path: str, content: str) -> str:
"""写入文件"""
try:
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
f.write(content)
return f"成功写入文件:{path}({len(content)} 字符)"
except Exception as e:
return f"写入失败:{str(e)}"
# 测试
if __name__ == "__main__":
tool = FileTool()
print(tool.run("write:test.txt:Hello, Agent World!"))
print(tool.run("read:test.txt"))
工具三:网络请求 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"网页内容(前{self.MAX_CONTENT_LENGTH}字符):\n{truncated}"
except requests.exceptions.Timeout:
return f"错误:请求超时(>{self.TIMEOUT}秒)"
except requests.exceptions.HTTPError as e:
return f"错误:HTTP 请求失败,状态码 {e.response.status_code}"
except Exception as e:
return f"请求失败:{str(e)}"
def _strip_html(self, html: str) -> str:
"""简单去除 HTML 标签"""
import re
# 去除 script 和 style 标签及其内容
html = re.sub(r"<(script|style)[^>]*>.*?</\1>", "", html, flags=re.DOTALL)
# 去除所有 HTML 标签
html = re.sub(r"<[^>]+>", " ", html)
# 合并多余空白
html = re.sub(r"\s+", " ", html).strip()
return 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)
if i < len(self._history) and self._history[i]["role"] == "assistant":
self._history.pop(i)
break
def get_messages(self) -> list[dict]:
""" 获取完整的消息列表(包含 system prompt)
Returns:
适合直接传给 LLM 的消息列表
"""
messages = []
if self.system_prompt:
messages.append({"role": "system", "content": self.system_prompt})
messages.extend(self._history)
return messages
def clear(self):
"""清空对话历史"""
self._history.clear()
def __len__(self) -> int:
return len(self._history)
def __repr__(self) -> str:
return f"ConversationMemory(turns={len(self._history)//2}, max={self.max_turns})"
这是整个教程最关键的部分。
新建 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}" for name, tool in self.tools.items()
)
# 初始化 LLM 和记忆
self.llm = LLMClient()
self.memory = ConversationMemory(
max_turns=20,
system_prompt=SYSTEM_PROMPT.format(tool_descriptions=tool_descriptions),
)
def run(self, task: str) -> str:
""" 执行一个任务
Args:
task: 用户的任务描述(自然语言)
Returns:
Agent 的最终回答
"""
print(f"\n{Fore.CYAN}{'='*60}")
print(f"🤖 任务开始:{task}")
print(f"{'='*60}{Style.RESET_ALL}\n")
# 将任务加入记忆
self.memory.add_message("user", task)
# ReAct 主循环
for iteration in range(1, self.MAX_ITERATIONS + 1):
print(f"{Fore.YELLOW}── 第 {iteration} 轮迭代 ──{Style.RESET_ALL}")
# 调用 LLM 获取下一步决策
response = self.llm.chat(self.memory.get_messages())
print(f"{Fore.GREEN}LLM 输出:\n{response}{Style.RESET_ALL}\n")
# 解析 LLM 输出
parsed = self._parse_response(response)
# 情况一:任务完成,返回最终答案
if parsed["type"] == "final_answer":
self.memory.add_message("assistant", response)
final = parsed["content"]
print(f"\n{Fore.CYAN}{'='*60}")
print(f"✅ 任务完成!")
print(f"最终答案:{final}")
print(f"Token 消耗:{self.llm.get_token_usage()}")
print(f"{'='*60}{Style.RESET_ALL}\n")
return final
# 情况二:需要调用工具
elif parsed["type"] == "action":
tool_name = parsed["tool"]
tool_input = parsed["input"]
print(f"{Fore.MAGENTA}🔧 调用工具:{tool_name}")
print(f" 输入:{tool_input}{Style.RESET_ALL}")
# 执行工具
observation = self._execute_tool(tool_name, tool_input)
print(f"{Fore.BLUE}📋 工具返回:{observation}{Style.RESET_ALL}\n")
# 将 LLM 输出和工具结果都加入记忆
self.memory.add_message("assistant", response)
self.memory.add_message(
"user", f"Observation: {observation}"
)
# 情况三:解析失败,提示 LLM 修正格式
else:
print(f"{Fore.RED}⚠️ 输出格式解析失败,提示 LLM 修正{Style.RESET_ALL}")
self.memory.add_message("assistant", response)
self.memory.add_message(
"user", "你的输出格式不正确。请严格按照 Thought/Action/Action Input 或 Final Answer 格式回复。",
)
# 超出最大迭代次数
return "任务未能在规定步骤内完成,请尝试简化任务描述。"
def _parse_response(self, response: str) -> dict:
""" 解析 LLM 的输出,提取 Action 或 Final Answer
Returns:
{"type": "action", "tool": "...", "input": "..."}
{"type": "final_answer", "content": "..."}
{"type": "unknown"}
"""
# 检查是否有 Final Answer
final_match = re.search(
r"Final Answer:\s*(.+?)(?:\n|$)", response, re.DOTALL
)
if final_match:
return {"type": "final_answer", "content": final_match.group(1).strip()}
# 检查是否有 Action
action_match = re.search(r"Action:\s*(\w+)", response)
input_match = re.search(
r"Action Input:\s*(.+?)(?:\nThought|\nAction|\nObservation|$)", response, re.DOTALL,
)
if action_match and input_match:
return {
"type": "action",
"tool": action_match.group(1).strip(),
"input": input_match.group(1).strip(),
}
return {"type": "unknown"}
def _execute_tool(self, tool_name: str, tool_input: str) -> str:
""" 执行指定工具
Args:
tool_name: 工具名称
tool_input: 工具输入
Returns:
工具执行结果
"""
if tool_name not in self.tools:
available = ", ".join(self.tools.keys())
return f"错误:工具 '{tool_name}' 不存在。可用工具:{available}"
try:
return self.tools[tool_name].run(tool_input)
except Exception as e:
return f"工具执行异常:{str(e)}"
def reset(self):
"""重置 Agent 状态(清空对话历史)"""
self.memory.clear()
print("Agent 状态已重置")
新建 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(f"已加载 {len(self._history)} 条历史记录")
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:工具调用很慢,怎么优化?
1. 对工具结果加缓存(相同输入直接返回缓存结果)
2. 对不依赖顺序的工具调用,改为并行执行
3. 减少 MAX_ITERATIONS,强制 Agent 更高效地规划
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
生成新的随机RSA私钥和公钥pem证书。 在线工具,RSA密钥对生成器在线工具,online
基于 Mermaid.js 实时预览流程图、时序图等图表,支持源码编辑与即时渲染。 在线工具,Mermaid 预览与可视化编辑在线工具,online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具,curl 转代码在线工具,online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具,Base64 字符串编码/解码在线工具,online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具,Base64 文件转换器在线工具,online
将 Markdown(GFM)转为 HTML 片段,浏览器内 marked 解析;与 HTML 转 Markdown 互为补充。 在线工具,Markdown 转 HTML在线工具,online