Memory Agent：task_mAIstro

0 整体架构一览

task_mAIstro 是一个 ToDo 管理助手，与前面的被动记忆 Chatbot 最大的区别在于：它自己决定何时、保存哪类记忆，而不是每轮对话后自动保存所有内容。

1 导入与依赖

import uuid
from datetime import datetime

from pydantic           import BaseModel, Field
from trustcall           import create_extractor
from typing              import Literal, Optional, TypedDict

from langchain_core.runnables import RunnableConfig
from langchain_core.messages  import merge_message_runs
from langchain_core.messages  import SystemMessage, HumanMessage

from langchain_deepseek  import ChatDeepSeek

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

import configuration

2 Spy 工具类：监听 Trustcall 的内部调用

Trustcall 在内部通过工具调用来操作记忆（PatchDoc 更新、schema 名新建）。Spy 是一个 LangChain Runnable 监听器，通过 with_listeners(on_end=spy) 挂载到提取器上，在调用结束后捕获所有内部工具调用记录。

class Spy:
    def __init__(self):
        self.called_tools = []

    def __call__(self, run):
        q = [run]
        while q:
            r = q.pop()
            if r.child_runs:
                q.extend(r.child_runs)
            if r.run_type == "chat_model":
                self.called_tools.append(
                    r.outputs["generations"][0][0]["message"]["kwargs"]["tool_calls"]
                )

extract_tool_info：将 Spy 数据转为可读字符串

该函数解析 Spy 捕获到的工具调用，区分"更新已有记忆（PatchDoc）"和"新建记忆（schema 名）"，将结果格式化为字符串，最终作为 ToolMessage 的 content 返回给 task_mAIstro 节点。

def extract_tool_info(tool_calls, schema_name="Memory"):
    changes = []

    for call_group in tool_calls:
        for call in call_group:
            if call['name'] == 'PatchDoc':
                changes.append({
                    'type': 'update',
                    'doc_id': call['args']['json_doc_id'],
                    'planned_edits': call['args']['planned_edits'],
                    'value': call['args']['patches'][0]['value']
                })
            elif call['name'] == schema_name:
                changes.append({'type': 'new', 'value': call['args']})

    result_parts = []
    for change in changes:
        if change['type'] == 'update':
            result_parts.append(
                f"Document {change['doc_id']} updated:\n"
                f"Plan: {change['planned_edits']}\n"
                f"Added content: {change['value']}"
            )
        else:
            result_parts.append(
                f"New {schema_name} created:\n"
                f"Content: {change['value']}"
            )

    return "\n\n".join(result_parts)

3 三类记忆 Schema：Profile、ToDo、UpdateMemory

用户画像：Profile

class Profile(BaseModel):
    """This is the profile of the user you are chatting with"""
    name: Optional[str]  = Field(description="The user's name",     default=None)
    location: Optional[str] = Field(description="The user's location", default=None)
    job: Optional[str]   = Field(description="The user's job",       default=None)
    connections: list[str] = Field(
        description="Personal connections: family, friends, coworkers",
        default_factory=list
    )
    interests: list[str] = Field(
        description="Interests that the user has",
        default_factory=list
    )

所有字段均为可选（Optional 或带 default_factory），避免首次提取时因信息不全而报错。Trustcall 的 JSON Patch 只更新有变化的字段，保留其他已知信息。

待办事项：ToDo

class ToDo(BaseModel):
    task: str = Field(description="The task to be completed.")
    time_to_complete: Optional[int] = Field(
        description="Estimated time to complete (minutes)."
    )
    deadline: Optional[datetime] = Field(
        description="When the task needs to be completed by", default=None
    )
    solutions: list[str] = Field(
        description="Specific, actionable solutions (ideas / providers / options)",
        min_items=1, default_factory=list
    )
    status: Literal["not started", "in progress", "done", "archived"] = Field(
        description="Current status of the task", default="not started"
    )

solutions 字段要求至少一条具体行动方案（min_items=1）。结合后面的 Instructions（用户可以要求"包含具体的本地商家"），这个字段会越来越丰富。

路由工具：UpdateMemory

class UpdateMemory(TypedDict):
    """Decision on what memory type to update"""
    update_type: Literal['user', 'todo', 'instructions']

初始化模型与 profile_extractor

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

# Profile 提取器：全局复用，无 Spy，不需要 enable_inserts（每用户只有一条记录）
profile_extractor = create_extractor(
    model,
    tools=[Profile],
    tool_choice="Profile",
)

profile_extractor 是全局单例：因为 Profile 更新不需要 Spy 监听，可以安全复用同一个实例。相比之下，ToDo 的 extractor 每次需要绑定新的 Spy() 实例，所以在节点函数内部创建。

4 三段系统提示词

整个代理的行为由三段提示词控制，分别用于三种不同的场景。

① MODEL_SYSTEM_MESSAGE — 主代理提示词

MODEL_SYSTEM_MESSAGE = """You are a helpful chatbot.
You are designed to be a companion to a user, helping them keep track of their ToDo list.

You have a long term memory which keeps track of three things:
1. The user's profile (general information about them)
2. The user's ToDo list
3. General instructions for updating the ToDo list

Here is the current User Profile (may be empty):
<user_profile>
{user_profile}
</user_profile>

Here is the current ToDo List (may be empty):
<todo>
{todo}
</todo>

Here are the current user-specified preferences (may be empty):
<instructions>
{instructions}
</instructions>

Instructions for reasoning:
1. Reason carefully about the user's messages.
2. Decide whether any long-term memory should be updated:
   - Personal info → UpdateMemory(update_type='user')
   - Tasks mentioned → UpdateMemory(update_type='todo')
   - Preferences for ToDo updates → UpdateMemory(update_type='instructions')
3. Do not tell the user when you update the profile or instructions.
   Tell the user when you update the todo list.
4. Err on the side of updating the todo list.
5. Respond naturally after tool calls."""

② TRUSTCALL_INSTRUCTION — 提取器指令

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.
System Time: {time}"""

这段提示词作为 SystemMessage 插入到 Trustcall 的消息列表最前面，指导提取器关注对话中的记忆价值。{time} 注入当前时间，为带时间戳的 ToDo deadline 提供上下文基准。

③ CREATE_INSTRUCTIONS — 指令更新提示词

CREATE_INSTRUCTIONS = """Reflect on the following interaction.
Based on this interaction, update your instructions for how to update ToDo list items.
Use any feedback from the user to update how they like to have items added, etc.

Your current instructions are:
<current_instructions>
{current_instructions}
</current_instructions>"""

仅在 update_instructions 节点中使用。不走 Trustcall，而是直接让 LLM 生成新的自然语言指令，覆写 Store 中的旧版本。

5 节点①：task_mAIstro — 读取记忆与路由决策

def task_mAIstro(state: MessagesState, config: RunnableConfig, store: BaseStore):
    # 从 config 解析 user_id（由 configuration.Configuration 定义）
    configurable = configuration.Configuration.from_runnable_config(config)
    user_id = configurable.user_id

    # 1. 读取用户画像
    namespace = ("profile", user_id)
    memories = store.search(namespace)
    user_profile = memories[0].value if memories else None

    # 2. 读取所有 ToDo，拼接为多行字符串
    namespace = ("todo", user_id)
    memories = store.search(namespace)
    todo = "\n".join(f"{mem.value}" for mem in memories)

    # 3. 读取操作偏好指令
    namespace = ("instructions", user_id)
    memories = store.search(namespace)
    instructions = memories[0].value if memories else ""

    # 4. 构建系统提示，注入三类记忆
    system_msg = MODEL_SYSTEM_MESSAGE.format(
        user_profile=user_profile, todo=todo, instructions=instructions
    )

    # 5. 绑定 UpdateMemory 工具，串行调用（确保路由逻辑简单）
    response = model.bind_tools(
        [UpdateMemory], parallel_tool_calls=False
    ).invoke([SystemMessage(content=system_msg)] + state["messages"])

    return {"messages": [response]}

6 节点②：update_profile — Trustcall 增量更新画像

def update_profile(state: MessagesState, config: RunnableConfig, store: BaseStore):
    configurable = configuration.Configuration.from_runnable_config(config)
    user_id = configurable.user_id
    namespace = ("profile", user_id)

    # 从 Store 读取已有 Profile，格式化为 Trustcall 三元组
    existing_items = store.search(namespace)
    tool_name = "Profile"
    existing_memories = (
        [(item.key, tool_name, item.value) for item in existing_items]
        if existing_items else None
    )

    # 注入当前时间，去掉最后一条消息（task_mAIstro 的工具调用消息）
    instr = TRUSTCALL_INSTRUCTION.format(time=datetime.now().isoformat())
    updated_messages = list(merge_message_runs(
        messages=[SystemMessage(content=instr)] + state["messages"][:-1]
    ))

    # Trustcall 提取：有 existing 时做 JSON Patch 增量更新
    result = profile_extractor.invoke({
        "messages": updated_messages,
        "existing": existing_memories
    })

    # 保存到 Store（json_doc_id 存在则更新同一条记录，否则新建）
    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"))

    # 必须返回 ToolMessage：响应 task_mAIstro 的 UpdateMemory 工具调用
    tool_calls = state['messages'][-1].tool_calls
    return {"messages": [{"role": "tool", "content": "updated profile",
                           "tool_call_id": tool_calls[0]['id']}]}

7 节点③：update_todos — 带 Spy 的 ToDo 更新

ToDo 更新与 Profile 更新的最大区别：需要 enable_inserts=True 支持新增任务，且需要 Spy 捕获变更详情放入 ToolMessage。

def update_todos(state: MessagesState, config: RunnableConfig, store: BaseStore):
    configurable = configuration.Configuration.from_runnable_config(config)
    user_id = configurable.user_id
    namespace = ("todo", user_id)

    existing_items = store.search(namespace)
    tool_name = "ToDo"
    existing_memories = (
        [(item.key, tool_name, item.value) for item in existing_items]
        if existing_items else None
    )

    instr = TRUSTCALL_INSTRUCTION.format(time=datetime.now().isoformat())
    updated_messages = list(merge_message_runs(
        messages=[SystemMessage(content=instr)] + state["messages"][:-1]
    ))

    # 每次调用创建新 Spy（防止跨调用数据混用）
    spy = Spy()

    # ToDo 提取器：enable_inserts=True 允许新建任务，挂载 Spy 监听器
    todo_extractor = create_extractor(
        model, tools=[ToDo], tool_choice=tool_name, enable_inserts=True
    ).with_listeners(on_end=spy)

    result = todo_extractor.invoke({
        "messages": updated_messages, "existing": existing_memories
    })

    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"))

    # 把 Spy 捕获的变更摘要作为 ToolMessage 内容，让 task_mAIstro 能告知用户
    tool_calls = state['messages'][-1].tool_calls
    todo_update_msg = extract_tool_info(spy.called_tools, tool_name)
    return {"messages": [{"role": "tool", "content": todo_update_msg,
                           "tool_call_id": tool_calls[0]['id']}]}

8 节点④：update_instructions — 纯文本指令更新

Instructions 是用户对"如何管理 ToDo"的自然语言偏好描述，没有固定 Schema，不走 Trustcall，直接让 LLM 生成文本后覆写。

def update_instructions(state: MessagesState, config: RunnableConfig, store: BaseStore):
    configurable = configuration.Configuration.from_runnable_config(config)
    user_id = configurable.user_id
    namespace = ("instructions", user_id)

    # 用 get（精确 key）而非 search（全量），因为 Instructions 始终只有一条
    existing_memory = store.get(namespace, "user_instructions")

    system_msg = CREATE_INSTRUCTIONS.format(
        current_instructions=existing_memory.value if existing_memory else None
    )

    # 直接调用 LLM 生成新指令文本（不用 Trustcall）
    new_memory = model.invoke(
        [SystemMessage(content=system_msg)] +
        state['messages'][:-1] +
        [HumanMessage(content="Please update the instructions based on the conversation")]
    )

    # 固定 key 覆写：Instructions 只有一条，无需追踪 json_doc_id
    store.put(namespace, "user_instructions", {"memory": new_memory.content})

    tool_calls = state['messages'][-1].tool_calls
    return {"messages": [{"role": "tool", "content": "updated instructions",
                           "tool_call_id": tool_calls[0]['id']}]}

9 条件边：route_message

def route_message(
    state: MessagesState, config: RunnableConfig, store: BaseStore
) -> Literal[END, "update_todos", "update_instructions", "update_profile"]:

    message = state['messages'][-1]          # task_mAIstro 的输出

    if len(message.tool_calls) == 0:           # 无工具调用 → 直接结束
        return END

    tool_call = message.tool_calls[0]          # 只取第一个（已强制串行）
    update_type = tool_call['args']['update_type']

    if   update_type == "user":         return "update_profile"
    elif update_type == "todo":         return "update_todos"
    elif update_type == "instructions": return "update_instructions"
    else:
        raise ValueError

10 图的组装与编译

# config_schema 将 Configuration（含 user_id）注册为可注入的运行时配置
builder = StateGraph(MessagesState, config_schema=configuration.Configuration)

# add_node(fn) 自动用函数名作为节点名
builder.add_node(task_mAIstro)
builder.add_node(update_todos)
builder.add_node(update_profile)
builder.add_node(update_instructions)

builder.add_edge(START, "task_mAIstro")
builder.add_conditional_edges("task_mAIstro", route_message)

# 更新节点执行完后，返回 task_mAIstro（携带 ToolMessage）
builder.add_edge("update_todos",        "task_mAIstro")
builder.add_edge("update_profile",     "task_mAIstro")
builder.add_edge("update_instructions", "task_mAIstro")

# 双存储模式：线程内短期记忆 + 跨线程长期记忆
across_thread_memory = InMemoryStore()
within_thread_memory = MemorySaver()
graph = builder.compile(
    checkpointer=within_thread_memory,
    store=across_thread_memory
)

11 完整运行演示

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