In [1]:
from dotenv import load_dotenv
load_dotenv()
Out[1]:
True
In [2]:
import uuid
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
from IPython.display import Image, display
from pydantic import BaseModel, Field
from datetime import datetime
from trustcall import create_extractor
from typing import Optional
from pydantic import BaseModel, Field
from langchain_core.runnables import RunnableConfig
from langchain_core.messages import merge_message_runs, HumanMessage, SystemMessage
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, MessagesState, END, START
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore
from trustcall import create_extractor
from langchain_deepseek import ChatDeepSeek
from langchain_deepseek import ChatDeepSeek
# Initialize the model
model = ChatDeepSeek(model="deepseek-chat", temperature=0)
In [3]:
class Memory(BaseModel):
content: str = Field(description="The main content of the memory. For example: User expressed interest in learning about French.")
class MemoryCollection(BaseModel):
memories: list[Memory] = Field(description="A list of memories about the user.")
In [4]:
# Inspect the tool calls made by Trustcall
class Spy:
def __init__(self):
self.called_tools = []
def __call__(self, run):
# Collect information about the tool calls made by the extractor.
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"]
)
# Initialize the spy
spy = Spy()
# Create the extractor
trustcall_extractor = create_extractor(
model,
tools=[Memory],
tool_choice="Memory",
enable_inserts=True,
)
# Add the spy as a listener
trustcall_extractor_see_all_tool_calls = trustcall_extractor.with_listeners(on_end=spy)
In [5]:
# Instruction
instruction = """Extract memories from the following conversation:"""
# 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.")]
# Invoke the extractor
result = trustcall_extractor.invoke({"messages": [SystemMessage(content=instruction)] + conversation})
In [6]:
# Messages contain the tool calls
for m in result["messages"]:
m.pretty_print()
================================== Ai Message ==================================
Tool Calls:
Memory (call_00_GVDYpfTsfoE4EkWsGlwQ0375)
Call ID: call_00_GVDYpfTsfoE4EkWsGlwQ0375
Args:
content: User's name is Lance.
Memory (call_01_aAOULvxdD6ZakHQ2Lr2v6129)
Call ID: call_01_aAOULvxdD6ZakHQ2Lr2v6129
Args:
content: Lance enjoyed a bike ride in San Francisco this morning.
In [27]:
print(result)
{'messages': [AIMessage(content='', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 78, 'prompt_tokens': 342, 'total_tokens': 420, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}, 'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 342}, 'model_provider': 'deepseek', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'fp_8b330d02d0_prod0820_fp8_kvcache_20260402', 'id': '4aaa5e94-f732-4751-b4a7-5070fd9d1c95', 'finish_reason': 'tool_calls', 'logprobs': None}, id='lc_run--019e4a3a-c6e2-7960-91fb-dc253751fdb4-0', tool_calls=[{'name': 'Memory', 'args': {'content': "User's name is Lance."}, 'id': 'call_00_GVDYpfTsfoE4EkWsGlwQ0375', 'type': 'tool_call'}, {'name': 'Memory', 'args': {'content': 'Lance enjoyed a bike ride in San Francisco this morning.'}, 'id': 'call_01_aAOULvxdD6ZakHQ2Lr2v6129', 'type': 'tool_call'}], invalid_tool_calls=[], usage_metadata={'input_tokens': 342, 'output_tokens': 78, 'total_tokens': 420, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}})], 'responses': [Memory(content="User's name is Lance."), Memory(content='Lance enjoyed a bike ride in San Francisco this morning.')], 'response_metadata': [{'id': 'call_00_GVDYpfTsfoE4EkWsGlwQ0375'}, {'id': 'call_01_aAOULvxdD6ZakHQ2Lr2v6129'}], 'attempts': 1}
In [28]:
print(result["messages"])
[AIMessage(content='', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 78, 'prompt_tokens': 342, 'total_tokens': 420, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}, 'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 342}, 'model_provider': 'deepseek', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'fp_8b330d02d0_prod0820_fp8_kvcache_20260402', 'id': '4aaa5e94-f732-4751-b4a7-5070fd9d1c95', 'finish_reason': 'tool_calls', 'logprobs': None}, id='lc_run--019e4a3a-c6e2-7960-91fb-dc253751fdb4-0', tool_calls=[{'name': 'Memory', 'args': {'content': "User's name is Lance."}, 'id': 'call_00_GVDYpfTsfoE4EkWsGlwQ0375', 'type': 'tool_call'}, {'name': 'Memory', 'args': {'content': 'Lance enjoyed a bike ride in San Francisco this morning.'}, 'id': 'call_01_aAOULvxdD6ZakHQ2Lr2v6129', 'type': 'tool_call'}], invalid_tool_calls=[], usage_metadata={'input_tokens': 342, 'output_tokens': 78, 'total_tokens': 420, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}})]
In [7]:
# Responses contain the memories that adhere to the schema
for m in result["responses"]:
print(m)
content="User's name is Lance." content='Lance enjoyed a bike ride in San Francisco this morning.'
In [8]:
# Metadata contains the tool call
for m in result["response_metadata"]:
print(m)
{'id': 'call_00_GVDYpfTsfoE4EkWsGlwQ0375'}
{'id': 'call_01_aAOULvxdD6ZakHQ2Lr2v6129'}
In [25]:
from pprint import pprint
pprint(result["messages"][0].tool_calls[0])
{'args': {'content': "User's name is Lance."},
'id': 'call_00_GVDYpfTsfoE4EkWsGlwQ0375',
'name': 'Memory',
'type': 'tool_call'}
In [31]:
# Update the conversation
updated_conversation = [AIMessage(content="That's great, what 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 my Japan trip, and going back this winter!"),]
# Update the instruction
system_msg = """Update existing memories and create new ones based on the following conversation:"""
# We'll save existing memories, giving them an ID, key (tool name), and value
tool_name = "Memory"
if result["responses"]:
existing_memories = []
for i, memory in enumerate(result["responses"]):
item = (
str(i),
tool_name,
memory.model_dump()
)
existing_memories.append(item)
else:
existing_memories = None
existing_memories
Out[31]:
[('0', 'Memory', {'content': "User's name is Lance."}), ('1', 'Memory', {'content': 'Lance enjoyed a bike ride in San Francisco this morning.'})]
In [32]:
# Invoke the extractor with our updated conversation and existing memories
result = trustcall_extractor_see_all_tool_calls.invoke({"messages": updated_conversation,
"existing": existing_memories})
In [34]:
# Metadata contains the tool call
for m in result["response_metadata"]:
print(m)
{'id': 'call_00_4Mv7P6xDxECLK7eJQO6a8382'}
{'id': 'call_01_YFdDTrR8eoucRxkYzILe3768'}
In [35]:
# Messages contain the tool calls
for m in result["messages"]:
m.pretty_print()
================================== Ai Message ==================================
Tool Calls:
Memory (call_00_4Mv7P6xDxECLK7eJQO6a8382)
Call ID: call_00_4Mv7P6xDxECLK7eJQO6a8382
Args:
content: User went to Tartine and ate a croissant.
Memory (call_01_YFdDTrR8eoucRxkYzILe3768)
Call ID: call_01_YFdDTrR8eoucRxkYzILe3768
Args:
content: User is thinking about their Japan trip and wants to go back this winter.
In [36]:
# Parsed responses
for m in result["responses"]:
print(m)
content='User went to Tartine and ate a croissant.' content='User is thinking about their Japan trip and wants to go back this winter.'
In [37]:
# Inspect the tool calls made by Trustcall
spy.called_tools
Out[37]:
[[{'name': 'Memory', 'args': {'content': 'User went to Tartine and ate a croissant.'}, 'id': 'call_00_4Mv7P6xDxECLK7eJQO6a8382', 'type': 'tool_call'}, {'name': 'Memory', 'args': {'content': 'User is thinking about their Japan trip and wants to go back this winter.'}, 'id': 'call_01_YFdDTrR8eoucRxkYzILe3768', 'type': 'tool_call'}]]
In [38]:
def extract_tool_info(tool_calls, schema_name="Memory"):
"""Extract information from tool calls for both patches and new memories.
Args:
tool_calls: List of tool calls from the model
schema_name: Name of the schema tool (e.g., "Memory", "ToDo", "Profile")
"""
# Initialize list of changes
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']
})
# Format results as a single string
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)
# Inspect spy.called_tools to see exactly what happened during the extraction
schema_name = "Memory"
changes = extract_tool_info(spy.called_tools, schema_name)
print(changes)
New Memory created:
Content: {'content': 'User went to Tartine and ate a croissant.'}
New Memory created:
Content: {'content': 'User is thinking about their Japan trip and wants to go back this winter.'}
In [39]:
from typing import TypedDict, Literal
# Update memory tool
class UpdateMemory(TypedDict):
""" Decision on what memory type to update """
update_type: Literal['user', 'todo', 'instructions']
In [40]:
# User profile schema
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 connection of the user, such as family members, friends, or coworkers",
default_factory=list
)
interests: list[str] = Field(description="Interests that the user has", default_factory=list)
# ToDo schema
class ToDo(BaseModel):
task: str = Field(description="The task to be completed.")
time_to_complete: Optional[int] = Field(description="Estimated time to complete the task (minutes).")
deadline: Optional[datetime] = Field(description="When the task needs to be completed by (if applicable)",default=None)
solutions: list[str] = Field(
description="List of specific, actionable solutions (e.g., specific ideas, service providers, or concrete options relevant to completing the task)",
min_items=1,
default_factory=list
)
status: Literal["not started", "in progress", "done", "archived"] = Field(
description="Current status of the task",
default="not started"
)
# Create the Trustcall extractor for updating the user profile
profile_extractor = create_extractor(
model,
tools=[Profile],
tool_choice="Profile"
)
/var/folders/br/58x9n8f93rx561lknnhtpmr00000gn/T/ipykernel_89082/2031475938.py:21: PydanticDeprecatedSince20: `min_items` is deprecated and will be removed, use `min_length` instead. Deprecated in Pydantic V2.0 to be removed in V3.0. See Pydantic V2 Migration Guide at https://errors.pydantic.dev/2.13/migration/ solutions: list[str] = Field(
In [ ]:
# Chatbot instruction for choosing what to update and what tools to call
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 if no information has been collected yet):
<user_profile>
{user_profile}
</user_profile>
Here is the current ToDo List (may be empty if no tasks have been added yet):
<todo>
{todo}
</todo>
Here are the current user-specified preferences for updating the ToDo list (may be empty if no preferences have been specified yet):
<instructions>
{instructions}
</instructions>
Here are your instructions for reasoning about the user's messages:
1. Reason carefully about the user's messages as presented below.
2. Decide whether any of your long-term memory should be updated:
- If personal information was provided about the user, update the user's profile by calling UpdateMemory tool with type `user`
- If tasks are mentioned, update the ToDo list by calling UpdateMemory tool with type `todo`
- If the user has specified preferences for how to update the ToDo list, update the instructions by calling UpdateMemory tool with type `instructions`
3. Tell the user that you have updated your memory, if appropriate:
- Do not tell the user you have updated the user's profile
- Tell the user when you update the todo list
- Do not tell the user that you have updated instructions
4. Err on the side of updating the todo list. No need to ask for explicit permission.
5. Respond naturally to the user after a tool call was made to save memories, or if no tool call was made."""
# 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}"""
# Instructions for updating the ToDo list
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 (may be empty if no instructions has been collected yet):
<current_instructions>
{current_instructions}
</current_instructions>"""
In [41]:
# Node definitions
def task_mAIstro(state: MessagesState, config: RunnableConfig, store: BaseStore):
"""Load memories from the store and use them to personalize the chatbot's response."""
# Get the user ID from the config
user_id = config["configurable"]["user_id"]
# Retrieve profile memory from the store
namespace = ("profile", user_id)
memories = store.search(namespace)
if memories:
user_profile = memories[0].value
else:
user_profile = None
# Retrieve task memory from the store
namespace = ("todo", user_id)
memories = store.search(namespace)
todo = "\n".join(f"{mem.value}" for mem in memories)
# Retrieve custom instructions
namespace = ("instructions", user_id)
memories = store.search(namespace)
if memories:
instructions = memories[0].value
else:
instructions = ""
system_msg = MODEL_SYSTEM_MESSAGE.format(user_profile=user_profile, todo=todo, instructions=instructions)
# Respond using memory as well as the chat history
response = model.bind_tools([UpdateMemory],
parallel_tool_calls=False).invoke([SystemMessage(content=system_msg)]+state["messages"])
return {"messages": [response]}
In [42]:
def update_profile(state: MessagesState, config: RunnableConfig, store: BaseStore):
"""Reflect on the chat history and update the memory collection."""
# Get the user ID from the config
user_id = config["configurable"]["user_id"]
# Define the namespace for the memories
namespace = ("profile", user_id)
# Retrieve the most recent memories for context
existing_items = store.search(namespace)
# Format the existing memories for the Trustcall extractor
tool_name = "Profile"
existing_memories = ([(existing_item.key, tool_name, existing_item.value)
for existing_item in existing_items]
if existing_items
else None
)
# Merge the chat history and the instruction
TRUSTCALL_INSTRUCTION_FORMATTED=TRUSTCALL_INSTRUCTION.format(time=datetime.now().isoformat())
updated_messages=list(merge_message_runs(messages=[SystemMessage(content=TRUSTCALL_INSTRUCTION_FORMATTED)] + state["messages"][:-1]))
# Invoke the extractor
result = profile_extractor.invoke({"messages": updated_messages,
"existing": existing_memories})
# Save the memories from Trustcall to the store
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"),
)
tool_calls = state['messages'][-1].tool_calls
return {"messages": [{"role": "tool", "content": "updated profile", "tool_call_id":tool_calls[0]['id']}]}
In [43]:
def update_todos(state: MessagesState, config: RunnableConfig, store: BaseStore):
"""Reflect on the chat history and update the memory collection."""
# Get the user ID from the config
user_id = config["configurable"]["user_id"]
# Define the namespace for the memories
namespace = ("todo", user_id)
# Retrieve the most recent memories for context
existing_items = store.search(namespace)
# Format the existing memories for the Trustcall extractor
tool_name = "ToDo"
existing_memories = ([(existing_item.key, tool_name, existing_item.value)
for existing_item in existing_items]
if existing_items
else None
)
# Merge the chat history and the instruction
TRUSTCALL_INSTRUCTION_FORMATTED=TRUSTCALL_INSTRUCTION.format(time=datetime.now().isoformat())
updated_messages=list(merge_message_runs(messages=[SystemMessage(content=TRUSTCALL_INSTRUCTION_FORMATTED)] + state["messages"][:-1]))
# Initialize the spy for visibility into the tool calls made by Trustcall
spy = Spy()
# Create the Trustcall extractor for updating the ToDo list
todo_extractor = create_extractor(
model,
tools=[ToDo],
tool_choice=tool_name,
enable_inserts=True
).with_listeners(on_end=spy)
# Invoke the extractor
result = todo_extractor.invoke({"messages": updated_messages,
"existing": existing_memories})
# Save the memories from Trustcall to the store
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"),
)
# Respond to the tool call made in task_mAIstro, confirming the update
tool_calls = state['messages'][-1].tool_calls
# Extract the changes made by Trustcall and add the the ToolMessage returned to task_mAIstro
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']}]}
In [44]:
def update_instructions(state: MessagesState, config: RunnableConfig, store: BaseStore):
"""Reflect on the chat history and update the memory collection."""
# Get the user ID from the config
user_id = config["configurable"]["user_id"]
namespace = ("instructions", user_id)
existing_memory = store.get(namespace, "user_instructions")
# Format the memory in the system prompt
system_msg = CREATE_INSTRUCTIONS.format(current_instructions=existing_memory.value if existing_memory else None)
new_memory = model.invoke([SystemMessage(content=system_msg)]+state['messages'][:-1] + [HumanMessage(content="Please update the instructions based on the conversation")])
# Overwrite the existing memory in the store
key = "user_instructions"
store.put(namespace, key, {"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']}]}
In [45]:
# Conditional edge
def route_message(state: MessagesState, config: RunnableConfig, store: BaseStore) -> Literal[END, "update_todos", "update_instructions", "update_profile"]:
"""Reflect on the memories and chat history to decide whether to update the memory collection."""
message = state['messages'][-1]
if len(message.tool_calls) ==0:
return END
else:
tool_call = message.tool_calls[0]
if tool_call['args']['update_type'] == "user":
return "update_profile"
elif tool_call['args']['update_type'] == "todo":
return "update_todos"
elif tool_call['args']['update_type'] == "instructions":
return "update_instructions"
else:
raise ValueError
In [46]:
# Create the graph + all nodes
builder = StateGraph(MessagesState)
# Define the flow of the memory extraction process
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)
builder.add_edge("update_todos", "task_mAIstro")
builder.add_edge("update_profile", "task_mAIstro")
builder.add_edge("update_instructions", "task_mAIstro")
# Store for long-term (across-thread) memory
across_thread_memory = InMemoryStore()
# Checkpointer for short-term (within-thread) memory
within_thread_memory = MemorySaver()
# We compile the graph with the checkpointer and store
graph = builder.compile(checkpointer=within_thread_memory, store=across_thread_memory)
# View
display(Image(graph.get_graph(xray=1).draw_mermaid_png()))
In [ ]:
# We supply a thread ID for short-term (within-thread) memory
# We supply a user ID for long-term (across-thread) memory
config = {"configurable": {"thread_id": "1", "user_id": "Lance"}}
# User input to create a profile memory
input_messages = [HumanMessage(content="My name is Lance. I live in SF with my wife. I have a 1 year old daughter.")]
# Run the graph
for chunk in graph.stream({"messages": input_messages}, config, stream_mode="values"):
chunk["messages"][-1].pretty_print()
In [ ]:
# User input for a ToDo
input_messages = [HumanMessage(content="My wife asked me to book swim lessons for the baby.")]
# Run the graph
for chunk in graph.stream({"messages": input_messages}, config, stream_mode="values"):
chunk["messages"][-1].pretty_print()
In [ ]:
# User input to update instructions for creating ToDos
input_messages = [HumanMessage(content="When creating or updating ToDo items, include specific local businesses / vendors.")]
# Run the graph
for chunk in graph.stream({"messages": input_messages}, config, stream_mode="values"):
chunk["messages"][-1].pretty_print()
In [ ]:
# Check for updated instructions
user_id = "Lance"
# Search
for memory in across_thread_memory.search(("instructions", user_id)):
print(memory.value)
In [ ]:
# User input for a ToDo
input_messages = [HumanMessage(content="I need to fix the jammed electric Yale lock on the door.")]
# Run the graph
for chunk in graph.stream({"messages": input_messages}, config, stream_mode="values"):
chunk["messages"][-1].pretty_print()
In [ ]:
# Namespace for the memory to save
user_id = "Lance"
# Search
for memory in across_thread_memory.search(("todo", user_id)):
print(memory.value)
In [ ]:
# User input to update an existing ToDo
input_messages = [HumanMessage(content="For the swim lessons, I need to get that done by end of November.")]
# Run the graph
for chunk in graph.stream({"messages": input_messages}, config, stream_mode="values"):
chunk["messages"][-1].pretty_print()
In [ ]:
# User input for a ToDo
input_messages = [HumanMessage(content="Need to call back City Toyota to schedule car service.")]
# Run the graph
for chunk in graph.stream({"messages": input_messages}, config, stream_mode="values"):
chunk["messages"][-1].pretty_print()
In [ ]:
# Namespace for the memory to save
user_id = "Lance"
# Search
for memory in across_thread_memory.search(("todo", user_id)):
print(memory.value)
In [ ]:
# We supply a thread ID for short-term (within-thread) memory
# We supply a user ID for long-term (across-thread) memory
config = {"configurable": {"thread_id": "2", "user_id": "Lance"}}
# Chat with the chatbot
input_messages = [HumanMessage(content="I have 30 minutes, what tasks can I get done?")]
# Run the graph
for chunk in graph.stream({"messages": input_messages}, config, stream_mode="values"):
chunk["messages"][-1].pretty_print()
In [ ]:
# Chat with the chatbot
input_messages = [HumanMessage(content="Yes, give me some options to call for swim lessons.")]
# Run the graph
for chunk in graph.stream({"messages": input_messages}, config, stream_mode="values"):
chunk["messages"][-1].pretty_print()