简介

本文深入探讨LangGraph框架下的长短期记忆管理机制，详细介绍了短期记忆的存储、修剪、删除和总结策略，以及长期记忆的存储、更新和语义检索方法。通过实战案例，展示了如何构建融合长短期记忆的智能Agent系统，帮助AI实现类人记忆能力，提供更连贯、个性化的交互体验。

---

在 AI 智能体持续发展的进程中，我们深刻认识到赋予其类人记忆能力是构建真正智能实用系统的关键所在。正如人类依靠记忆理解世界、积累经验并维持连贯对话，AI 智能体同样需要记忆系统来超越简单反应式交互，实现更高级的智能水平和用户体验。

本文将深入探讨 Agent Memory 的核心概念，并聚焦于LangGraph框架下的长短期记忆实现，详解短期会话与长期知识的存储、管理、语义检索等技巧。更进一步地，通过一个实战案例，手把手带你构建一个真实的融合长记忆机制的Agent系统，直观展示中断、记忆与协作的融合。

要创建真正智能且实用的 AI 智能体，关键在于为其配备能够模仿人类认知特性的记忆系统。正如人类利用短期记忆和长期记忆来进行有效的互动和学习一样，AI智能体也同样需要利用记忆来演进。

关于 Agent Memory 我们需要考虑什么？

• • 如何获取记忆：通过和用户交互、环境交互等…
• • 怎么组织记忆：模型参数、模型上下文、数据库
• • 怎么利用记忆：RAG、Few-shot等…

有哪些Memory类型？

关于Memory的分类，有许多种分类体系，这里我们只讨论最常见及最易于理解的。

正如人类利用长短期记忆进行有效的交互和学习一样，Agent的记忆机制通常划分为短期记忆（Short-Term Memory）和长期记忆(Long-Term Memory)，短期记忆决定了Agent在微观任务上的即时表现，而长期记忆则作为持久知识库，决定了Agent在宏观时间尺度上的智能深度和个性化水平，通过两者配合，Agent才能表现出连贯性、上下文感知能力，才会显得更智能。

Agent Memory如何工作？

Agent通常通过以下几步来有效地管理记忆，使得每次于用户的交互都更加精准智能：

• • 记忆存储：通过设计一系列策略来存储重要的交互信息，这些信息可能来源于对话内容、历史数据或任务要求等等。
• • 记忆更新：记忆会随着交互的发生，不断地进行更新，例如用户的偏好、最新的近况等等。记忆更新使得Agent能够不断优化其响应。
• • 记忆检索：Agent根据当下的需求，去记忆中检索需要的记忆内容，从而提供更加智能的回复。

Agent Memory怎么实现？

• • 物理外挂：即外置数据库和 RAG，需要检索当前query相关的内容，例如：Mem0、ACE。好处是即插即用，坏处是不够end-to-end。
• • Memory as Reasoning / Tool：通过训练Reasoning或Tool的方式动态更新context，例如：MemAgent、memory-R1。好处是更接近end-to-end，但不是很灵活。
• • 参数更新：LLM本身就是一个Memory体，所有参数都是它的Memory，通过更新参数来更新记忆，这种方式是最本质的，但也是最难实现的。

短期记忆与长期记忆

短期记忆的核心价值在于保障对话的即时性。它使 AI 智能体 能够专注于当前对话情境，准确理解用户输入的细微差别，并维持对话的连贯性。缺乏短期记忆将导致每轮对话彼此割裂，丧失上下文关联。

◎ 对话历史记录：按时间顺序完整记录用户与智能体之间交换的所有消息，为交互过程提供即时历史轨迹。
◎ 当前任务上下文：明确记录用户当前目标的详细参数和要求，包括智能体正在处理的具体任务及其相关参数或约束条件。
◎ 用户意图（会话特定）：智能体对用户在本次对话中所表达需求和目标的动态理解，这一理解会随着对话轮次不断深化和完善。
◎ 中间结果（会话特定） ：在对话过程中生成的临时数据和输出，包括计算中间值、检索结果或工具输出等即时处理所需的信息。

长期记忆则提供了持续发展的基础。它使AI 智能体能够突破单次交互的局限，建立稳定的身份特征，从历史经验中持续学习，最终提供日益个性化和精准的服务。没有长期记忆的智能体将停滞不前，难以与用户建立深度互动关系。

◎ 语义记忆（Semantic Memory）—— 结构化知识库，存储概念、关系等事实性知识。包括用户资料、产品目录、领域知识等。
◎ 情景记忆（Episodic Memory）—— 自传式记录，存储智能体历史互动、经验和事件，支持特定对话回溯，从过去的成功和失败中学习，并基于历史数据改进策略。
◎ 程序记忆（Procedural Memory）—— 技能知识库，存储核心系统提示、其底层代码和算法，以及任何已学习的有效互动和任务执行的最佳实践。

LangGraph中的记忆管理

LangGraph作为一款面向多智能体协作与状态管理的框架，其设计了巧妙的记忆管理系统，旨在为Agent提供在不同交互中存储、检索和利用信息的能力。它区分了两种主要的记忆类型：短期记忆和长期记忆。在实际使用中，通过这两种记忆协同，既能保障实时任务的高效执行，又支持了跨任务、跨周期的经验复用。

• • Short-Term Memory（通过Checkpointer实现）：针对单个对话线程，核心价值在于保障对话的临时性，使得Agent能够跟踪会话中的多轮对话，可以在该线程内的任何时刻被回忆。

• • Long-Term Memory（通过BaseStore实现）：可以跨对话线程共享，可以在任何时间，任何线程中被回忆，而不像短期记忆局限于单个对话。

• 

  

Checkpointer

LangGraph有一个内置的持久化（Persistence）层，通过checkpointer实现，能够持久化存储图状态，这使得开发记忆功能和人类干预功能成为可能。

当使用检查点编译一个图时，检查点会在每个super-step保存图状态的checkpoint，这些checkpoint被保存到一个thread中，可以在图执行后访问。因为threads允许在执行后访问图的状态，所以可以实现记忆、人机协作、时间旅行、容错等多种强大的功能。

BaseStore

图状态可以由checkpointer在每个super-step写入线程，从而实现状态的持久化。但是，如果想在多个线程之间保留一些信息的话，那么就需要用到Store。Store本质上是一个暴露给图节点和工具的键值数据库，与checkpointer的自动化快照不同，Store需要显式和主动的进行操作。

完整的大模型学习和面试资料已经上传带到CSDN的官方了，有需要的朋友可以扫描下方二维码免费领取【保证100%免费】👇👇
​​

短期记忆详解

InMemorySaver内存会话临时存储

对于开发、原型设计和测试阶段，最简单快捷的方式是使用InMemorySaver。它将所有的对话状态存储在内存中的一个Python字典里。

from langchain_openai import ChatOpenAIfrom langchain.chat_models import init_chat_modelfrom langgraph.checkpoint.memory import InMemorySaverfrom langgraph.prebuilt import create_react_agent#1.设置记忆管理检查点# 初始化检查点保存器checkpointer = InMemorySaver()#2.定义大模型并创建agentBASE_URL=""TOKEN=""MODEL_NAME=""model = init_chat_model(    model=MODEL_NAME,    model_provider="openai",     base_url=BASE_URL,    api_key=TOKEN,    temperature=0,)agent = create_react_agent(    model=model,    tools=[],    # 传入检查点，是将持久化能力“注入”图的关键步骤。编译后的graph对象现在具备了状态管理的能力。    checkpointer=checkpointer)#3.短期记忆-内存后端config = {"configurable": {"thread_id": "1"}}  # 激活记忆机制的核心。如果没有提供thread_id，每次invoke调用都将是无状态的，只要使用相同的thread_id，LangGraph就会在多次调用之间维持对话状态response = agent.invoke(    {"messages": [{"role": "user", "content": "你好，我叫ada！"}]},    config)print(f"thread1_bot_answer：{response['messages'][-1].content}")response = agent.invoke(    {"messages": [{"role": "user", "content": "你好，请问你还记得我叫什么名字么？"}]},    config)print('------------线程1------------------')print(f"thread1_bot_answer：{response['messages'][-1].content}")new_config = {"configurable": {"thread_id": "2"}}response = agent.invoke(    {"messages": [{"role": "user", "content": "你好，请问你还记得我叫什么名字么？"}]},    new_config)print('------------线程2------------------')print(f"thread2_bot_answer：{response['messages'][-1].content}")#输出如下：thread1_bot_answer：你好，Ada！很高兴认识你！😊这是一个很美的名字呢！有什么我可以帮助你的吗？无论是想聊聊天，还是有任何问题需要解答，我都很乐意为你提供帮助。------------线程1------------------thread1_bot_answer：当然记得！你刚才告诉我你叫 Ada～很高兴再次和你打招呼！😊------------线程2------------------thread2_bot_answer：你好！很抱歉，我无法记住之前对话中的个人信息，比如你的名字。这是为了保护你的隐私，所以我不会保留这类数据。你可以告诉我你的名字，或者任何你想让我称呼你的方式，我会很乐意在这次的对话中使用它！😊

短期记忆与线程相关，在对话时，需要在配置中传入thread_id。通过上面的结果我们可以看到，当我们传入相同的thread_id时，agent就可以记住用户的名字，然而当我们更换thread_id时，agent就不记得用户的名字了。

需要注意的是，InMemorySaver将所有状态都保存在内存中，一旦程序终止，那么所有对话历史都会消失。

短期记忆管理策略

随着对话的进行，短期记忆（对话历史）会不断增长，可能会超出LLM的上下文窗口，导致请求调用失败，或者使LLM反应变慢、变差。这时，就需要对记忆进行管理了。常见的解决办法有：

• • 修剪消息（trim messages）：移除前 N 条或后 N 条消息（在调用 LLM 之前）。最简单直接，但信息丢失严重，适合短期任务、无状态问答机器人、近期上下文最重要的应用。

from langchain_core.messages.utils import (    trim_messages,    count_tokens_approximately)from langchain.chat_models import init_chat_modelfrom langgraph.checkpoint.memory import InMemorySaverfrom langgraph.graph import StateGraph, START, MessagesStateBASE_URL=""TOKEN=""MODEL_NAME=""model = init_chat_model(    model=MODEL_NAME,    model_provider="openai",     base_url=BASE_URL,    api_key=TOKEN,    temperature=0,)summarization_model = model.bind(max_tokens=128)def call_model(state: MessagesState):    # 保留最近消息，总 token ≤ 128    messages = trim_messages(   state["messages"],   strategy="last",  # 保留最后的消息   token_counter=count_tokens_approximately,   max_tokens=128,   start_on="human",  # 确保第一条消息（不包括系统消息）是从human消息开始保留   end_on=("human", "tool"),  # 保留到human或tool消息为止   allow_partial=False,  # 不允许分割消息内容   include_system=True  # 保留system prompt    )    # --- 在这里打印传入模型的内容 ---    print("-" * 20)    print(f"Messages being sent to the model (trimmed to <= 128 tokens): {len(messages)}")    for msg in messages:   print(f"  [{msg.type.upper()}]: {msg.content}")    print("-" * 20)    response = model.invoke(messages)    return {"messages": [response]}checkpointer = InMemorySaver()builder = StateGraph(MessagesState)builder.add_node(call_model)builder.add_edge(START, "call_model")graph = builder.compile(checkpointer=checkpointer)

• • 删除消息（delete messages）：从LangGraph状态中永久删除消息。可以精确的控制移除内容，但需要自定义逻辑来判断哪些消息需要删除，适合用于移除不再需要的冗余系统消息、工具输出或错误信息。

from langchain_core.messages import RemoveMessagefrom langchain.chat_models import init_chat_modelfrom langgraph.checkpoint.memory import InMemorySaverfrom langgraph.graph import StateGraph, START, MessagesStateBASE_URL=""TOKEN=""MODEL_NAME=""model = init_chat_model(    model=MODEL_NAME,    model_provider="openai",     base_url=BASE_URL,    api_key=TOKEN,    temperature=0,)def delete_messages(state):    messages = state["messages"]    if len(messages) > 2:   # 删除最早的两条消息   return {"messages": [RemoveMessage(id=m.id) for m in messages[:2]]}    def call_model(state: MessagesState):    response = model.invoke(state["messages"])    return {"messages": response}builder = StateGraph(MessagesState)builder.add_sequence([call_model, delete_messages])builder.add_edge(START, "call_model")checkpointer = InMemorySaver()app = builder.compile(checkpointer=checkpointer)config = {"configurable": {"thread_id": "1"}}for event in app.stream(    {"messages": [{"role": "user", "content": "你好，我是ada"}]},    config,    stream_mode="values"):    print([(message.type, message.content) for message in event["messages"]])for event in app.stream(    {"messages": [{"role": "user", "content": "我叫什么名字"}]},    config,    stream_mode="values"):    print([(message.type, message.content) for message in event["messages"]])

• • 总结消息（summarize messages）：汇总历史记录中的早期消息并将其替换为摘要。保留了核心语义信息，但计算成本高，实现相对复杂，适合用于长期连续对话、需要维持深度长期上下文的智能体。

#首先，安装LangMem，这是一个由LangChain维护的库，提供了用于在agent中管理记忆的工具。pip install -U langmem
``````plaintext
#langmem库提供了一个预构建的SummarizationNode，可以极大地简化实现过程import tiktokenfrom typing import Any, TypedDictfrom langchain.chat_models import init_chat_modelfrom langchain_core.messages import AnyMessage, BaseMessage, SystemMessagefrom langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholderfrom langgraph.graph import StateGraph, START, MessagesStatefrom langgraph.checkpoint.memory import InMemorySaverfrom langmem.short_term import SummarizationNode, RunningSummarysummary_prompt = ChatPromptTemplate.from_messages(    [   MessagesPlaceholder(variable_name="messages"),   # 使用 HumanMessage 模拟用户在最后发出总结指令   ("human", "请根据以上对话，生成一段简洁、连贯的中文摘要，注意不要丢失细节"),    ])update_summary_prompt = ChatPromptTemplate.from_messages(    [   MessagesPlaceholder(variable_name="messages"),   # 使用 HumanMessage 模拟用户在最后发出总结指令   ("human", "以下是目前为止的对话摘要：{existing_summary}\n\n请根据以上新消息扩展此摘要："),    ])BASE_URL=""TOKEN=""MODEL_NAME=""model = init_chat_model(    model=MODEL_NAME,    model_provider="openai",     base_url=BASE_URL,    api_key=TOKEN,    temperature=0,)summarization_model = model.bind(max_tokens=128)# count_tokens_approximately更适合英文分词，中文这里使用tiktoken来计算token数量encoding = tiktoken.get_encoding("cl100k_base")def count_tokens_accurately(messages: list[BaseMessage]) -> int:    """使用 tiktoken 精确计算消息列表的 token 总数"""    # 注意：langmem 的 token_counter 期望的输入是消息列表    text_content = " ".join([msg.content for msg in messages if isinstance(msg.content, str)])    return len(encoding.encode(text_content))class State(MessagesState):    context: dict[str, RunningSummary]  class LLMInputState(TypedDict):      summarized_messages: list[AnyMessage]    context: dict[str, RunningSummary]summarization_node = SummarizationNode(  #token_counter=count_tokens_approximately,    token_counter=count_tokens_accurately,  # 更换为自定义的token计算工具    model=summarization_model,    max_tokens=256,    max_tokens_before_summary=256,    max_summary_tokens=128,    initial_summary_prompt=summary_prompt,  # 使用自定义prompt，默认为英文    existing_summary_prompt=update_summary_prompt)def call_model(state: LLMInputState):    response = model.invoke(state["summarized_messages"])    return {"messages": [response]}checkpointer = InMemorySaver()builder = StateGraph(State)builder.add_node(call_model)builder.add_node("summarize", summarization_node)  builder.add_edge(START, "summarize")builder.add_edge("summarize", "call_model")graph = builder.compile(checkpointer=checkpointer)# Invoke the graphconfig = {"configurable": {"thread_id": "1"}}graph.invoke({"messages": "你好，我叫ada"}, config)graph.invoke({"messages": "请写一首诗，关于小狗的"}, config)graph.invoke({"messages": "再写一首关于小猫的"}, config)final_response = graph.invoke({"messages": "我叫什么名字？"}, config)final_response["messages"][-1].pretty_print()# 检查摘要是否生成if"running_summary"in final_response["context"]:    print("\n生成的摘要:", final_response["context"]["running_summary"].summary)else:    print("\n对话较短，尚未生成摘要。")输出如下：你叫Ada。生成的摘要：用户自我介绍说她叫Ada，并请求写一首关于小狗的诗。助手回应问候并写了一首描述小狗在阳光下欢快奔跑、充满喜悦和纯真爱的诗。随后，用户请求再写一首关于小猫的诗。助手回应并写了一首描述小猫轻盈如风、充满好奇心和温暖陪伴的诗。

• • 自定义策略：例如消息过滤等。

数据库持久化存储

可以发现，上面一小节的代码在应用程序结束后再启动，记忆就又消失了。这是因为InMemorySaver仅仅是把记忆保存在内存中，应用程序结束后释放内存记忆就消失了。在生产环境中常常使用数据库支持的检查点记录器持久化保存记忆，以保证数据的可靠性和服务的连续性。

这里我们以postgres数据库为例来说明，怎么持久化地保存记忆数据。

1.首先安装以下依赖：

pip install -U "psycopg[binary,pool]" langgraph-checkpoint-postgres

2.安装postgres数据库，具体的安装方法可以参考：Linux下安装PostgreSQL_linux安装postgresql-CSDN博客。这里选择以源码的方式进行安装，安装包从官网（PostgreSQL: Downloads）下载，选择最新的。
3.安装数据库成功后，编码如下代码。DB_URI是数据库连接的URL。想要自动保存在数据库中的State需要在PostgresSaver.from_conn_string(DB_URI)上下文中操作。

from langchain.chat_models import init_chat_modelfrom langgraph.graph import StateGraph, MessagesState, STARTfrom langgraph.checkpoint.postgres import PostgresSaverBASE_URL=""TOKEN=""MODEL_NAME=""model = init_chat_model(    model=MODEL_NAME,    model_provider="openai",     base_url=BASE_URL,    api_key=TOKEN,    temperature=0,)DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"with PostgresSaver.from_conn_string(DB_URI) as checkpointer:    checkpointer.setup()  # 第一次调用时必须要setup()        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")        graph = builder.compile(checkpointer=checkpointer)        config = {   "configurable": {"thread_id": "1"   }    }    response = graph.invoke(   {"messages": [{"role": "user", "content": "你好，我叫ada！"}]},   config    )    print(response['messages'][-1].content)    response = graph.invoke(   {"messages": [{"role": "user", "content": "你好，请问你还记得我叫什么名字么？"}]},   config    )    print(response['messages'][-1].content)

我们检查数据库可以发现，postgres数据库中出现了四个表：

除了PostgreSQL之外，LangGraph还支持MongoDB、Redis等数据库。

子图中的记忆

当构建复杂的、由多个子图嵌套而成的应用时，需要更灵活的记忆管理策略。

• • 记忆继承（默认）：默认情况下，子图会继承其父图的checkpointer。这意味着整个嵌套图共享同一个对话状态，数据可以在父子图之间无缝流动。这对于将一个大型任务分解为多个模块化子任务非常有用。
• • 记忆隔离：在某些场景下，例如构建多智能体系统，希望每个智能体（由一个子图表示）拥有自己独立的内存空间，互不干扰。此时，可以在编译子图时设置checkpointer=True。

长期记忆详解

LangGraph中的长期记忆允许系统在不同对话中保留信息，是跨对话线程共享的，可以在任何时间、任何线程中被回忆。与短期记忆不同，长期记忆保存在自定义的命名空间中，每个记忆都组织在一个自定义的namespace和一个唯一的key下。

记忆存储：LangGraph将长期记忆存储为JSON文档，使用Store进行管理，允许存储结构化和非结构化的数据。

LangGraph 的记忆存储以 BaseStore 类及其实现为基础，为长期记忆管理提供了抽象层支持。作为开源且可扩展的持久化键值存储接口，它能让开发者根据自身特定需求选择适配的存储后端 —— 从开发阶段适用的轻量级内存解决方案，到生产环境所需的高性能数据库均可覆盖，同时确保记忆管理 API 的一致性。

记忆更新

put 方法是实现信息存储的基础操作，需要提供以下三个核心参数。
（1）命名空间（元组）。
功能：作为记忆的逻辑分组机制，类比文件系统中的文件夹。
格式：采用元组定义，支持分层组织结构。
命名规范：建议包含用户标识符和记忆类别，如(“user_123”,“chat_history”)。
（2）键（字符串）。
功能：命名空间内记忆条目的唯一标识符。
命名建议：应具备描述性和易检索性。
生成方式：推荐使用Python 的uuid 库生成通用唯一标识符。
（3）值（字典）。
格式要求：必须为Python 字典对象。
优势：支持结构化数据和元数据存储。
应用示例：用户资料可存储为包含姓名、偏好等键值对的字典。

记忆检索

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

1、store.get()：根据命名空间和键精确获取记忆。

get(namespace, key)。
功能：基于精确键值的直接记忆检索。
参数说明如下。
◎ namespace（元组）：目标记忆命名空间。
◎ key（字符串）：目标记忆唯一键。
特点：针对已知记忆的高效精准检索。

from langchain.chat_models import init_chat_modelfrom langchain_core.runnables import RunnableConfigfrom langgraph.store.memory import InMemoryStorefrom langgraph.config import get_storefrom langgraph.prebuilt import create_react_agentstore = InMemoryStore()BASE_URL=""TOKEN=""MODEL_NAME=""model = init_chat_model(    model=MODEL_NAME,    model_provider="openai",     base_url=BASE_URL,    api_key=TOKEN,    temperature=0,)store.put(    ("users",),  # 命名空间：元组类型，类比文件系统中的文件夹，支持分层组织结构    "user_123",  # 键: 字符串，是命名空间内的唯一标识符，一般推荐使用uuid库生成唯一标识符    {   "name": "ada",   "language": "中文",    }  # 值：Python字典类型，比如保存公共角色资料时可以是包含姓名、偏好等键值对的字典)def get_user_info(config: RunnableConfig) -> str:    """查找用户信息的函数，可以查看长期记忆中储存的用户信息"""    store = get_store()  # 获取上下文中可用的store实例    user_id = config["configurable"].get("user_id")    user_info = store.get(("users",), user_id)  # 输入命名空间和键进行精确查询    return str(user_info.value) if user_info else"Unknown user"agent = create_react_agent(    model=model,    tools=[get_user_info],    # 传入store    store=store)response = agent.invoke(    {"messages": [{"role": "user", "content": "帮我查找长期记忆中储存的用户信息"}]},    config={"configurable": {"user_id": "user_123"}})print(response['messages'])

为了让记忆真正”长期“，生产环境必须使用数据库支持的Store，LangGraph目前主要支持PostgresStore和RedisStore。

2、store.search()：在指定命名空间内实现灵活记忆检索，不但可以通过命名空间和标识符，更可以通过语义检索到记忆内容。通常需要Store配置一个embed来支持语义搜索。

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

启用语义搜索时需要进行索引配置：

（1） 基本配置参数。
◎ embed 函数：处理从文本到向量嵌入的转换，支持各类嵌入模型（如
OpenAI、Cohere 等）。
◎ dims 参数：指定嵌入向量的维度，必须与所用嵌入模型的输出维度一致。
（2）可选配置参数。
◎ fields 参数：指定记忆字典中需要索引的特定字段（默认索引整个字典）。

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

from langchain.embeddings import OpenAIEmbeddingsfrom langgraph.store.memory import InMemoryStore# 初始化BGE-M3 向量化模型embeddings = OpenAIEmbeddings(model="BAAI/bge-m3") # 或其他向量化模型# 使用语义搜索索引配置InMemoryStorestore_with_semantic_search = InMemoryStore(index={"embed": embeddings.embed_documents,"dims": 1024, # BGE-M3 的向量维度"fields": ["memory_content"] # 仅向量化memory_content 字段（可选）})# 保存将为语义搜索索引的记忆（默认行为）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, # 禁用此记忆的索引)# 保存一个记忆，覆盖默认索引字段并仅索引contextstore_with_semantic_search.put(("user_789", "restaurant_reviews"),"memory_3",{"memory_content": " 服务很慢，但食物很好。", "context": " 对 'The Italian Place' 餐厅的评论"},index=["context"] # 仅索引 context 字段)

使用 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)Plaintext:查询的语义搜索结果：该用户喜欢哪种食物？记忆键：memory_1，相似度评分：0.3914929762045983记忆内容：{'memory_content': ' 我真的很喜欢辛辣的印度咖喱。'}==============================

航班预订：多步骤任务或复杂工作流的 AI 智能体

接下来，我们模拟一个用户去旅游的航班预订场景，引导用户完成航班预订流程，记住任务状态中的预订详细信息。

import jsonfrom typing import Dict, Any, Listfrom langgraph.graph import StateGraph, END, STARTfrom langgraph.store.memory import InMemoryStore, BaseStorefrom langchain_core.messages import BaseMessage, HumanMessage, AIMessagefrom langchain_core.runnables import RunnableConfigfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_openai import ChatOpenAI# 定义任务状态类型class TaskState(dict):    """任务状态管理"""    def __init__(self, **kwargs):        super().__init__(kwargs)        if 'messages' not in self:            self['messages'] = []        if 'task_data' not in self:            self['task_data'] = {}        if 'current_step' not in self:            self['current_step'] = 0        if 'task_type' not in self:            self['task_type'] = None        if 'task_complete' not in self:            self['task_complete'] = Falseclass TaskCompletionAgent:    """多步骤任务完成智能体"""        def __init__(self):        # 定义不同类型的任务模板        self.task_templates = {            "flight_booking": {                "steps": [                    {"field": "departure_city", "prompt": "请告诉我您的出发城市：", "required": True},                    {"field": "arrival_city", "prompt": "请告诉我您的目的地城市：", "required": True},                    {"field": "departure_date", "prompt": "请提供出发日期（格式：YYYY-MM-DD）：", "required": True},                    {"field": "passengers", "prompt": "请告诉我乘客人数：", "required": True}                ],                "completion_message": "✈️ 航班信息已收集完毕！正在为您查找最佳航班..."            },            "restaurant_reservation": {                "steps": [                    {"field": "cuisine_type", "prompt": "您想预订哪种类型的餐厅？（如：中餐、西餐、日料等）", "required": True},                    {"field": "date_time", "prompt": "请提供用餐日期和时间：", "required": True},                    {"field": "party_size", "prompt": "请告诉我用餐人数：", "required": True}                ],                "completion_message": "🍽️ 餐厅预订信息已收集完毕！正在为您查找合适的餐厅..."            }        }def process_task(state: TaskState, config: RunnableConfig, store: BaseStore):    """处理任务的主要逻辑"""    print(f"🔄 处理任务 - 当前状态: step={state.get('current_step', 0)}, type={state.get('task_type')}")        agent = TaskCompletionAgent()        # 如果还没有任务类型，先检测任务类型    if not state.get('task_type'):        if not state['messages']:            return state                last_message = state['messages'][-1].content.lower()                # 检测任务类型        if any(keyword in last_message for keyword in ['机票', '航班', '飞机', '预订机票']):            task_type = "flight_booking"        elif any(keyword in last_message for keyword in ['餐厅', '预订', '吃饭', '订餐']):            task_type = "restaurant_reservation"        else:            task_type = "flight_booking"                print(f"📋 检测到任务类型: {task_type}")                # 初始化任务状态        state['task_type'] = task_type        state['current_step'] = 0        state['task_data'] = {}        state['task_complete'] = False                # 添加确认消息和第一个问题        template = agent.task_templates[task_type]        first_step = template['steps'][0]                confirmation_messages = {            "flight_booking": f"我来帮您预订航班。{first_step['prompt']}",            "restaurant_reservation": f"我来帮您预订餐厅。{first_step['prompt']}"        }                ai_message = AIMessage(content=confirmation_messages[task_type])        state['messages'].append(ai_message)        return state        # 如果已经完成任务，直接返回    if state.get('task_complete'):        return state        # 处理信息收集    task_type = state['task_type']    template = agent.task_templates[task_type]    steps = template['steps']    current_step = state['current_step']        # 检查是否还有步骤需要完成    if current_step >= len(steps):        # 所有信息已收集完毕        state['task_complete'] = True        response = template['completion_message']        response += "\n\n📋 收集到的信息：\n"        for key, value in state['task_data'].items():            response += f"• {key}: {value}\n"                ai_message = AIMessage(content=response)        state['messages'].append(ai_message)        return state        # 检查是否有新的用户消息需要处理    if len(state['messages']) >= 2 and isinstance(state['messages'][-1], HumanMessage):        # 获取当前步骤信息        step_info = steps[current_step]        field_name = step_info['field']        user_message = state['messages'][-1].content                print(f"🎯 处理用户输入: {user_message} -> 提取字段: {field_name}")                # 尝试提取信息        extracted_value = extract_field_value(user_message, field_name)                if extracted_value and extracted_value != "NOT_FOUND":            # 成功提取信息            state['task_data'][field_name] = extracted_value            state['current_step'] += 1                        print(f"✅ 成功提取 {field_name}: {extracted_value}")                        # 检查是否还有更多步骤            if state['current_step'] < len(steps):                next_step = steps[state['current_step']]                next_prompt = next_step['prompt']                response = f"已记录您的{field_name}：{extracted_value}。{next_prompt}"            else:                # 所有信息已收集完毕                state['task_complete'] = True                response = template['completion_message']                response += "\n\n📋 收集到的信息：\n"                for key, value in state['task_data'].items():                    response += f"• {key}: {value}\n"        else:            # 提取失败，重新询问            response = f"抱歉，我没能理解您提供的{field_name}信息。{step_info['prompt']}"                # 添加AI响应        ai_message = AIMessage(content=response)        state['messages'].append(ai_message)        return statedef extract_field_value(user_message: str, field_name: str) -> str:    """简化的信息提取函数"""    message_lower = user_message.lower()        if field_name == "departure_city":        cities = ["北京", "上海", "广州", "深圳", "杭州", "成都", "重庆", "西安", "南京", "武汉"]        for city in cities:            if city in user_message:                return city        elif field_name == "arrival_city":        cities = ["北京", "上海", "广州", "深圳", "杭州", "成都", "重庆", "西安", "南京", "武汉", "大连", "青岛"]        for city in cities:            if city in user_message:                return city        elif field_name == "departure_date":        import re        date_patterns = [r'\d{4}-\d{2}-\d{2}', r'\d{1,2}月\d{1,2}日', r'\d{1,2}/\d{1,2}']        for pattern in date_patterns:            match = re.search(pattern, user_message)            if match:                return match.group()        elif field_name == "passengers":        import re        number_match = re.search(r'(\d+)人?', user_message)        if number_match:            return number_match.group(1)        elif field_name == "cuisine_type":        cuisine_types = ["中餐", "西餐", "日料", "韩料", "意大利", "法餐", "泰餐", "印度"]        for cuisine in cuisine_types:            if cuisine in user_message:                return cuisine        elif field_name == "party_size":        import re        number_match = re.search(r'(\d+)人?', user_message)        if number_match:            return number_match.group(1)        elif field_name == "date_time":        if any(word in message_lower for word in ["明天", "今天", "后天"]):            return user_message        import re        if re.search(r'\d+点|\d+:\d+', user_message):            return user_message        return "NOT_FOUND"def save_task_memory(state: TaskState, config: RunnableConfig, store: BaseStore):    """保存任务完成记录到长期记忆"""    print("💾 保存任务记录到记忆...")        if not state['task_data'] or not state['task_type']:        return state        user_id = config.get("configurable", {}).get("user_id", "default_user")        # 创建任务记录    task_record = {        "task_type": state['task_type'],        "task_data": state['task_data'],        "completion_time": "2024-02-15T10:30:00Z",        "status": "completed"    }        # 保存到记忆存储    import uuid    record_key = str(uuid.uuid4())    namespace = (user_id, "completed_tasks")        store.put(namespace, record_key, task_record)    print(f"✅ 任务记录已保存: {record_key}")        # 添加确认消息    confirmation_message = AIMessage(content="✨ 任务已完成并保存到您的记录中！如需查看历史记录或开始新任务，请随时告诉我。")    state['messages'].append(confirmation_message)        return state# 创建简化的任务完成图def create_task_completion_graph():    """创建任务完成流程图"""    store = InMemoryStore()        # 创建状态图    workflow = StateGraph(dict)        # 添加节点    workflow.add_node("process_task", process_task)    workflow.add_node("save_memory", save_task_memory)        # 条件函数：检查是否需要保存记忆    def should_save_memory(state):        if state.get('task_complete', False) and state.get('task_data'):            return "save_memory"        else:            return "end"        # 添加边    workflow.add_edge(START, "process_task")    workflow.add_conditional_edges(        "process_task",        should_save_memory,        {            "save_memory": "save_memory",            "end": END        }    )    workflow.add_edge("save_memory", END)        # 编译图    app = workflow.compile(store=store)    return app, store# 演示任务完成系统print("=== 情境化任务完成系统演示 ===")# 创建应用和存储app, store = create_task_completion_graph()# 配置config = {"configurable": {"user_id": "user_456"}}print("\n--- 航班预订演示 ---")# 演示完整的多轮对话conversation_steps = [    "我想预订一张机票",    "我从北京出发",     "我要去上海",    "2024-03-15",    "1人"]current_state = TaskState()for i, user_input in enumerate(conversation_steps):    print(f"\n👤 用户: {user_input}")        # 添加用户消息    current_state['messages'].append(HumanMessage(content=user_input))        # 处理任务    result = app.invoke(current_state, config=config)        # 更新状态    current_state = result        # 显示AI响应    if current_state['messages'] and isinstance(current_state['messages'][-1], AIMessage):        print(f"🤖 助手: {current_state['messages'][-1].content}")        # 显示当前任务状态    print(f"📊 当前步骤: {current_state.get('current_step', 0)}")    print(f"🏁 任务完成: {current_state.get('task_complete', False)}")    if current_state.get('task_data'):        print(f"📋 已收集数据: {current_state['task_data']}")# 检查保存的记忆print("\n--- 检查保存的任务记忆 ---")saved_memories = store.search(("user_456", "completed_tasks"))print(f"📋 用户完成的任务数量: {len(saved_memories)}")for memory in saved_memories:    task_data = memory.value    print(f"✅ {task_data['task_type']}: {task_data['task_data']}")

输出：

=== 情境化任务完成系统演示 ===--- 航班预订演示 ---👤 用户: 我想预订一张机票🔄 处理任务 - 当前状态: step=0, type=None📋 检测到任务类型: flight_booking🤖 助手: 我来帮您预订航班。请告诉我您的出发城市：📊 当前步骤: 0🏁 任务完成: False👤 用户: 我从北京出发🔄 处理任务 - 当前状态: step=0, type=flight_booking🎯 处理用户输入: 我从北京出发 -> 提取字段: departure_city✅ 成功提取 departure_city: 北京🤖 助手: 已记录您的departure_city：北京。请告诉我您的目的地城市：📊 当前步骤: 1🏁 任务完成: False📋 已收集数据: {'departure_city': '北京'}👤 用户: 我要去上海🔄 处理任务 - 当前状态: step=1, type=flight_booking🎯 处理用户输入: 我要去上海 -> 提取字段: arrival_city✅ 成功提取 arrival_city: 上海🤖 助手: 已记录您的arrival_city：上海。请提供出发日期（格式：YYYY-MM-DD）：📊 当前步骤: 2🏁 任务完成: False📋 已收集数据: {'departure_city': '北京', 'arrival_city': '上海'}👤 用户: 2025-12-10🔄 处理任务 - 当前状态: step=2, type=flight_booking🎯 处理用户输入: 2025-12-10 -> 提取字段: departure_date✅ 成功提取 departure_date: 2025-12-10🤖 助手: 已记录您的departure_date：2025-12-10。请告诉我乘客人数：📊 当前步骤: 3🏁 任务完成: False📋 已收集数据: {'departure_city': '北京', 'arrival_city': '上海', 'departure_date': '2025-12-10'}👤 用户: 1人🔄 处理任务 - 当前状态: step=3, type=flight_booking🎯 处理用户输入: 1人 -> 提取字段: passengers✅ 成功提取 passengers: 1💾 保存任务记录到记忆...✅ 任务记录已保存: de22b6ec-2859-4fbd-ab61-c4d1c3898f40🤖 助手: ✨ 任务已完成并保存到您的记录中！如需查看历史记录或开始新任务，请随时告诉我。📊 当前步骤: 4🏁 任务完成: True📋 已收集数据: {'departure_city': '北京', 'arrival_city': '上海', 'departure_date': '2025-12-10', 'passengers': '1'}--- 检查保存的任务记忆 ---📋 用户完成的任务数量: 1✅ flight_booking: {'departure_city': '北京', 'arrival_city': '上海', 'departure_date': '2025-12-10', 'passengers': '1'}

在上述的示例（航班预订）中，我们仅将用户预定信息直接进行存储、检索，但对于Agent来说，记住和不同用户的过往交互是非常重要的，在后续的交互中，才能针对不同的用户，给出更合适的回应。让Agent能自主决定记忆的存储、遗忘、更新和检索优先级，才能真正模拟人类的记忆过程。因此，未来需要改进设计更智能的记忆管理策略。

​最后

我在一线科技企业深耕十二载，见证过太多因技术卡位而跃迁的案例。那些率先拥抱 AI 的同事，早已在效率与薪资上形成代际优势，我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在大模型的学习中的很多困惑。

我整理出这套 AI 大模型突围资料包：

• ✅AI大模型学习路线图
• ✅Agent行业报告
• ✅100集大模型视频教程
• ✅大模型书籍PDF
• ✅DeepSeek教程
• ✅AI产品经理入门资料

完整的大模型学习和面试资料已经上传带到CSDN的官方了，有需要的朋友可以扫描下方二维码免费领取【保证100%免费】👇👇
​​

为什么说现在普通人就业/升职加薪的首选是AI大模型？

人工智能技术的爆发式增长，正以不可逆转之势重塑就业市场版图。从DeepSeek等国产大模型引发的科技圈热议，到全国两会关于AI产业发展的政策聚焦，再到招聘会上排起的长队，AI的热度已从技术领域渗透到就业市场的每一个角落。

智联招聘的最新数据给出了最直观的印证：2025年2月，AI领域求职人数同比增幅突破200% ，远超其他行业平均水平；整个人工智能行业的求职增速达到33.4%，位居各行业榜首，其中人工智能工程师岗位的求职热度更是飙升69.6%。

AI产业的快速扩张，也让人才供需矛盾愈发突出。麦肯锡报告明确预测，到2030年中国AI专业人才需求将达600万人，人才缺口可能高达400万人，这一缺口不仅存在于核心技术领域，更蔓延至产业应用的各个环节。

​​

资料包有什么？

①从入门到精通的全套视频教程⑤⑥

包含提示词工程、RAG、Agent等技术点

② AI大模型学习路线图（还有视频解说）

全过程AI大模型学习路线

③学习电子书籍和技术文档

市面上的大模型书籍确实太多了，这些是我精选出来的

④各大厂大模型面试题目详解

⑤ 这些资料真的有用吗?

这份资料由我和鲁为民博士共同整理，鲁为民博士先后获得了北京清华大学学士和美国加州理工学院博士学位，在包括IEEE Transactions等学术期刊和诸多国际会议上发表了超过50篇学术论文、取得了多项美国和中国发明专利，同时还斩获了吴文俊人工智能科学技术奖。目前我正在和鲁博士共同进行人工智能的研究。

所有的视频教程由智泊AI老师录制，且资料与智泊AI共享，相互补充。这份学习大礼包应该算是现在最全面的大模型学习资料了。

资料内容涵盖了从入门到进阶的各类视频教程和实战项目，无论你是小白还是有些技术基础的，这份资料都绝对能帮助你提升薪资待遇，转行大模型岗位。

智泊AI始终秉持着“让每个人平等享受到优质教育资源”的育人理念‌，通过动态追踪大模型开发、数据标注伦理等前沿技术趋势‌，构建起"前沿课程+智能实训+精准就业"的高效培养体系。

课堂上不光教理论，还带着学员做了十多个真实项目。学员要亲自上手搞数据清洗、模型调优这些硬核操作，把课本知识变成真本事‌！

​​​​

如果说你是以下人群中的其中一类，都可以来智泊AI学习人工智能，找到高薪工作，一次小小的“投资”换来的是终身受益！

应届毕业生‌：无工作经验但想要系统学习AI大模型技术，期待通过实战项目掌握核心技术。

零基础转型‌：非技术背景但关注AI应用场景，计划通过低代码工具实现“AI+行业”跨界‌。

业务赋能 ‌突破瓶颈：传统开发者（Java/前端等）学习Transformer架构与LangChain框架，向AI全栈工程师转型‌。

👉获取方式：

😝有需要的小伙伴，可以保存图片到wx扫描二v码免费领取【保证100%免费】🆓**

​