Memory Schema: Profile

1Why you need structured memory Schema

The memory from the previous section (Lesson 1) is a plain text string, e.g."User likes biking in SF". This method is simple, but has obvious flaws:

• It is not convenient to programmatically access specific fields (such as "only get user name")
• LLM must regenerate the entire string every time it is updated, and it is easy to miss existing information.
• Field-level validation cannot be done (for example, interests must be a list)

The solution is to useStructured memory Schema: Define the memory as a TypedDict or Pydantic model and let LLM extract information according to Schema.

2TypedDict defines Profile Schema

PythonTypedDictIt is the simplest way to define structured data, and fields have type annotations.

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']}, ...}

3with_structured_output: extract structured output

LangChain'swith_structured_output()The method can force the model to output structured data that conforms to the specified Schema and automatically complete the parsing.

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']}

Used in write_memory node

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)

4Limitations of with_structured_output: Complex Schema will fail

For complex Pydantic Schemas that are deeply nested, even highly capable models such as deepseek-v4-pro are prone to validation errors:

# 复杂嵌套 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

5Trustcall: create_extractor in-depth explanation

Trustcall is an open source structured extraction library developed by the LangChain team, specifically designed to solvewith_structured_outputof the above issues. It uses the underlyingTool callto extract structured data and built-inself-correctionandIncremental patch updatemechanism.

Basic usage: 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's handling of complex Schema

SameTelegramAndTrustFallPreferencesFor complex schemas, Trustcall can be extracted successfully - because it will try to retry and correct itself:

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

6Incremental update of Trustcall: JSON Patch mechanism

The core advantage of Trustcall isincremental update:Pass"existing"The parameters are passed into the existing memory, and Trustcall will generate aJSON Patchto modify only the changed parts instead of regenerating the entire 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'，名字保留，无需重新生成

json_doc_id in response_metadata

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 表示更新同一文档）

7Complete Chatbot: a combination of Trustcall + Store implementation

Schema definition

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: Read the structured Profile and inject prompt words

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 incremental update 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)

8Running the demo and cross-thread memory verification

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

9Profile vs Collection: When to use which mode?