AI Agent的规划与记忆机制

🏷️ L3 📊 advanced ⏱️ 45分钟 🏷️ Agent,ReAct,规划,记忆,工具调用,前沿

# AI Agent的规划与记忆机制

概述

本课程深入解析AI Agent的核心推理与记忆架构,重点覆盖ReAct规划模式、Tree-of-Thought高级推理、短期/长期/向量记忆管理以及工具调用机制。学完本课后,你将能够设计具备复杂任务分解、多步推理和持久化记忆能力的Agent系统,适用于自动化客服、智能助手、科研辅助等需要连续交互与决策的场景。

一、核心知识讲解

1. ReAct规划模式:推理与行动的结合

原理说明 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》

在博海学习网开始学习 →