AI智能体（九）：智能体记忆
神译局是36氪旗下编译团队，关注科技、商业、职场、生活等领域，重点介绍国外的新技术、新观点、新风向。
编者按：2025年是AI智能体元年。本系列文章旨在介绍AI智能体的概念、类型、原理、架构、开发等，为进一步了解AI智能体提供入门知识。本文为系列文章的第九篇，文章来自编译。
想象一下：每次和朋友聊天，对方都会清空记忆——每段对话都从零开始。没有记忆，没有上下文，没有进展。这种交流只会让人尴尬、疲惫又疏离。而遗憾的是，这正是当下大多数AI系统的现状。它们足够聪明，却缺失了关键要素：记忆能力。
让我们先探讨AI领域"记忆"的本质及其重要性。
1 记忆的本质
1.1 当前AI的"伪记忆"假象
ChatGPT或编程助手看似智能，直到你发现需要重复输入相同的指令或偏好。要打造具备学习进化能力和协作能力的智能体，真正的记忆系统不可或缺——它不只是锦上添花，而是核心基建。
通过上下文窗口和提示词工程营造的"伪记忆"，让许多人误以为AI已有记忆能力。实则当今绝大多数智能体都是无状态的，既无法从历史交互中学习，也无法随时间演进适应。
要实现从无状态工具到真正智能的自主化智能体（有状态的）的跨越，我们需要赋予它们记忆能力——而非仅靠扩大提示框或优化检索。
1.2 AI智能体的记忆指什么？
在AI智能体语境中，记忆是跨时间、跨任务、多轮交互中保留并调用相关信息的能力。它使智能体能够追溯过往事件，并运用这些信息优化未来行为。
记忆不等于存储聊天记录或向提示框塞入更多字符，而是构建持续演进的持久化内部状态——即使间隔数周数月，仍能影响智能体的每次交互。
智能体记忆由三大支柱构成：
状态感知：理解当下发生的事件
持久存储：跨会话保留知识
信息筛选：判断记忆价值
三者共同成就了前所未有的能力：连续性。
1.3 记忆在智能体架构中的位置
无状态智能体（无记忆）与有状态智能体（有记忆）之别
典型智能体组件：
负责推理生成的LLM
策略规划模块（如ReAct/AutoGPT式）
工具/API调用接口
文档/历史数据检索器
核心问题在于：这些组件都无法记住昨日之事。没有内部状态，没有演进认知，没有记忆能力。
引入记忆层后：
智能体架构中的记忆层 | 来源：mem0
该架构将智能体从一次性工具转变为持续成长的协作伙伴。
1.4 上下文窗口 ≠ 记忆系统
常见误区是认为扩大上下文窗口就能替代记忆系统。但这种方法存在本质缺陷：
成本制约：更多字符=更高费用与延迟
功能局限：上下文窗口仅维持单会话一致性，记忆系统实现跨会话智能。即使百万级字符窗口，缺乏持久化、优先级和显著性判断，仍无法达成真正智能。
1.5 RAG与记忆系统的本质差异
虽然检索增强生成（RAG）和记忆系统都通过检索支持LLM，但解决的问题截然不同：
RAG推理时导入外部知识到提示框，用于基于文档事实生成回答
RAG本质上属于无状态：无历史交互感知，无视用户身份，不关联当前查询与过往对话
而记忆系统创造连续性：捕获用户偏好、历史查询、决策结果及失败记录，使其作用于未来交互。
本质区别：
RAG让智能体回答更准确
记忆系统让智能体行为更智能
理想方案是双轨并行：RAG提供知识支持，记忆系统塑造行为模式。
2 智能体的记忆类型
AI智能体的记忆基础分为两类：
短期记忆：单次交互中的即时上下文
长期记忆：跨会话、跨任务、跨时间的持久化知识
就像人类认知机制，这两类记忆功能各异：短期记忆维持当下对话连贯性，长期记忆实现学习进化与个性化适应。
👉 简易区分法则：
短期记忆 = AI此刻对话中的"即时记忆"
长期记忆 = AI多次对话后"习得并能回溯"的知识
记忆类型对比
2.1 短期记忆（工作记忆）
AI系统最基础的记忆形式，如同人记住对话中的上一句话。包含了：
对话历史：近期消息及顺序
工作记忆：临时变量与状态
注意力上下文：当前对话焦点
2.2 长期记忆
进阶AI应用通过长期记忆实现跨会话信息留存，具体包含：
2.2.1 程序性记忆
定义智能体固有行为逻辑的"肌肉记忆"，直接编码于系统中。
从简单模板到复杂推理流程的底层逻辑
例：自动执行"优先处理API文档邮件"或"技术问题回复采用更友好语气"等内化流程
2.2.2 情景记忆（案例库）
记录用户特定历史交互的"数字相册"，支撑连续性与个性化服务。
例：记忆"上次回复客户延期请求时语气生硬引发摩擦"
例：识别"含'简单疑问'的邮件往往需要长篇技术解答"
2.2.3 语义记忆（事实库）
通过向量检索或RAG获取的客观知识"百科全书"。
例：存储"Alice负责API文档"或"John偏好晨会"等事实
特征：脱离具体场景的独立知识留存
3 记忆管理实战
跨会话共享上下文需专业工具支持，LangGraph提供两类核心记忆管理方案：
短期记忆：跟踪单会话内的消息流
长期记忆：跨会话存储用户/应用级数据
当长对话超出LLM上下文窗口时，解决方案包括：
截断法：删除首尾N条消息（调用LLM前）
摘要法：浓缩早期消息为摘要替代
永久删除：从LangGraph状态中清除消息
自定义策略：消息过滤等创新方案
4 记忆写入机制
人类在睡眠中巩固长期记忆，智能体则需特殊设计。记忆写入时机与方式主要有两类，一类是热路径写入，一类是后台写入。
4.1 热路径写入（运行时）
优势：
实时更新，新记忆立即可用
高透明度：用户可获知记忆创建状态
案例：
ChatGPT通过save_memories工具动态更新记忆，根据消息内容决策调用时机
4.2 后台写入（异步任务）
优势：
消除主应用延迟
隔离应用逻辑与记忆管理
灵活调度避免冗余操作
5 实现短期记忆
短期记忆（线程级持久化）让智能体能追踪多轮对话。实现步骤：
Python
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.memory import InMemorySaver
model = init_chat_model(model="anthropic:claude-3-5-haiku-latest")
def call_model(state: MessagesState):
response = model.invoke(state["messages"])
return {"messages": response}
builder = StateGraph(MessagesState)
builder.add_node(call_model)
builder.add_edge(START, "call_model")
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "1"}}
# 首次对话
for chunk in graph.stream(
{"messages": [{"role": "user", "content": "hi! I'm bob"}]},
config,
stream_mode="values",
):
chunk["messages"][-1].pretty_print()
# 后续对话（记忆生效）
for chunk in graph.stream(
{"messages": [{"role": "user", "content": "what's my name?"}]},
config,
stream_mode="values",
):
chunk["messages"][-1].pretty_print()
▸ 执行效果：
================================ Human Message =================================
hi! I'm bob
================================== Ai Message ==================================
Hi Bob! How are you doing today? Is there anything I can help you with?
================================ Human Message =================================
what's my name?
================================== Ai Message ==================================
Your name is Bob.
5.1 生产环境方案
使用数据库支持的检查点（如MongoDB）：
Python
pip install -U pymongo langgraph langgraph-checkpoint-mongodb
Python
from langgraph.checkpoint.mongodb import MongoDBSaver
DB_URI = "localhost:27017"
with MongoDBSaver.from_conn_string(DB_URI) as checkpointer:
# 构建流程图（同前）
graph = builder.compile(checkpointer=checkpointer)
# 对话流执行（同前）
5.2 子图处理
主图编译时自动传播检查点到子图：
Python
# 子图构建
subgraph_builder = StateGraph(State)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph = subgraph_builder.compile()
# 主图构建（关键：仅主图需checkpointer）
builder.add_node("node_1", subgraph)
graph = builder.compile(checkpointer=checkpointer) # 检查点仅在此设置
5.3 检查点管理
查看线程状态：
Python
config = {"configurable": {"thread_id": "1"}}
graph.get_state(config)
# 返回示例：
StateSnapshot(
values={'messages': [对话历史记录]},
metadata={'thread_id': '1', 'step': 4},
created_at='2025-05-05T16:01:24.680462+00:00'
)
查看历史检查点：
Python
list(graph.get_state_history(config))
# 返回按时间倒序的检查点列表
删除线程所有检查点：
Python
thread_id = "1"
checkpointer.delete_thread(thread_id) # 彻底清除相关记忆
6 实现长期记忆
长期记忆（跨线程持久化）存储用户/应用级数据，适用于需记忆用户偏好的场景（如聊天机器人）。核心实现：
Python
import uuid
from langgraph.store.memory import InMemoryStore
from langgraph.store.base import BaseStore
# 构建流程图时注入store
def call_model(
state: MessagesState,
config: RunnableConfig,
*, # 关键：通过关键字参数接收store
store: BaseStore,
):
user_id = config["configurable"]["user_id"]
namespace = ("memories", user_id)
# 语义检索历史记忆
memories = store.search(namespace, query=str(state["messages"][-1].content))
info = "\n".join([d.value["data"] for d in memories])
system_msg = f"用户背景信息：{info}" # 注入系统提示
# 主动记忆机制：当用户要求记住时
last_message = state["messages"][-1]
if "remember" in last_message.content.lower():
memory = "用户名叫Bob"
store.put(namespace, str(uuid.uuid4()), {"data": memory}) # 存储新记忆
# 生成响应（含记忆上下文）
response = model.invoke(
[{"role": "system", "content": system_msg}] + state["messages"]
)
return {"messages": response}
# 初始化存储与检查点
store = InMemoryStore()
graph = builder.compile(
checkpointer=checkpointer,
store=store, # 注入长期记忆存储
)
# 跨线程测试（同一用户不同会话）
config_user1 = {"configurable": {"user_id": "1", "thread_id": "1"}}
graph.stream(
{"messages": [{"role": "user", "content": "记住：我叫Bob"}]},
config_user1
)
config_user1_new_thread = {"configurable": {"user_id": "1", "thread_id": "2"}}
graph.stream(
{"messages": [{"role": "user", "content": "我叫什么名字？"}]},
config_user1_new_thread
)
执行效果：
================================ Human Message =================================
记住：我叫Bob
================================== Ai Message ==================================
好的Bob，我会记住你叫Bob。今天有什么需要帮忙的吗？
================================ Human Message =================================
我叫什么名字？
================================== Ai Message ==================================
你叫Bob。
6.1 生产环境方案
使用PostgreSQL存储长期记忆：
Python
pip install -U "psycopg[binary,pool]" langgraph langgraph-checkpoint-postgres
Python
from langgraph.store.postgres import PostgresStore
from langgraph.checkpoint.postgres import PostgresSaver
DB_URI = "postgresql://用户:密码@地址:端口/数据库"
with PostgresStore.from_conn_string(DB_URI) as store, \
PostgresSaver.from_conn_string(DB_URI) as checkpointer:
# 首次使用需初始化表（取消注释运行一次）
# store.setup()
# checkpointer.setup()
# 构建流程图（同前，含store注入逻辑）
graph = builder.compile(
checkpointer=checkpointer,
store=store
)
# 执行对话流（同前）
6.2 语义搜索增强
启用向量检索实现记忆的语义匹配：
Python
from langchain.embeddings import init_embeddings
# 创建带语义索引的存储
embeddings = init_embeddings("openai:text-embedding-3-small")
store = InMemoryStore(
index={
"embed": embeddings, # 绑定嵌入模型
"dims": 1536 # 向量维度
}
)
# 预存记忆（实际场景中动态写入）
store.put(("user_123", "memories"), "1", {"text": "我喜欢披萨"})
store.put(("user_123", "memories"), "2", {"text": "我是水管工"})
def chat(state, *, store: BaseStore):
# 基于最新消息语义检索
items = store.search(
("user_123", "memories"),
query=state["messages"][-1].content, # 用最新用户消息检索
limit=2
)
# 整合记忆到系统提示
memories = "\n".join(item.value["text"] for item in items)
system_prompt = f"## 用户记忆片段\n{memories}" if memories else ""
# 后续调用LLM（略）
智能体记忆设计指南
规划记忆系统时需思考：
记忆内容类型
事实知识 / 过往事件摘要 / 行为规则与风格？
记忆形成时机
何时创建记忆？由用户触发还是智能体自主判断？
记忆存储位置
存于提示词？向量数据库？这直接决定了检索方式。
7 构建邮件智能体：全流程实现
通过融合三类长期记忆，我们将打造能自主进化、深度个性化的邮件助手：
自动识别：发现特定客户邮件需在24小时内跟进
风格适应：对外正式邮件 vs 团队轻松沟通
场景记忆：无需重复解释项目背景
智能预判：自动分类需人工处理/可自动回复的邮件
工作流
工作流：
开始 → 邮件分诊(情景+程序记忆) → 决策 → 回复生成(语义+程序记忆) → 结束
7.1 环境配置
Python
!pip install langmem langchain_community python-dotenv --quiet
import os
import warnings
from dotenv import load_dotenv
from typing import TypedDict, Literal, Annotated
# 核心组件
from langgraph.graph import StateGraph, START, END, add_messages
from langgraph.store.memory import InMemoryStore
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langchain.prompts import PromptTemplate
# 记忆管理工具
from langmem import create_manage_memory_tool, create_search_memory_tool
# 环境初始化
load_dotenv()
warnings.filterwarnings("ignore", category=RuntimeWarning)
# 配置用户和模型
USER_ID = "test_user"
CONFIG = {"configurable": {"langgraph_user_id": USER_ID}}
llm = init_chat_model("openai:gpt-4o-mini")
store = InMemoryStore(index={"embed": "openai:text-embedding-3-small"}) # 语义记忆存储
7.2 定义智能体"大脑"：状态机
Python
class State(TypedDict):
"""邮件处理流程的实时状态"""
email_input: dict # 原始邮件数据
messages: Annotated[list, add_messages] # 对话历史
triage_result: str # 分诊结果：ignore/notify/respond
7.3 邮件分诊中心（情景记忆驱动）
步骤1：定义决策结构
Python
from pydantic import BaseModel, Field
class Router(BaseModel):
"""邮件分类结构化输出"""
reasoning: str = Field(description="决策推理过程")
classification: Literal["ignore", "respond", "notify"] = Field(
description="邮件处理类型：忽略/回复/通知人工"
)
llm_router = llm.with_structured_output(Router) # 绑定结构化输出
步骤2：情景记忆格式化
Python
def format_few_shot_examples(examples):
"""将情景记忆转化为示例样本"""
return "\n\n".join(
f"发件人：{eg.value['email']['author']}\n"
f"主题：{eg.value['email']['subject']}\n"
f"内容：{eg.value['email']['email_thread'][:300]}...\n\n"
f"历史分类：{eg.value['label']}"
for eg in examples
)
步骤3：分诊函数实现
Python
def triage_email(state: State, config: dict, store: InMemoryStore) -> dict:
"""基于情景记忆的邮件分诊"""
email = state["email_input"]
user_id = config["configurable"]["langgraph_user_id"]
# 检索相似历史邮件（情景记忆）
examples = store.search(("email_assistant", user_id, "examples"), str(email))
# 构建动态提示模板
prompt = PromptTemplate.from_template("""
你是一名邮件分诊助手，请分类以下邮件：
发件人：{author}
收件人：{to}
主题：{subject}
内容：{email_thread}
分类选项：忽略(ignore)/回复(respond)/通知人工(notify)
历史参考案例：
{examples}
""").format(examples=format_few_shot_examples(examples), **email)
# 调用分类模型
result = llm_router.invoke([HumanMessage(content=prompt)])
return {"triage_result": result.classification}
7.4 工具定义（语义记忆支持）
基础工具
Python
@tool
def write_email(to: str, subject: str, content: str) -> str:
"""邮件撰写工具（对接实际邮件API）"""
print(f"[邮件发送] 给 {to}\n主题：{subject}\n内容：{content}")
return "邮件已发送"
@tool
def check_calendar_availability(day: str) -> str:
"""日历检查工具（对接日历API）"""
return f"{day}的空闲时段：09:00, 14:00, 16:00"
记忆增强工具
Python
# 语义记忆管理工具
manage_memory_tool = create_manage_memory_tool(
namespace=("email_assistant", "{langgraph_user_id}", "collection")
)
search_memory_tool = create_search_memory_tool(
namespace=("email_assistant", "{langgraph_user_id}", "collection")
)
# 工具全集
tools = [write_email, check_calendar_availability, manage_memory_tool, search_memory_tool]
7.5 构建图：把各个部分连接起来
现在，咱们把所有东西整合到一个统一的工作流里：
```python
def create_basic_email_agent(store):
"""
一个用来创建基础邮件处理代理的工厂函数，这个代理只用到了情景记忆。
这个函数创建了一个更简单的工作流，主要是为了做对比，它只用到了：
1. 情景记忆：过去的邮件例子，用来做小样本学习
Args:
store: 包含情景记忆的 InMemoryStore
Returns:
CompiledGraph: 具有基础记忆能力的可执行工作流
"""
# 定义工作流
workflow = StateGraph(State)
# 使用基础的分类函数（只用情景记忆）
workflow.add_node("triage", lambda state, config: triage_email(state, config, store))
# 创建一个使用静态提示词的基础响应代理
response_agent = create_react_agent(
tools=tools,
prompt=create_agent_prompt,
store=store,
model=llm
)
workflow.add_node("response_agent", response_agent)
def route_based_on_triage(state):
if state["triage_result"] == "respond":
return "response_agent"
else:
return END
# 路由逻辑保持不变
workflow.add_edge(START, "triage")
workflow.add_conditional_edges("triage", route_based_on_triage,
{
"response_agent": "response_agent",
END: END
})
# 编译并返回图
return workflow.compile(store=store)
我们的智能体看起来是这样子的：
```python
from langchain_core.runnables.graph import MermaidDrawMethod
from IPython.display import display, Image
display(
Image(
create_basic_email_agent(store).get_graph().draw_mermaid_png(
draw_method=MermaidDrawMethod.API,
)
)
)
这个工作流为我=智能体定义了如今的逻辑路径：
首先，用情景记忆对收到的邮件进行分类
如果需要回复，就启动带语义记忆的响应代理
否则，就结束流程（针对“忽略”或“通知”类的邮件）
这就像在工厂里设置流水线上的工位一样——每个部分都有自己的活儿，但它们会协同工作，一起造出最终的产品。
7.6 跑起来试试！（顺便存点记忆）
是时候把我们的智能体拉出来遛遛了：
```python
# 示例数据和初始化
# ===============================================================================
# 用于测试的示例邮件
email_input = {
"author": "Alice Smith <alice.smith@company.com>",
"to": "John Doe <john.doe@company.com>",
"subject": "关于API文档的一个小问题",
"email_thread": """嗨，John,
我正在看API文档，发现有几个端点不见了。你能帮忙看看吗？
谢谢,
Alice""",
}
# 初始提示词
initial_triage_prompt = """你是一个邮件分类助理。请对下面的邮件进行分类：
发件人: {author}
收件人: {to}
主题: {subject}
正文: {email_thread}
分类为 'ignore'（忽略）, 'notify'（通知）, 或 'respond'（回复）。
这里有一些以前的分类例子：
{examples}
"""
initial_response_prompt = """你是一个乐于助人的助理。请使用包括记忆工具在内的可用工具来帮助用户。"""
```
咱们再往情景记忆里加个训练样本，帮智能体以后能认出垃圾邮件：
```python
def initialize_memory():
"""用默认的例子和提示词来初始化记忆"""
# 把小样本例子加到情景记忆里
example1 = {
"email": {
"author": "垃圾邮件营销员 <spam@example.com>",
"to": "John Doe <john.doe@company.com>",
"subject": "大减价！！！",
"email_thread": "立即购买我们的产品，享受5折优惠！",
},
"label": "ignore",
}
store.put(("email_assistant", USER_ID, "examples"), "spam_example", example1)
# 用默认的提示词初始化程序化记忆
store.put(("email_assistant", USER_ID, "prompts"), "triage_prompt", initial_triage_prompt)
store.put(("email_assistant", USER_ID, "prompts"), "response_prompt", initial_response_prompt)
```
这就好比在训练一个新来的助理：“看到这种邮件没？这些都可以放心忽略。” 我们给的例子越多，智能体的理解就会变得越精细。
7.7 添加程序化记忆（更新指令）——最后润色一下！
现在，轮到最复杂的记忆系统了，程序化记忆能让我们的代理根据反馈来改进它自己的指令。
咱们来创建一个能从记忆里提取指令的分类函数版本：
```python
def triage_email_with_procedural_memory(state: State, config: dict, store: InMemoryStore) -> dict:
"""
使用情景记忆和程序化记忆进行高级邮件分类。
这是自适应学习系统的核心，它结合了：
1. 程序化记忆：可以随时间改进的动态提示词
2. 情景记忆：来自过去分类的小样本例子
Args:
state: 包含 email_input 的当前工作流状态
config: 带有 user_id 用于记忆命名空间的配置
store: 包含两种记忆类型的 InMemoryStore
Returns:
dict: 带有 triage_result 分类结果的更新后状态
"""
email = state["email_input"]
user_id = config["configurable"]["langgraph_user_id"]
print(f"分类：正在分析来自 {email['author']} 的邮件，主题是：'{email['subject']}'")
# 检索当前的分类提示词（程序化记忆）
current_prompt_template = store.get(("email_assistant", user_id, "prompts"), "triage_prompt")
print("程序化记忆：已检索到当前分类提示词")
# 确保提示词模板是字符串
if not isinstance(current_prompt_template, str):
current_prompt_template = str(current_prompt_template) # 如果是 Item 对象，则转换为字符串
# 从记忆中检索相关例子（情景记忆）
namespace = ("email_assistant", user_id, "examples")
examples = store.search(namespace, query=str(email))
formatted_examples = format_few_shot_examples(examples)
print(f"情景记忆：从过去的分类中找到了 {len(examples)} 个相关例子")
# 格式化提示词
prompt = PromptTemplate.from_template(current_prompt_template).format(examples=formatted_examples, **email)
messages = [HumanMessage(content=prompt)]
result = llm_router.invoke(messages)
print(f"分类结果：{result.classification}")
print(f"推理过程：{result.reasoning}")
return {"triage_result": result.classification}
```
这个函数把程序化记忆（当前的提示词模板）和情景记忆（相关的例子）结合起来，用来做分类决策。
现在，咱们来创建一个函数，它能根据反馈来改进我们的提示词：
```python
def optimize_prompts(feedback: str, config: dict, store: InMemoryStore):
"""
程序化记忆学习：根据性能反馈优化提示词。
这个函数实现了核心的学习机制，可以实现逐步改进：
1. 通过反馈分析当前提示词的表现
2. 使用AI驱动的优化来改进提示词
3. 用更好的提示词更新程序化记忆
4. 未来的代理实例会自动使用改进后的提示词
Args:
feedback: 描述性能问题的人类反馈
config: 带有 user_id 用于访问记忆的配置
store: 用于更新程序化记忆的 InMemoryStore
Returns:
str: 关于所做改进的确认信息
演变示例：
初始: "今天我能为您做点什么？"
反馈后: "关于API文档，我能为您提供什么帮助？"
"""
print("\n优化：开始提示词改进流程...")
user_id = config["configurable"]["langgraph_user_id"]
# 获取当前提示词
print("检索：从程序化记忆中检索当前提示词")
triage_prompt = store.get(("email_assistant", user_id, "prompts"), "triage_prompt").value
response_prompt = store.get(("email_assistant", user_id, "prompts"), "response_prompt").value
# 根据我们的实际邮件创建一个更相关的测试例子
sample_email = {
"author": "Alice Smith <alice.smith@company.com>",
"to": "John Doe <john.doe@company.com>",
"subject": "关于API文档的一个小问题",
"email_thread": "嗨，John, 我正在看API文档，发现有几个端点不见了。你能帮忙看看吗？谢谢, Alice",
}
print("分析：创建带有反馈的对话轨迹")
# 创建优化器
optimizer = create_multi_prompt_optimizer(llm)
# 创建一个更相关的带反馈的对话轨迹
conversation = [
{"role": "system", "content": response_prompt},
{"role": "user", "content": f"我收到了这封邮件: {sample_email}"},
{"role": "assistant", "content": "今天我能为您做点什么？"}
]
# 格式化提示词
prompts = [
{"name": "triage", "prompt": triage_prompt},
{"name": "response", "prompt": response_prompt}
]
# 更相关的轨迹
trajectories = [(conversation, {"feedback": feedback})]
print("优化：使用AI根据反馈改进提示词...")
result = optimizer.invoke({"trajectories": trajectories, "prompts": prompts})
# 提取改进后的提示词
improved_triage_prompt = next(p["prompt"] for p in result if p["name"] == "triage")
improved_response_prompt = next(p["prompt"] for p in result if p["name"] == "response")
# 为API文档问题追加具体指令
improved_triage_prompt = improved_triage_prompt + "\n\n要特别注意关于API文档或缺失端点的邮件——这些是高优先级的，应该总是被分类为'respond'（回复）。"
improved_response_prompt = improved_response_prompt + "\n\n在回复关于文档或API问题的邮件时，要确认提到的具体问题，并提供具体的帮助，而不是泛泛的回应。"
print("存储：在程序化记忆中更新提示词")
# 存储改进后的提示词
store.put(("email_assistant", user_id, "prompts"), "triage_prompt", improved_triage_prompt)
store.put(("email_assistant", user_id, "prompts"), "response_prompt", improved_response_prompt)
print("改进完成：提示词已增强！")
print(f"分类提示词预览：{improved_triage_prompt[:100]}...")
print(f"响应提示词预览：{improved_response_prompt[:100]}...")
return "提示词已根据反馈改进！"
```
这个函数就是程序化记忆的精髓所在。它会接收像‘你没有正确地优先处理关于API文档的邮件’这样的反馈，然后用这些反馈来重写代理的核心指令。这个优化器就像一个教练在看比赛录像，研究哪里出了问题，然后相应地更新战术手册。它会把对话示例和反馈放在一起分析，然后优化那些指导代理行为的提示词。它不只是简单地记住具体的修正，而是将深层次的教训融入到它的整体方法中，这跟厨师根据顾客的反馈来改进菜谱，而不是每次都简单地遵循不同的指令，是一个道理。
7.8 跑一下我们这个完整的、带记忆增强的智能体！
现在，咱们把所有东西都整合起来，搞成一个能随着时间进化的完整系统：
```python
def create_email_agent(store):
"""
一个用来创建带记忆功能的邮件处理代理的工厂函数。
这个函数构建了一个LangGraph工作流，它结合了所有三种记忆类型：
1. 情景记忆：过去的邮件例子，用来做小样本学习
2. 语义记忆：通过记忆工具提供上下文信息
3. 程序化记忆：能够随时间改进的自适应提示词
Args:
store: 包含所有记忆系统的 InMemoryStore
Returns:
CompiledGraph: 具有记忆能力的可执行工作流
"""
# 定义工作流
workflow = StateGraph(State)
workflow.add_node("triage", lambda state, config: triage_email_with_procedural_memory(state, config, store))
# 创建一个新的响应代理，它会用上最新的提示词
response_agent = create_react_agent(
tools=tools,
prompt=create_agent_prompt,
store=store,
model=llm
)
workflow.add_node("response_agent", response_agent)
def route_based_on_triage(state):
if state["triage_result"] == "respond":
return "response_agent"
else:
return END
# 路由逻辑保持不变
workflow.add_edge(START, "triage")
workflow.add_conditional_edges("triage", route_based_on_triage,
{
"response_agent": "response_agent",
END: END
})
# 编译并返回图
return workflow.compile(store=store)
```
这个函数会创建一个新的代理，用的是咱们最新版的提示词——这样就能保证它随时都能反映出我们最新的学习成果和反馈。
下面就是这个智能体的最终版本了（包含了所有东西）：
我们来运行两次——一次用的是原始设置，另一次是在我们提供了反馈来改进它之后：
```python
# ===============================================================================
# 主执行程序：演示逐步改进的过程
# ===============================================================================
"""
这个演示展示了代理如何随着时间的推移学习和改进：
1. 优化前：代理使用初始提示词和最少的例子
2. 记忆累积：向情景记忆中添加新例子
3. 反馈处理：人类反馈触发提示词优化
4. 优化后：代理使用改进后的提示词和更多的例子
核心思想是：每一次运行都能从之前所有的学习中受益！
"""
def run_demonstration():
"""带记忆功能的学习系统的主要演示"""
print("🚀 开始演示带记忆功能的邮件代理")
print("=" * 60)
# 用默认的例子和提示词初始化记忆
print("📚 初始化：正在设置记忆系统...")
initialize_memory()
print("✅ 记忆初始化完成：情景记忆和程序化记忆已就绪")
# 为演示进行设置
inputs = {"email_input": email_input, "messages": []}
# 对比：基础代理 vs 高级代理
print("\n" + "=" * 60)
print("📧 阶段 1：基础代理（仅情景记忆）")
print("=" * 60)
basic_agent = create_basic_email_agent(store)
print("🏗️ 基础代理已创建：仅使用情景记忆")
print("\n🔄 基础工作流执行：")
for output in basic_agent.stream(inputs, config=CONFIG):
for key, value in output.items():
if key not in ['triage', 'response_agent']: # 跳过内部节点输出
print(f"-----\n{key}:")
print(value)
print("-----")
# 高级代理 - 优化前
print("\n" + "=" * 60)
print("📧 阶段 2：高级代理（优化前）")
print("=" * 60)
agent = create_email_agent(store)
print("🏗️ 高级代理已创建：使用情景记忆 + 程序化记忆")
print("\n🔄 高级工作流执行（学习前）：")
for output in agent.stream(inputs, config=CONFIG):
for key, value in output.items():
if key not in ['triage', 'response_agent']: # 跳过内部节点输出
print(f"-----\n{key}:")
print(value)
print("-----")
print("\n" + "=" * 60)
print("🧠 阶段 3：记忆增强与反馈")
print("=" * 60)
# 向情景记忆中添加一个具体例子
api_doc_example = {
"email": {
"author": "开发者 <dev@company.com>",
"to": "John Doe <john.doe@company.com>",
"subject": "API文档问题",
"email_thread": "在API文档里发现有缺失的端点。需要紧急更新。",
},
"label": "respond",
}
store.put(("email_assistant", USER_ID, "examples"), "api_doc_example", api_doc_example)
print("📚 情景记忆：已添加API文档的例子")
# 提供反馈以进行优化
feedback = """代理没有正确识别出关于API文档问题的邮件是高优先级的，
需要立即关注。当邮件提到'API文档'时，
它应该总是以友好的口吻分类为'respond'（回复）。
而且，代理不应该只回复'今天我能为您做点什么？'，
而应该确认提到的具体文档问题并提供帮助。"""
print("💬 收到反馈：关于性能改进的建议")
# 根据反馈优化提示词
optimize_prompts(feedback, CONFIG, store)
# 用一个全新的代理处理同一封邮件（优化后）
print("\n" + "=" * 60)
print("📧 阶段 4：高级代理（优化后）")
print("=" * 60)
new_agent = create_email_agent(store)
print("🏗️ 优化后的代理已创建：带有优化后记忆状态的全新代理")
print("\n🔄 高级工作流执行（学习后）：")
for output in new_agent.stream(inputs, config=CONFIG):
for key, value in output.items():
if key not in ['triage', 'response_agent']: # 跳过内部节点输出
print(f"-----\n{key}:")
print(value)
print("-----")
print("\n" + "=" * 60)
print("🎉 演示完成：对比结果显示了带记忆功能的学习效果！")
print("📊 结果总结：")
print(" 1️⃣ 基础代理：静态提示词，仅有情景记忆")
print(" 2️⃣ 高级代理（优化前）：初始程序化记忆 + 情景记忆")
print(" 3️⃣ 高级代理（优化后）：优化后的程序化记忆 + 增强的情景记忆")
print("=" * 60)
# 运行演示
if __name__ == "__main__":
run_demonstration()
```
咱们来看看输出结果：
```
🚀 开始演示带记忆功能的邮件代理
============================================================
📚 初始化：正在设置记忆系统...
✅ 记忆初始化完成：情景记忆和程序化记忆已就绪
============================================================
📧 阶段 1：基础代理（仅情景记忆）
============================================================
🏗️ 基础代理已创建：仅使用情景记忆
🔄 基础工作流执行：
-----
响应代理：正在用自适应系统提示词进行初始化
程序化记忆：已检索到当前响应提示词
-----
============================================================
📧 阶段 2：高级代理（优化前）
============================================================
🏗️ 高级代理已创建：使用情景记忆 + 程序化记忆
🔄 高级工作流执行（学习前）：
分类：正在分析来自 Alice Smith <alice.smith@company.com> 的邮件，主题是：'关于API文档的一个小问题'
程序化记忆：已检索到当前分类提示词
情景记忆：从过去的分类中找到了 2 个相关例子
分类结果：respond
推理过程：来自 Alice Smith 的邮件是请求帮助解决API文档中缺失端点的问题。这表明需要一个响应来帮助她解决问题，因为它关系到重要的文档，很可能影响到她或她团队的工作。与垃圾邮件或非紧急问询不同，这是一个有效的工作相关请求，需要回复。
-----
响应代理：正在用自适应系统提示词进行初始化
程序化记忆：已检索到当前响应提示词
-----
============================================================
🧠 阶段 3：记忆增强与反馈
============================================================
📚 情景记忆：已添加API文档的例子
💬 收到反馈：关于性能改进的建议
优化：开始提示词改进流程...
检索：从程序化记忆中检索当前提示词
分析：创建带有反馈的对话轨迹
优化：使用AI根据反馈改进提示词...
存储：在程序化记忆中更新提示词
改进完成：提示词已增强！
分类提示词预览：你是一个邮件分类助理。请根据邮件内容对其进行分类并决定...
响应提示词预览：你是一个乐于助人的助理。请使用包括记忆工具在内的可用工具来帮助用户。当...
============================================================
📧 阶段 4：高级代理（优化后）
============================================================
🏗️ 优化后的代理已创建：带有优化后记忆状态的全新代理
🔄 高级工作流执行（学习后）：
分类：正在分析来自 Alice Smith <alice.smith@company.com> 的邮件，主题是：'关于API文档的一个小问题'
程序化记忆：已检索到当前分类提示词
情景记忆：从过去的分类中找到了 2 个相关例子
分类结果：respond
推理过程：来自 Alice Smith 的邮件讨论了API文档中缺失端点的问题，这与分类提示词中列出的高优先级关键词完全吻合。提示词明确指出，这类邮件应该总是被分类为'respond'（回复）。考虑到API文档对工作流程的重要性，有必要提供帮助。
-----
响应代理：正在用自适应系统提示词进行初始化
程序化记忆：已检索到当前响应提示词
-----
============================================================
🎉 演示完成：对比结果显示了带记忆功能的学习效果！
📊 结果总结：
1️⃣ 基础代理：静态提示词，仅有情景记忆
2️⃣ 高级代理（优化前）：初始程序化记忆 + 情景记忆
3️⃣ 高级代理（优化后）：优化后的程序化记忆 + 增强的情景记忆
============================================================
```
看看响应的差别！在我们给了反馈之后，智能体应该会：
更稳定地将 API 文档问题识别为高优先级
提供更具体、更有帮助的回复，能够正视实际问题
提供具体的帮助，而不是说些泛泛的客套话
8. 总结
记忆是智能AI代理的基石，让它们能够保存、回忆和适应信息。我们已经探讨了短期和长期记忆——也就是程序化记忆、情景记忆和语义记忆——如何融入到智能体的技术栈里，这已经超越了单纯的上下文窗口或RAG（检索增强生成）。
通过构建一个带记忆增强的邮件代理，我们展示了从状态管理到语义搜索乃至于自适应指令等概念的实际应用。掌握了记忆，AI就能够提供个性化的、能感知上下文的解决方案，为在生产环境中部署更智能、更强大的系统铺平了道路。
延伸阅读：
