GEOZ

本文手把手教你用Python在2小时内构建一个能自主规划、调

2026/8/12
本文手把手教你用Python在2小时内构建一个能自主规划、调

AIAI Summary (BLUF)

本文手把手教你用Python在2小时内构建一个能自主规划、调用工具、完成任务的AI Agent,涵盖ReAct模式、三个核心组件、环境配置以及完整代码实现,适合想入门Agent开发的开发者。

核心洞察

这篇文章最有意思的点是它把 Agent 从热词拆成了三块:模型、工具、记忆。工具那部分尤其实在,代码直接给到,能跑。唯一我持保留意见的地方是计算器工具里用了 eval,自己折腾没问题,真要上生产得把表达式解析换成更安全的方式。

核心结论

  1. Agent 的核心结构由模型、工具、记忆三个组件构成:模型负责推理规划,工具负责执行操作,记忆负责保存中间状态,三者缺一不可。
  2. Agent 的运行基于 ReAct 循环:自主规划 → 调用工具 → 观察结果 → 继续规划,直到完成目标;当前主流 Agent 框架大多遵循这一思路。
  3. 教程实现了计算器、文件读写、网络请求三个工具,其中文件工具通过 ALLOWED_DIRabspath 校验限制读写范围,能防路径穿越,但 startswith 判断存在前缀绕过缺陷(如 workspace_evil 目录会被误放行)。
  4. 计算器工具使用 eval 执行数学表达式,仅适合学习演示;上生产环境前必须替换为安全解析方案。
  5. 教程推荐新手使用 DeepSeek 作为模型接口:价格约为 OpenAI 的二十分之一,API 格式完全兼容,国内访问稳定。

保姆级教程:从零搭建你的第一个 AI Agent(附完整可运行代码)

手把手教你,用 Python 在 2 小时内构建一个能自主规划、调用工具、完成任务的 AI Agent

预计完成时间:2 小时
所需技能:基础 Python、会用命令行
适合人群:想入门 AI Agent 开发的同学,不限工作年限

前言:为什么 2026 年你必须懂 Agent?

2024 年是大模型跑出来的年份,2026 年轮到 Agent 了。

GitHub 上 AI 相关仓库已经超过 430 万个,Agent 框架类项目涨得最快。可很多人提到 Agent,第一反应还是“ChatGPT 加了几个插件”。

这篇文章想把 Agent 怎么工作这件事拆开讲,然后带着你从零写一个能跑的 Agent。不调现成框架,每一行代码都自己敲,搞明白它为什么这样设计。以后你再去碰那些 Agent 框架,上手会快很多。

一、先搞清楚:AI Agent 到底是什么?

很多教程上来就贴代码。代码贴完了,你还是不知道刚才发生了什么。咱们先把概念理清楚。

1.1 普通大模型调用和 Agent 的区别

普通大模型调用长这样:

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

你问一句,它答一句。模型不会主动去查资料,也不会自己动手算。

Agent 长这样:

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

你给它一个目标,它自己拆步骤、调工具、看结果,不行就再来一轮。这个循环有个名字,叫 ReAct,简单说就是推理和行动交替进行。现在市面上的 Agent 框架,大部分都是照着这个思路做的。

1.2 Agent 的三个核心组件

一个 Agent 要跑起来,缺三样东西:

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

模型负责想,工具负责做,记忆负责记住中间过程。这三个凑齐,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 接口密钥?可以用国内的兼容接口,比如 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 的二十分之一,API 格式完全兼容,国内访问也稳。

第二阶段:构建核心组件(50 分钟)

第 4 步:封装大模型客户端

新建 src/core/llm_client.py

"""
大模型客户端封装
统一管理模型调用,支持多种 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)},请检查表达式格式"

另外两个工具(文件读写和网络请求)放到下一部分,写法一样:每个工具就是一个类,带上 name、description、run。计算器跑通了,Agent 主循环就有东西可以调了。

先把计算器这段跑一遍,确认上一节写的代码没毛病:

# 测试
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

四个表达式全部通过。eval 那点事上一节说过了,自己折腾没事,上生产前记得换掉。

工具二:文件读写

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

这个工具最值得看的是沙箱设计。ALLOWED_DIR 把读写范围圈在 ./workspace 里,run 方法里那段 abspath 校验防路径穿越。你传 read:../../etc/passwd,os.path.join 拼出 ./workspace/../../etc/passwd,abspath 解析后落在 workspace 外面,直接拒绝。

不过 startswith 这个判断有个小毛病。假如存在一个叫 ./workspace_evil 的目录,它的绝对路径同样以 ./workspace 开头,这个校验会放行。更严谨的做法是用 os.path.commonpath 或者 Path.resolve() 再比较。教程代码就不改了,你心里有数就行。

跑一下测试,能正常写入并读回文件。命令用冒号分段,split 的 maxsplit 设成 2,所以文件内容里带冒号也没事,会整体留在第三段。

工具三:网络请求

新建 src/tools/web_tool.py。requests 不是标准库,先 pip install requests

"""
网络请求工具
让 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

这个工具把写爬虫最容易踩的三个坑提前填了。

一个是 User-Agent。很多网站看到 requests 默认的 UA 会直接拒掉,这里伪装成 Chrome 浏览器,大部分站点能过。再讲究还有 cookie、验证码,教程不展开。

一个是超时。requests 不设 timeout,遇到慢网站能挂好几分钟。这里限制 10 秒,超时就把错误返回给 Agent,让模型自己决定下一步。

一个是编码。直接用 response.text,碰上没在响应头里标 charset 的网站,中文会乱码。用 apparent_encoding 让 requests 从页面内容里猜编码,再赋回去。

HTML 提取用的正则。先删掉 script 和 style 整段,其余标签换成空格,再合并连续空白。够用,但也只是够用。真要好好抓正文,换 BeautifulSoup 吧。

第 6 步:构建对话记忆

工具齐了,还差记忆。

没有记忆的 Agent,每一次对话都是白纸一张。用户上一句说了什么,这一句再问,它全不知道。就像每次打电话都要先自我介绍一遍,烦死。

所以做一个最简单的对话记忆:一个列表存消息,一个上限防爆。新建 src/memory/conversation.py

"""
对话记忆模块
管理 Agent 的上下文历史,支持长度限制和摘要压缩
"""

class ConversationMemory:
    """对话历史管理"""

    def __init__(self, max_messages=20):
        self.history = []
        self.max_messages = max_messages

    def add_message(self, role, content):
        """添加一条消息"""
        self.history.append({"role": role, "content": content})
        if len(self.history) > self.max_messages:
            self.history = self.history[-self.max_messages:]

    def add_user(self, content):
        """添加用户消息"""
        self.add_message("user", content)

    def add_bot(self, content):
        """添加助手消息"""
        self.add_message("assistant", content)

    def get_history(self):
        """返回完整的消息列表"""
        return self.history

    def get_context(self):
        """把历史拼成一段文本,作为上下文给模型用"""
        return "\n".join(
            f"{msg['role']}: {msg['content']}" for msg in self.history
        )

    def clear(self):
        """清空历史"""
        self.history = []

    def __len__(self):
        return len(self.history)

实现就三块。add_message 负责追加,超长时丢掉最老的。get_context 把整个历史拼成一段文本,作为上下文塞给模型。clear 一键清空。

max_messages 默认 20 条,日常聊天够支撑十几轮。真要上生产,按条数限制不科学,得按 token 数来。或者学文件名里那个方向,对早期对话做摘要压缩,只留摘要。后话了。

到这里工具和记忆都齐了,下一步把这些零件组装起来,拼成一个真正能跑的 Agent。

刚才那段计算器测试跑通之后,Agent 还缺个记性。大模型自己记不住之前的对话,每轮都得把历史一块发过去。先建一个 src/memory/conversation.py,把记忆封装起来。

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

这里有个小坑:max_turns 按轮算,不按条算。一轮对话有 user 和 assistant 两条消息,所以判断条件里乘了 2。删的时候先找第一条 user 记录,把它和后面那条 assistant 一起 pop 掉,这样系统消息能留在前面。


这一段最绕。ReAct 的循环看着复杂,拆开就三步:想,做,看结果。把这个流程记住,代码就好写了。

第 7 步:实现 ReAct Agent

新建 src/core/agent.py,先把文件头部和系统提示词写进去。后面要用的 import 也一次备齐,省得等会儿来回改。

"""
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: [给用户的最终回答]
"""

主循环的代码下一步再写,{tool_descriptions} 也会在运行前填成真实工具列表。这里先把提示词和工具装配好。

工具和记忆备齐了,到这一步该把它们串起来了。ReAct 的循环用一句话就能说清:先让模型思考,再让模型选工具,拿到结果接着思考,反复走。

第 7 步:实现 ReAct Agent

新建 src/core/agent.py,内容分两块,一是系统提示词,二是 ReActAgent 类。

SYSTEM_PROMPT 长,但值得仔细看。它把工具列表动态编进去,模型才知道自己能调什么。然后规定了严格输出格式,每轮回复必须按照 Thought / Action / Action Input / Observation 的顺序走。里面还特意写了条规则:计算问题必须用 calculator,不许心算。这条对模型这种动不动就自己硬算的家伙很有必要。

ReActAgent 类里,MAX_ITERATIONS = 10 是兜底,防止模型陷入调工具的死循环。主循环里做的事就四步:

  1. 把记忆里的消息发给 LLM,拿到回复
  2. 检查有没有 Final Answer,有就直接返回
  3. 检查有没有 Action 和 Action Input,有就执行工具
  4. 格式都不对,就把提示语加进记忆,让模型重新输出

有个细节:工具执行结果是拿 user 的身份追加到历史的。这个姿势看着怪,但模型训练时习惯的对话结构就是 user 说完 assistant 回,把工具结果塞进 user 位,下一轮模型读起来顺畅。

_parse_response 用的是正则提取,两个模式,一个抓 Final Answer,一个抓 Action 和 Action Input。这个方案够用,等以后工具多了想传更复杂的参数,再换结构化输出。_execute_tool 包了一层异常处理,工具不存在会返回可用工具列表,内部报错也会转成字符串,反正不让 Agent 直接崩掉。

第 8 步:创建入口文件

新建 main.py,实例化一个 ReActAgent,打印启动信息。就这么点事:

"""
AI Agent 入口文件
"""

from src.core.agent import ReActAgent

def main():
    agent = ReActAgent()

    print("=" * 60)
    print("🤖 AI Agent 已启动!输入 'quit' 退出,'reset' 重置对话")
    print("=" * 60)

代码里只写了 main 函数的基本几行。交互部分就是在 main 里加一个 while 循环,判断 quit 和 reset,分别触发退出和 agent.reset()。这段跟 Agent 核心没关系,不占篇幅。

第 8 步:创建命令行入口

代理写完了,得给它一个能对话的地方。新建 main.py,补上命令行交互。

# main.py
# 命令行交互入口

import os
from dotenv import load_dotenv

# 加载 .env 配置文件
load_dotenv()

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再见!")
            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()

逻辑不复杂:读输入,quit 就退出,reset 清空记忆,正常任务直接丢给 agent.run

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

三步迭代,模型自己决定先算后写,中间没人插过手。


第五阶段:扩展与优化(自选)

这个阶段按需做。想让 Agent 更能干,下面几个方向都值得折腾。

第 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)}"

注册也简单,在 ReActAgent__init__ 里加两行:

# 在 ReActAgent.__init__ 中添加
self.tools["get_time"] = TimeTool()
self.tools["get_weather"] = WeatherTool()

两行注册,Agent 立刻多出两个技能。

第 11 步:添加流式输出

不加流式也能跑,就是每次得等模型把整段话生成完才显示。交互起来有点闷。开个流式,字会一个个蹦出来,体验好不少。

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 步:持久化对话历史

之前的 ConversationMemory 把消息存在内存里,程序一停,对话记录就没了。想让 Agent 重启后还能接着聊,把它落盘到 JSON 文件:

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)} 条历史记录")

用的时候把 ReActAgent 里的 memory 换成 PersistentMemory 就行。


六、常见问题与排错指南

问:运行报错 AuthenticationError

.env 文件里的 API Key 检查一遍。最常见的是复制的时候多带了空格,或者末尾留着引号。

问:Agent 陷入死循环,一直调用同一个工具?

多半是 system prompt 写得不够明确,工具描述也有歧义。有个土办法,在 system prompt 里加一条:如果同一个工具连续调用 3 次仍然失败,请直接输出 Final Answer 说明原因。

问:LLM 输出格式不对,解析总是失败?

换更强的模型,比如 gpt-4o 替代 gpt-4o-mini。或者在 system prompt 里多给几个格式示例,也就是 few-shot。这两个办法效果都不错。

问:工具调用很慢,怎么优化?

三个方向:

  1. 给工具结果加缓存,相同输入直接返回缓存结果
  2. 不依赖顺序的工具调用改成并行执行
  3. 调低 MAX_ITERATIONS,逼 Agent 更高效地规划

问:想用本地模型(Ollama)怎么接入?

Ollama 提供的是 OpenAI 兼容接口,改三行 .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                   # 入口文件

八、总结

恭喜,Agent 的核心部件你已经全部搭了一遍。梳理一下掌握的东西:

知识点 掌握程度
AI Agent 的核心概念(LLM + Tools + Memory) 已掌握
ReAct 范式(Thought → Action → Observation) 已掌握
如何封装 LLM 调用客户端 已掌握
如何设计和实现工具系统 已掌握
如何管理对话上下文记忆 已掌握
如何解析 LLM 的结构化输出 已掌握
如何扩展 Agent 能力(添加新工具) 已掌握

接下来想深入的话,有几个方向可以走:

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

常见问题(FAQ)

AI Agent的三个核心组件分别是什么?

模型、工具和记忆。模型负责推理规划,工具执行具体操作(如计算、读写文件、网络请求),记忆存储历史对话和中间结果。三者配合实现Agent的自主任务执行,缺一不可。

用Python构建AI Agent需要哪些环境依赖?

需要安装openai、python-dotenv、requests和colorama等库。建议创建虚拟环境,配置API密钥,可使用OpenAI官方或DeepSeek、智谱GLM等兼容接口,设置base_url和model名称即可。

ReAct模式在AI Agent中起什么作用?

ReAct是推理和行动交替进行的循环。Agent根据目标自主规划步骤,调用工具执行,观察结果后继续规划,直到完成目标。它是AI Agent框架的核心设计思路,大部分框架都遵循此模式。

晓婷深圳
本文由 晓婷 审核,最后更新于 2026年8月12日
联系编辑 →
← 返回文章列表
分享到:微博

版权与免责声明:本文仅用于信息分享与交流,不构成任何形式的法律、投资、医疗或其他专业建议,也不构成对任何结果的承诺或保证。

文中提及的商标、品牌、Logo、产品名称及相关图片/素材,其权利归各自合法权利人所有。本站内容可能基于公开资料整理,亦可能使用 AI 辅助生成或润色;我们尽力确保准确与合规,但不保证完整性、时效性与适用性,请读者自行甄别并以官方信息为准。

若本文内容或素材涉嫌侵权、隐私不当或存在错误,请相关权利人/当事人联系本站,我们将及时核实并采取删除、修正或下架等处理措施。也请勿在评论或联系信息中提交身份证号、手机号、住址等个人敏感信息。