源码级别解析 · 源码解析 · 智能体架构模式
2026-05-16 | 每日技术深度解读
ReAct代表了智能体技术的重要突破
ReAct = Reasoning + Acting = 智能体的核心能力
ReAct平衡了智能体的智能性和实用性
ReAct提供了标准化的智能体开发范式
这个循环构成了ReAct智能体的核心工作机制
ReAct的成功依赖于多个技术组件的协同工作
class ReActAgent:
def __init__(self, llm, tools):
self.llm = llm
self.tools = tools
self.thoughts = []
self.actions = []
self.observations = []
def step(self, observation):
# Step 1: 观察当前状态
current_state = self.format_state(observation)
# Step 2: LLM推理
prompt = self.create_react_prompt(current_state)
response = self.llm.generate(prompt)
# Step 3: 执行行动
action = self.parse_action(response)
result = self.execute_action(action)
# 记录过程
self.thoughts.append(response)
self.actions.append(action)
self.observations.append(result)
return resultReAct智能体的基本实现框架
LLM是ReAct智能体的'大脑'
丰富的工具集是ReAct智能体的'手和脚'
良好的状态管理是智能体可靠性的关键
健壮的错误处理让智能体更可靠
好的Prompt设计是ReAct成功的核心
SYSTEM: 你是一个ReAct智能体，遵循以下步骤进行推理和行动：
1. 观察当前环境状态
2. 基于观察进行推理分析
3. 选择合适的工具执行行动
4. 观察行动结果
5. 重复直到问题解决
可用工具：
- search(query): 网络搜索
- calculate(expression): 数学计算
- code_execute(code): 代码执行
- api_call(endpoint): API调用
用户问题：{user_query}标准的ReAct提示模板
可视化帮助理解智能体的工作原理
ReAct特别适合需要结合推理和行动的任务
ReAct提供了更完整的智能体能力
实现高效ReAct智能体面临多项技术挑战
优化策略显著提升智能体的响应速度
可解释性是ReAct的重要特性
安全性是ReAct智能体部署的重要考虑
良好的扩展性让ReAct适应不同需求
ReAct已经在多个领域成功应用
这些组件共同构成了ReAct的完整框架
LLM推理引擎是ReAct的大脑和决策中心
class LLMReasoningEngine:
def __init__(self, model_name="gpt-4"):
self.model = model_name
self.cache = {}
self.max_tokens = 2000
self.temperature = 0.7
def reason(self, observation, context):
# 检查缓存
cache_key = f"{observation}_{context}"
if cache_key in self.cache:
return self.cache[cache_key]
# 构建推理prompt
prompt = self.build_reasoning_prompt(observation, context)
# 调用LLM
response = self.call_llm(prompt)
# 缓存结果
self.cache[cache_key] = response
return response
def call_llm(self, prompt):
# 实际调用LLM API
# 这里使用OpenAI API作为示例
import openai
response = openai.ChatCompletion.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
max_tokens=self.max_tokens,
temperature=self.temperature
)
return response.choices[0].message.contentLLM推理引擎的核心实现
工具管理器让ReAct能够灵活集成各种外部能力
class ToolManager:
def __init__(self):
self.tools = {}
self.tool_stats = {}
def register_tool(self, name, tool_func, description=""):
"""注册新工具"""
self.tools[name] = {
'function': tool_func,
'description': description,
'stats': {'calls': 0, 'success': 0, 'errors': 0}
}
def call_tool(self, name, *args, **kwargs):
"""调用工具"""
if name not in self.tools:
raise ValueError(f"Tool {name} not found")
tool = self.tools[name]
try:
result = tool['function'](*args, **kwargs)
tool['stats']['calls'] += 1
tool['stats']['success'] += 1
return result
except Exception as e:
tool['stats']['calls'] += 1
tool['stats']['errors'] += 1
raise e
def get_tool_stats(self, name):
"""获取工具统计信息"""
return self.tools[name]['stats'] if name in self.tools else None工具管理器的核心功能实现
状态跟踪器确保智能体的连续性和可靠性
class StateTracker:
def __init__(self):
self.dialogue_history = []
self.tool_usage = {}
self.intermediate_results = {}
self.error_states = set()
def add_to_history(self, role, content):
"""添加到对话历史"""
self.dialogue_history.append({"role": role, "content": content})
def record_tool_usage(self, tool_name, args, result):
"""记录工具使用情况"""
if tool_name not in self.tool_usage:
self.tool_usage[tool_name] = []
self.tool_usage[tool_name].append({
'timestamp': datetime.now(),
'args': args,
'result': result
})
def store_result(self, key, value):
"""存储中间结果"""
self.intermediate_results[key] = value
def get_result(self, key):
"""获取存储的结果"""
return self.intermediate_results.get(key)状态跟踪器的实现代码
循环控制器确保ReAct智能体的安全运行
class LoopController:
def __init__(self, max_iterations=10, timeout_seconds=30):
self.max_iterations = max_iterations
self.timeout_seconds = timeout_seconds
self.current_iteration = 0
self.start_time = None
def start_loop(self):
"""开始循环"""
self.current_iteration = 0
self.start_time = time.time()
def should_continue(self, current_result=None):
"""判断是否应该继续循环"""
# 检查迭代次数
if self.current_iteration >= self.max_iterations:
return False, "达到最大迭代次数"
# 检查超时
if time.time() - self.start_time > self.timeout_seconds:
return False, "执行超时"
# 检查任务是否完成
if self.is_task_complete(current_result):
return False, "任务已完成"
return True, "继续执行"
def increment_iteration(self):
"""增加迭代计数"""
self.current_iteration += 1
def is_task_complete(self, result):
"""判断任务是否完成"""
# 这里可以根据具体任务定义完成条件
return result and "完成" in result or "解决" in result循环控制器的核心逻辑
完善的错误处理确保智能体的健壮性
class ErrorHandler:
def __init__(self, max_retries=3):
self.max_retries = max_retries
def handle_tool_error(self, error, tool_name, args):
"""处理工具调用错误"""
if self.max_retries <= 0:
return None, "工具调用失败，已达到最大重试次数"
# 根据错误类型采取不同策略
if "timeout" in str(error):
return self.retry_with_timeout(tool_name, args)
elif "connection" in str(error):
return self.retry_with_backoff(tool_name, args)
else:
return None, f"工具调用失败: {str(error)}"
def retry_with_timeout(self, tool_name, args):
"""带超时的重试"""
import time
time.sleep(1) # 等待1秒
# 实现重试逻辑
pass
def retry_with_backoff(self, tool_name, args):
"""指数退避重试"""
import time
time.sleep(2) # 等待2秒
# 实现退避重试
pass错误处理的实现策略
性能优化让ReAct智能体更高效
class PerformanceOptimizer:
def __init__(self):
self.cache = {}
self.tool_parallelism = True
self.max_parallel_tools = 3
def cache_result(self, key, value, ttl=3600):
"""缓存结果"""
from datetime import datetime, timedelta
self.cache[key] = {
'value': value,
'expires': datetime.now() + timedelta(seconds=ttl)
}
def get_cached_result(self, key):
"""获取缓存结果"""
if key in self.cache:
entry = self.cache[key]
if entry['expires'] > datetime.now():
return entry['value']
else:
del self.cache[key]
return None
def execute_parallel_tools(self, tools, args_list):
"""并行执行多个工具"""
if not self.tool_parallelism:
# 串行执行
return [self.call_tool(tool, *args) for tool, args in zip(tools, args_list)]
# 并行执行（简化示例）
import concurrent.futures
results = []
with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_parallel_tools) as executor:
futures = [executor.submit(self.call_tool, tool, *args) for tool, args in zip(tools, args_list)]
for future in concurrent.futures.as_completed(futures):
results.append(future.result())
return results性能优化的核心实现
监控和日志帮助优化智能体性能
合理的架构确保ReAct智能体的稳定运行
扩展能力让ReAct适应不同场景需求
ReAct技术正在快速发展，未来可期
好的实践让ReAct开发更高效
丰富的学习资源帮助掌握ReAct技术
ReAct代表了智能体技术的重要发展方向
感谢阅读！
访问 https://atcfu.com/ai-articles/react-agent-framework/ 回顾本文