1. 环境准备¶
首先导入必要的库并设置环境。
In [ ]:
Copied!
# 安装依赖（如果需要）
!pip install openai sentence-transformers rank-bm25 -q
import os
import re
import json
from typing import Dict, List, Tuple, Callable, Optional, Any
from dataclasses import dataclass
from datetime import datetime
# 设置OpenAI API Key（替换为你的key）
# os.environ['OPENAI_API_KEY'] = 'your-api-key-here'
print("✅ 环境准备完成")
# 安装依赖（如果需要）
!pip install openai sentence-transformers rank-bm25 -q
import os
import re
import json
from typing import Dict, List, Tuple, Callable, Optional, Any
from dataclasses import dataclass
from datetime import datetime
# 设置OpenAI API Key（替换为你的key）
# os.environ['OPENAI_API_KEY'] = 'your-api-key-here'
print("✅ 环境准备完成")
2. ReAct模式核心概念¶
2.1 什么是ReAct？¶
ReAct = Reasoning（推理） + Acting（行动）
核心循环：
Thought（思考） → Action（行动） → Observation（观察）
↑ ↓
└──────────────────────────────────────────────┘
直到得出最终答案
示例：
Question: "Python和JavaScript哪个性能更好？"
Thought 1: 我需要比较两种语言的性能特点
Action 1: Search["Python vs JavaScript performance comparison"]
Observation 1: Python通常在计算密集型任务上较慢，JavaScript在V8引擎下表现优异...
Thought 2: 我还需要了解它们在不同场景下的表现
Action 2: Search["Python JavaScript benchmark numeric operations"]
Observation 2: 在数值计算中，Python（使用NumPy）可以接近C的速度...
Thought 3: 我现在有足够信息来回答了
Action 3: Finish["两者各有优势：Python适合数据科学，JavaScript适合Web开发..."]
In [ ]:
Copied!
@dataclass
class Tool:
"""Agent工具"""
name: str
description: str
func: Callable
def __call__(self, *args, **kwargs) -> str:
"""调用工具"""
try:
result = self.func(*args, **kwargs)
return str(result)
except Exception as e:
return f"Error: {str(e)}"
class ToolRegistry:
"""工具注册表"""
def __init__(self):
self.tools: Dict[str, Tool] = {}
def register(self, tool: Tool):
"""注册工具"""
self.tools[tool.name] = tool
print(f"✅ 注册工具: {tool.name}")
def get_tool(self, name: str) -> Optional[Tool]:
"""获取工具"""
return self.tools.get(name)
def list_tools(self) -> List[str]:
"""列出所有工具"""
return list(self.tools.keys())
def get_tool_descriptions(self) -> str:
"""获取工具描述（用于prompt）"""
descriptions = []
for tool in self.tools.values():
descriptions.append(f"- {tool.name}: {tool.description}")
return "\n".join(descriptions)
# 创建工具注册表
registry = ToolRegistry()
print("✅ 工具系统初始化完成")
@dataclass
class Tool:
"""Agent工具"""
name: str
description: str
func: Callable
def __call__(self, *args, **kwargs) -> str:
"""调用工具"""
try:
result = self.func(*args, **kwargs)
return str(result)
except Exception as e:
return f"Error: {str(e)}"
class ToolRegistry:
"""工具注册表"""
def __init__(self):
self.tools: Dict[str, Tool] = {}
def register(self, tool: Tool):
"""注册工具"""
self.tools[tool.name] = tool
print(f"✅ 注册工具: {tool.name}")
def get_tool(self, name: str) -> Optional[Tool]:
"""获取工具"""
return self.tools.get(name)
def list_tools(self) -> List[str]:
"""列出所有工具"""
return list(self.tools.keys())
def get_tool_descriptions(self) -> str:
"""获取工具描述（用于prompt）"""
descriptions = []
for tool in self.tools.values():
descriptions.append(f"- {tool.name}: {tool.description}")
return "\n".join(descriptions)
# 创建工具注册表
registry = ToolRegistry()
print("✅ 工具系统初始化完成")
3.2 实现具体工具¶
In [ ]:
Copied!
# 工具1：模拟搜索
knowledge_base = {
"Python性能": "Python是一种解释型语言，通常执行速度较慢，但使用NumPy等库可以显著提升性能。适合数据科学和机器学习任务。",
"JavaScript性能": "JavaScript运行在V8引擎上，JIT编译使其在Web开发中性能优异。Node.js使其可用于服务器端。",
"RAG": "RAG（Retrieval-Augmented Generation）结合检索和生成，提供准确的上下文相关答案。",
"Agent": "Agent是能够自主感知环境、做出决策并执行行动的智能系统。"
}
def search_tool(query: str) -> str:
"""模拟搜索引擎"""
# 简单的关键词匹配
results = []
for keyword, info in knowledge_base.items():
if keyword.lower() in query.lower() or query.lower() in info.lower():
results.append(f"{keyword}: {info}")
if results:
return "\n".join(results)
return f"未找到关于'{query}'的信息"
registry.register(Tool(
name="Search",
description="搜索知识库。输入：搜索关键词",
func=search_tool
))
# 工具2：计算器
import operator
ops = {
'+': operator.add,
'-': operator.sub,
'*': operator.mul,
'/': operator.trued,
'**': operator.pow
}
def calculator_tool(expression: str) -> str:
"""简单的计算器"""
try:
# 安全地计算简单表达式
result = eval(expression, {"__builtins__": {}}, {})
return f"计算结果: {result}"
except:
return f"无法计算表达式: {expression}"
registry.register(Tool(
name="Calculator",
description="执行数学计算。输入：数学表达式，如 '2 + 2' 或 '10 ** 2'",
func=calculator_tool
))
# 工具3：文档检索（模拟）
documents = [
"RAG通过检索相关文档片段来增强生成能力",
"向量检索使用语义相似度查找文档",
"BM25检索使用词频统计查找文档"
]
def rag_tool(query: str) -> str:
"""RAG检索工具"""
# 简单的关键词匹配
results = []
for i, doc in enumerate(documents, 1):
if any(word in doc.lower() for word in query.lower().split()):
results.append(f"文档{i}: {doc}")
if results:
return "\n".join(results[:3])
return "未找到相关文档"
registry.register(Tool(
name="RAG",
description="检索相关文档。输入：查询问题",
func=rag_tool
))
print("\n📦 可用工具:")
for tool_name in registry.list_tools():
print(f" - {tool_name}")
# 工具1：模拟搜索
knowledge_base = {
"Python性能": "Python是一种解释型语言，通常执行速度较慢，但使用NumPy等库可以显著提升性能。适合数据科学和机器学习任务。",
"JavaScript性能": "JavaScript运行在V8引擎上，JIT编译使其在Web开发中性能优异。Node.js使其可用于服务器端。",
"RAG": "RAG（Retrieval-Augmented Generation）结合检索和生成，提供准确的上下文相关答案。",
"Agent": "Agent是能够自主感知环境、做出决策并执行行动的智能系统。"
}
def search_tool(query: str) -> str:
"""模拟搜索引擎"""
# 简单的关键词匹配
results = []
for keyword, info in knowledge_base.items():
if keyword.lower() in query.lower() or query.lower() in info.lower():
results.append(f"{keyword}: {info}")
if results:
return "\n".join(results)
return f"未找到关于'{query}'的信息"
registry.register(Tool(
name="Search",
description="搜索知识库。输入：搜索关键词",
func=search_tool
))
# 工具2：计算器
import operator
ops = {
'+': operator.add,
'-': operator.sub,
'*': operator.mul,
'/': operator.trued,
'**': operator.pow
}
def calculator_tool(expression: str) -> str:
"""简单的计算器"""
try:
# 安全地计算简单表达式
result = eval(expression, {"__builtins__": {}}, {})
return f"计算结果: {result}"
except:
return f"无法计算表达式: {expression}"
registry.register(Tool(
name="Calculator",
description="执行数学计算。输入：数学表达式，如 '2 + 2' 或 '10 ** 2'",
func=calculator_tool
))
# 工具3：文档检索（模拟）
documents = [
"RAG通过检索相关文档片段来增强生成能力",
"向量检索使用语义相似度查找文档",
"BM25检索使用词频统计查找文档"
]
def rag_tool(query: str) -> str:
"""RAG检索工具"""
# 简单的关键词匹配
results = []
for i, doc in enumerate(documents, 1):
if any(word in doc.lower() for word in query.lower().split()):
results.append(f"文档{i}: {doc}")
if results:
return "\n".join(results[:3])
return "未找到相关文档"
registry.register(Tool(
name="RAG",
description="检索相关文档。输入：查询问题",
func=rag_tool
))
print("\n📦 可用工具:")
for tool_name in registry.list_tools():
print(f" - {tool_name}")
In [ ]:
Copied!
class ReActAgent:
"""
ReAct Agent实现
核心循环：Thought → Action → Observation
"""
def __init__(self,
tool_registry: ToolRegistry,
llm_client: Any = None,
max_iterations: int = 5):
self.tool_registry = tool_registry
self.llm_client = llm_client # 可以是真实的LLM客户端
self.max_iterations = max_iterations
self.history = [] # 执行历史
def _build_prompt(self,
query: str,
steps: List[Dict]) -> str:
"""构建提示词"""
tool_desc = self.tool_registry.get_tool_descriptions()
prompt = f"""你是一个智能助手，可以使用工具来回答问题。
可用工具:
{tool_desc}
使用以下格式：
Thought: 你的思考
Action: 工具名称[输入]
当你知道答案时，使用：
Action: Finish[答案]
问题: {query}
"""
# 添加之前的步骤
for step in steps:
if 'thought' in step:
prompt += f"\nThought {step['step_num']}: {step['thought']}"
if 'action' in step:
prompt += f"\nAction {step['step_num']}: {step['action']}"
if 'observation' in step:
prompt += f"\nObservation {step['step_num']}: {step['observation']}"
prompt += "\n\nThought:"
return prompt
def _parse_response(self, response: str) -> Tuple[str, str, str]:
"""解析LLM响应
Returns:
(thought, action, action_input)
"""
# 提取Thought
thought_match = re.search(r'Thought:\s*(.+?)(?=\nAction:|$)', response, re.DOTALL)
thought = thought_match.group(1).strip() if thought_match else ""
# 提取Action
action_match = re.search(r'Action:\s*(\w+)\[(.*)\]', response)
if action_match:
action = action_match.group(1)
action_input = action_match.group(2).strip()
else:
action = "Finish"
action_input = response.strip()
return thought, action, action_input
def _simulate_llm(self, prompt: str, query: str, step_num: int) -> Tuple[str, str, str]:
"""模拟LLM响应（用于演示）
在实际应用中，替换为真实的LLM调用
"""
# 基于查询和步骤数的简单规则
if step_num == 1:
if "性能" in query or "比较" in query:
thought = "我需要搜索相关信息来比较"
action = "Search"
action_input = query
elif "计算" in query or any(op in query for op in ['+', '-', '*', '/', '**']):
thought = "这是一个计算问题"
action = "Calculator"
action_input = query
else:
thought = "我需要查找相关文档"
action = "RAG"
action_input = query
else:
thought = "我现在有足够信息来回答了"
action = "Finish"
action_input = "基于搜索结果，我可以回答这个问题。"
return thought, action, action_input
def run(self, query: str, verbose: bool = True) -> Dict:
"""运行Agent
Args:
query: 用户查询
verbose: 是否打印详细过程
Returns:
{'answer': str, 'steps': List[Dict], 'iterations': int}
"""
steps = []
thought = ""
for iteration in range(self.max_iterations):
step_num = len(steps) + 1
step = {'step_num': step_num}
# 构建提示
prompt = self._build_prompt(query, steps)
# 获取LLM响应
if self.llm_client:
response = self.llm_client.generate(prompt)
thought, action, action_input = self._parse_response(response)
else:
thought, action, action_input = self._simulate_llm(prompt, query, step_num)
step['thought'] = thought
step['action'] = f"{action}[{action_input}]"
if verbose:
print(f"\n--- 步骤 {step_num} ---")
print(f"💭 Thought: {thought}")
print(f"⚡ Action: {action}[{action_input}]")
# 检查是否完成
if action == "Finish":
if verbose:
print(f"\n✅ 最终答案: {action_input}")
return {
'answer': action_input,
'steps': steps,
'iterations': iteration + 1
}
# 执行工具
tool = self.tool_registry.get_tool(action)
if not tool:
observation = f"错误：找不到工具 '{action}'"
else:
observation = tool(action_input)
step['observation'] = observation
steps.append(step)
if verbose:
print(f"👁️ Observation: {observation[:200]}..." if len(observation) > 200 else f"👁️ Observation: {observation}")
# 更新思考
thought = f"Observation: {observation}"
# 达到最大迭代次数
return {
'answer': "未能在限制步骤内完成",
'steps': steps,
'iterations': self.max_iterations
}
print("✅ ReAct Agent类定义完成")
class ReActAgent:
"""
ReAct Agent实现
核心循环：Thought → Action → Observation
"""
def __init__(self,
tool_registry: ToolRegistry,
llm_client: Any = None,
max_iterations: int = 5):
self.tool_registry = tool_registry
self.llm_client = llm_client # 可以是真实的LLM客户端
self.max_iterations = max_iterations
self.history = [] # 执行历史
def _build_prompt(self,
query: str,
steps: List[Dict]) -> str:
"""构建提示词"""
tool_desc = self.tool_registry.get_tool_descriptions()
prompt = f"""你是一个智能助手，可以使用工具来回答问题。
可用工具:
{tool_desc}
使用以下格式：
Thought: 你的思考
Action: 工具名称[输入]
当你知道答案时，使用：
Action: Finish[答案]
问题: {query}
"""
# 添加之前的步骤
for step in steps:
if 'thought' in step:
prompt += f"\nThought {step['step_num']}: {step['thought']}"
if 'action' in step:
prompt += f"\nAction {step['step_num']}: {step['action']}"
if 'observation' in step:
prompt += f"\nObservation {step['step_num']}: {step['observation']}"
prompt += "\n\nThought:"
return prompt
def _parse_response(self, response: str) -> Tuple[str, str, str]:
"""解析LLM响应
Returns:
(thought, action, action_input)
"""
# 提取Thought
thought_match = re.search(r'Thought:\s*(.+?)(?=\nAction:|$)', response, re.DOTALL)
thought = thought_match.group(1).strip() if thought_match else ""
# 提取Action
action_match = re.search(r'Action:\s*(\w+)\[(.*)\]', response)
if action_match:
action = action_match.group(1)
action_input = action_match.group(2).strip()
else:
action = "Finish"
action_input = response.strip()
return thought, action, action_input
def _simulate_llm(self, prompt: str, query: str, step_num: int) -> Tuple[str, str, str]:
"""模拟LLM响应（用于演示）
在实际应用中，替换为真实的LLM调用
"""
# 基于查询和步骤数的简单规则
if step_num == 1:
if "性能" in query or "比较" in query:
thought = "我需要搜索相关信息来比较"
action = "Search"
action_input = query
elif "计算" in query or any(op in query for op in ['+', '-', '*', '/', '**']):
thought = "这是一个计算问题"
action = "Calculator"
action_input = query
else:
thought = "我需要查找相关文档"
action = "RAG"
action_input = query
else:
thought = "我现在有足够信息来回答了"
action = "Finish"
action_input = "基于搜索结果，我可以回答这个问题。"
return thought, action, action_input
def run(self, query: str, verbose: bool = True) -> Dict:
"""运行Agent
Args:
query: 用户查询
verbose: 是否打印详细过程
Returns:
{'answer': str, 'steps': List[Dict], 'iterations': int}
"""
steps = []
thought = ""
for iteration in range(self.max_iterations):
step_num = len(steps) + 1
step = {'step_num': step_num}
# 构建提示
prompt = self._build_prompt(query, steps)
# 获取LLM响应
if self.llm_client:
response = self.llm_client.generate(prompt)
thought, action, action_input = self._parse_response(response)
else:
thought, action, action_input = self._simulate_llm(prompt, query, step_num)
step['thought'] = thought
step['action'] = f"{action}[{action_input}]"
if verbose:
print(f"\n--- 步骤 {step_num} ---")
print(f"💭 Thought: {thought}")
print(f"⚡ Action: {action}[{action_input}]")
# 检查是否完成
if action == "Finish":
if verbose:
print(f"\n✅ 最终答案: {action_input}")
return {
'answer': action_input,
'steps': steps,
'iterations': iteration + 1
}
# 执行工具
tool = self.tool_registry.get_tool(action)
if not tool:
observation = f"错误：找不到工具 '{action}'"
else:
observation = tool(action_input)
step['observation'] = observation
steps.append(step)
if verbose:
print(f"👁️ Observation: {observation[:200]}..." if len(observation) > 200 else f"👁️ Observation: {observation}")
# 更新思考
thought = f"Observation: {observation}"
# 达到最大迭代次数
return {
'answer': "未能在限制步骤内完成",
'steps': steps,
'iterations': self.max_iterations
}
print("✅ ReAct Agent类定义完成")
In [ ]:
Copied!
# 创建Agent
agent = ReActAgent(
tool_registry=registry,
max_iterations=5
)
print("✅ ReAct Agent已创建")
print(f"📦 可用工具: {', '.join(registry.list_tools())}")
# 创建Agent
agent = ReActAgent(
tool_registry=registry,
max_iterations=5
)
print("✅ ReAct Agent已创建")
print(f"📦 可用工具: {', '.join(registry.list_tools())}")
5.2 实验1：性能比较问题¶
In [ ]:
Copied!
# 实验1：比较Python和JavaScript性能
query1 = "Python和JavaScript哪个性能更好？"
result1 = agent.run(query1, verbose=True)
print("\n" + "="*50)
print("📊 执行摘要:")
print(f" 迭代次数: {result1['iterations']}")
print(f" 步骤数: {len(result1['steps'])}")
print(f" 最终答案: {result1['answer'][:100]}...")
# 实验1：比较Python和JavaScript性能
query1 = "Python和JavaScript哪个性能更好？"
result1 = agent.run(query1, verbose=True)
print("\n" + "="*50)
print("📊 执行摘要:")
print(f" 迭代次数: {result1['iterations']}")
print(f" 步骤数: {len(result1['steps'])}")
print(f" 最终答案: {result1['answer'][:100]}...")
5.3 实验2：计算问题¶
In [ ]:
Copied!
# 实验2：数学计算
query2 = "计算 15 ** 2 + 10"
result2 = agent.run(query2, verbose=True)
print("\n" + "="*50)
print("📊 执行摘要:")
print(f" 迭代次数: {result2['iterations']}")
print(f" 答案: {result2['answer']}")
# 实验2：数学计算
query2 = "计算 15 ** 2 + 10"
result2 = agent.run(query2, verbose=True)
print("\n" + "="*50)
print("📊 执行摘要:")
print(f" 迭代次数: {result2['iterations']}")
print(f" 答案: {result2['answer']}")
5.4 实验3：RAG检索¶
In [ ]:
Copied!
# 实验3：文档检索
query3 = "什么是RAG？"
result3 = agent.run(query3, verbose=True)
print("\n" + "="*50)
print("📊 执行摘要:")
print(f" 迭代次数: {result3['iterations']}")
print(f" 答案: {result3['answer']}")
# 实验3：文档检索
query3 = "什么是RAG？"
result3 = agent.run(query3, verbose=True)
print("\n" + "="*50)
print("📊 执行摘要:")
print(f" 迭代次数: {result3['iterations']}")
print(f" 答案: {result3['answer']}")
In [ ]:
Copied!
class ReActAgentWithMemory(ReActAgent):
"""带记忆的ReAct Agent"""
def __init__(self, *args, memory_size: int = 5, **kwargs):
super().__init__(*args, **kwargs)
self.memory = [] # 存储历史查询和结果
self.memory_size = memory_size
def _build_prompt(self, query: str, steps: List[Dict]) -> str:
"""构建包含记忆的提示"""
prompt = super()._build_prompt(query, steps)
# 添加相关记忆
if self.memory:
memory_text = "\n\n历史对话（可能相关）:\n"
for i, (old_query, old_answer) in enumerate(self.memory[-3:], 1):
memory_text += f"{i}. Q: {old_query}\n A: {old_answer[:100]}...\n"
prompt = prompt.replace("问题:", memory_text + "\n当前问题:")
return prompt
def run(self, query: str, verbose: bool = True) -> Dict:
"""运行并保存记忆"""
result = super().run(query, verbose)
# 保存到记忆
self.memory.append((query, result['answer']))
if len(self.memory) > self.memory_size:
self.memory.pop(0)
return result
# 创建带记忆的Agent
agent_with_memory = ReActAgentWithMemory(
tool_registry=registry,
max_iterations=5,
memory_size=5
)
print("✅ 带记忆的Agent已创建")
class ReActAgentWithMemory(ReActAgent):
"""带记忆的ReAct Agent"""
def __init__(self, *args, memory_size: int = 5, **kwargs):
super().__init__(*args, **kwargs)
self.memory = [] # 存储历史查询和结果
self.memory_size = memory_size
def _build_prompt(self, query: str, steps: List[Dict]) -> str:
"""构建包含记忆的提示"""
prompt = super()._build_prompt(query, steps)
# 添加相关记忆
if self.memory:
memory_text = "\n\n历史对话（可能相关）:\n"
for i, (old_query, old_answer) in enumerate(self.memory[-3:], 1):
memory_text += f"{i}. Q: {old_query}\n A: {old_answer[:100]}...\n"
prompt = prompt.replace("问题:", memory_text + "\n当前问题:")
return prompt
def run(self, query: str, verbose: bool = True) -> Dict:
"""运行并保存记忆"""
result = super().run(query, verbose)
# 保存到记忆
self.memory.append((query, result['answer']))
if len(self.memory) > self.memory_size:
self.memory.pop(0)
return result
# 创建带记忆的Agent
agent_with_memory = ReActAgentWithMemory(
tool_registry=registry,
max_iterations=5,
memory_size=5
)
print("✅ 带记忆的Agent已创建")
6.2 记忆演示¶
In [ ]:
Copied!
# 第一次查询
print("🔍 第一次查询:")
result1 = agent_with_memory.run("什么是Agent？", verbose=True)
# 第二次查询（可以使用记忆）
print("\n\n🔍 第二次查询（上下文相关）:")
result2 = agent_with_memory.run("它有什么特点？", verbose=True)
print("\n" + "="*50)
print(f"💾 记忆中的对话数: {len(agent_with_memory.memory)}")
# 第一次查询
print("🔍 第一次查询:")
result1 = agent_with_memory.run("什么是Agent？", verbose=True)
# 第二次查询（可以使用记忆）
print("\n\n🔍 第二次查询（上下文相关）:")
result2 = agent_with_memory.run("它有什么特点？", verbose=True)
print("\n" + "="*50)
print(f"💾 记忆中的对话数: {len(agent_with_memory.memory)}")
In [ ]:
Copied!
import matplotlib.pyplot as plt
import matplotlib.patches as mpatches
from matplotlib.patches import FancyBboxPatch, FancyArrowPatch
def visualize_agent_execution(result: Dict, query: str):
"""可视化Agent执行轨迹"""
fig, ax = plt.subplots(figsize=(12, 6))
ax.set_xlim(0, 10)
ax.set_ylim(0, 6)
ax.axis('off')
# 标题
ax.text(5, 5.5, f"Agent执行轨迹: {query}",
ha='center', fontsize=14, weight='bold')
# 绘制步骤
steps = result['steps']
n_steps = len(steps)
colors = ['#FFE5E5', '#E5F5FF', '#E5FFE5', '#FFFFE5', '#F0E5FF']
for i, step in enumerate(steps):
x = 1 + (i % 5) * 1.8
y = 4 - (i // 5) * 2.5
# 步骤框
box = FancyBboxPatch((x-0.7, y-0.6), 1.4, 1.2,
boxstyle="round,pad=0.1",
facecolor=colors[i % 5],
edgecolor='black', linewidth=2)
ax.add_patch(box)
# 步骤编号
ax.text(x, y+0.3, f"步骤{i+1}", ha='center', weight='bold')
# Action
action = step.get('action', 'N/A')
ax.text(x, y-0.1, f"{action[:20]}...",
ha='center', fontsize=8, style='italic')
# 绘制箭头
if i < n_steps - 1:
next_x = 1 + ((i+1) % 5) * 1.8
next_y = 4 - ((i+1) // 5) * 2.5
arrow = FancyArrowPatch((x, y-0.6), (next_x, next_y+0.6),
arrowstyle='->', mutation_scale=20,
color='#666666', linewidth=2)
ax.add_patch(arrow)
# 最终答案
if result['answer']:
ax.text(5, 0.3, f"✅ 答案: {result['answer'][:60]}...",
ha='center', fontsize=10,
bbox=dict(boxstyle='round', facecolor='#E5FFE5', alpha=0.8))
plt.tight_layout()
plt.show()
# 可视化之前的执行
visualize_agent_execution(result1, query1)
import matplotlib.pyplot as plt
import matplotlib.patches as mpatches
from matplotlib.patches import FancyBboxPatch, FancyArrowPatch
def visualize_agent_execution(result: Dict, query: str):
"""可视化Agent执行轨迹"""
fig, ax = plt.subplots(figsize=(12, 6))
ax.set_xlim(0, 10)
ax.set_ylim(0, 6)
ax.axis('off')
# 标题
ax.text(5, 5.5, f"Agent执行轨迹: {query}",
ha='center', fontsize=14, weight='bold')
# 绘制步骤
steps = result['steps']
n_steps = len(steps)
colors = ['#FFE5E5', '#E5F5FF', '#E5FFE5', '#FFFFE5', '#F0E5FF']
for i, step in enumerate(steps):
x = 1 + (i % 5) * 1.8
y = 4 - (i // 5) * 2.5
# 步骤框
box = FancyBboxPatch((x-0.7, y-0.6), 1.4, 1.2,
boxstyle="round,pad=0.1",
facecolor=colors[i % 5],
edgecolor='black', linewidth=2)
ax.add_patch(box)
# 步骤编号
ax.text(x, y+0.3, f"步骤{i+1}", ha='center', weight='bold')
# Action
action = step.get('action', 'N/A')
ax.text(x, y-0.1, f"{action[:20]}...",
ha='center', fontsize=8, style='italic')
# 绘制箭头
if i < n_steps - 1:
next_x = 1 + ((i+1) % 5) * 1.8
next_y = 4 - ((i+1) // 5) * 2.5
arrow = FancyArrowPatch((x, y-0.6), (next_x, next_y+0.6),
arrowstyle='->', mutation_scale=20,
color='#666666', linewidth=2)
ax.add_patch(arrow)
# 最终答案
if result['answer']:
ax.text(5, 0.3, f"✅ 答案: {result['answer'][:60]}...",
ha='center', fontsize=10,
bbox=dict(boxstyle='round', facecolor='#E5FFE5', alpha=0.8))
plt.tight_layout()
plt.show()
# 可视化之前的执行
visualize_agent_execution(result1, query1)
7.2 性能统计¶
In [ ]:
Copied!
# 统计多个查询的性能
queries = [
"什么是RAG？",
"计算 100 * 5",
"Python性能如何",
"Agent的特点"
]
stats = []
for query in queries:
result = agent.run(query, verbose=False)
stats.append({
'query': query,
'iterations': result['iterations'],
'steps': len(result['steps'])
})
# 显示统计
print("\n📊 性能统计:")
print("-" * 50)
for stat in stats:
print(f"查询: {stat['query']}")
print(f" 迭代次数: {stat['iterations']}")
print(f" 步骤数: {stat['steps']}")
print()
# 平均值
avg_iterations = sum(s['iterations'] for s in stats) / len(stats)
avg_steps = sum(s['steps'] for s in stats) / len(stats)
print(f"平均迭代次数: {avg_iterations:.1f}")
print(f"平均步骤数: {avg_steps:.1f}")
# 统计多个查询的性能
queries = [
"什么是RAG？",
"计算 100 * 5",
"Python性能如何",
"Agent的特点"
]
stats = []
for query in queries:
result = agent.run(query, verbose=False)
stats.append({
'query': query,
'iterations': result['iterations'],
'steps': len(result['steps'])
})
# 显示统计
print("\n📊 性能统计:")
print("-" * 50)
for stat in stats:
print(f"查询: {stat['query']}")
print(f" 迭代次数: {stat['iterations']}")
print(f" 步骤数: {stat['steps']}")
print()
# 平均值
avg_iterations = sum(s['iterations'] for s in stats) / len(stats)
avg_steps = sum(s['steps'] for s in stats) / len(stats)
print(f"平均迭代次数: {avg_iterations:.1f}")
print(f"平均步骤数: {avg_steps:.1f}")
8. 与真实LLM集成（可选）¶
如果你有OpenAI API密钥，可以尝试真实的LLM：
In [ ]:
Copied!
class OpenAIClient:
"""简单的OpenAI客户端封装"""
def __init__(self, api_key: str):
self.api_key = api_key
def generate(self, prompt: str) -> str:
"""生成响应"""
# 这里使用真实的OpenAI API
# 实际实现需要导入openai库
return "模拟LLM响应"
# 使用示例（需要API密钥）
# api_key = os.environ.get('OPENAI_API_KEY')
# if api_key:
# llm_client = OpenAIClient(api_key)
# agent_llm = ReActAgent(
# tool_registry=registry,
# llm_client=llm_client,
# max_iterations=5
# )
# result = agent_llm.run("什么是RAG？")
print("ℹ️ 要使用真实LLM，请设置OPENAI_API_KEY环境变量")
class OpenAIClient:
"""简单的OpenAI客户端封装"""
def __init__(self, api_key: str):
self.api_key = api_key
def generate(self, prompt: str) -> str:
"""生成响应"""
# 这里使用真实的OpenAI API
# 实际实现需要导入openai库
return "模拟LLM响应"
# 使用示例（需要API密钥）
# api_key = os.environ.get('OPENAI_API_KEY')
# if api_key:
# llm_client = OpenAIClient(api_key)
# agent_llm = ReActAgent(
# tool_registry=registry,
# llm_client=llm_client,
# max_iterations=5
# )
# result = agent_llm.run("什么是RAG？")
print("ℹ️ 要使用真实LLM，请设置OPENAI_API_KEY环境变量")