Memory Schema: Collection

1Review: Limitations of Profile Solution and Motivation of Collection

In Lesson 2, we summarize all the user’s information into a fixedProfile Schema(such asUserProfile(name=..., interests=..., location=...)), using Trustcall to overwrite updates after each round of dialogue. This solution has a clear structure and clear fields, but it also has obvious limitations:

• Field fixed: Profile fields need to be defined in advance and cannot adapt to "new dimensions that suddenly appear" (for example, the user mentioned a certain book or a certain travel destination)
• information compression: Multiple points of interest will be compressed into the sameinterestsfields, details are easily lost
• single time dimension: Profile is always "current state" and cannot retain timing details such as "X was mentioned last week and Y was mentioned this week"

This method is closer to the "free association memory" of the human brain: you won't force "Friends like biking" and "Friends live in San Francisco" into a fixed grid, but write them down separately on different sticky notes and turn them out when needed.

2Collection Schema design idea: each memory is stored independently

core design principles

There are two key design decisions for the Collection approach:

1. 1

   Schema is extremely simplified: each memory has only one content field

   No memory type is predetermined; use a piece of free text to describe any information. LLM is responsible for summarizing the key information in the conversation into one sentence and storing itcontent。

2. 2

   Storage layer: Each memory uses an independent UUID as the key

   Different from the Profile scheme (all information is stored under the same key), each memory in the Collection scheme occupies an exclusive storage slot, which can be accumulated infinitely and a single record can be accurately updated.

3Defining Memory and MemoryCollection Schema with Pydantic

Schema that defines a single memory

The Schema of the Collection solution is extremely simple: one Pydantic BaseModel, only onecontentFields to store any type of user information in natural language:

from pydantic import BaseModel, Field

# 单条记忆的 Schema：极简设计，只有 content 一个字段
class Memory(BaseModel):
    content: str = Field(
        description="The main content of the memory. "
                    "For example: User expressed interest in learning about French."
    )

# 用于 with_structured_output 的容器 Schema（一次提取多条）
class MemoryCollection(BaseModel):
    memories: list[Memory] = Field(
        description="A list of memories about the user."
    )

Test extraction with with_structured_output

from langchain_core.messages import HumanMessage
from langchain_deepseek import ChatDeepSeek

model = ChatDeepSeek(model="deepseek-v4-pro", temperature=0)

# 将 MemoryCollection 绑定为结构化输出格式
model_with_structure = model.with_structured_output(MemoryCollection)

# 从一句话里提取出两条独立记忆
memory_collection = model_with_structure.invoke(
    [HumanMessage("My name is Lance. I like to bike.")]
)
memory_collection.memories

Each original message is split into independentMemoryobject.If there are several independent facts in a sentence, extract a few memories, this is the flexibility of the Collection scheme.

Serialization: model_dump() to dictionary

Before saving to the Store, the Pydantic object needs to be serialized into a Python dictionary:

# 把第一条记忆转换为字典格式，用于 store.put()
memory_collection.memories[0].model_dump()

4store.put() stores each memory into a different key (UUID)

LangGraph'sInMemoryStore(or persistent storage) providesnamespace + key + valueThree-layer positioning mechanism. The core of the Collection solution lies in:Each memory uses a randomly generated UUID as key, ensuring that each memory does not interfere with each other and can be updated independently.

import uuid
from langgraph.store.memory import InMemoryStore

# 初始化存储
in_memory_store = InMemoryStore()

# 命名空间：(user_id, "memories") — 按用户隔离
user_id = "1"
namespace_for_memory = (user_id, "memories")

# 存第一条记忆：生成唯一 key，存入字典值
key1 = str(uuid.uuid4())            # "e1c4e5ab-ab0f-..."
value1 = memory_collection.memories[0].model_dump()
in_memory_store.put(namespace_for_memory, key1, value1)

# 存第二条记忆：另一个 UUID，独立槽位
key2 = str(uuid.uuid4())            # "e132a1ea-6202-..."
value2 = memory_collection.memories[1].model_dump()
in_memory_store.put(namespace_for_memory, key2, value2)

Visualize storage results

Each memory has its own independentcreated_at、updated_atTimestamp supports subsequent sorting and filtering by time.

5Trustcall extractor: enable_inserts and json_doc_id mechanisms

Use directlywith_structured_outputThere is a problem with retrieving memories:The full amount is extracted every time, so it is impossible to know which ones are new and which ones need to be updated.. Trustcall specifically solves this problem.

Create Collection-specific extractors

from trustcall import create_extractor

# 创建以 Memory 为工具单元的提取器
# enable_inserts=True 允许提取器向集合中插入全新的记忆条目
trustcall_extractor = create_extractor(
    model,
    tools=[Memory],          # 每次工具调用产出一条 Memory
    tool_choice="Memory",    # 强制模型必须调用 Memory 工具
    enable_inserts=True,     # 关键：允许插入新条目（默认只能更新已有）
)

First extraction: Extracting memories from conversations

from langchain_core.messages import HumanMessage, SystemMessage, AIMessage

instruction = """Extract memories from the following conversation:"""

conversation = [
    HumanMessage(content="Hi, I'm Lance."),
    AIMessage(content="Nice to meet you, Lance."),
    HumanMessage(content="This morning I had a nice bike ride in San Francisco."),
]

# 调用提取器，传入带 SystemMessage 的消息列表
result = trustcall_extractor.invoke({
    "messages": [SystemMessage(content=instruction)] + conversation
})

The extractor returns three key fields:

• result["messages"]: Contains tool call messages (tool_calls) issued by AI
• result["responses"]: Extracted Memory object list
• result["response_metadata"]: Metadata of each memory, includingid(tool call ID) and possiblyjson_doc_id

Incremental update: passing in existing memory as context

When a new round of dialogue ends and the memory collection needs to be updated, the existing memory is passed to Trustcall in a specific format, and it will intelligently determine whether to update existing entries or insert new entries:

# 把现有记忆整理为 (id, tool_name, value) 三元组列表
tool_name = "Memory"
existing_memories = [
    (str(i), tool_name, memory.model_dump())
    for i, memory in enumerate(result["responses"])
]
# 结果：[('0', 'Memory', {'content': 'Lance had a nice bike ride...'})]

# 新一轮对话
updated_conversation = [
    AIMessage(content="That's great, did you do after?"),
    HumanMessage(content="I went to Tartine and ate a croissant."),
    AIMessage(content="What else is on your mind?"),
    HumanMessage(content="I was thinking about Japan, and going back this winter!"),
]

# 传入新对话 + 已有记忆，让 Trustcall 决定如何更新
result = trustcall_extractor.invoke({
    "messages": updated_conversation,
    "existing": existing_memories,   # 告知提取器已有的记忆内容
})

Understanding json_doc_id: update vs insert

# store.put 时自动决定用旧 key（更新）还是新 UUID（插入）
for r, rmeta in zip(result["responses"], result["response_metadata"]):
    store.put(
        namespace,
        rmeta.get("json_doc_id", str(uuid.uuid4())),  # 有则更新，无则插入
        r.model_dump(mode="json"),
    )

6Complete Chatbot: Workflow for write_memory node

Graph structure: two nodes are executed sequentially

Every time the user sends a message, the graph will: let firstcall_modelThe node reads the memory and generates a reply, and then letswrite_memoryThe node analyzes the conversation history and updates the memory collection.

call_model node: read memory, inject system prompts

# 系统提示模板：把记忆注入到 LLM 的上下文中
MODEL_SYSTEM_MESSAGE = """You are a helpful chatbot. You are designed to be a companion to a user.

You have a long term memory which keeps track of information you learn about the user over time.

Current Memory (may include updated memories from this conversation):

{memory}"""

def call_model(state: MessagesState, config: RunnableConfig, store: BaseStore):
    # 从 config 获取用户 ID
    user_id = config["configurable"]["user_id"]

    # 从 Store 搜索该用户的所有记忆
    namespace = ("memories", user_id)
    memories = store.search(namespace)

    # 格式化记忆列表，注入系统提示
    info = "\n".join(f"- {mem.value['content']}" for mem in memories)
    system_msg = MODEL_SYSTEM_MESSAGE.format(memory=info)

    # 用记忆 + 对话历史生成回复
    response = model.invoke([SystemMessage(content=system_msg)] + state["messages"])
    return {"messages": response}

write_memory node: Trustcall extracts and updates the collection

# Trustcall 指令：反思对话，决定需要记住什么
TRUSTCALL_INSTRUCTION = """Reflect on following interaction.

Use the provided tools to retain any necessary memories about the user.

Use parallel tool calling to handle updates and insertions simultaneously:"""

def write_memory(state: MessagesState, config: RunnableConfig, store: BaseStore):
    # 获取用户 ID 和命名空间
    user_id = config["configurable"]["user_id"]
    namespace = ("memories", user_id)

    # 从 Store 读取已有记忆（用于 Trustcall 的 existing 参数）
    existing_items = store.search(namespace)
    tool_name = "Memory"
    existing_memories = (
        [(existing_item.key, tool_name, existing_item.value)
         for existing_item in existing_items]
        if existing_items else None
    )

    # 合并对话历史 + Trustcall 指令
    updated_messages = list(merge_message_runs(
        messages=[SystemMessage(content=TRUSTCALL_INSTRUCTION)] + state["messages"]
    ))

    # 调用 Trustcall 提取器
    result = trustcall_extractor.invoke({
        "messages": updated_messages,
        "existing": existing_memories,
    })

    # 保存提取结果：有 json_doc_id 则更新原记录，否则用 UUID 插入新记录
    for r, rmeta in zip(result["responses"], result["response_metadata"]):
        store.put(
            namespace,
            rmeta.get("json_doc_id", str(uuid.uuid4())),
            r.model_dump(mode="json"),
        )

Compile the graph and run

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.store.memory import InMemoryStore
from langgraph.checkpoint.memory import MemorySaver

builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model)
builder.add_node("write_memory", write_memory)
builder.add_edge(START, "call_model")
builder.add_edge("call_model", "write_memory")
builder.add_edge("write_memory", END)

# 跨线程长期记忆：InMemoryStore（生产环境可换为持久化 DB）
across_thread_memory = InMemoryStore()
# 线程内短期记忆：MemorySaver（保存对话历史）
within_thread_memory = MemorySaver()

graph = builder.compile(
    checkpointer=within_thread_memory,
    store=across_thread_memory
)

# 运行 — thread_id 区分会话，user_id 区分用户
config = {"configurable": {"thread_id": "1", "user_id": "1"}}
for chunk in graph.stream(
    {"messages": [HumanMessage("Hi, my name is Lance")]},
    config, stream_mode="values"
):
    chunk["messages"][-1].pretty_print()

A collection of memories after two rounds of dialogue

7store.search() semantic retrieval related memory

store.search(namespace)is a method for retrieving memories from the Store. at presentInMemoryStorescenario, it returns all memory entries under this namespace:

# 搜索用户 1 的所有记忆
user_id = "1"
namespace = ("memories", user_id)
memories = across_thread_memory.search(namespace)

for m in memories:
    print(m.dict())

Each search result contains the following fields:

Format memory injection in call_model

The retrieved memory is formatted as an unordered list, injecting LLM's system prompt:

# 从每条记忆的 value 字典中取出 content，拼成多行文本
info = "\n".join(f"- {mem.value['content']}" for mem in memories)
# 注入后的效果（注入进系统提示的 {memory} 占位符）：
# - User's name is Lance.
# - User likes to bike around San Francisco.
# - User enjoys going to bakeries.
system_msg = MODEL_SYSTEM_MESSAGE.format(memory=info)

8Collection vs Profile: How to choose a memory solution

Now we have learned two memory schema solutions. How to make a choice in actual projects? The following is a systematic comparison:

Selection decision reference

9Complete process summary and mental model

The complete data flow of the Collection memory system

Quick review of core concepts

A complete cross-session interaction example

# 第一轮：告知姓名
config = {"configurable": {"thread_id": "1", "user_id": "1"}}
graph.stream({"messages": [HumanMessage("Hi, my name is Lance")]}, config)
# Store 写入：{"content": "User's name is Lance."}

# 第二轮：告知爱好
graph.stream({"messages": [HumanMessage("I like to bike around San Francisco")]}, config)
# Store 追加：{"content": "User likes to bike around San Francisco."}

# 第三轮：追加爱好
graph.stream({"messages": [HumanMessage("I also enjoy going to bakeries")]}, config)
# Store 追加：{"content": "User enjoys going to bakeries."}

# 切换新会话（thread_id="2"），但 user_id 不变
config = {"configurable": {"thread_id": "2", "user_id": "1"}}
graph.stream({"messages": [HumanMessage("What bakeries do you recommend for me?")]}, config)
# call_model 从 Store 读取三条记忆，知道：
#   - 用户叫 Lance
#   - 喜欢在旧金山骑车
#   - 喜欢逛面包店
# 推荐：Tartine、Arsicault、B. Patisserie 等适合骑行顺路拜访的面包店