从零搭建你的第一个 AI Agent

从零搭建你的第一个 AI Agent

一、先搞清楚：AI Agent 到底是什么？

很多教程上来就贴代码，但如果你不理解 Agent 的核心思想，代码写完了也是一头雾水。

1.1 普通 LLM 调用 vs Agent

普通 LLM 调用是这样的：

用户输入 → 大模型 → 输出结果 

一问一答，模型不会主动做任何事，你问什么它答什么，仅此而已。

AI Agent 是这样的：

用户给目标 → Agent 自主规划步骤 → 调用工具执行 → 观察结果 → 继续规划 → ... → 完成目标 

Agent 的核心在于一个循环：思考（Think）→ 行动（Act）→ 观察（Observe），不断迭代，直到任务完成。

这个循环有个专业名字，叫 ReAct（Reasoning + Acting），是目前最主流的 Agent 范式。

1.2 Agent 的三个核心组件

一个完整的 Agent 由三部分组成：

组件	作用	类比
大脑（LLM）	负责推理、规划、决策	人的大脑
工具（Tools）	执行具体操作（搜索、计算、读写文件等）	人的双手
记忆（Memory）	存储历史对话和中间结果	人的记忆

理解了这三个组件，你就理解了 Agent 的本质。

1.3 我们要做什么？

本教程将带你从零构建一个任务助手 Agent，它能够：

• ✅ 接收自然语言任务描述
• ✅ 自主拆解任务步骤
• ✅ 调用工具（计算器、文件读写、网络请求）
• ✅ 根据工具返回结果调整计划
• ✅ 最终输出完整的任务结果

---

第一阶段：环境准备（15 分钟）

第 1 步：安装依赖

# 创建虚拟环境（强烈推荐，避免依赖冲突）
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 即可，后面会说。

第 2 步：项目结构初始化

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 # 入口文件

第 3 步：配置 API Key

编辑 .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 格式，国内访问稳定。

---

第二阶段：构建核心组件（50 分钟）

第 4 步：封装 LLM 客户端

新建 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()}")

第 5 步：构建工具系统

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

第 6 步：构建对话记忆

新建 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})"

---

第三阶段：实现 Agent 核心逻辑（40 分钟）

这是整个教程最关键的部分。

第 7 步：实现 ReAct Agent

新建 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 状态已重置")

---

第四阶段：组装并运行（20 分钟）

第 8 步：创建入口文件

新建 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()

第 9 步：运行！

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'}
============================================================

---

第五阶段：扩展与优化（自选）

第 10 步：添加更多工具

工具系统是 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 立刻获得了新能力。

第 11 步：添加流式输出

让 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

第 12 步：持久化对话历史

让 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 能力（添加新工具）	✅

下一步可以探索的方向：

• 🔍 RAG（检索增强生成）：给 Agent 接入知识库，让它能回答私有文档的问题
• 🤝 Multi-Agent：多个 Agent 协作完成复杂任务
• 🧠 长期记忆：用向量数据库存储历史，让 Agent 真正'记住'你
• 🌐 Web Agent：给 Agent 接入浏览器控制能力，自动操作网页
• ⚡ 异步 Agent：用 asyncio 实现并发工具调用，大幅提升效率