LangGraph中的跨线程长期记忆存储
LangGraph 中跨线程的长期记忆功能：命名空间存储、插入/获取/搜索、基于记忆的语义搜索、去重写入以及由 Postgres 支持的持久化存储。
来源: Long Term Memory in LangGraph| Part 16| class 24 of LangGraph | cswithshahil
为何短期线程内存不够用
检查点化的聊天记录仅覆盖单个线程。真实用户会在不同线程间切换——周一调试LangChain，周二规划旅行——因此关于他们的信息会分散在各处。长期记忆（LTM）提供了一个持久存储机制：提取器会在对话轮次中记录个人详细信息，之后在回复查询时先从该存储中获取信息再作回答。用户一旦指定偏好（如优先使用Python），之后再询问斐波那契数列，助手便无需再次被告知即可直接使用该偏好。
存储作为LangGraph的记忆接口
LangGraph通过存储机制来实现长期记忆功能。BaseStore负责定义数据的创建、搜索、编辑和删除操作。常见的后端包括：用于RAM中原型开发的InMemoryStore；以及在重启时不能丢失数据的PostgresStore和RedisStore。
独立存储操作
将命名空间视为文件夹。它们是 Python 元组——如 ("users", "U1")，或是更深的层级结构 ("users", "U1", "profile")。
使用 put(namespace, key, values) 创建命名空间；当已知键值时，用 get 获取对应项；用 search 列出命名空间中的所有内容。
from langgraph.store.memory import InMemoryStore
# creating a store
store = InMemoryStore()
# creating a namespace
namespace = ("user", "u1")
创建记忆
# Adding memories
store.put(namespace, "1", {"data": "User likes pizza"})
store.put(namespace, "2", {"data": "User prefers dark mode"})
# another namespace
namespace2 = ("user", "u2")
# Add memories
store.put(namespace2, "1", {"data": "User likes pasta"})
store.put(namespace2, "2", {"data": "User prefers grid style navigation"})
检索某条记忆
# store.get(namespace, key)
store.get(namespace2, "2")
Item(namespace=['user', 'u2'], key='2', value={'data': 'User prefers grid style navigation'}, created_at='2026-01-06T15:05:16.376069+00:00', updated_at='2026-01-06T15:05:16.376070+00:00')
检索命名空间中的所有内容
items = store.search(namespace2)
for item in items:
print(item.value)
{'data': 'User likes pasta'}
{'data': 'User prefers grid style navigation'}
当数据量增加时的语义搜索
在提示词中输入上百条个人资料信息会让模型不堪重负。应使用嵌入模型（例如 text-embedding-3-small）对存储内容进行索引，这样 search 函数就能根据 query 和 limit 参数，仅返回与当前问题相近的向量。
Semantic Search
from langchain_openai import OpenAIEmbeddings
from dotenv import load_dotenv
load_dotenv()
embedding_model = OpenAIEmbeddings(model='text-embedding-3-small')
store = InMemoryStore(index={'embed': embedding_model, 'dims':1536})
namespace = ('users', 'u1')
store.put(namespace, "1", {"data": "User prefers concise answers over long explanations"})
store.put(namespace, "2", {"data": "User likes examples in Python"})
store.put(namespace, "3", {"data": "User usually works late at night"})
store.put(namespace, "4", {"data": "User prefers dark mode in applications"})
store.put(namespace, "5", {"data": "User is learning machine learning"})
store.put(namespace, "6", {"data": "User dislikes overly theoretical explanations"})
store.put(namespace, "7", {"data": "User prefers step-by-step reasoning"})
store.put(namespace, "8", {"data": "User is based in India"})
store.put(namespace, "9", {"data": "User likes real-world analogies"})
store.put(namespace, "10", {"data": "User prefers bullet points over paragraphs"})
items = store.search(namespace, query="what is the user currently learning", limit=1)
for item in items:
print(item.value)
# output :{'data': 'User is learning machine learning'}
items = store.search(namespace, query="what are user's preferences", limit=3)
for item in items:
print(item.value)
# output :
# {'data': 'User prefers dark mode in applications'}
# {'data': 'User prefers bullet points over paragraphs'}
# {'data': 'User prefers concise answers over long explanations'}
聊天节点中的只读个性化功能
将存储内容连接到 LangGraph 聊天节点，该节点会接收 state、config（用户或对话线程标识）以及 store。根据用户标识构建命名空间，检索相关记忆内容，将其整合到系统提示词中，并指示模型使用已知的名称、项目及语气风格，同时添加一些相关的补充内容。
Chatbot Reading Existing Memories
from dotenv import load_dotenv
load_dotenv()
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnableConfig
from langchain_core.messages import SystemMessage
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.store.memory import InMemoryStore
from langgraph.store.base import BaseStore
# ----------------------------
# 1) Create LTM store + seed memories (done BEFORE running the graph)
# ----------------------------
store = InMemoryStore()
user_id = "u1"
# Store user details as a single blob (simple for teaching)
# You can also split into multiple records; this keeps it easy.
user_details = ("user", user_id, "details")
store.put(user_details, "profile_1", {"data": "Name: Nitish"})
store.put(user_details, "profile_2", {"data": "Profession: Teaches AI on YouTube"})
store.put(user_details, "preference_1", {"data": "Prefers concise answers"})
store.put(user_details, "preference_2", {"data": "Likes examples in Python"})
store.put(user_details, "project_1", {"data": "Building MCP servers (Python-based project)"})
# ----------------------------
# 2) System prompt template (your prompt)
# ----------------------------
SYSTEM_PROMPT_TEMPLATE = """You are a helpful assistant with memory capabilities.
If user-specific memory is available, use it to personalize
your responses based on what you know about the user.
Your goal is to provide relevant, friendly, and tailored
assistance that reflects the user’s preferences, context, and past interactions.
If the user’s name or relevant personal context is available, always personalize your responses by:
– Always Address the user by name (e.g., "Sure, Nitish...") when appropriate
– Referencing known projects, tools, or preferences (e.g., "your MCP server python based project")
– Adjusting the tone to feel friendly, natural, and directly aimed at the user
Avoid generic phrasing when personalization is possible. For example, instead of "In TypeScript apps..."
say "Since your project is built with TypeScript..."
Use personalization especially in:
– Greetings and transitions
– Help or guidance tailored to tools and frameworks the user uses
– Follow-up messages that continue from past context
Always ensure that personalization is based only on known user details and not assumed.
In the end suggest 3 relevant further questions based on the current response and user profile
The user’s memory (which may be empty) is provided as: {user_details_content}
"""
# ----------------------------
# 3) Build graph: START -> chat -> END (read-only LTM)
# ----------------------------
llm = ChatOpenAI(model="gpt-4o-mini")
def chat_node(state: MessagesState, config: RunnableConfig, store: BaseStore):
user_id = config["configurable"]["user_id"]
# Read-only: fetch user details memory (no writes)
user_details = ("user", user_id, "details")
items = store.search(user_details)
# Convert memory items into a string blob for {user_details_content}
# Keep it dead simple for teaching.
if items:
user_details_content = "\n".join(f"- {it.value.get('data', '')}" for it in items)
else:
user_details_content = "" # prompt says it may be empty
system_prompt = SYSTEM_PROMPT_TEMPLATE.format(
user_details_content=user_details_content
)
system_msg = SystemMessage(content=system_prompt)
response = llm.invoke([system_msg] + state["messages"])
return {"messages": [response]}
builder = StateGraph(MessagesState)
builder.add_node("chat", chat_node)
builder.add_edge(START, "chat")
builder.add_edge("chat", END)
graph = builder.compile(store=store)
graph
# ----------------------------
# 4) Run it (provide user_id in config)
# ----------------------------
config = {"configurable": {"user_id": "u1"}}
result = graph.invoke(
{"messages": [{"role": "user", "content": "Explain gen ai in simple terms."}]},
config,
)
print(result["messages"][-1].content)
# Sure, Nitish! Generative AI, or Gen AI, refers to a type of artificial intelligence that creates new content or data instead of just analyzing existing information. It can generate text, images, music, and more by learning patterns from large sets of data. For example, it can write a story, design an artwork, or compose a song based on the styles of various creators it has learned from.
#If you're considering using examples or applications of Gen AI for your YouTube teaching, there are plenty of interesting case studies you could explore, especially in creative fields.
#Here are a few questions to consider:
#1. Would you like to see specific examples of Gen AI in Python?
#2. Are you interested in exploring tools or libraries for building Gen AI projects?
#3. How would you like to incorporate Gen AI concepts into your teaching materials?
在聊天过程中创建新记忆
Chatbot Creating New Memories
from dotenv import load_dotenv
load_dotenv()
import uuid
from typing import List
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage
from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.store.memory import InMemoryStore
from langgraph.store.base import BaseStore
# ----------------------------
# 1) LTM store
# ----------------------------
store = InMemoryStore()
# ----------------------------
# 2) LLM that decides what to remember (structured output)
# ----------------------------
extractor_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
class MemoryDecision(BaseModel):
should_write: bool = Field(description="Whether to store any memories")
memories: List[str] = Field(default_factory=list, description="Atomic user memories to store")
memory_extractor = extractor_llm.with_structured_output(MemoryDecision)
# ----------------------------
# 3) Graph: START -> remember -> END
# (Creates memories, but does NOT use them to answer)
# ----------------------------
def remember_only_node(state: MessagesState, config: RunnableConfig, store: BaseStore):
user_id = config["configurable"]["user_id"]
namespace = ("user", user_id, "details")
# take latest user message
last_msg = state["messages"][-1].content
# LLM decides what to store
decision: MemoryDecision = memory_extractor.invoke(
[
SystemMessage(
content=(
"Extract LONG-TERM memories from the user's message.\n"
"Only store stable, user-specific info (identity, preferences, ongoing projects).\n"
"Do NOT store transient info.\n"
"Return should_write=false if nothing is worth storing.\n"
"Each memory should be a short atomic sentence."
)
),
{"role": "user", "content": last_msg},
]
)
# Write to store (LTM)
if decision.should_write:
for mem in decision.memories:
store.put(namespace, str(uuid.uuid4()), {"data": mem})
# IMPORTANT: we are NOT using memory, not even responding with the LLM.
# We just return a fixed acknowledgement.
return {"messages": [{"role": "assistant", "content": "Noted."}]}
builder = StateGraph(MessagesState)
builder.add_node("remember", remember_only_node)
builder.add_edge(START, "remember")
builder.add_edge("remember", END)
graph = builder.compile(store=store)
graph
# ----------------------------
# 4) Demo
# ----------------------------
config = {"configurable": {"user_id": "u1"}}
res = graph.invoke({"messages": [{"role": "user", "content": "Hi my name is nitish"}]},config)
print("Assistant:", res["messages"][-1].content)
#output : Assistant: Noted.
res = graph.invoke({"messages": [{"role": "user", "content": "I teach AI on youtube"}]},config)
print("Assistant:", res["messages"][-1].content)
#output :Assistant: Noted.
res = graph.invoke({"messages": [{"role": "user", "content": "My favorite programming language is Python"}]},config)
print("Assistant:", res["messages"][-1].content)
# output :Assistant: Noted.
items = store.search(("user", "u1", "details"))
for item in items:
print(item.value['data'])
#output:
# User's name is Nitish.
# User teaches AI on YouTube.
# User's favorite programming language is Python.
#User's name is Nitish.
# User teaches AI on YouTube.
# User's favorite programming language is Python.
自动写入功能需要在聊天之前先有一个记忆节点。专门的提取模型会生成结构化的MemoryDecision输出，指示是否要写入以及写入哪些内容。如果没有去重机制，反复提及“我的名字是……”就会产生重复记录。可通过is_new标志区分新内容与现有记忆，仅保留新的信息。
Chatbot Creating New Memories(without Duplication)
from dotenv import load_dotenv
load_dotenv()
import uuid
from typing import List
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage
from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.store.memory import InMemoryStore
from langgraph.store.base import BaseStore
# ----------------------------
# 1) LTM store
# ----------------------------
store = InMemoryStore()
# ----------------------------
# 2) LLMs
# - memory_llm: extracts candidate memories + tells if each is NEW (no duplicate_of needed)
# ----------------------------
memory_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
class MemoryItem(BaseModel):
text: str = Field(description="Atomic user memory as a short sentence")
is_new: bool = Field(description="True if this memory is NEW and should be stored. False if duplicate/already known.")
class MemoryDecision(BaseModel):
should_write: bool = Field(description="Whether to store any memories")
memories: List[MemoryItem] = Field(default_factory=list, description="Atomic user memories to store")
memory_extractor = memory_llm.with_structured_output(MemoryDecision)
MEMORY_PROMPT = """You are responsible for updating and maintaining accurate user memory.
CURRENT USER DETAILS (existing memories):
{user_details_content}
TASK:
- Review the user's latest message.
- Extract user-specific info worth storing long-term (identity, stable preferences, ongoing projects/goals).
- For each extracted item, set is_new=true ONLY if it adds NEW information compared to CURRENT USER DETAILS.
- If it is basically the same meaning as something already present, set is_new=false.
- Keep each memory as a short atomic sentence.
- No speculation; only facts stated by the user.
- If there is nothing memory-worthy, return an empty list.
"""
def chat_creates_memory_node(state: MessagesState, config: RunnableConfig, store: BaseStore):
user_id = config["configurable"]["user_id"]
namespace = ("user", user_id, "details")
# A) Load existing memories
existing_items = store.search(namespace)
existing_texts = [it.value.get("data", "") for it in existing_items if it.value.get("data")]
user_details_content = "\n".join(f"- {t}" for t in existing_texts) if existing_texts else "(empty)"
# B) Latest user message
last_text = state["messages"][-1]
# C) LLM extracts memories + marks new vs duplicate
decision: MemoryDecision = memory_extractor.invoke(
[
SystemMessage(content=MEMORY_PROMPT.format(user_details_content=user_details_content)),
{"role": "user", "content": f"USER MESSAGE:\n{last_text}"},
]
)
# D) Store ONLY new memories
if decision.should_write:
for mem in decision.memories:
if mem.is_new:
store.put(namespace, str(uuid.uuid4()), {"data": mem.text})
return {"messages": [{"role": "assistant", "content": "Noted."}]}
# ----------------------------
# 4) Build graph: START -> chat -> END
# ----------------------------
builder = StateGraph(MessagesState)
builder.add_node("chat", chat_creates_memory_node)
builder.add_edge(START, "chat")
builder.add_edge("chat", END)
graph = builder.compile(store=store)
config = {"configurable": {"user_id": "u1"}}
r1 = graph.invoke({"messages": [{"role": "user", "content": "My name is Nitish"}]}, config)
print("Assistant:", r1["messages"][-1].content)
# output: Assistant: Noted.
# Likely duplicate of the first memory -> is_new should be false -> saved=0
r2 = graph.invoke({"messages": [{"role": "user", "content": "I like Python for programming."}]}, config)
print("\nAssistant:", r2["messages"][-1].content)
# output:Assistant: Noted.
for it in store.search(("user", "u1", "details")):
print(it.value['data'])
# output:The user's name is Nitish.
# output:Nitish likes Python for programming.
合并后的图结构与持久化存储
最终流程为：开始 → 记忆（提取并保存）→ 聊天（读取并回复）→ 结束。智能体会在对话过程中不断学习，并立即运用新获取的信息。
Merged Workflow
from dotenv import load_dotenv
load_dotenv()
import uuid
from typing import List
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage
from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.store.memory import InMemoryStore
from langgraph.store.base import BaseStore
# ----------------------------
# 1) LTM store (START EMPTY)
# ----------------------------
store = InMemoryStore()
# ----------------------------
# 2) System prompt
# ----------------------------
SYSTEM_PROMPT_TEMPLATE = """You are a helpful assistant with memory capabilities.
If user-specific memory is available, use it to personalize
your responses based on what you know about the user.
Your goal is to provide relevant, friendly, and tailored
assistance that reflects the user’s preferences, context, and past interactions.
If the user’s name or relevant personal context is available, always personalize your responses by:
– Always Address the user by name (e.g., "Sure, Nitish...") when appropriate
– Referencing known projects, tools, or preferences (e.g., "your MCP server python based project")
– Adjusting the tone to feel friendly, natural, and directly aimed at the user
Avoid generic phrasing when personalization is possible.
Use personalization especially in:
– Greetings and transitions
– Help or guidance tailored to tools and frameworks the user uses
– Follow-up messages that continue from past context
Always ensure that personalization is based only on known user details and not assumed.
In the end suggest 3 relevant further questions based on the current response and user profile
The user’s memory (which may be empty) is provided as: {user_details_content}
"""
# ----------------------------
# 3) Memory extraction LLM
# ----------------------------
memory_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
class MemoryItem(BaseModel):
text: str = Field(description="Atomic user memory")
is_new: bool = Field(description="True if new, false if duplicate")
class MemoryDecision(BaseModel):
should_write: bool
memories: List[MemoryItem] = Field(default_factory=list)
memory_extractor = memory_llm.with_structured_output(MemoryDecision)
MEMORY_PROMPT = """You are responsible for updating and maintaining accurate user memory.
CURRENT USER DETAILS (existing memories):
{user_details_content}
TASK:
- Review the user's latest message.
- Extract user-specific info worth storing long-term (identity, stable preferences, ongoing projects/goals).
- For each extracted item, set is_new=true ONLY if it adds NEW information compared to CURRENT USER DETAILS.
- If it is basically the same meaning as something already present, set is_new=false.
- Keep each memory as a short atomic sentence.
- No speculation; only facts stated by the user.
- If there is nothing memory-worthy, return should_write=false and an empty list.
"""
# ----------------------------
# 4) Node 1: remember
# ----------------------------
def remember_node(state: MessagesState, config: RunnableConfig, *, store: BaseStore):
user_id = config["configurable"]["user_id"]
ns = ("user", user_id, "details")
# existing memory
items = store.search(ns)
existing = "\n".join(it.value["data"] for it in items) if items else "(empty)"
# last user message
last_msg = state["messages"][-1].content
decision: MemoryDecision = memory_extractor.invoke(
[
SystemMessage(content=MEMORY_PROMPT.format(user_details_content=existing)),
{"role": "user", "content": last_msg},
]
)
if decision.should_write:
for mem in decision.memories:
if mem.is_new:
store.put(ns, str(uuid.uuid4()), {"data": mem.text})
return {} # no message change
# ----------------------------
# 5) Node 2: chat
# ----------------------------
chat_llm = ChatOpenAI(model="gpt-4o-mini")
def chat_node(state: MessagesState, config: RunnableConfig, *, store: BaseStore):
user_id = config["configurable"]["user_id"]
ns = ("user", user_id, "details")
items = store.search(ns)
user_details = "\n".join(it.value["data"] for it in items) if items else ""
system_msg = SystemMessage(
content=SYSTEM_PROMPT_TEMPLATE.format(
user_details_content=user_details or "(empty)"
)
)
response = chat_llm.invoke([system_msg] + state["messages"])
return {"messages": [response]}
# ----------------------------
# 6) Graph
# ----------------------------
builder = StateGraph(MessagesState)
builder.add_node("remember", remember_node)
builder.add_node("chat", chat_node)
builder.add_edge(START, "remember")
builder.add_edge("remember", "chat")
builder.add_edge("chat", END)
graph = builder.compile(store=store)
graph
# ----------------------------
# 7) Demo
# ----------------------------
config = {"configurable": {"user_id": "u1"}}
result = graph.invoke({"messages": [{"role": "user", "content": "Hi, my name is Nitish"}]}, config)
result['messages'][-1].content
# output:'Hi Nitish! It’s great to meet you. How can I assist you today?'
for it in store.search(("user", "u1", "details")):
print(it.value["data"])
result = graph.invoke({"messages": [{"role": "user", "content": "I teach AI on YouTube"}]}, config)
print(result['messages'][-1].content)
# output:That's great, Nitish! Teaching AI on YouTube must be really fulfilling. What topics do you cover in your videos? Are there any specific projects or concepts you find your audience is particularly interested in?
for it in store.search(("user", "u1", "details")):
print(it.value["data"])
# output:User's name is Nitish.
# output:Nitish teaches AI on YouTube.
result = graph.invoke({"messages": [{"role": "user", "content": "Explain GenAI simply"}]}, config)
print(result['messages'][-1].content)
# output:Sure, Nitish! Generative AI, often referred to as GenAI, is a type of artificial intelligence that can create new content based on the data it's trained on. This includes generating text, images, music, and even video.
# output:Here's a simple breakdown:
# output:1. **Learning from Examples:** GenAI is trained on vast amounts of data, learning the patterns, styles, and structures of the content it sees.
# output:2. **Creating Content:** Once trained, it can generate new content that resembles what it learned. For example, it can write a story in a specific style, create artwork, or compose music.
# output:3. **Applications:** It's used in various fields, from helping writers brainstorm ideas to producing artwork and even customer service chatbots that can engage with users in a conversational manner.
# output:If you're planning to teach this topic on your YouTube channel, emphasizing practical examples and real-world applications could be very engaging for your audience!
Do you want to explore specific applications of GenAI, tools for using it, or perhaps the ethics involved?
for it in store.search(("user", "u1", "details")):
print(it.value["data"])
# output:User's name is Nitish.
# output:Nitish teaches AI on YouTube.
InMemoryStore会在进程退出时失效。建议运行Postgres（通常通过Docker实现），安装Postgres的检查点/存储相关包，然后将存储方式改为基于连接的实现。完全重启后，查询数据库仍应能获取到已保存的配置信息。
services:
postgres:
image: postgres:16
environment:
POSTGRES_USER: postgres
POSTGRES_PASSWORD: postgres
POSTGRES_DB: postgres
ports:
- "5442:5432"
! pip install -U "psycopg[binary,pool]" langgraph langgraph-checkpoint-postgres
from dotenv import load_dotenv
load_dotenv()
import uuid
from typing import List
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage
from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.store.postgres import PostgresStore
from langgraph.store.base import BaseStore
# ----------------------------
# 2) System prompt
# ----------------------------
SYSTEM_PROMPT_TEMPLATE = """You are a helpful assistant with memory capabilities.
If user-specific memory is available, use it to personalize
your responses based on what you know about the user.
Your goal is to provide relevant, friendly, and tailored
assistance that reflects the user’s preferences, context, and past interactions.
If the user’s name or relevant personal context is available, always personalize your responses by:
– Always Address the user by name (e.g., "Sure, Nitish...") when appropriate
– Referencing known projects, tools, or preferences (e.g., "your MCP server python based project")
– Adjusting the tone to feel friendly, natural, and directly aimed at the user
Avoid generic phrasing when personalization is possible.
Use personalization especially in:
– Greetings and transitions
– Help or guidance tailored to tools and frameworks the user uses
– Follow-up messages that continue from past context
Always ensure that personalization is based only on known user details and not assumed.
In the end suggest 3 relevant further questions based on the current response and user profile
The user’s memory (which may be empty) is provided as: {user_details_content}
"""
# ----------------------------
# 3) Memory extraction LLM
# ----------------------------
memory_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
class MemoryItem(BaseModel):
text: str = Field(description="Atomic user memory")
is_new: bool = Field(description="True if new, false if duplicate")
class MemoryDecision(BaseModel):
should_write: bool
memories: List[MemoryItem] = Field(default_factory=list)
memory_extractor = memory_llm.with_structured_output(MemoryDecision)
MEMORY_PROMPT = """You are responsible for updating and maintaining accurate user memory.
CURRENT USER DETAILS (existing memories):
{user_details_content}
TASK:
- Review the user's latest message.
- Extract user-specific info worth storing long-term (identity, stable preferences, ongoing projects/goals).
- For each extracted item, set is_new=true ONLY if it adds NEW information compared to CURRENT USER DETAILS.
- If it is basically the same meaning as something already present, set is_new=false.
- Keep each memory as a short atomic sentence.
- No speculation; only facts stated by the user.
- If there is nothing memory-worthy, return should_write=false and an empty list.
"""
# ----------------------------
# 3) Nodes
# ----------------------------
def remember_node(state: MessagesState, config: RunnableConfig, *, store: BaseStore):
user_id = config["configurable"]["user_id"]
ns = ("user", user_id, "details")
# existing memory (all items under namespace)
items = store.search(ns)
existing = "\n".join(it.value.get("data", "") for it in items) if items else "(empty)"
# latest user message
last_text = state["messages"][-1].content
decision: MemoryDecision = memory_extractor.invoke(
[
SystemMessage(content=MEMORY_PROMPT.format(user_details_content=existing)),
{"role": "user", "content": last_text},
]
)
if decision.should_write:
for mem in decision.memories:
if mem.is_new and mem.text.strip():
store.put(ns, str(uuid.uuid4()), {"data": mem.text.strip()})
return {}
chat_llm = ChatOpenAI(model="gpt-4o-mini")
def chat_node(state: MessagesState, config: RunnableConfig, *, store: BaseStore):
user_id = config["configurable"]["user_id"]
ns = ("user", user_id, "details")
items = store.search(ns)
user_details = "\n".join(it.value.get("data", "") for it in items) if items else ""
system_msg = SystemMessage(
content=SYSTEM_PROMPT_TEMPLATE.format(user_details_content=user_details or "(empty)")
)
response = chat_llm.invoke([system_msg] + state["messages"])
return {"messages": [response]}
# ----------------------------
# 4) Build graph
# ----------------------------
builder = StateGraph(MessagesState)
builder.add_node("remember", remember_node)
builder.add_node("chat", chat_node)
builder.add_edge(START, "remember")
builder.add_edge("remember", "chat")
builder.add_edge("chat", END)
<langgraph.graph.state.StateGraph at 0x110ce4a90>
# ----------------------------
# 5) Use PostgresStore (PERSISTENT LTM)
# ----------------------------
DB_URI = "postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable"
with PostgresStore.from_conn_string(DB_URI) as store:
# IMPORTANT: run ONCE the first time you use this database
store.setup()
graph = builder.compile(store=store)
config = {"configurable": {"user_id": "u1"}}
graph.invoke({"messages": [{"role": "user", "content": "Hi, my name is Nitish"}]}, config)
graph.invoke({"messages": [{"role": "user", "content": "I teach AI on YouTube"}]}, config)
out = graph.invoke({"messages": [{"role": "user", "content": "Explain GenAI simply"}]}, config)
print(out["messages"][-1].content)
print("\n--- Stored Memories (from Postgres) ---")
for it in store.search(("user", "u1", "details")):
print(it.value["data"])
# output:Sure, Nitish! Generative AI, or GenAI, refers to a type of artificial intelligence that can create new content, like images, text, music, or even code, on its own. It learns from a large dataset and uses patterns it recognizes to generate something that resembles what it has seen before.
# output:For example, if you input a topic or a few words into a generative AI model, it can produce a coherent piece of writing or generate a unique piece of art. Popular examples include models like GPT-3 for text and DALL-E for images.
# output:The key point is that GenAI is designed to take what it has learned and create new, original outputs based on that learning. If you’re discussing this concept in your YouTube teachings, it might be interesting for your audience to see how GenAI can assist in creative processes or even in automating certain tasks.
# output:Would you like to dive deeper into a specific application of GenAI, or perhaps explore its implications for teaching AI?
# output:--- Stored Memories (from Postgres) ---
# output:Nitish teaches AI on YouTube.
# output:User's name is Nitish.
Check Persistence
from langgraph.store.postgres import PostgresStore
DB_URI = "postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable"
with PostgresStore.from_conn_string(DB_URI) as store:
ns = ("user", "u1", "details")
items = store.search(ns)
for it in items:
print(it.value["data"])
# output: Nitish teaches AI on YouTube.
# output: User's name is Nitish.
LTM接线时的实际隐患
在两个功能共享(“users”, user_id)并互相覆盖键值时，命名空间看似简单却容易出问题。建议使用以功能为范围的元组，如(“users”, user_id, “prefs”)而非(“users”, user_id, “facts”)，并明确记录哪些用户可以修改各字段。键值应作为稳定的标识符，而非自由文本句子；句子内容应放在值的数据部分中。
语义搜索的质量取决于嵌入方式以及记忆内容的表述方式。只有当嵌入空间能将“喜欢Python”与“首选编程语言”这类短语归类到同一组时，后续查询才能有效。模型升级后需定期重新进行嵌入处理，虽然这项工作较为繁琐，但能避免检索性能的隐性下降。
写入路径需要设置速率限制。那种每当用户发出语音就会触发数据提取的机制，即便经过了is_new检查，仍可能向Postgres注入大量近乎重复的数据，尤其是在模型表现不确定时更为严重。应限制每轮的写入量，批量处理会话结束时的汇总数据，并监控每位用户的存储增长情况。在正式环境中，应将长期存储与短期检查点结合使用，这样即使线程本地对话数据与跨线程的上下文内存共享同一基础设施，也能在概念上保持分离。
读取路径的规范性
在读取方面，一旦演示版本实现了个性化功能，就绝不能将整个命名空间一次性加载到系统提示中。应限制可检索的记忆量，标明是哪些命名空间键影响了结果，并允许操作员在回忆内容出现错误时针对特定用户禁用LTM功能。无法关闭的个性化功能只会带来极大的维护难题。此外，应在存储中将“个人资料信息”与“对话摘要”分开，这样糟糕的摘要替换就不会导致用户长期设定的偏好被清除。从InMemoryStore迁移到Postgres时，应像练习检查点恢复一样提前做好内存表的备份与恢复准备——LTM功能的丢失会对用户造成可见的记忆丧失，而不仅仅是系统层面的异常。
教会提取器何为“重要”信息
那些要求“保存任何有用信息”的提示会生成垃圾内容。建议使用结构化的清单：身份信息、持久性偏好、进行中的项目、严格限制条件，以及明确指定的提醒事项。其他所有内容都应保留在对话上下文中的检查点中。向提取工具展示几组包含噪声的对话样本，以及你期望的最简记忆结构JSON格式，然后在提示发生变化时在持续集成环境中重新运行这些测试用例。长期记忆的质量是提示与架构共同作用的结果，应将其视为一个整体来处理。
操作人员还应当练习“错误记忆”的修复方法：通过关键标识删除相关内容，从修正后的表述中重新提取信息，并确认聊天节点不再引用过时的数据。如果没有这样的训练，长期记忆系统就会变成一个只能写入数据的垃圾存储空间。