AI智能体的记忆系统

Posted on 2025-11-09 Edited on 2025-12-07 In 人工智能 , LangGraph Views: Word count in article: 35k Reading time ≈ 32 mins.

1、短期记忆与长期记忆

要创建真正智能且实用的 AI 智能体，关键在于为其配备能够模仿人类认知特性的记忆系统。正如人类利用短期记忆和长期记忆来进行有效的互动和学习一样，AI 智能体也同样需要利用记忆来演进，如图所示。本节将探讨 LangGraph 框架中短期记忆和长期记忆的作用及其在构建 AI 智能体中的协同效应。
短期记忆与长期记忆的共存必要性源于智能交互的多维特性。试想，如果 AI 智能体只能记住刚发生的对话，或者不加区分地记住所有信息而忽视当前交流重点，那么都将导致对话效果大打折扣。这两种极端情况都无法实现有效的智能交互。

短期记忆的核心价值在于保障对话的即时性。它使 AI 智能体能够专注于当前对话情境，准确理解用户输入的细微差别，并维持对话的连贯性。缺乏短期记忆将导致每轮对话彼此割裂，丧失上下文关联。
长期记忆则提供了持续发展的基础。它使 AI 智能体能够突破单次交互的局限，建立稳定的身份特征，从历史经验中持续学习，最终提供日益个性化和精准的服务。没有长期记忆的智能体将停滞不前，难以与用户建立深度互动关系。
因此，一个完善的 AI 智能体记忆架构需要采取差异化策略，通过短期记忆和长期记忆的有机结合，既保障即时对话的流畅性，又实现智能体能力的持续进化。

1.1 短期记忆：维持对话的连贯性

短期记忆(Short-Term Memory)是 AI 智能体实现连贯性对话的基础架构。它使智能体能够在完整对话过程中维持上下文理解能力。从本质上说，短期记忆构成了智能体在特定对话期间的“工作上下文”，其核心功能是动态保存和更新与当前交互直接相关的信息，主要包括以下关键要素。
- 对话历史记录：按时间顺序完整记录用户与智能体之间交换的所有消息，为交互过程提供即时历史轨迹。
- 当前任务上下文：明确记录用户当前目标的详细参数和要求，包括智能体正在处理的具体任务及其相关参数或约束条件。
- 用户意图(会话特定)：智能体对用户在本次对话中所表达需求和目标的动态理解，这一理解会随着对话轮次不断深化和完善。
- 中间结果(会话特定)：在对话过程中生成的临时数据和输出，包括计算中间值、检索结果或工具输出等即时处理所需的信息。

在 LangGraph 框架中，对话历史记录作为短期记忆的核心组成部分，通常通过状态管理系统进行维护。典型的 AgentState 结构会包含专门用于存储进行中对话的 messages 列表，以此实现短期记忆功能。

from typing import TypedDict, List, Dict, Any
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage


class AgentState(TypedDict):
    messages: List[BaseMessage]  # 消息列表，作为对话历史记录
    intermediate_results: Dict[str, Any]  # 短期记忆中的其他数据

向此历史记录添加新消息，无论是来自用户还是 AI 智能体，都只需在图执行期间更新 AgentState 中的 messages 列表即可。

def add_user_message(state: AgentState, user_message: str) -> Dict[str, List[BaseMessage]]:
    """ 向对话历史记录添加用户消息 """
    new_message = HumanMessage(content=user_message)
    return {"messages": state["messages"] + [new_message]}


def add_ai_message(state: AgentState, ai_response: str) -> Dict[str, List[BaseMessage]]:
    """ 向对话历史记录添加 AI 响应 """
    new_message = AIMessage(content=ai_response)
    return {"messages": state["messages"] + [new_message]}

随着对话时间的延长，有效管理短期记忆变得至关重要。LLM 上下文窗口的限制使得我们需要采取以下优化技术来防止性能下降、延迟增加，以及记录对话成本上升等问题。

1.1.1 截断

保持时效性最直接的方法是截断对话历史记录，丢弃较旧的消息，以确保上下文窗口在可接受的限制范围内。然而，截断必须谨慎进行，以避免丢失必要的对话语境。

对于处理词元的截断，LangChain 提供了一个名为 trim_messages 的实用程序。它采用了一种更精细的方法，同时尊重词元限制和消息类型约束，如示例所示。

from langchain_core.messages import trim_messages
from langchain_openai import ChatOpenAI


def truncate_history(state: AgentState, max_messages: int) -> Dict[str, List[BaseMessage]]:
    """ 截断对话历史记录，仅保留最近的消息 """
    truncated_messages = state["messages"][-max_messages:]
    return {"messages": truncated_messages}


def trim_message_history_by_token(state: AgentState, max_tokens: int) -> Dict[str, List[BaseMessage]]:
    """ 使用 LangChain 的 trim_messages 根据词元计数修剪消息历史记录"""
    trimmed_messages = trim_messages(
        state["messages"],
        strategy="last",  # 保留最后 <= max_tokens 个词元
        token_counter=ChatOpenAI(model="gpt-4o"),  # 根据模型调整
        max_tokens=max_tokens,
        start_on="human",  # 大多数聊天模型期望聊天历史记录以 HumanMessage 或 SystemMessage 后跟 HumanMessage 开头
        end_on=("human", "tool"),  # 大多数聊天模型期望聊天历史记录以 HumanMessage 或 ToolMessage 结尾
        include_system=True,  # 通常希望保留系统消息(如果它存在于原始历史记录中)。系统消息对模型有特殊指令
    )
    return {"messages": trimmed_messages}

1.1.2 摘要

通过使用 LLM 将对话历史记录浓缩为语义上的精华，显著降低了词元数量，同时保留了核心的语境信息。

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI


def summarize_history(state: AgentState, llm) -> Dict[str, List[BaseMessage]]:
    """使用 LLM 总结对话历史记录"""
    messages = state["messages"]
    prompt = ChatPromptTemplate.from_template(" 总结以下对话以供参考 :\n{conversation}")
    conversation_string = "\n".join([f"{m.role}: {m.content}" for m in messages])
    summarization_chain = prompt | llm | StrOutputParser()  # 确保 llm 已被正确定义
    summary = summarization_chain.invoke({"conversation": conversation_string})
    # 将历史记录替换为摘要和最新的用户消息
    summary_message = AIMessage(content=summary)
    # 从消息列表中获取最后一条人类消息
    last_human_message = [m for m in messages if m.type == "human"][-1] if any(
        m.type == "human" for m in messages) else None
    new_messages = [summary_message]
    if last_human_message:
        new_messages.append(last_human_message)
    return {"messages": new_messages}

还可以有选择性地仅保留对话中最关键的信息片段，以最有效地保留关键语境和最大限度地减少词元使用，但这需要更复杂的逻辑来确定消息的显著性和相关性。
在 LangGraph 中处理短期记忆时需特别注意：记忆的更新与截断操作必须谨慎执行，以避免丢失关键上下文信息。具体的记忆处理策略应根据实际应用需求进行调整，既可采用示例中展示的 LangChain 作为辅助工具，也可选择其他 LLM 开发框架(如 OpenAI SDK 或 Anthropic SDK 等)。需要强调的是，LangGraph 框架本身与 LangChain 并不存在强制绑定关系。

1.2 长期记忆：实现跨会话

长期记忆(Long-Term Memory)使 AI 智能体突破单次对话的局限，构建持久的知识体系，实现跨会话学习能力，最终提供真正个性化的用户体验。长期记忆不仅是简单的信息存储，更是构建具备以下核心能力的智能系统的基础。
- 深度个性化体验：跨会话记忆用户偏好、交互历史和细粒度上下文，打造定制化服务。
- 持续学习进化：通过积累的知识持续优化技能，基于历史交互动态调整行为。
- 身份一致性维护：在所有交互中保持稳定可识别的角色特征，建立用户信任。
- 持久关系培养：促进超越单次交流的深度互动体验。
受人类记忆模型的启发，AI 智能体的长期记忆可以有效地分为三种相互关联的类型。
- 语义记忆(Semantic Memory)——结构化知识库，存储概念、关系等事实性知识。包括用户资料、产品目录、领域知识等。
- 情景记忆(Episodic Memory)——自传式记录，存储智能体历史互动、经验和事件，支持特定对话回溯，从过去的成功和失败中学习，并基于历史数据改进策略。
- 程序记忆(Procedural Memory)——技能知识库，存储核心系统提示、其底层代码和算法，以及任何已学习的有效互动和任务执行的最佳实践。

为了充分利用长期记忆，要精心设计提示结构。以下代码演示了如何基于 YAML 的提示整合这三类记忆并创建具有情境感知能力的 AI 智能体。

- role: system
	content: |
		你是一个乐于助人的 AI 助手，名叫 AssistantBot。
		
		## 语义记忆:用户资料
		{semantic_memory_retrieval(user_id)}
		
		## 程序记忆:指令
		在每次互动中遵循以下准则:
		1. 始终保持礼貌和专业。
		2. 利用来自语义记忆的用户资料来个性化响应。
		3. 利用来自先前对话(情景记忆)的示例来提供具有情境感知的帮助。 
		4.如果用户询问你的名字，请介绍自己为 AssistantBot。
		
- role: user
	content: |
		## 情景记忆:相关的过去互动 
		以下是与该用户先前对话的一些示例，这些示例可能有所帮助: 
		{episodic_memory_retrieval(user_id, query="relevant past interactions")}
		
		## 当前查询:用户输入 
		{user_input}

系统角色(YAML 列表项第一项)为每次互动提供基础框架，定义智能体的持久身份、知识库和核心操作规则。
- 智能体身份定义：明确身份定位(如名为 AssistantBot 的 AI 助手)。
- 语义记忆集成：通过占位符 {semantic_memory_retrieval(user_id)} 注入特定于用户的知识，可能包含如偏好、购买历史记录、过去的支持问题等详细信息。
- 程序记忆嵌入：以指令集形式编码操作规范，指导交互风格与记忆使用方式。
用户角色(YAML 列表项第二项)注入于特定语境：用户消息提供动态交互上下文，补充系统消息的静态基础。
- 情景记忆占位符：通过 {episodic_memory_retrieval(user_id,query=”relevant past interactions”)} 占位符回忆过去相关对话片段。运行时替换为相关历史交互片段，确保返回与当前对话最相关的历史数据。
- 当前用户查询占位符：{user_input} 占位符表示用户的即时消息或查询。

以下是对应前述提示的完整 YAML 提示示例：

- role: system
	content: |
		你是一个乐于助人的 AI 助手，名叫 AssistantBot。
		## 语义记忆:用户资料
		{“name”: “Alice”, “location”: “San Francisco”, “preferred_language”: “Python”, “expertise”: [“Data Science”, “Machine Learning”]}
		## 程序记忆:指令
		在每次互动中遵循以下准则:
		1. 始终保持礼貌和专业。
		2. 利用来自语义记忆的用户资料来进行个性化响应。
		3. 利用来自先前对话(情景记忆)的示例来提供具有情境感知的帮助。 
		4. 如果用户询问你的名字，请介绍自己为 AssistantBot。
		
- role: user
	content: |
		## 情景记忆:相关的过去互动 以下是与该用户先前对话的一些示例，这些示例可能有所帮助: 
		- user: “什么是 Transformer 架构?”
			assistant: “Transformer架构是一种神经网络架构......”
			relevance_score: 0.8
		- user: “你能帮我调试我的 Python 代码吗?” 
			assistant: “请提供代码片段和错误消息......”
			relevance_score: 0.9
		## 当前查询:用户输入
		我遇到了 Python 脚本问题。它没有运行。你能帮忙吗?

短期记忆和长期记忆在 AI 智能体中不是孤立的，而是协同运作的有机体。短期记忆负责管理即时对话流程，为智能体提供处理每轮对话所需的临时存储空间；长期记忆提供持久的知识库、历史语境和行为准则。二者相互配合，共同实现交互的丰富性、个性化塑造、一致性保持，以及智能体的持续优化。高效的 AI 智能体架构只有通过这种精心设计的协同机制，才能创造出真正智能且实用的对话体验。

2、记忆存储

在明确长期记忆对实现持久化、个性化 AI 智能体的关键作用之后，我们将探讨其在 LangGraph 中的实现机制。本节将深入探讨 LangGraph 记忆存储系统——这是一个多功能抽象层，旨在管理 AI 智能体的持久化跨线程知识。本节内容包括不同类型存储配置方法，信息存储机制，以及如何通过检索和利用存储的知识来提升 AI 智能体的行为表现。

2.1 记忆存储的基本操作

本节将介绍 LangGraph 记忆存储的基本操作，重点介绍 InMemoryStore 实现和 BaseStore 接口的核心功能。
InMemoryStore 将数据存储在内存中，具有快速访问和设置简易的特点。

使用 InMemoryStore 实例化记忆存储，只需导入并初始化该类。

1
2
3

from langgraph.store.memory import InMemoryStore

in_memory_store = InMemoryStore()

虽然 InMemoryStore 适用于初期开发和原型设计，但 LangGraph 的架构设计已为未来集成更强大、持久的存储解决方案做好准备。基于统一的 BaseStore 接口，后续实现将支持无缝迁移至生产级数据库，而无须大幅修改现有代码。
LangGraph 的记忆存储以 BaseStore 类及其实现为基础，为长期记忆管理提供了抽象层支持。作为开源且可扩展的持久化键值存储接口，它能让开发者根据自身特定需求选择适配的存储后端 —— 从开发阶段适用的轻量级内存解决方案，到生产环境所需的高性能数据库均可覆盖，同时确保记忆管理 API 的一致性。
put 方法是实现信息存储的基础操作，需要提供以下三个核心参数。
- 命名空间(元组)。
  - 功能：作为记忆的逻辑分组机制，类比文件系统中的文件夹。
  - 格式：采用元组定义，支持分层组织结构。
  - 命名规范：建议包含用户标识符和记忆类别，如 (“user_123”,”chat_history”)。
- 键(字符串)。
  - 功能：命名空间内记忆条目的唯一标识符。
  - 命名建议：应具备描述性和易检索性。
  - 生成方式：推荐使用 Python 的 uuid 库生成通用唯一标识符。
- 值(字典)。
  - 格式要求：必须为 Python 字典对象。
  - 优势：支持结构化数据和元数据存储。
  - 应用示例：用户资料可存储为包含姓名、偏好等键值对的字典。

以下示例演示了如何使用 put 方法将用户的姓名保存到 InMemoryStore 中。

import uuid
from langgraph.store.memory import InMemoryStore

in_memory_store = InMemoryStore()
# 定义用户特定数据的命名空间
user_id = "example_user"
namespace_for_user_data = (user_id, "user_info")

# 为记忆条目生成唯一键
memory_key = str(uuid.uuid4())

# 创建一个字典来保存用户姓名作为记忆值
memory_value = {"user_name": "Example User"}

# 使用put将记忆存储在InMemoryStore中
in_memory_store.put(namespace_for_user_data, memory_key, memory_value)
print(f" 记忆已保存，键为 {memory_key}，命名空间为 {namespace_for_user_data}")
# 记忆已保存，键为 e1e95160-ce18-4d95-97d2-a0db1bfdbcc2，命名空间为 ('example_user', 'user_info')

此示例说明了记忆存储的核心流程：首先定义一个命名空间来组织记忆，然后生成一个唯一键来标识记忆，创建字典来保存记忆内容(例如，用户名)，最后使用 put 方法将此信息存储在 InMemoryStore 中。

同时，LangGraph 提供了两种主要方法从存储中检索记忆：search 和 get，每种方法都针对不同的检索场景进行了优化。

search(namespace, query=None, filter=None, limit=None, index_ name=”default”, **kwargs)。功能：在指定命名空间内实现灵活记忆检索。特点：支持语义搜索、内容过滤等高级检索功能。参数说明如下：
- namespace(元组)：必需的参数，指定检索目标命名空间。
- query(可选字符串)：启用语义搜索功能(需预先配置索引)。
- filter(可选字典)：基于记忆值字典的键值对过滤。
- limit(可选整数)：限制返回结果数量。
- index_name(字符串)：指定语义搜索索引(默认为 default)。
- **kwargs：支持传递后端特定参数。
get(namespace, key)。功能：基于精确键值的直接记忆检索。参数说明如下。
- namespace(元组)：目标记忆命名空间。
- key(字符串)：目标记忆唯一键。

# 假设 in_memory_store、namespace_for_user_data和 memory_key 已定义
# 检索user_info命名空间中的所有记忆(没有查询或过滤器)
all_user_memories = in_memory_store.search(namespace_for_user_data)
print(" 命名空间中的所有用户记忆 :")
for record in all_user_memories:
    print(record.dict())  # 打印 MemoryRecord 字典表示
#  命名空间中的所有用户记忆 :
# {'namespace': ['example_user', 'user_info'], 'key': 'e1e95160-ce18-4d95-97d2-a0db1bfdbcc2', 'value': {'user_name': 'Example User'}, 'created_at': '2025-11-08T07:55:33.657137+00:00', 'updated_at': '2025-11-08T07:55:33.657139+00:00', 'score': None}

# 使用get通过键检索记忆
retrieved_memory_record = in_memory_store.get(namespace_for_user_data, memory_key)
print(f"\n使用键 '{memory_key}' 检索到的记忆:")
print(retrieved_memory_record.dict())  # 打印 MemoryRecord 字典表示
# 使用键 'e1e95160-ce18-4d95-97d2-a0db1bfdbcc2' 检索到的记忆:
# {'namespace': ['example_user', 'user_info'], 'key': 'e1e95160-ce18-4d95-97d2-a0db1bfdbcc2', 'value': {'user_name': 'Example User'}, 'created_at': '2025-11-08T07:55:33.657137+00:00', 'updated_at': '2025-11-08T07:55:33.657139+00:00'}

2.2 通过语义搜索增强记忆检索

LangGraph 记忆存储的语义搜索功能显著提升了记忆检索的有效性，使 AI 智能体能够基于语义相似性而非简单的关键词匹配来检索信息。这种机制通过向量嵌入技术实现，能够捕捉文本的深层语义特征，从而获得更符合语境的检索结果。
语义搜索的核心优势如下所述。
- 突破传统关键词检索的局限性，有效处理同义词、近义词和概念相关但表述不同的内容。
- 利用向量嵌入技术将文本转换为数值表示，通过计算向量间的相似度(如余弦相似度)确定语义相关性。
- 实现概念层面的信息关联，例如将“附近有什么好吃的餐馆”与“当地美食推荐”等表述不同但语义相近的内容进行关联。
在 InMemoryStore 中启用语义搜索时需要进行索引配置。
- 基本配置参数。
  - embed 函数：处理从文本到向量嵌入的转换，支持各类嵌入模型(如 OpenAI、Cohere 等)。
  - dims 参数：指定嵌入向量的维度，必须与所用嵌入模型的输出维度一致。
- 可选配置参数。
  - fields 参数：指定记忆字典中需要索引的特定字段(默认索引整个字典)。

以下配置示例将展示如何集成硅基流动平台上提供的 BAAI BGE-M3 文本向量化模型实现语义搜索功能。


from langchain_openai import OpenAIEmbeddings
from langgraph.store.memory import InMemoryStore
import os
import dotenv

dotenv.load_dotenv()
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY1")
os.environ["OPENAI_BASE_URL"] = os.getenv("OPENAI_BASE_URL")

# 初始化text-embedding-ada-002向量化模型
embeddings = OpenAIEmbeddings(model="text-embedding-ada-002")  # 或其他向量化模型

# 使用语义搜索索引配置InMemoryStore
store_with_semantic_search = InMemoryStore(
    index={
        "embed": embeddings.embed_documents,
        "dims": 1536,  # text-embedding-ada-002的向量维度
        "fields": ["memory_content"]  # 仅向量化 memory_content 字段(可选)
    })

在此示例中，我们使用 OpenAIEmbeddings 模型的 embed_documents 方法作为嵌入函数，并设置 dims=1536 以匹配 text-embedding-ada-002 向量化模型的输出维度。通过指定 fields=[“memory_content”] 参数，我们限定仅对记忆字典中 memory_content 键对应的内容进行嵌入和索引处理。

配置隐语义索引的 InMemoryStore 后，所有通过 put 方法保存的记忆都将自动建立索引。索引过程在 put 操作时自动完成，开发者无须额外干预。put 方法支持通过 index 参数来灵活控制索引行为。
- index=True(默认设置)：使用存储的默认索引配置对记忆进行索引。
- index=False：仅存储记忆内容而不建立隐语义索引，适用于系统元数据等不需要语义检索的场景。
- index=[…fields]：为特定 put 操作指定自定义索引字段，实现精细化的索引控制。

以下示例将具体展示如何在启用语义搜索的存储中执行记忆保存操作。

# Assuming store_with_semantic_search 已使用 IndexConfig 初始化
# 保存将为语义搜索索引的记忆(默认行为) 
store_with_semantic_search.put(
    ("user_789", "food_memories"),
    "memory_1",
    {"memory_content": " 我真的很喜欢辛辣的印度咖喱。"},
)

# 保存另一个记忆，显式禁用此条目的索引
store_with_semantic_search.put(
    ("user_789", "system_metadata"),
    "memory_2",
    {"memory_content": " 用户入职已完成。", "status": "completed"}, index=False,  # 禁用此记忆的索引
)

# 保存一个记忆，覆盖默认索引字段并仅索引context
store_with_semantic_search.put(
    ("user_789", "restaurant_reviews"),
    "memory_3",
    {"memory_content": " 服务很慢，但食物很好。", "context": " 对 'The Italian Place' 餐厅的评论 "},
    index=["context"]  # 仅索引 context 字段
)

执行语义搜索时，可通过 search 方法传入 query 字符串参数实现。该方法的工作流程包含三个关键步骤。
- 向量生成：使用 IndexConfig 配置的嵌入函数将查询文本转换为向量表示。
- 相似度计算：采用余弦相似度等度量方法，计算查询向量与命名空间内所有已索引记忆向量的相似程度。
- 结果排序：根据相似度评分对记忆进行排序，返回前 limit 个最相关的检索结果对象。返回的每个检索结果对象均包含 score 属性，用于表示该检索与查询的语义相似度评分。

以下示例将展示如何执行与食物主题相关的语义记忆检索。

# 假设store_with_semantic_search已初始化并且记忆已保存
# 语义搜索食物偏好
search_query = " 该用户喜欢哪种食物? "
semantic_memory_results = store_with_semantic_search.search(
    ("user_789", "food_memories"), query=search_query, limit=2
)

print(" 查询的语义搜索结果:", search_query) 
for record in semantic_memory_results:
    print(f" 记忆键:{record.key}，相似度评分:{record.score}") 
    print(f" 记忆内容:{record.value}")
print("=" * 30)
#  查询的语义搜索结果:  该用户喜欢哪种食物? 
#  记忆键:memory_1，相似度评分:0.8044571384246496
#  记忆内容:{'memory_content': ' 我真的很喜欢辛辣的印度咖喱。'}
# ==============================

该代码示例展示了语义搜索的实际应用。通过定义与食物偏好相关的 search_ query，并将其与命名空间参数及结果数量限制(limit=2)一同传入 search 方法，系统将返回与语义相关的记忆结果。代码遍历返回的 MemoryRecord 对象，输出每个匹配记忆的键值、相似度评分及具体内容。其中 score 属性量化反映了记忆与查询的语义关联程度，为结果相关性评估提供客观依据。

语义搜索技术通过理解查询的深层含义(而非表面关键词)，实现了概念层面的信息检索，为智能体系统带来三大核心优势。
- 检索精度提升：获取更多与语境相关的记忆，使智能体响应更精准有效。
- 用户体验优化：准确理解多样化表述的隐含需求，提供个性化服务。
- 智能水平进阶：基于广谱语义知识进行类人推理，展现更自然的对话能力。
这项技术使 LangGraph 智能体突破传统关键词匹配的局限，充分利用向量嵌入蕴含的丰富语义信息，在记忆检索效能和整体智能表现上实现质的飞跃。

2.3 构建自定义记忆存储

LangGraph 的 InMemoryStore 虽然为开发和基础场景提供了便利，但其记忆存储架构的真正优势在于出色的可扩展性。针对生产环境部署、专业应用开发或特色数据库集成等需求，开发者可以通过继承抽象基类 BaseStore 来实现自定义记忆存储，从而精准满足各类智能体系统的独特需求。
在实现支持语义搜索的自定义 BaseStore 时，正确处理 IndexConfig 配置并与向量数据库 / 索引库集成是关键所在。具体实现要点包括：
- 索引配置处理。
  - 存储初始化时需内部保存 IndexConfig 参数(嵌入函数、维度、索引字段等)。
  - 实现嵌入函数与维度参数的校验机制。
- 数据存储时的索引处理。
  - 在 put/aput 方法中自动生成向量嵌入(当 index 参数不为 False 时)。
  - 优先使用 put 调用指定的字段，其次采用默认字段。
  - 将生成的嵌入向量存入专用索引结构(向量数据库或内存索引)。
- 语义搜索实现。
  - 在 search/asearch 方法中处理 query 参数时自动触发语义搜索流程。
  - 使用配置的嵌入函数将查询文本转换为向量。
  - 执行向量相似度计算并排序结果(如余弦相似度)。
  - 返回带相似度评分的检索结果对象列表。
开发者可根据实际需求选择集成专业向量数据库(如 Milvus、Pinecone)或实现轻量级内存索引。自定义 BaseStore 实现的主要优势包括：
- 数据库集成灵活性。
  - 支持对接各类数据库系统(PostgreSQL/MongoDB 等)。
  - 复用现有数据基础设施。
- 搜索功能扩展。
  - 实现混合搜索策略。
  - 支持自定义过滤和排序算法。
  - 开发领域特定的检索逻辑。
- 功能增强空间。
  - 添加批量导入接口。
  - 实现记忆生命周期管理。
  - 集成数据分析能力。

下面示例将通过实现 TextFileStore 具体说明自定义存储的实现方法。

from langgraph.store.base import BaseStore, Item, SearchItem, Op, Result, IndexConfig, NamespacePath
from typing import Optional, Union, Literal, Any, Iterable, List
import json, os, uuid, datetime


class TextFileStore(BaseStore):  # 继承 BaseStore 

    def __init__(self, base_dir: str):
        self.base_dir = base_dir  # 存储文件的基本目录

    def put(
        self,
        namespace: tuple[str, ...],
        key: str,
        value: dict[str, Any],
        index: Literal[False] | list[str] | None = None,
        *,
        ttl: float | None | NotProvided = NOT_PROVIDED,
    ) -> None:
        """将值保存到 namespace/key.json 中的文本文件"""
        namespace_path = os.path.join(self.base_dir, *namespace)  # 创建命名空间路径
        os.makedirs(namespace_path, exist_ok=True)  # 确保命名空间目录存在
        file_path = os.path.join(namespace_path, f"{key}.json")  # 键的文件路径
        with open(file_path, 'w') as f:
            json.dump(value, f)  # 将值作为 JSON 保存到文件

    def get(
        self,
        namespace: tuple[str, ...],
        key: str,
        *,
        refresh_ttl: bool | None = None,
    ) -> Item | None:
        """从文本文件中检索值，返回 Item 或 None(如果未找到)"""
        file_path = os.path.join(self.base_dir, *namespace, f"{key}.json")
        if not os.path.exists(file_path):  # 检查文件是否存在
            return None
        with open(file_path, 'r') as f:
            value = json.load(f)  # 从文件加载 JSON
        created_at = datetime.datetime.fromtimestamp(os.path.getctime(file_path),tz=datetime.timezone.utc)  # 从文件元数据中获取创建时间和更新时间
        updated_at = datetime.datetime.fromtimestamp(os.path.getmtime(file_path), tz=datetime.timezone.utc)
        return Item(namespace=namespace, key=key, value=value,created_at=created_at,updated_at=updated_at)  # 返回 Item 对象
    
    # 实现其他抽象方法，如 search、delete、batch、alist_namespaces 等
    # 为简单起见，异步方法(如 aput、aget、asearch、adelete、alist_ namespaces)也需要实现，以获得完整的存储
    
# 自定义TextFileStore的示例用法
file_store = TextFileStore(base_dir="./my_filestore")
file_store.put(("user_data",), "user_profile_1", {"name": " 自定义存储用户 ", "preference": "files"})
retrieved_item = file_store.get(("user_data",), "user_profile_1")
if retrieved_item:
    print(f" 检索到的条目:{retrieved_item.value}")

此示例展示了自定义 TextFileStore 的基本结构。完整的实现除此之外还需要实现 BaseStore 的所有抽象方法，支持同步和异步版本，并处理错误条件和边缘情况。此示例仅为开发者理解如何通过扩展 BaseStore 创建定制化记忆存储解决方案提供一个基础框架。

3、记忆系统的实际应用

在系统阐述短期记忆与长期记忆的核心概念，并深入解析 LangGraph 记忆存储机制后，我们将聚焦实际应用场景。本节重点探讨记忆系统如何显著增强 AI 智能体的功能表现，针对每个典型用例，我们将详细说明记忆应用价值分析、概念性代码示例，以及实际集成实施方案。其中，我们会特别关注记忆数据更新时机选择、信息提取技术，以及如何通过 TrustCall 机制提升记忆数据提取的可靠性。

3.1 个性化推荐

长期记忆最具价值的应用场景之一是实现个性化服务。通过持续记录用户偏好、兴趣点及历史交互数据，AI 智能体能够提供精准定制的推荐内容，从而显著提升用户参与度与满意度。以下示例主要依托语义记忆实现用户画像和偏好的跨会话持久化存储。

import json
from langchain_core.messages import AIMessage, HumanMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableConfig
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, StateGraph, MessagesState
from langgraph.store.memory import BaseStore, InMemoryStore
import os
import dotenv

dotenv.load_dotenv()
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY1")
os.environ["OPENAI_BASE_URL"] = os.getenv("OPENAI_BASE_URL")

# 假设fetch_product_recommendations、format_recommendation_message、 UserProfile 已在其他地方定义
recommendation_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个乐于助人的推荐引擎。根据用户资料，提供个性化的产品推荐。"),
    ("human", "{user_profile_summary}")
])

recommendation_chain = recommendation_prompt | ChatOpenAI(model="gpt-4-turbo") | (
    lambda x: {"messages": [AIMessage(content=x.content)]})


def recommend_products(state: MessagesState, config: RunnableConfig, store: BaseStore):
    """ 根据用户存储的偏好向用户推荐产品 """
    user_id = config["configurable"]["user_id"]
    namespace = ("user_profiles", user_id)
    user_profile_record = store.get(namespace, "profile")
    user_profile = user_profile_record.value if user_profile_record else {}
    user_profile_summary = format_user_profile_summary(user_profile)  # 用于格式化提示的用户资料字典的函数
    # 调用推荐链
    result = recommendation_chain.invoke({"user_profile_summary": user_profile_summary})
    return result


def format_user_profile_summary(user_profile: dict) -> str:
    """ 将用户资料字典格式化为字符串以进行提示注入 """
    name = user_profile.get("preferred_name", "用户")
    categories = ",".join(user_profile.get("preferred_product_categories", ["产品"]))
    return f"用户名为 {name}。他们偏好的产品类别是:{categories}。"


def extract_preference_updates(state: MessagesState) -> dict:
    """ 从最新的用户消息中提取用户偏好更新 """
    latest_message_content = state["messages"][-2].content
    # 示例:使用 LLM 提取偏好，替换为实际的提取逻辑 
    extraction_prompt = ChatPromptTemplate.from_messages([
        ("system",
         "从用户消息中提取用户的产品类别偏好。以JSON字典形式返回，外层不要包裹 '''json'''，键为 'preferred_product_categories'，值为类别列表。如果没有表达偏好，则返回一个空字典。"),
        ("human", "{user_message}")
    ])
    extraction_chain = extraction_prompt | ChatOpenAI(model="gpt-4-turbo") # 如果需要结构化输出，则请替换为合适的链
    preferences_json = extraction_chain.invoke({"user_message":latest_message_content})
    try:
        preferences = json.loads(preferences_json.content) # 假设 LLM 返回JSON 字符串
        return preferences
    except json.JSONDecodeError:
        return {} # 如果提取失败，则返回空字典
    
def update_user_profile_node(state: MessagesState, config: RunnableConfig, store: BaseStore):
    """ 在当前会话中触发的记忆存储中更新用户资料 """
    user_id = config["configurable"]["user_id"]
    namespace = ("user_profiles", user_id)
    user_profile_record = store.get(namespace, "profile") 
    user_profile = user_profile_record.value if user_profile_record else {}
    preference_updates = extract_preference_updates(state) # 从当前轮次提取偏好
    updated_profile = user_profile.copy() # 创建副本以避免修改原始字典
    if "preferred_product_categories" in preference_updates: # 合并或更新偏好
        updated_profile["preferred_product_categories"] = list(set(updated_profile.get("preferred_product_categories", []) + preference_updates["preferred_product_categories"])) # 示例:合并列表
    store.put(namespace, "profile", updated_profile) # 保存更新后的资料
    return {} # 节点应返回字典

memory_store = InMemoryStore()
builder = StateGraph(MessagesState)
builder.add_node("recommend_products", recommend_products)
builder.add_node("update_profile", update_user_profile_node) # 更新资料的节点
builder.add_edge(START, "recommend_products")
builder.add_edge("recommend_products", "update_profile") # 在推荐后更新资料
builder.add_edge("update_profile", END)
graph = builder.compile(store=memory_store)

# 初始化用户资料
user_id = "user_123" 
memory_store.put(
    ("user_profiles", user_id),
    "profile",
    {
        "preferred_name": "张三",
        "preferred_product_categories": ["电子产品", "书籍"]
    }
)
# 执行图:第一次交互(获取推荐)
config = {"configurable": {"user_id": user_id}}
result = graph.invoke({"messages": [HumanMessage(content="你好")]}, config=config)
print("初始推荐:")
print(result["messages"][-1].content)
print("=" * 50)
# 初始推荐:
# 根据张三的偏好，推荐以下产品：
# 
# **电子产品：**
# 1. **Apple MacBook Pro**: 适合需要高性能电脑的用户，尤其适合喜欢最新技术的用户。
# 2. **Kindle Paperwhite**: 这款电子书阅读器不仅适用于阅读电子书，同时其长久的电池寿命和防水功能也非常出色。
# 3. **Sony WH-1000XM4 无线降噪耳机**: 对于喜欢音乐或经常需要集中注意力工作的用户，这款顶级的降噪耳机非常合适。
# 
# **书籍：**
# 1. **《追风筝的人》（卡勒德·胡赛尼著）**: 这本书深受读者喜爱，适合喜欢深入人心故事的读者。
# 2. **《人类简史》（尤瓦尔·赫拉利著）**: 对历史感兴趣的张三可能会喜欢这本概述人类历史的书。
# 3. **《失控》（凯文·凯利著）**: 如果张三对科技和未来发展趋势感兴趣，这本书是一个不错的选择。
# 
# 希望张三会喜欢这些推荐！
# ==================================================

# 模拟用户表达新的偏好
user_message = "我最近对户外装备和运动鞋很感兴趣。"
result = graph.invoke(
    {"messages": result["messages"] + [HumanMessage(content=user_message)]},config=config)

# 检查更新后的用户资料
updated_profile = memory_store.get(("user_profiles", user_id), "profile").value
print("更新后的用户资料:")
print(json.dumps(updated_profile, ensure_ascii=False, indent=2))
print("=" * 50)
# 更新后的用户资料:
# {
#   "preferred_name": "张三",
#   "preferred_product_categories": [
#     "运动鞋",
#     "户外装备",
#     "书籍",
#     "电子产品"
#   ]
# }
# ==================================================

# 再次获取推荐，应该包含新的偏好
result = graph.invoke({"messages": [HumanMessage(content="我又来了")]}, config=config)
print("基于更新后资料的推荐:")
print(result["messages"][-1].content)
# 基于更新后资料的推荐:
# 基于张三的喜好，我会推荐以下产品：
# 
# 1. 运动鞋:
#    - 耐克 Air Zoom Pegasus 运动鞋：适合长时间跑步，轻便且吸震。
#    - 阿迪达斯 UltraBoost 21：提供极佳的能量回馈和支撑。
# 
# 2. 户外装备:
#    - 野营者4人自动帐篷：快速搭建，结实耐用，适合户外野营。
#    - 奥斯普雷登山背包：适合多日徒步旅行，有良好的负重系统和通风功能。
# 
# 3. 书籍:
#    - 《活出全新自己：高效能人士的七个习惯》（史蒂芬·柯维著）：帮助提升个人效率和人际关系的经典之作。
#    - 《穷查理宝典》（查理·芒格著）：展示查理·芒格的投资哲学和生活智慧。
# 
# 4. 电子产品:
#    - 苹果 MacBook Pro：适合需要高性能处理和便携性的用户。
#    - 索尼 WH-1000XM4 无线降噪耳机：顶级的降噪功能，适合旅行和日常使用，提升音质体验。
# 
# 这些推荐会帮助张三发现符合他兴趣和需求的高品质产品。希望他会喜欢这些精选的项!

该示例代码的核心实现逻辑包含以下关键设计要素。
- 记忆更新时机控制。
  - update_user_profile_node 节点被策略性地放置在 recommend_products 节点之后，形成即时更新机制。这种设计确保在生成推荐后立即更新用户资料，使同一对话线程的后续交互能够基于最新偏好进行决策。
- 智能信息提取实现。extract_preference_updates(state) 函数展示了 LLM 在信息提取中的应用。
  - 输入：当前对话状态(state)。
  - 处理：采用 ChatPromptTemplate 指导 LLM 解析最新用户消息。
  - 输出：结构化提取产品类别偏好(JSON 格式，键为 preferred_product_categories)。
  - 容错机制：包含 JSON 解析异常处理，支持扩展更健壮的提取逻辑(如函数调用)。
- 资料更新工作流。update_user_profile_node 执行完整的资料更新流程：
  - 检索：获取现有用户资料。
  - 提取：调用 extract_preference_updates() 获取当前轮次偏好更新。
  - 合并：将新偏好与现有资料智能融合。
  - 持久化：通过 store.put() 将更新后的资料存入记忆存储。

3.2 多步骤的情境化任务

对于需要引导用户完成多步骤任务或复杂工作流的 AI 智能体来说，记忆至关重要。短期记忆在跟踪任务的当前阶段、记住先前步骤中的用户输入，以及在整个过程中保持语境方面起着至关重要的作用。长期记忆可用于存储模板或成功的任务完成示例，以指导 AI 智能体的行为。

# 假设 book_flight_api、format_flight_confirmation、extract_departure_ city_from_message、extract_arrival_city_from_message 已定义
def book_flight(state: MessagesState, config: RunnableConfig):
    """ 引导用户完成航班预订流程，记住任务状态中的预订详细信息 """
    task_state = state.get("flight_booking_state", {})  # 从短期记忆中检索任务状态
    if not task_state.get("departure_city"):
        response = model.invoke([HumanMessage(content="您想从哪里出发?")] + state["messages"])
        return {"messages": response,
                "flight_booking_state": {"departure_city": "pending"}}  # 在短期记忆中更新任务状态，将 departure_city 标记为 pending
    elif task_state.get("departure_city") == "pending" and not task_state.get("arrival_city"):
        departure_city = extract_departure_city_from_message(state["messages"][-1].content)  # 从用户消息中提取信息
        if departure_city:  # 确保提取成功
            response = model.invoke([HumanMessage(content=f"您要从{departure_city}飞往哪里?")] + state["messages"])
            return {"messages": response,
                    "flight_booking_state": {"departure_city": departure_city, "arrival_city": "pending"}}  # 更新短期记忆
        else:  # 处理提取失败: 再次提示
            response = model.invoke(
                [HumanMessage(content="抱歉，我没有听清您的出发城市。您能换一种说法吗? ")] + state["messages"])
            return {"messages": response}  # 重新提示，任务状态保持不变
    elif task_state.get("arrival_city") == "pending":
        arrival_city = extract_arrival_city_from_message(state["messages"][-1].content)  # 提取到达城市
        if arrival_city:  # 确保提取成功
            flight_details = book_flight_api(departure_city=task_state["departure_city"],
                                             arrival_city=arrival_city)  #API 调用
            confirmation_message = format_flight_confirmation(flight_details)
            response = model.invoke([HumanMessage(content=confirmation_message)] + state["messages"])
            return {"messages": response, "flight_booking_state": {}} # 任务完成时清除任务状态
        else: # 处理提取失败:再次提示
            response = model.invoke([HumanMessage(content="抱歉，我没有听清您的到达城市。您能换一种说法吗?")]+state["messages"])
            return {"messages": response} # 重新提示，任务状态保持不变
    
    # 启动预订流程的初始提示
    response = model.invoke([HumanMessage(content="让我们为您预订航班。您要从哪里起飞? ")]+state["messages"])
    return {"messages": response, "flight_booking_state": {"departure_city": "pending"}} # 在短期记忆中初始化任务状态

builder = StateGraph(MessagesState)
builder.add_node("book_flight", book_flight)
builder.add_edge(START, "book_flight")
# ... ( 图定义的其余部分 )

该示例的核心实现逻辑包含以下关键设计要素。
- 记忆更新机制：book_flight 节点函数在任务每个步骤完成后立即更新短期记忆中的任务状态，确保分步工作流程中智能体能准确记录当前进度和后续步骤。
- 信息提取实现：extract_departure_city_from_message(message_content) 和 extract_arrival_city_from_message(message_content) 作为提取函数，有两种提取方法。
  - 正则表达式：适用于结构化 / 可预测的输入格式。
  - LLM 提示工程：处理自然语言输入的复杂情况。
- 容错处理：包含输入模糊情况的异常处理，支持用户重新输入提示，以增强流程的健壮性。

3.3 TrustCall：信息提取和记忆更新

LangGraph 与 TrustCall 的无缝集成为记忆管理提供了更可靠的解决方案。传统手动提取对话信息并更新智能体记忆的方式很复杂且容易出错，而 TrustCall 这一由 LangChain 团队开发的开源库，通过模式驱动的信息提取机制有效解决了这些问题。
TrustCall 的核心价值体现在以下方面。
- 结构化数据提取简化：支持从 LLM 输出中高效提取符合预定义模式(Pydantic模型 /JSON Schema)的结构化数据，确保记忆数据的类型安全与一致性。
- 严格模式强制执行：利用 LLM 工具调用机制强制输出符合预定义模式的结构化数据，显著减少传统字符串解析方法导致的格式错误。
- 增量更新优化：采用 JSON Patch 操作实现局部模式更新，支持非破坏性的记忆数据修改，提升用户画像等场景下的更新效率。
- 系统可靠性增强：通过模式驱动的方法提升记忆操作的可信度，确保数据结构的可预测性和一致性。
图将具体展示 TrustCall 的核心工作流程。

以下示例说明 TrustCall 从对话中提取用户资料信息(包括姓名和兴趣)，并使用 Pydantic 模式对其进行结构化。

from pydantic import BaseModel, Field
from typing import List
class UserProfile(BaseModel):
    """ 具有类型化字段的用户资料模式 """
    user_name: str = Field(description="用户的首选姓名") 
    interests: List[str] = Field(description="用户兴趣列表")

初始化 LangChain 聊天模型并使用 create_extractor() 创建 TrustCall 提取器，将模型和 UserProfile 模式作为工具传递。

1	pip install -U trustcall

from trustcall import create_extractor
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4-turbo", temperature=0)  # 初始化 LLM
trustcall_extractor = create_extractor(
    model,
    tools=[UserProfile],  # 将 Pydantic 模式作为工具传递 tool_choice="UserProfile" # 强制 TrustCall 使用 UserProfile 工具进行输出
)

现在，我们可以使用对话和指令调用 trustcall_extractor 以提取用户资料，这里以一个简单的对话作为 LangChain HumanMessage 和 AIMessage 对象的列表。

from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

conversation = [
    HumanMessage(content="嗨，我是 Alice。"),
    AIMessage(content="很高兴认识你，Alice !"),
    HumanMessage(content="我的爱好包括远足和阅读科幻小说。")]

instruction_prompt = "从以下对话中提取用户资料。"
result = trustcall_extractor.invoke({"messages": [SystemMessage(content=instruction_prompt)] + conversation})

trustcall_extractor.invoke() 方法返回一个字典，其中包含从 responses 键中提取的结构化输出。我们可以访问提取的 UserProfile 对象及其属性。

1
2
3

extracted_profile = result["responses"][0] # 访问提取的 UserProfile 对象
print(extracted_profile)
# user_name='Alice' interests=['hiking', 'reading science fiction novels']

该示例展示了 TrustCall 在结构化信息提取中的基本工作流程。TrustCall 通过封装 LLM 提示构造与响应解析的复杂性，使开发者能够直接使用符合预定义模式的 Python 对象，大幅简化了 LangGraph 智能体中的记忆更新操作。
这些应用场景充分证明了记忆系统对 AI 智能体的变革性影响。通过系统性地整合短期记忆与长期记忆机制，并借助 TrustCall 等工具实现高效模式管理，开发者能够构建出具备深度情境感知能力、个性化交互体验、持续学习进化机制和主动辅助功能的 AI 智能体系统。
记忆系统绝非简单附加组件，而是构建真正智能化、高交互性 AI 智能体的核心基础架构，使智能体能够与用户建立持久的互动关系。

4、LangMem

在系统阐述 AI 智能体记忆理论基础，并全面解析 LangGraph 记忆存储架构后，本节将详细介绍如何运用 LangChain 团队最新推出的 LangMem 工具库，将先进的记忆管理功能无缝集成至 LangGraph 智能体。

4.1 LangMem的核心组件

LangMem 是一个经过深思熟虑设计的双层架构，包括功能核心层和有状态集成层，如图所示。
- 功能核心层：记忆转换原语。该层提供一组无状态的功能性构建模块，支持开发者设计复杂的记忆操作流程，具有高模块化和可测试性的特点。核心原语包括:
  - 记忆管理器(create_memory_manager)：实现长期记忆的提取与更新逻辑抽象，使开发者能够专注记忆内容的处理策略，而无须关注底层存储实现细节。
  - 提示优化器(create_prompt_optimizer/create_multi_prompt_optimizer)：提供多种优化策略，持续改进智能体的程序记忆质量，确保核心指令随使用场景动态优化。
- 有状态集成层：LangGraph 记忆实现。在功能核心层之上构建的高阶集成组件，提供与 LangGraph 记忆存储深度整合的有状态服务。
  - 存储管理器(create_memory_store_manager)：自动化处理记忆的持久化存储与检索，抽象化底层存储管理复杂度，使开发者聚焦于业务逻辑设计。
  - 记忆管理工具有两种。
    - create_manage_memory_tool：支持智能体在对话过程中实时更新长期记忆。
    - create_search_memory_tool：实现记忆库的实时查询检索。

4.2 LangMem应用实例

以下为使用 LangMem 构建记忆增强型智能体的实际示例，重点涵盖短期记忆和长期记忆的实现技术，以及高效的记忆更新技术。
在编码前，请确保已正确设置以下环境：
1
pip install -U langmem
LangMem 支持在会话期间动态更新和利用记忆。
通过 create_manage_memory_tool 实现用户指定信息的持久化的存储。例如，用户指令：“记住我每周五下午 3 点有团队会议”，AI 智能体便会将这一信息存储到长期记忆中，以便后续调用。
通过 create_search_memory_tool 实现情境化信息检索。例如，用户询问“我今天有哪些任务?”，AI 智能体可以从记忆中检索出所有相关的任务信息，并给出精准的答复。

以下是使用记忆工具配置 ReAct 智能体的代码片段。

from langchain_core.messages import HumanMessage
from langgraph.prebuilt import create_react_agent
from langgraph.store.memory import InMemoryStore
from langmem import create_manage_memory_tool, create_search_memory_tool

agent = create_react_agent(
    "gpt-4-turbo",  # 选择 LLM 
    tools=[
        # 为智能体配备工具以在“热路径”中管理自己的记忆
        create_manage_memory_tool(namespace=("memories",)),  # 用于创建、更新、删除记忆的工具
        create_search_memory_tool(namespace=("memories",)),  # 用于搜索现有记忆的工具
    ],
    store=InMemoryStore(),  # 提供记忆存储以实现持久化 
)

# 执行示例:使用智能体进行简单对话
response = agent.invoke({"messages": [HumanMessage(content="请记住我, 喜欢编程。")]})
print("智能体响应:", response["messages"][-1].content)
# 智能体响应: 好的，我已经记住了你喜欢编程。如果还有其他事情要我记住，随时告诉我！

# 检索记忆以验证存储
search_result = agent.invoke({"messages": [HumanMessage(content="回忆一下我喜欢什么? ")]})
print("记忆检索结果:", search_result["messages"][-1].content)
# 记忆检索结果: 你喜欢编程。

在此示例中，我们使用 create_react_agent 创建了一个智能体，并为其配备了两个记忆工具：create_manage_memory_tool 和 create_search_memory_tool。这些工具均配置了命名空间 (“memories”)，作为记忆容器。InMemoryStore 本身已初始化，具有用于语义搜索的索引功能。

为了确保智能体不仅记住而且可以学习和改进，LangMem 的提示优化实用程序提供了一种强大的机制，用于自动改进程序记忆。想象一下，一个 AI 写作助手需要不断磨炼其写作风格。通过 create_prompt_optimizer 建立反馈循环，实现互动分析、写作风格优化点识别(如语气、清晰度、简洁性等)，并自动更新其系统提示(其核心程序记忆)以整合这些学习成果。

以下示例展示如何为这样的写作助手实现提示优化。

from langmem import create_prompt_optimizer

optimizer = create_prompt_optimizer(
    "gpt-4-turbo",  # 选择 LLM 以进行优化
    kind="gradient",  # 选择成功案例优化策略
    config={
        "max_reflection_steps": 5,
        "min_reflection_steps": 1
    },  #配置优化行为
)

# 带有反馈的示例对话轨迹
trajectories = [
    # 没有标注的对话(仅对话内容)
    (
        [
            {"role": "user", "content": "请告诉我Python的优点"},
            {"role": "assistant", "content": "Python是一种易于学习和使用的编程语言......"},
            {"role": "user", "content": "能详细说说它的库支持吗?"},
        ],
        None,
    ),
    # 带有反馈的对话
    (
        [
            {"role": "user", "content": "Python有哪些流行的库?"},
            {"role": "assistant", "content": "Python有许多流行的库，如NumPy、Pandas等", }
        ],
        {
            "score": 0.8,
            "comment": "可以增加库的应用场景和更多细节",
        }
    ),
    # 标注内容可以是不同类型的，例如编辑或修订原内容
    (
        [
            {"role": "user", "content": "Python和Java相比如何?"},
            {"role": "assistant", "content": "Python和Java在语法和应用领域上有许多不同......"},
        ],
        {"revised": "Python和Java在语法、性能和应用领域上各有优劣......"},
    ),
]

# 定义写作助手的初始系统提示
initial_prompt = "您是一位乐于助人的写作中文助手"
# 调用优化器以根据训练数据改进初始提示
optimized_prompt = optimizer.invoke(
    {"trajectories": trajectories, "prompt": initial_prompt}  # 提供轨迹和要优化的初始提示
)
print(optimized_prompt)  # 输出优化的提示
# 您是一位乐于助人的写作中文助手。在讨论技术话题时，请提供详尽的例子或情景，以便更全面地说明或比较不同技术。若用户的提问广泛或含糊，请主动询问具体化的问题，确保回答尽可能详细和具体。

4.3 LangMem关键函数解析

4.3.1 create_memory_manager函数解析

此函数是 LangMem 功能 API 的核心，用于创建记忆管理器 Runnable，以实现无状态记忆提取和更新。关键参数如下所示。
- model(str | BaseChatModel)：指定记忆管理器要使用的 LLM，可以是模型名称字符串或初始化的 LangChain BaseChatModel 实例。
- schemas(Optional[Union[type[str],list[type[BaseModel]],list[dict]]])：定义提取记忆模式支持以下三个值。
  - None(默认值)：非结构化字符串记忆。
  - Pydantic BaseModel：定义具有类型化字段的结构化记忆。
  - List[Pydantic BaseModel]：结构化记忆集合。
- instructions(str)：指导记忆提取和更新过程的 LLM 指令。
- enable_inserts(bool)：一个布尔标志，控制是否允许新增记忆(默认为True值)。
- enable_deletes(bool)：控制是否允许删除记忆。
- enable_updates(bool)：控制是否运行更新记忆。

4.3.2 create_prompt_optimizer函数解析

该函数用于创建提示优化器，实现自动提示改进功能。其关键参数如下所示。
- model (str | BaseChatModel)：指定用于提示优化的 LLM。建议选择功能强大的 LLM ，优化效果直接依赖底层模型的推理能力。
- kind (Literal[‘’gradient’’,’’metaprompt’’,’’prompt_memory’’])：选择优化策略，具体如下所示。
  - gradient：基于梯度的方法，通过多轮反思迭代改进提示，优化计算量。
  - metaprompt：利用元学习技术直接生成提示更新，在优化效果和计算效率之间取得平衡。
  - prompt_memory：利用历史成功模式指导提示改进，提供更简单且具有情境感知能力的优化方法。
- config (Optional[dict])：允许进行特定于策略的配置，微调优化过程参数，例如，控制反思步骤的数量、调整元提示指令等。

4.3.3 create_manage_memory_tool函数解析

此函数创建 manage_memory 工具，用于操作长期记忆。关键参数如下所示。
- namespace (tuple[str, …] | str)：定义记忆存储的命名空间。命名空间对于组织记忆和确保数据隔离至关重要(例如，按用户、按智能体、按应用程序)。此参数接受字符串元组作为分层命名空间，或接受单个字符串作为平面命名空间。
- instructions (str)：提供 LLM 智能体使用此工具的指导说明，应清晰简洁。
- schema (type)：指定由此工具管理记忆模式，支持使用 Python str、Pydantic BaseModel 和 JSON 模式。
- actions_permitted (tuple[Literal[‘’create’’, ‘’update’’, ‘’delete’’], …])：控制允许执行的操作，包括“create”“update”“delete” 记忆。也可以限制这些操作以创建更受控的记忆管理工作流。
- store (Optional[BaseStore])：可显式指定该工具要使用的 BaseStore 实例。未指定时，该工具将默认使用在 LangGraph 中配置的 BaseStore，确保与应用程序无缝集成。

4.3.4 create_search_memory_tool函数解析

此函数用于创建 search_memory 工具，使智能体能够查询其长期记忆。关键参数如下所示。
- namespace (tuple[str, …] | str)：定义要搜索的命名空间。建议与 create_ manage_memory_tool 使用的命名空间匹配，以确保智能体可以搜索其正在管理的相同记忆空间。
- instructions (str)：提供使用此工具的指导说明。要求指令清晰明确。
  - response_format (Literal[‘’content’’,’’content_and_artifact’’])：控制搜索结果响应的格式。
  - content( 默认值 ):仅返回记忆内容，为对话智能体提供简洁且用户友好的输出。
- content_and_artifact：返回一个元组，其中包含 content 和原始检索结果对象，适用于需要返回记忆属性的高级场景。
- store(Optional[BaseStore])：显式指定 BaseStore 实例。未指定时，该工具将默认为使用在 LangGraph 环境中配置的 BaseStore。