guyu09081. 项目概述：从零构建ReAct Agent的核心架构
在当今AI应用开发领域，ReAct（Reasoning + Acting）模式已经成为构建智能Agent的主流范式。与传统的单次问答不同，ReAct通过"思考-行动-观察"的循环机制，使AI系统能够像人类一样逐步解决问题。这种模式特别适合需要多步骤推理、工具调用和实时数据获取的复杂场景。
本项目将完全从零开始实现一个功能完整的ReAct Agent系统，不使用LangChain等现成框架，而是基于原生Python和OpenAI API构建。这种底层实现方式能让我们更深入理解Agent的工作原理，为后续定制开发打下坚实基础。
整个系统由三个核心模块组成：
- Agent核心框架：负责与大模型交互，管理对话历史
- 工具函数集合：提供具体功能实现（如计算器、数据查询）
- ReAct执行引擎：驱动"思考-行动-观察"的循环流程
这种架构设计具有以下优势：
- 轻量级：不依赖复杂框架，核心代码不到300行
- 易扩展：工具系统采用插件式设计，新增功能只需添加工具函数
- 可调试：每个推理步骤都清晰可见，便于问题排查
- 低成本：基于开源模型API，无需昂贵基础设施
需要模型API调用？ 免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 核心模块实现详解
2.1 Agent核心框架设计
Agent类是整个系统的大脑，负责与大模型API交互并维护对话上下文。其核心实现如下：
python复制# agent.py
import os
import dotenv
from openai import OpenAI
# 环境变量配置
dotenv.load_dotenv()
client = OpenAI(
api_key=os.getenv("OPENAI_API_KEY"),
base_url=os.getenv("OPENAI_API_BASE")
)
DEFAULT_MODEL = os.getenv("AI_MODEL", "deepseek-chat")
class Agent:
"""Agent推理框架"""
def __init__(self, system_prompt="", max_history=10):
self._messages = []
self.max_history = max_history
if system_prompt:
self._messages.append({"role": "system", "content": system_prompt})
def invoke(self, query: str) -> str:
"""调用Agent进行推理"""
self._messages.append({"role": "user", "content": query})
result = self.exec()
self._messages.append({"role": "assistant", "content": result})
# 历史消息压缩
if len(self._messages) > self.max_history:
self._messages = self._messages[:1] + self._messages[-self.max_history:]
return result
def exec(self) -> str:
"""执行推理，返回结果"""
completion = client.chat.completions.create(
model=DEFAULT_MODEL,
messages=self._messages,
temperature=0 # 确定性输出
)
return completion.choices[0].message.content
关键技术解析：
-
temperature参数控制：
- 设置为0确保输出确定性，这对工具调用至关重要
- 不同场景下的推荐值：
python复制# 推理任务（工具调用、逻辑推理） temperature = 0 # 创意任务（故事写作、头脑风暴） temperature = 0.7 - 1.0 # 平衡任务（需要一定创造性但也要准确） temperature = 0.3 - 0.5
-
对话历史管理：
- 采用滑动窗口机制控制历史消息长度
- 保留system消息（角色定义）和最近的N条对话
- 避免上下文过长导致的API成本增加和性能下降
-
消息格式规范：
- 严格遵循OpenAI的消息角色系统（system/user/assistant）
- 确保模型能正确理解上下文关系
2.2 工具系统实现
工具是Agent能力的延伸，每个工具都是一个独立的Python函数：
python复制# tools.py
import ast
from functools import lru_cache
@lru_cache(maxsize=100) # 添加缓存提高性能
def calculate(expression: str) -> str:
"""安全计算数学表达式
使用ast.literal_eval替代eval防止代码注入
"""
try:
node = ast.parse(expression, mode='eval')
if not all(isinstance(n, ast.NameConstant) or
isinstance(n, ast.Num) or
isinstance(n, ast.BinOp) for n in ast.walk(node)):
raise ValueError("Unsafe expression")
return str(eval(expression))
except Exception as e:
return f"计算错误: {str(e)}"
def ask_fruit_unit_price(fruit: str) -> str:
"""查询水果单价（模拟数据库查询）"""
price_map = {
"apple": 10,
"banana": 6,
"orange": 8,
"pear": 7
}
fruit = fruit.lower()
if fruit in price_map:
return f"{fruit.capitalize()}单价是 {price_map[fruit]}元/公斤"
return f"未找到{fruit}的价格信息"
工具设计最佳实践：
-
输入验证：
- 对传入参数进行类型和内容检查
- 使用安全函数（如ast.literal_eval）处理用户输入
-
错误处理：
- 捕获所有可能的异常
- 返回友好的错误信息供Agent理解
-
性能优化：
- 对频繁调用的工具添加缓存（如lru_cache）
- 避免重复计算和冗余操作
-
文档规范：
- 每个工具都应有清晰的docstring
- 说明参数、返回值和可能出现的错误
3. ReAct引擎实现
3.1 核心执行流程
ReAct引擎是系统的协调中心，驱动整个推理循环：
python复制# main.py
import re
from agent import Agent
from tools imp