Memory Schema: Profile

1 为什么需要结构化记忆 Schema

上一节（Lesson 1）的记忆是纯文本字符串，例如 "User likes biking in SF"。这种方式简单，但有明显缺陷：

• 不便于程序化访问特定字段（如"只取用户名字"）
• 每次更新时必须让 LLM 重新生成整个字符串，容易遗漏已有信息
• 无法做字段级别的验证（例如 interests 必须是列表）

解决方案是使用结构化记忆 Schema：将记忆定义为 TypedDict 或 Pydantic 模型，让 LLM 按 Schema 提取信息。

2 TypedDict 定义 Profile Schema

Python 的 TypedDict 是定义结构化数据的最简方式，字段带类型注解。

from typing import TypedDict, List

class UserProfile(TypedDict):
    """User profile schema with typed fields"""
    user_name: str          # The user's preferred name
    interests: List[str]    # A list of the user's interests

# TypedDict 实例就是普通字典
user_profile: UserProfile = {
    "user_name": "Lance",
    "interests": ["biking", "technology", "coffee"]
}

# TypedDict 实例可以直接作为 store.put 的 value
user_id = "1"
namespace_for_memory = (user_id, "memory")
in_memory_store.put(namespace_for_memory, "user_profile", user_profile)

# 检索验证
for m in in_memory_store.search(namespace_for_memory):
    print(m.dict())
# {'value': {'user_name': 'Lance', 'interests': ['biking', 'technology', 'coffee']}, ...}

3 with_structured_output：提取结构化输出

LangChain 的 with_structured_output() 方法可以强制模型输出符合指定 Schema 的结构化数据，并自动完成解析。

from pydantic import BaseModel, Field
from langchain_core.messages import HumanMessage

# 绑定 Schema 到模型
model_with_structure = model.with_structured_output(UserProfile)

# 调用后直接得到符合 Schema 的字典
structured_output = model_with_structure.invoke(
    [HumanMessage("My name is Lance, I like to bike.")]
)
# 输出：{'user_name': 'Lance', 'interests': ['biking']}

在 write_memory 节点中使用

def write_memory(state, config, store):
    user_id = config["configurable"]["user_id"]
    namespace = ("memory", user_id)
    existing_memory = store.get(namespace, "user_memory")

    # 格式化已有记忆作为上下文
    if existing_memory and existing_memory.value:
        memory_dict = existing_memory.value
        formatted_memory = (
            f"Name: {memory_dict.get('user_name', 'Unknown')}\n"
            f"Interests: {', '.join(memory_dict.get('interests', []))}"
        )
    else:
        formatted_memory = None

    system_msg = CREATE_MEMORY_INSTRUCTION.format(memory=formatted_memory)

    # 用 with_structured_output 从对话中提取 UserProfile
    new_memory = model_with_structure.invoke(
        [SystemMessage(content=system_msg)] + state['messages']
    )

    # 覆写存入 Store（每次重新生成整个 Profile）
    store.put(namespace, "user_memory", new_memory)

4 with_structured_output 的局限：复杂 Schema 会失败

对于嵌套较深的复杂 Pydantic Schema，即使是高能力模型（如 deepseek-v4-pro）也容易产生验证错误：

# 复杂嵌套 Schema 示例（TelegramAndTrustFallPreferences）
class OutputFormat(BaseModel):
    preference: str
    sentence_preference_revealed: str

class TelegramPreferences(BaseModel):
    preferred_encoding: Optional[List[OutputFormat]] = None
    favorite_telegram_operators: Optional[List[OutputFormat]] = None
    preferred_telegram_paper: Optional[List[OutputFormat]] = None

# ... 更多嵌套层级

# 用 with_structured_output 调用会报 ValidationError：
# 1 validation error for TelegramAndTrustFallPreferences
# semaphore: Input should be a valid dictionary or instance of Semaphore

5 Trustcall：create_extractor 深度讲解

Trustcall 是 LangChain 团队开源的结构化提取库，专门解决 with_structured_output 的上述问题。它底层使用工具调用来提取结构化数据，并内置自我纠错和增量 patch 更新机制。

基础用法：create_extractor

from trustcall import create_extractor

# 使用 Pydantic 模型定义 Schema（支持 Field 描述）
class UserProfile(BaseModel):
    """User profile schema with typed fields"""
    user_name: str = Field(description="The user's preferred name")
    interests: List[str] = Field(description="A list of the user's interests")

# 创建提取器：model + schema + tool_choice（强制使用该 schema）
trustcall_extractor = create_extractor(
    model,
    tools=[UserProfile],
    tool_choice="UserProfile"
)

# 调用：传入消息列表
result = trustcall_extractor.invoke({
    "messages": [SystemMessage(content="Extract the user profile")] + conversation
})

# result 包含三个键：
# - "messages"：包含工具调用的 AIMessage 列表
# - "responses"：解析后的 Pydantic 模型实例列表
# - "response_metadata"：各响应对应的元数据（含 json_doc_id）

result["responses"][0]
# UserProfile(user_name='Lance', interests=['biking around San Francisco'])

result["responses"][0].model_dump()
# {'user_name': 'Lance', 'interests': ['biking around San Francisco']}

Trustcall 对复杂 Schema 的处理

同样的 TelegramAndTrustFallPreferences 复杂 Schema，Trustcall 可以成功提取——因为它会尝试重试并自我纠错：

bound = create_extractor(model, tools=[TelegramAndTrustFallPreferences], tool_choice="TelegramAndTrustFallPreferences")
result = bound.invoke(conversation_text)
result["responses"][0]  # 成功提取完整嵌套 Schema，不再报 ValidationError

6 Trustcall 的增量更新：JSON Patch 机制

Trustcall 最核心的优势是增量更新：通过 "existing" 参数传入已有记忆，Trustcall 会生成一个 JSON Patch 来只修改有变化的部分，而不是重新生成整个 Schema。

# 1. 先提取初始 Profile
result = trustcall_extractor.invoke({"messages": initial_conversation})
schema = result["responses"]  # [UserProfile(user_name='Lance', interests=['biking'])]

# 2. 对话继续：用户说他还喜欢去面包店
updated_conversation = [...]

# 3. 将已有 Schema 作为 existing 传入（key=schema 名, value=dict）
result = trustcall_extractor.invoke(
    {"messages": [SystemMessage(content="Update the memory to incorporate new information")] + updated_conversation},
    {"existing": {"UserProfile": schema[0].model_dump()}}
)

result["responses"][0].model_dump()
# {'user_name': 'Lance', 'interests': ['biking', 'visiting bakeries']}
# → 只增加了 'visiting bakeries'，名字保留，无需重新生成

response_metadata 中的 json_doc_id

result["response_metadata"]
# [{'id': 'call_WeZl0ACfQStxblim0ps8LNKT'}]
# 新建时：只有 id，没有 json_doc_id
# 更新时：包含 json_doc_id 指向被更新的文档

# 在 Store 中保存时，用 json_doc_id（如果有）作为 key：
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")
    )
# 对 Profile：key 通常是固定的（Trustcall 返回同一 json_doc_id 表示更新同一文档）

7 完整 Chatbot：Trustcall + Store 的组合实现

Schema 定义

class UserProfile(BaseModel):
    """Profile of a user"""
    user_name: str = Field(description="The user's preferred name")
    user_location: str = Field(description="The user's location")
    interests: list = Field(description="A list of the user's interests")

trustcall_extractor = create_extractor(
    model, tools=[UserProfile], tool_choice="UserProfile"
)

call_model：读取结构化 Profile 并注入提示词

def call_model(state, config, store):
    user_id = config["configurable"]["user_id"]
    namespace = ("memory", user_id)
    existing_memory = store.get(namespace, "user_memory")

    # 结构化记忆可以精确提取各字段
    if existing_memory and existing_memory.value:
        memory_dict = existing_memory.value
        formatted_memory = (
            f"Name: {memory_dict.get('user_name', 'Unknown')}\n"
            f"Location: {memory_dict.get('user_location', 'Unknown')}\n"
            f"Interests: {', '.join(memory_dict.get('interests', []))}"
        )
    else:
        formatted_memory = None

    system_msg = MODEL_SYSTEM_MESSAGE.format(memory=formatted_memory)
    response = model.invoke([SystemMessage(content=system_msg)] + state["messages"])
    return {"messages": response}

write_memory：Trustcall 增量更新 Profile

def write_memory(state, config, store):
    user_id = config["configurable"]["user_id"]
    namespace = ("memory", user_id)
    existing_memory = store.get(namespace, "user_memory")

    # 将已有 Profile 格式化为 Trustcall 的 existing 参数
    existing_profile = {"UserProfile": existing_memory.value} if existing_memory else None

    # Trustcall 提取：有 existing 时做增量 patch，无 existing 时新建
    result = trustcall_extractor.invoke({
        "messages": [SystemMessage(content=TRUSTCALL_INSTRUCTION)] + state["messages"],
        "existing": existing_profile
    })

    # 保存更新后的 Profile（覆写同一 key）
    updated_profile = result["responses"][0].model_dump()
    store.put(namespace, "user_memory", updated_profile)

8 运行演示与跨线程记忆验证

config = {"configurable": {"thread_id": "1", "user_id": "1"}}

9 Profile vs Collection：何时用哪种模式