本文手把手教你用Python在2小时内构建一个能自主规划、调
AIAI Summary (BLUF)
本文手把手教你用Python在2小时内构建一个能自主规划、调用工具、完成任务的AI Agent,涵盖ReAct模式、三个核心组件、环境配置以及完整代码实现,适合想入门Agent开发的开发者。
核心洞察
这篇文章最有意思的点是它把 Agent 从热词拆成了三块:模型、工具、记忆。工具那部分尤其实在,代码直接给到,能跑。唯一我持保留意见的地方是计算器工具里用了 eval,自己折腾没问题,真要上生产得把表达式解析换成更安全的方式。
核心结论
- Agent 的核心结构由模型、工具、记忆三个组件构成:模型负责推理规划,工具负责执行操作,记忆负责保存中间状态,三者缺一不可。
- Agent 的运行基于 ReAct一种代理执行模式,结合推理和行动(Reasoning + Acting),使代理能够在思考后调用工具并观察结果。 循环:自主规划 → 调用工具 → 观察结果 → 继续规划,直到完成目标;当前主流 Agent 框架大多遵循这一思路。
- 教程实现了计算器、文件读写、网络请求三个工具,其中文件工具通过
ALLOWED_DIR和abspath校验限制读写范围,能防路径穿越,但startswith判断存在前缀绕过缺陷(如workspace_evil目录会被误放行)。 - 计算器工具使用
eval执行数学表达式,仅适合学习演示;上生产环境前必须替换为安全解析方案。 - 教程推荐新手使用 DeepSeekA high-performance code search and analysis tool designed for developers, utilizing advanced indexing and semantic analysis algorithms. 作为模型接口:价格约为 OpenAI 的二十分之一,API 格式完全兼容,国内访问稳定。
保姆级教程:从零搭建你的第一个 AI Agent人工智能代理,指能够感知环境、自主决策并执行任务的智能软件实体。在微软框架中,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 是兜底,防止模型陷入调工具的死循环。主循环里做的事就四步:
- 把记忆里的消息发给 LLMLarge Language Model, the underlying technology for generative AI.,拿到回复
- 检查有没有 Final Answer,有就直接返回
- 检查有没有 Action 和 Action Input,有就执行工具
- 格式都不对,就把提示语加进记忆,让模型重新输出
有个细节:工具执行结果是拿 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跨会话记忆,允许 AI 助手跨不同对话线程记住用户信息,通过 memory 参数启用。 换成 PersistentMemory 就行。
六、常见问题与排错指南
问:运行报错 AuthenticationError?
.env 文件里的 API Key 检查一遍。最常见的是复制的时候多带了空格,或者末尾留着引号。
问:Agent 陷入死循环,一直调用同一个工具?
多半是 system prompt 写得不够明确,工具描述也有歧义。有个土办法,在 system prompt 里加一条:如果同一个工具连续调用 3 次仍然失败,请直接输出 Final Answer 说明原因。
问:LLM 输出格式不对,解析总是失败?
换更强的模型,比如 gpt-4o 替代 gpt-4o-mini。或者在 system prompt 里多给几个格式示例,也就是 few-shot。这两个办法效果都不错。
问:工具调用很慢,怎么优化?
三个方向:
- 给工具结果加缓存,相同输入直接返回缓存结果
- 不依赖顺序的工具调用改成并行执行
- 调低 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工具,Agent用于执行具体操作的模块,如计算器、文件读写、网络请求等。 + 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用于从.env文件加载环境变量的Python库。、requests和colorama等库。建议创建虚拟环境Python开发中的隔离环境,用于管理项目依赖,避免不同项目间的包版本冲突。,配置API密钥,可使用OpenAI官方或DeepSeek、智谱GLM智谱AI提供的大模型API,兼容OpenAI格式,国内可用。等兼容接口,设置base_url和model名称即可。
ReAct模式在AI Agent中起什么作用?
ReAct是推理和行动交替进行的循环。Agent根据目标自主规划步骤,调用工具执行,观察结果后继续规划,直到完成目标。它是AI Agent框架的核心设计思路,大部分框架都遵循此模式。
版权与免责声明:本文仅用于信息分享与交流,不构成任何形式的法律、投资、医疗或其他专业建议,也不构成对任何结果的承诺或保证。
文中提及的商标、品牌、Logo、产品名称及相关图片/素材,其权利归各自合法权利人所有。本站内容可能基于公开资料整理,亦可能使用 AI 辅助生成或润色;我们尽力确保准确与合规,但不保证完整性、时效性与适用性,请读者自行甄别并以官方信息为准。
若本文内容或素材涉嫌侵权、隐私不当或存在错误,请相关权利人/当事人联系本站,我们将及时核实并采取删除、修正或下架等处理措施。也请勿在评论或联系信息中提交身份证号、手机号、住址等个人敏感信息。



