在 第一部分 中，我们理解了 ReAct 智能体的记忆问题。现在，让我们动手构建一个真正能"记住"的智能体。
我们将从零开始实现一个 ReAct 循环，然后逐步加入 Mem0 记忆层。
ReAct 循环的核心结构
在开始之前，让我们先理解 ReAct 循环的本质。它是一个 Thought → Action → Observation 的迭代过程。
while not task_complete and steps < max_steps:
# 1. Thought: 智能体分析当前状态并制定计划
thought = llm.generate(f"当前状态: {state}\n下一步应该做什么？")
# 2. Action: 选择并执行工具
action = parse_action(thought)
observation = execute_tool(action)
# 3. Update State: 更新智能体的状态
state += f"\nObservation: {observation}"
实现基础 ReAct 智能体（无记忆）
让我们先实现一个简单的 ReAct 智能体，它可以使用两个工具：search 和 calculator。
import openai
from typing import Dict, List
class ReActAgent:
def __init__(self, model="gpt-4"):
self.model = model
self.tools = {
"search": self._search,
"calculator": self._calculator
}
def _search(self, query: str) -> str:
"""模拟搜索工具"""
# 实际应用中，这里会调用真实的搜索 API
return f"搜索结果：关于 '{query}' 的信息..."
def _calculator(self, expression: str) -> str:
"""简单的计算器"""
try:
result = eval(expression)
return f"计算结果：{result}"
except:
return "计算错误"
def run(self, task: str, max_steps: int = 5) -> str:
state = f"任务：{task}\n"
for step in range(max_steps):
# Thought: 生成下一步行动
prompt = f"""你是一个 ReAct 智能体。当前状态：
{state}
可用工具：
- search(query): 搜索信息
- calculator(expression): 计算数学表达式
请按照以下格式回复：
Thought: [你的思考过程]
Action: [工具名称](参数)
如果任务完成，回复：
Thought: 任务完成
Answer: [最终答案]
"""
response = openai.ChatCompletion.create(
model=self.model,
messages=[{"role": "user", "content": prompt}]
)
text = response.choices[0].message.content
state += f"\n{text}\n"
# 检查是否完成
if "Answer:" in text:
return text.split("Answer:")[1].strip()
# 解析并执行 Action
if "Action:" in text:
action_line = text.split("Action:")[1].strip()
tool_name, params = self._parse_action(action_line)
if tool_name in self.tools:
observation = self.tools[tool_name](params)
state += f"Observation: {observation}\n"
return "达到最大步数限制"
def _parse_action(self, action_str: str):
# 简单的解析逻辑: tool_name(params)
tool_name = action_str.split("(")[0]
params = action_str.split("(")[1].rstrip(")")
return tool_name, params
# 使用示例
agent = ReActAgent()
result = agent.run("今天纽约的天气是多少度？如果是华氏 60 度，换算成摄氏度是多少？")
print(result)
执行过程示例：
Thought: 我需要先搜索纽约的天气
Action: search(纽约天气)
Observation: 搜索结果：纽约今天 60°F
Thought: 现在我需要将华氏转摄氏，公式是 (F-32)*5/9
Action: calculator((60-32)*5/9)
Observation: 计算结果：15.56
Thought: 任务完成
Answer: 纽约今天 60°F，约 15.6°C
问题：智能体没有记忆
现在，如果我们问智能体："刚才的温度是多少？"，它会回答："我不知道"。
因为每次调用 run() 都是全新的上下文。
加入 Mem0 记忆层
现在让我们升级这个智能体，给它加上"长期记忆"。
1. 安装 Mem0
pip install mem0ai
2. 改造 ReActAgent
from mem0 import Memory
class MemoryEnabledReActAgent(ReActAgent):
def __init__(self, model="gpt-4", user_id="default_user"):
super().__init__(model)
self.memory = Memory()
self.user_id = user_id
def run(self, task: str, max_steps: int = 5) -> str:
# **第 1 步：检索相关记忆**
memories = self.memory.search(
query=task,
user_id=self.user_id,
limit=5
)
# 将记忆注入到初始状态中
memory_context = "\n".join([
f"- {m['memory']}" for m in memories
]) if memories else "无相关记忆"
state = f"""任务：{task}
相关记忆：
{memory_context}
"""
for step in range(max_steps):
# ... [ReAct 循环代码，与之前相同] ...
# **第 2 步：保存新知识**
# 在每次 Observation 后，保存新信息
if "Observation:" in state:
self.memory.add(
messages=[{
"role": "user",
"content": f"任务：{task}\n观察到：{observation}"
}],
user_id=self.user_id
)
return "..."
3. 测试记忆功能
agent = MemoryEnabledReActAgent(user_id="alice")
# 第一次对话
result = agent.run("帮我搜索 Python 装饰器的定义")
# 智能体会搜索并保存结果到记忆中
# 5 分钟后...
result = agent.run("刚才我问的那个概念，能再详细解释一下吗？")
# 智能体会从记忆中检索到"Python 装饰器"的定义！
记忆的智能去重
Mem0 的强大之处在于它不会简单地"追加"记忆。它会：
- 检测重复：如果智能体再次搜索"Python 装饰器"，Mem0 会识别出这是重复信息。
- 更新而不是追加：如果新搜索到的定义更详细，旧记录会被更新。
- 删除过时信息：如果智能体发现"Python 3.11 已发布"，之前的"Python 3.10 是最新版本"会被标记为过时。
这一切都是 自动 完成的，通过 Mem0 内部的 LLM 驱动的记忆管理管道。
下一步是什么？
我们已经实现了一个基础的带记忆 ReAct 智能体。但在生产环境中，你通常不会从零手写 ReAct 循环。
在 下一篇文章 中，我们将探索如何将 Mem0 集成到 LangChain 这样的成熟框架中，获得更强大的工具调用、更优雅的代码结构，以及对复杂任务的支持。