用集合（Collection）代替单一 Profile，让 AI 记忆随对话自然积累——
Trustcall 如何智能地插入、更新和检索每一条独立记忆。
在 Lesson 2 中，我们把用户的所有信息归纳进一个固定的 Profile Schema（如 UserProfile(name=..., interests=..., location=...)），用 Trustcall 在每轮对话后覆盖更新。这种方案结构清晰、字段一目了然，但也有明显的局限：
interests 字段，细节容易丢失Collection（集合）方案把每一条记忆作为独立记录存储，不预先规定字段，让记忆库随着对话自然生长。每次提取出新信息，就往集合里追加一条新记录；旧信息有变化，就更新对应那条记录。
这种方式更接近人类大脑的"自由联想记忆"：你不会把"朋友喜欢骑车"和"朋友住在旧金山"强行压进一个固定格子，而是分开记在不同的便签纸上，需要时再翻出来。
Collection 方案有两个关键设计决策：
不预先规定记忆类型，用一段自由文本描述任何信息。LLM 负责把对话中的关键信息归纳成一句话存入 content。
不同于 Profile 方案（所有信息存在同一个 key 下），Collection 方案中每条记忆独占一个存储槽，可以无限累积，也可以精准更新单条记录。
Collection 方案的 Schema 极其简单：一个 Pydantic BaseModel，只有一个 content 字段，用自然语言存储任何类型的用户信息：
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."
)Memory 是单条记忆的结构，也是后面 Trustcall 提取器的工具（tool）单元。MemoryCollection 是用于 with_structured_output 时的外层容器，让模型一次提取出多条记忆。在 Trustcall 流程里，只需要 Memory 这一层即可。
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每条原始信息被拆分为独立的 Memory 对象。一句话里有几个独立事实，就提取几条记忆，这正是 Collection 方案的灵活之处。
存入 Store 之前，需要把 Pydantic 对象序列化为 Python 字典：
# 把第一条记忆转换为字典格式，用于 store.put()
memory_collection.memories[0].model_dump()LangGraph 的 InMemoryStore（或持久化存储）提供了 namespace + key + value 三层定位机制。Collection 方案的核心在于：每条记忆用一个随机生成的 UUID 作为 key，确保各条记忆互不干扰、可独立更新。
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)store.put(namespace, key, value)
("1", "memories")，按用户和业务类型分区str(uuid.uuid4()) 生成model_dump() 结果每条记忆都有独立的 created_at、updated_at 时间戳，支持后续按时间排序和筛选。
直接用 with_structured_output 提取记忆有一个问题：每次都是全量提取，无法知道哪些是新增的、哪些需要更新已有记录。Trustcall 专门解决了这个问题。
from trustcall import create_extractor
# 创建以 Memory 为工具单元的提取器
# enable_inserts=True 允许提取器向集合中插入全新的记忆条目
trustcall_extractor = create_extractor(
model,
tools=[Memory], # 每次工具调用产出一条 Memory
tool_choice="Memory", # 强制模型必须调用 Memory 工具
enable_inserts=True, # 关键：允许插入新条目（默认只能更新已有）
)默认情况下 Trustcall 只会修改已有条目（update/overwrite）。设置 enable_inserts=True 后，当提取器遇到现有集合中没有的新信息，它会创建全新的记忆条目（insert）。这正是 Collection 方案的核心能力。
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
})提取器返回三个关键字段：
result["messages"]：包含 AI 发出的工具调用消息（tool_calls）result["responses"]：提取出的 Memory 对象列表result["response_metadata"]：每条记忆的元数据，包含 id（工具调用 ID）和可能的 json_doc_id当新一轮对话结束，需要更新记忆集合时，把现有记忆以特定格式传给 Trustcall，它会智能判断是更新已有条目还是插入新条目：
# 把现有记忆整理为 (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, # 告知提取器已有的记忆内容
})
有 json_doc_id：这条 response 对应已有记忆的更新，存储时应使用该记忆在 Store 中的原始 key。
无 json_doc_id：这是一条全新的记忆，存储时应生成新的 UUID 作为 key。
# 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"),
)每次用户发送消息，图都会：先让 call_model 节点读取记忆并生成回复，再让 write_memory 节点分析对话历史、更新记忆集合。
# 系统提示模板：把记忆注入到 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}# 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"),
)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()切换到 thread_id: "2" 但保持相同的 user_id: "1"，call_model 仍然能从 Store 检索到这两条记忆，并在新对话中做到个性化回复（比如推荐骑行路线旁边的面包店）。这正是跨线程长期记忆的价值所在。
store.search(namespace) 是从 Store 中检索记忆的方法。在当前 InMemoryStore 场景下，它返回该命名空间下的所有记忆条目：
# 搜索用户 1 的所有记忆
user_id = "1"
namespace = ("memories", user_id)
memories = across_thread_memory.search(namespace)
for m in memories:
print(m.dict())每条搜索结果包含以下字段：
| 字段 | 类型 | 说明 |
|---|---|---|
value | dict | 存储的原始值，即 Memory.model_dump() 的结果 |
key | str | 该条记忆的唯一标识符（UUID），用于精准更新 |
namespace | list | 命名空间，如 ['memories', '1'] |
created_at | str | 创建时间（ISO 8601 格式） |
updated_at | str | 最近更新时间 |
当记忆集合规模很大时，全量返回所有记忆会导致 context 过长。更高级的实现会为 Store 配置向量嵌入（embeddings），让 store.search(namespace, query="用户喜欢什么运动？") 只返回最相关的 Top-K 条记忆，节省 token、提升精度。LangGraph 的持久化 Store（如配合 PostgreSQL + pgvector）原生支持此能力。
检索到的记忆被格式化为无序列表，注入 LLM 的系统提示：
# 从每条记忆的 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)现在我们已经学完了两种记忆 Schema 方案。如何在实际项目中做选择？以下是系统性对比：
| 维度 | Profile 方案（Lesson 2） | Collection 方案（本节课） |
|---|---|---|
| 结构 | 固定字段的 Pydantic Schema（如 name, location, interests） | 极简单条 Schema（只有 content），记录数量动态增长 |
| 存储 | 所有信息存在同一个 key 下，覆盖更新 | 每条记忆独立 key（UUID），新增或精准更新单条 |
| 灵活性 | 低：只能记录预先设计好的字段 | 高：可以记录任何维度的信息，无需预定义 |
| 信息密度 | 高：结构化字段，检索时直接使用字段名 | 中：需要语义匹配才能找到相关记忆 |
| 可扩展性 | 差：加新字段需要改 Schema，可能影响历史数据 | 好：直接追加新记忆，无需改动 Schema |
| 适合场景 | 用户信息维度固定、结构化查询场景 | 开放域对话、信息维度不可预测场景 |
Profile 负责存储高价值的结构化核心信息（姓名、位置、核心偏好），Collection 负责存储细节性的自由文本记忆（某次对话提到的具体事件、临时喜好）。在同一个 chatbot 里同时维护两个 namespace，是生产系统常见的做法。
Collection 记忆 = 无限延伸的便签板。每次对话后，AI 把学到的新事情写在新便签纸上（store.put + 新 UUID），遇到已知信息有更新时，找到对应的旧便签修改（store.put + 原 key）。下次对话开始，AI 先把所有便签翻出来读一遍（store.search），然后带着这些背景知识来回应用户。Trustcall 就是那个负责"判断要不要写新便签、要修改哪张旧便签"的智能助手。
# 第一轮：告知姓名
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 等适合骑行顺路拜访的面包店