# AI Agent的规划与记忆机制
原理说明 ReAct(Reasoning + Acting)是一种将推理链(Thought)与工具调用(Action)交替进行的规划模式。Agent在每一步先思考当前状态(Thought),然后决定执行什么操作(Action),再根据观察结果(Observation)更新思考,形成循环。这种设计让Agent能够边推理边行动,动态调整策略。
典型示例 假设Agent需要回答“今天北京到上海的机票价格”。它会:
关键代码
import json
class ReActAgent:
def __init__(self, tools):
self.tools = {t.name: t for t in tools}
self.history = []
def step(self, thought):
# 模拟LLM输出:从thought生成action
if "查询" in thought:
action = {"name": "search_flights", "args": {"from": "北京", "to": "上海"}}
else:
action = {"name": "respond", "args": {"message": "无法处理"}}
return action
def run(self, user_input):
thought = f"用户输入: {user_input}"
while True:
action = self.step(thought)
tool = self.tools[action["name"]]
observation = tool.execute(**action["args"])
self.history.append({"thought": thought, "action": action, "observation": observation})
if action["name"] == "respond":
return observation
thought = f"基于观察: {observation},下一步思考"
# 工具定义
class Tool:
def __init__(self, name, func):
self.name = name
self.func = func
def execute(self, **kwargs):
return self.func(**kwargs)
def search_flights(from_city, to_city):
return f"{from_city}到{to_city}的机票价格为800元"
agent = ReActAgent([Tool("search_flights", search_flights), Tool("respond", lambda message: message)])
print(agent.run("今天北京到上海的机票价格"))
2. Tree-of-Thought(ToT)推理:多路径探索
原理说明
ToT将推理过程建模为树结构,每个节点代表一个中间思考状态。Agent同时探索多条推理路径,通过评估函数选择最有希望的路径继续扩展,避免陷入局部最优。适用于数学证明、代码生成等需要深度推理的任务。
典型示例
解决“24点游戏”:给定4个数字,用加减乘除得到24。ToT会生成多种运算顺序(如(1+2)×(3+4) vs 1+2×3+4),评估每条路径的潜力。
关键代码
import random
class ToTNode:
def __init__(self, state, parent=None):
self.state = state
self.parent = parent
self.children = []
self.value = 0
def expand(self, generate_fn, max_children=3):
for _ in range(max_children):
new_state = generate_fn(self.state)
child = ToTNode(new_state, self)
self.children.append(child)
def evaluate(self, eval_fn):
self.value = eval_fn(self.state)
return self.value
def tot_search(initial_state, generate_fn, eval_fn, max_depth=3):
root = ToTNode(initial_state)
best_node = root
queue = [root]
for depth in range(max_depth):
next_queue = []
for node in queue:
node.expand(generate_fn, max_children=2)
for child in node.children:
child.evaluate(eval_fn)
if child.value > best_node.value:
best_node = child
next_queue.append(child)
queue = next_queue
return best_node.state
# 示例:数字组合游戏
def generate_next(state):
if len(state) == 1:
return state
idx = random.randint(0, len(state)-2)
op = random.choice(['+', '-', '*', '/'])
new_val = eval(f"{state[idx]}{op}{state[idx+1]}")
new_state = state[:idx] + [new_val] + state[idx+2:]
return new_state
def evaluate(state):
return 1.0 / (abs(state[0] - 24) + 0.01) if len(state)==1 else 0.5
result = tot_search([1, 2, 3, 4], generate_next, evaluate, max_depth=3)
print(f"最佳状态: {result}")
3. 短期与长期记忆管理
原理说明
短期记忆(工作记忆)存储当前对话上下文,容量有限;长期记忆通过向量数据库持久化重要信息。Agent需要策略性地将短期记忆中的关键信息(如用户偏好、任务状态)转化为长期记忆,并在需要时检索。
典型示例
客服Agent:短期记忆保存当前对话轮次,长期记忆存储用户历史订单、投诉记录。当用户说“我之前投诉过”,Agent从向量库检索相关记录。
关键代码
import numpy as np
from collections import deque
class MemoryManager:
def __init__(self, short_term_capacity=10):
self.short_term = deque(maxlen=short_term_capacity)
self.long_term = {} # 简化版:字典模拟向量库
def add_to_short(self, item):
self.short_term.append(item)
def add_to_long(self, key, value, embedding):
# embedding为numpy向量
self.long_term[key] = {"value": value, "embedding": embedding}
def retrieve_from_long(self, query_embedding, top_k=3):
similarities = []
for key, data in self.long_term.items():
sim = np.dot(query_embedding, data["embedding"])
similarities.append((sim, key))
similarities.sort(reverse=True)
return [self.long_term[key]["value"] for _, key in similarities[:top_k]]
def get_context(self):
return list(self.short_term)
# 使用示例
mem = MemoryManager()
mem.add_to_short("用户: 查询订单")
mem.add_to_long("order_123", {"status": "已发货"}, np.array([0.1, 0.2, 0.3]))
query = np.array([0.15, 0.25, 0.35])
results = mem.retrieve_from_long(query, top_k=1)
print(f"检索结果: {results}")
print(f"短期记忆: {mem.get_context()}")
4. 向量记忆:语义检索
原理说明
向量记忆将文本转换为高维向量(Embedding),通过余弦相似度或欧氏距离进行语义检索。相比关键词搜索,它能理解“手机”和“移动设备”的语义关联,适用于开放域问答和知识库查询。
典型示例
Agent需要回答“如何重置密码”,从向量库中检索到“密码找回流程”文档。
关键代码
from sentence_transformers import SentenceTransformer
import numpy as np
class VectorMemory:
def __init__(self, model_name='all-MiniLM-L6-v2'):
self.model = SentenceTransformer(model_name)
self.vectors = []
self.texts = []
def add(self, text):
vec = self.model.encode(text)
self.vectors.append(vec)
self.texts.append(text)
def search(self, query, top_k=3):
query_vec = self.model.encode(query)
similarities = []
for i, vec in enumerate(self.vectors):
sim = np.dot(query_vec, vec) / (np.linalg.norm(query_vec)*np.linalg.norm(vec))
similarities.append((sim, i))
similarities.sort(reverse=True)
return [self.texts[i] for _, i in similarities[:top_k]]
# 使用示例
vm = VectorMemory()
vm.add("重置密码需要点击'忘记密码'链接")
vm.add("订单取消后款项将在3个工作日内退回")
vm.add("如何修改账户邮箱")
results = vm.search("密码找不到了", top_k=1)
print(f"最相关文档: {results}")
5. 工具调用模式
原理说明
Agent通过函数调用的方式与外部系统交互,典型模式包括:直接调用(同步)、异步回调、组合调用(多个工具串联)。关键在于定义统一的工具接口(名称、参数、返回值),并让LLM理解如何选择合适的工具。
典型示例
天气查询Agent:用户问“明天上海会下雨吗?” → Agent调用get_weather("上海", "2024-01-15") → 返回天气数据 → 生成回答。
关键代码
import requests
import json
class ToolRegistry:
def __init__(self):
self.tools = {}
def register(self, name, func, description, parameters):
self.tools[name] = {
"func": func,
"description": description,
"parameters": parameters
}
def call(self, name, **kwargs):
if name not in self.tools:
return {"error": f"Tool {name} not found"}
return self.tools[name]["func"](**kwargs)
def get_schema(self):
schema = []
for name, tool in self.tools.items():
schema.append({
"name": name,
"description": tool["description"],
"parameters": tool["parameters"]
})
return schema
# 注册工具
registry = ToolRegistry()
def get_weather(city, date):
# 模拟API调用
return f"{city}在{date}的天气:晴,25°C"
registry.register("get_weather", get_weather, "查询天气", {"city": "城市名", "date": "日期"})
# Agent推理选择工具
def agent_think(user_input):
if "天气" in user_input:
return {"tool": "get_weather", "args": {"city": "上海", "date": "2024-01-15"}}
return None
user_input = "明天上海天气怎么样?"
action = agent_think(user_input)
if action:
result = registry.call(action["tool"], **action["args"])
print(f"工具调用结果: {result}")
else:
print("无法处理")
二、实操步骤
项目:构建一个带记忆的智能客服Agent
1. 环境准备
- 安装依赖:`pip install sentence-transformers numpy`
- 创建项目文件夹`agent_project/`
2. 实现记忆模块
- 创建`memory.py`,包含短期记忆(deque)和向量记忆(基于sentence-transformers)
- 初始化向量库,添加示例知识文档(如“退货政策”、“物流查询”)
3. 实现工具模块
- 创建`tools.py`,定义`search_order`、`get_return_policy`等函数
- 使用ToolRegistry管理工具,并生成JSON Schema供LLM参考
4. 实现规划模块
- 创建`planner.py`,实现ReAct循环:
- 接收用户输入 → 生成Thought → 选择Action → 调用工具 → 获取Observation → 更新记忆 → 循环直到生成最终回答
5. 集成与测试
- 创建`main.py`,实例化所有组件
- 测试对话:
agent.run("我想退货,订单号是12345")
# 预期输出:检索订单信息 → 调用退货政策 → 生成回答
6. 优化与扩展
- 添加ToT模式处理复杂查询(如多商品比较)
- 实现长期记忆的自动清理策略(如LRU淘汰)
三、常见问题与故障排查
1. 问题:Agent陷入无限循环
- **原因**:推理链没有终止条件,或工具调用返回空结果导致重复思考
- **解决**:设置最大步数限制(如max_steps=10),并在工具返回空时强制生成最终回答
2. 问题:向量检索结果不相关
- **原因**:Embedding模型不适合领域(如通用模型对专业术语理解差)
- **解决**:更换领域专用模型(如`BAAI/bge-large-zh`),或增加文档预处理(分词、关键词提取)
3. 问题:工具参数解析错误
- **原因**:LLM生成的JSON格式不正确,或参数名与工具定义不匹配
- **解决**:在调用工具前添加JSON校验和参数映射,使用`json.loads`并捕获异常
4. 问题:短期记忆溢出
- **原因**:对话过长导致deque自动丢弃早期信息
- **解决**:实现摘要机制,定期将短期记忆压缩为摘要存入长期记忆
5. 问题:ToT搜索空间爆炸
- **原因**:树的宽度和深度设置过大,导致计算量指数增长
- **解决**:使用剪枝策略(如只保留价值最高的前3个节点),或限制最大节点数
四、总结与扩展学习
核心要点
- ReAct模式通过Thought-Action-Observation循环实现动态推理与行动
- ToT通过多路径探索提升复杂问题的解决能力
- 短期记忆管理对话上下文,长期记忆通过向量检索实现语义级知识复用
- 工具调用需要统一的接口定义和参数校验机制
扩展学习方向
- **多智能体协作**:研究AutoGen、CrewAI等框架,学习多个Agent如何分工协作
- **强化学习优化**:使用RLHF微调Agent的规划策略,使其更符合人类偏好
- **记忆压缩技术**:探索Memory-Sandwich、Retrieval-Augmented Generation等高级记忆管理方法
- **推荐资源**:
- 论文:`ReAct: Synergizing Reasoning and Acting in Language Models`
- 框架:LangChain(工具调用与记忆)、AutoGPT(自主Agent)
- 课程:DeepLearning.AI的《Building Systems with ChatGPT》