LangGraph (3/4): memoria a largo plazo y human-in-the-loop

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
import uuid
from langgraph.store.memory import InMemoryStore
 
in_memory_store = InMemoryStore()
 
# Namespace for the memory to save
user_id = "1"
namespace_for_memory = (user_id, "memories")
 
# Save a memory to namespace as key and value
key = str(uuid.uuid4())
 
# The value needs to be a dictionary  
value = {"food_preference" : "I like pizza"}
 
# Save the memory
in_memory_store.put(namespace_for_memory, key, value)
	
	
Copied

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
# Search 
memories = in_memory_store.search(namespace_for_memory)
type(memories), len(memories)
	
	
Copied

			

				
(list, 1)
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
value = memories[0]
value.dict()
	
	
Copied

			

				
{'namespace': ['1', 'memories'],
'key': '70006131-948a-4d7a-bdce-78351c44fc4d',
'value': {'food_preference': 'I like pizza'},
'created_at': '2025-05-11T07:24:31.462465+00:00',
'updated_at': '2025-05-11T07:24:31.462468+00:00',
'score': None}
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
# The key, value
memories[0].key, memories[0].value
	
	
Copied

			

				
('70006131-948a-4d7a-bdce-78351c44fc4d', {'food_preference': 'I like pizza'})
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
# Get the memory by namespace and key
memory = in_memory_store.get(namespace_for_memory, key)
memory.dict()
	
	
Copied

			

				
{'namespace': ['1', 'memories'],
'key': '70006131-948a-4d7a-bdce-78351c44fc4d',
'value': {'food_preference': 'I like pizza'},
'created_at': '2025-05-11T07:24:31.462465+00:00',
'updated_at': '2025-05-11T07:24:31.462468+00:00'}
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.graph.message import add_messages
from langchain_huggingface import HuggingFaceEndpoint, ChatHuggingFace
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langgraph.checkpoint.memory import MemorySaver # Short-term memory
from langgraph.store.base import BaseStore # Long-term memory
from langchain_core.runnables.config import RunnableConfig
from langgraph.store.memory import InMemoryStore
from huggingface_hub import login
from IPython.display import Image, display
import os
import dotenv
 
dotenv.load_dotenv()
HUGGINGFACE_TOKEN = os.getenv("HUGGINGFACE_LANGGRAPH")
 
class State(TypedDict):
    messages: Annotated[list, add_messages]
 
os.environ["LANGCHAIN_TRACING_V2"] = "false" # Disable LangSmith tracing
 
# Create the LLM model
login(token=HUGGINGFACE_TOKEN) # Login to HuggingFace to use the model
MODEL = "Qwen/Qwen2.5-72B-Instruct"
model = HuggingFaceEndpoint(
    repo_id=MODEL,
    task="text-generation",
    max_new_tokens=512,
    do_sample=False,
    repetition_penalty=1.03,
)
# Create the chat model
llm = ChatHuggingFace(llm=model)
 
# Chatbot instruction
MODEL_SYSTEM_MESSAGE = """You are a helpful assistant that can answer questions and help with tasks.
You have access to a long-term memory that you can use to answer questions and help with tasks.
Here is the memory (it may be empty): {memory}"""
 
# Create new memory from the chat history and any existing memory
CREATE_MEMORY_INSTRUCTION = """You are a helpful assistant that gets information from the user to personalize your responses.
 
# INFORMATION FROM THE USER:
{memory}
 
# INSTRUCTIONS:
1. Carefully review the chat history
2. Identify new information from the user, such as:
   - Personal details (name, location)
   - Preferences (likes, dislikes)
   - Interests and hobbies
   - Past experiences
   - Goals or future plans
3. Combine any new information with the existing memory
4. Format the memory as a clear, bulleted list
5. If new information conflicts with existing memory, keep the most recent version
 
Remember: Only include factual information directly stated by the user. Do not make assumptions or inferences.
 
Based on the chat history below, please update the user information:"""
 
# Nodes
def call_model(state: MessagesState, config: RunnableConfig, store: BaseStore):
 
    """Load memory from the store and use it to personalize the chatbot's response."""
    
    # Get the user ID from the config
    user_id = config["configurable"]["user_id"]
 
    # Retrieve memory from the store
    namespace = ("memory", user_id)
    key = "user_memory"
    existing_memory = store.get(namespace, key)
 
    # Extract the actual memory content if it exists and add a prefix
    if existing_memory:
        # Value is a dictionary with a memory key
        existing_memory_content = existing_memory.value.get('memory')
    else:
        existing_memory_content = "No existing memory found."
    if isinstance(existing_memory_content, str):
        print(f"	[Call model debug] Existing memory: {existing_memory_content}")
    else:
        print(f"	[Call model debug] Existing memory: {existing_memory_content.content}")
 
    # Format the memory in the system prompt
    system_msg = MODEL_SYSTEM_MESSAGE.format(memory=existing_memory_content)
    
    # Respond using memory as well as the chat history
    response = llm.invoke([SystemMessage(content=system_msg)]+state["messages"])
 
    return {"messages": response}
 
def write_memory(state: MessagesState, config: RunnableConfig, store: BaseStore):
 
    """Reflect on the chat history and save a memory to the store."""
    
    # Get the user ID from the config
    user_id = config["configurable"]["user_id"]
 
    # Retrieve existing memory from the store
    namespace = ("memory", user_id)
    existing_memory = store.get(namespace, "user_memory")
        
    # Extract the memory
    if existing_memory:
        existing_memory_content = existing_memory.value.get('memory')
    else:
        existing_memory_content = "No existing memory found."
    if isinstance(existing_memory_content, str):
        print(f"	[Write memory debug] Existing memory: {existing_memory_content}")
    else:
        print(f"	[Write memory debug] Existing memory: {existing_memory_content.content}")
 
    # Format the memory in the system prompt
    system_msg = CREATE_MEMORY_INSTRUCTION.format(memory=existing_memory_content)
    new_memory = llm.invoke([SystemMessage(content=system_msg)]+state['messages'])
    if isinstance(new_memory, str):
        print(f"
	[Write memory debug] New memory: {new_memory}")
    else:
        print(f"
	[Write memory debug] New memory: {new_memory.content}")
 
    # Overwrite the existing memory in the store 
    key = "user_memory"
 
    # Write value as a dictionary with a memory key
    store.put(namespace, key, {"memory": new_memory.content})
 
# Create graph builder
graph_builder = StateGraph(State)
 
# Add nodes
graph_builder.add_node("call_model", call_model)
graph_builder.add_node("write_memory", write_memory)
 
# Connect nodes
graph_builder.add_edge(START, "call_model")
graph_builder.add_edge("call_model", "write_memory")
graph_builder.add_edge("write_memory", END)
 
# Store for long-term (across-thread) memory
long_term_memory = InMemoryStore()
 
# Checkpointer for short-term (within-thread) memory
short_term_memory = MemorySaver()
 
# Compile the graph
graph = graph_builder.compile(checkpointer=short_term_memory, store=long_term_memory)
 
display(Image(graph.get_graph().draw_mermaid_png()))
	
	
Copied

			

				
<IPython.core.display.Image object>
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
# 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": "1"}}
 
# User input 
input_messages = [HumanMessage(content="Hi, my name is Maximo")]
 
# Run the graph
for chunk in graph.stream({"messages": input_messages}, config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()
	
	
Copied

			

				
================================ Human Message =================================
Hi, my name is Maximo
[Call model debug] Existing memory: No existing memory found.
================================== Ai Message ==================================
Hello Maximo! It's nice to meet you. How can I assist you today?
[Write memory debug] Existing memory: No existing memory found.
[Write memory debug] New memory:
Here's the updated information I have about you:
- Name: Maximo
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
# User input 
input_messages = [HumanMessage(content="I like to bike around San Francisco")]
 
# Run the graph
for chunk in graph.stream({"messages": input_messages}, config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()
	
	
Copied

			

				
================================ Human Message =================================
I like to bike around San Francisco
[Call model debug] Existing memory:
Here's the updated information I have about you:
- Name: Maximo
================================== Ai Message ==================================
That sounds like a great way to explore the city! San Francisco has some fantastic biking routes. Are there any specific areas or routes you enjoy biking the most, or are you looking for some new recommendations?
[Write memory debug] Existing memory:
Here's the updated information I have about you:
- Name: Maximo
[Write memory debug] New memory:
Here's the updated information about you:
- Name: Maximo
- Location: San Francisco
- Interest: Biking around San Francisco
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
# Namespace for the memory to save
user_id = "1"
namespace = ("memory", user_id)
existing_memory = long_term_memory.get(namespace, "user_memory")
existing_memory.dict()
	
	
Copied

			

				
{'namespace': ['memory', '1'],
'key': 'user_memory',
'value': {'memory': " 

Here's the updated information about you:
- Name: Maximo
- Location: San Francisco
- Interest: Biking around San Francisco"},
'created_at': '2025-05-11T09:41:26.739207+00:00',
'updated_at': '2025-05-11T09:41:26.739211+00:00'}
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
print(existing_memory.value.get('memory'))
	
	
Copied

			

				
Here's the updated information about you:
- Name: Maximo
- Location: San Francisco
- Interest: Biking around San Francisco
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
# We supply a user ID for across-thread memory as well as a new thread ID
config = {"configurable": {"thread_id": "2", "user_id": "1"}}
 
# User input 
input_messages = [HumanMessage(content="Hi! Where would you recommend that I go biking?")]
 
# Run the graph
for chunk in graph.stream({"messages": input_messages}, config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()
	
	
Copied

			

				
================================ Human Message =================================
Hi! Where would you recommend that I go biking?
[Call model debug] Existing memory:
Here's the updated information about you:
- Name: Maximo
- Location: San Francisco
- Interest: Biking around San Francisco
================================== Ai Message ==================================
Hi there! Given my interest in biking around San Francisco, I'd recommend a few great routes:
1. **Golden Gate Park**: This is a fantastic place to bike, with wide paths that are separated from vehicle traffic. You can start at the eastern end near Stow Lake and bike all the way to the western end at Ocean Beach. There are plenty of scenic spots to stop and enjoy along the way.
2. **The Embarcadero**: This route follows the waterfront from Fisherman’s Wharf to the Bay Bridge. It’s relatively flat and offers beautiful views of the San Francisco Bay and the city skyline. You can also stop by the Ferry Building for some delicious food and drinks.
3. **Presidio**: The Presidio is a large park with numerous trails that offer diverse landscapes, from forests to coastal bluffs. The Crissy Field area is especially popular for its views of the Golden Gate Bridge.
4. **Golden Gate Bridge**: Riding across the Golden Gate Bridge is a must-do experience. You can start from the San Francisco side, bike across the bridge, and then continue into Marin County for a longer ride with stunning views.
5. **Lombard Street**: While not a long ride, biking down the famous crooked section of Lombard Street can be a fun and memorable experience. Just be prepared for the steep hill on the way back up!
Each of these routes offers a unique experience, so you can choose based on your interests and the type of scenery you enjoy. Happy biking!
[Write memory debug] Existing memory:
Here's the updated information about you:
- Name: Maximo
- Location: San Francisco
- Interest: Biking around San Francisco
[Write memory debug] New memory:  😊
Let me know if you have any other questions or if you need more recommendations!
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
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
	
	
Copied

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.graph.message import add_messages
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langgraph.checkpoint.memory import MemorySaver # Short-term memory
from langgraph.store.base import BaseStore # Long-term memory
from langchain_core.runnables.config import RunnableConfig
from langgraph.store.memory import InMemoryStore
from IPython.display import Image, display
from pydantic import BaseModel, Field
import os
import dotenv
 
dotenv.load_dotenv()
ANTHROPIC_TOKEN = os.getenv("ANTHROPIC_LANGGRAPH_API_KEY")
 
class State(TypedDict):
    messages: Annotated[list, add_messages]
 
os.environ["LANGCHAIN_TRACING_V2"] = "false" # Disable LangSmith tracing
 
# Create the LLM model
llm = ChatAnthropic(model="claude-3-7-sonnet-20250219", api_key=ANTHROPIC_TOKEN)
llm_with_structured_output = llm.with_structured_output(UserProfile)
 
# Chatbot instruction
MODEL_SYSTEM_MESSAGE = """You are a helpful assistant with memory that provides information about the user. 
If you have memory for this user, use it to personalize your responses.
Here is the memory (it may be empty): {memory}"""
 
# Create new memory from the chat history and any existing memory
CREATE_MEMORY_INSTRUCTION = """Create or update a user profile memory based on the user's chat history. 
This will be saved for long-term memory. If there is an existing memory, simply update it. 
Here is the existing memory (it may be empty): {memory}"""
 
# Nodes
def call_model(state: MessagesState, config: RunnableConfig, store: BaseStore):
 
    """Load memory from the store and use it to personalize the chatbot's response."""
    
    # Get the user ID from the config
    user_id = config["configurable"]["user_id"]
 
    # Retrieve memory from the store
    namespace = ("memory", user_id)
    existing_memory = store.get(namespace, "user_memory")
 
    # Format the memories for the system prompt
    if existing_memory and existing_memory.value:
        memory_dict = existing_memory.value
        formatted_memory = (
            f"Name: {memory_dict.get('user_name', 'Unknown')}
"
            f"Interests: {', '.join(memory_dict.get('interests', []))}"
        )
    else:
        formatted_memory = None
    # if isinstance(existing_memory_content, str):
    print(f"	[Call model debug] Existing memory: {formatted_memory}")
    # else:
    #     print(f"	[Call model debug] Existing memory: {existing_memory_content.content}")
 
    # Format the memory in the system prompt
    system_msg = MODEL_SYSTEM_MESSAGE.format(memory=formatted_memory)
 
    # Respond using memory as well as the chat history
    response = llm.invoke([SystemMessage(content=system_msg)]+state["messages"])
 
    return {"messages": response}
 
def write_memory(state: MessagesState, config: RunnableConfig, store: BaseStore):
 
    """Reflect on the chat history and save a memory to the store."""
    
    # Get the user ID from the config
    user_id = config["configurable"]["user_id"]
 
    # Retrieve existing memory from the store
    namespace = ("memory", user_id)
    existing_memory = store.get(namespace, "user_memory")
 
    # Format the memories for the system prompt
    if existing_memory and existing_memory.value:
        memory_dict = existing_memory.value
        formatted_memory = (
            f"Name: {memory_dict.get('user_name', 'Unknown')}
"
            f"Interests: {', '.join(memory_dict.get('interests', []))}"
        )
    else:
        formatted_memory = None
    print(f"	[Write memory debug] Existing memory: {formatted_memory}")
        
    # Format the existing memory in the instruction
    system_msg = CREATE_MEMORY_INSTRUCTION.format(memory=formatted_memory)
 
    # Invoke the model to produce structured output that matches the schema
    new_memory = llm_with_structured_output.invoke([SystemMessage(content=system_msg)]+state['messages'])
    print(f"	[Write memory debug] New memory: {new_memory}")
 
    # Overwrite the existing use profile memory
    key = "user_memory"
    store.put(namespace, key, new_memory)
 
# Create graph builder
graph_builder = StateGraph(MessagesState)
 
# Add nodes
graph_builder.add_node("call_model", call_model)
graph_builder.add_node("write_memory", write_memory)
 
# Connect nodes
graph_builder.add_edge(START, "call_model")
graph_builder.add_edge("call_model", "write_memory")
graph_builder.add_edge("write_memory", END)
 
# Store for long-term (across-thread) memory
long_term_memory = InMemoryStore()
 
# Checkpointer for short-term (within-thread) memory
short_term_memory = MemorySaver()
 
# Compile the graph
graph = graph_builder.compile(checkpointer=short_term_memory, store=long_term_memory)
 
display(Image(graph.get_graph().draw_mermaid_png()))
	
	
Copied

			

				
<IPython.core.display.Image object>
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
# 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": "1"}}
 
# User input 
input_messages = [HumanMessage(content="Hi, my name is Maximo and I like to bike around Madrid and eat salads.")]
 
# Run the graph
for chunk in graph.stream({"messages": input_messages}, config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()
	
	
Copied

			

				
================================ Human Message =================================
Hi, my name is Maximo and I like to bike around Madrid and eat salads.
[Call model debug] Existing memory: None
================================== Ai Message ==================================
Hello Maximo! It's nice to meet you. I see you enjoy biking around Madrid and eating salads - those are great healthy habits! Madrid has some beautiful areas to explore by bike, and the city has been improving its cycling infrastructure in recent years.
Is there anything specific about Madrid's cycling routes or perhaps some good places to find delicious salads in the city that you'd like to know more about? I'd be happy to help with any questions you might have.
[Write memory debug] Existing memory: None
[Write memory debug] New memory: {'user_name': 'Maximo', 'interests': ['biking', 'Madrid', 'salads']}
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
# Namespace for the memory to save
user_id = "1"
namespace = ("memory", user_id)
existing_memory = long_term_memory.get(namespace, "user_memory")
existing_memory.value
	
	
Copied

			

				
{'user_name': 'Maximo', 'interests': ['biking', 'Madrid', 'salads']}
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
from langchain_core.messages import HumanMessage, AIMessage
 
# Conversation
conversation = [HumanMessage(content="Hi, I'm Maximo."),
                AIMessage(content="Nice to meet you, Maximo."),
                HumanMessage(content="I really like playing soccer.")]
	
	
Copied

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
from pydantic import BaseModel, Field
from typing import List
 
# Schema
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")
 
from langchain_anthropic import ChatAnthropic
import os
import dotenv
 
dotenv.load_dotenv()
ANTHROPIC_TOKEN = os.getenv("ANTHROPIC_LANGGRAPH_API_KEY")
os.environ["LANGCHAIN_TRACING_V2"] = "false" # Disable LangSmith tracing
 
# Create the LLM model
llm = ChatAnthropic(model="claude-3-7-sonnet-20250219", api_key=ANTHROPIC_TOKEN)
	
	
Copied

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
from trustcall import create_extractor
 
# Create the extractor
trustcall_extractor = create_extractor(
    llm,
    tools=[UserProfile],
    tool_choice="UserProfile"
)
	
	
Copied

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
from langchain_core.messages import SystemMessage
 
# Instruction
system_msg = "Extract the user profile from the following conversation"
 
# Invoke the extractor
result = trustcall_extractor.invoke({"messages": [SystemMessage(content=system_msg)]+conversation})
result
	
	
Copied

			

				
{'messages': [AIMessage(content=[{'id': 'toolu_01WfgbD1fG3rJYAXGrjqjfVY', 'input': {'user_name': 'Maximo', 'interests': ['soccer']}, 'name': 'UserProfile', 'type': 'tool_use'}], additional_kwargs={}, response_metadata={'id': 'msg_01TEB3FeDKLAeHJtbKo5noyW', 'model': 'claude-3-7-sonnet-20250219', 'stop_reason': 'tool_use', 'stop_sequence': None, 'usage': {'cache_creation_input_tokens': 0, 'cache_read_input_tokens': 0, 'input_tokens': 497, 'output_tokens': 56}, 'model_name': 'claude-3-7-sonnet-20250219'}, id='run-8a15289b-fd39-4a2d-878a-fa6feaa805c5-0', tool_calls=[{'name': 'UserProfile', 'args': {'user_name': 'Maximo', 'interests': ['soccer']}, 'id': 'toolu_01WfgbD1fG3rJYAXGrjqjfVY', 'type': 'tool_call'}], usage_metadata={'input_tokens': 497, 'output_tokens': 56, 'total_tokens': 553, 'input_token_details': {'cache_read': 0, 'cache_creation': 0}})],
'responses': [UserProfile(user_name='Maximo', interests=['soccer'])],
'response_metadata': [{'id': 'toolu_01WfgbD1fG3rJYAXGrjqjfVY'}],
'attempts': 1}
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
for m in result["messages"]:
    m.pretty_print()
	
	
Copied

			

				
================================== Ai Message ==================================
[{'id': 'toolu_01WfgbD1fG3rJYAXGrjqjfVY', 'input': {'user_name': 'Maximo', 'interests': ['soccer']}, 'name': 'UserProfile', 'type': 'tool_use'}]
Tool Calls:
  UserProfile (toolu_01WfgbD1fG3rJYAXGrjqjfVY)
Call ID: toolu_01WfgbD1fG3rJYAXGrjqjfVY
  Args:
    user_name: Maximo
    interests: ['soccer']
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
schema = result["responses"]
schema
	
	
Copied

			

				
[UserProfile(user_name='Maximo', interests=['soccer'])]
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
type(schema[0])
	
	
Copied

			

				
__main__.UserProfile
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
schema[0].model_dump()
	
	
Copied

			

				
{'user_name': 'Maximo', 'interests': ['soccer']}
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
# Update the conversation
updated_conversation = [HumanMessage(content="Hi, I'm Maximo."),
                AIMessage(content="Nice to meet you, Maximo."),
                HumanMessage(content="I really like playing soccer."),
                AIMessage(content="It is great to play soccer! Where do you go after playing soccer?"),
                HumanMessage(content="I really like to go to a bakery after playing soccer."),]
	
	
Copied

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
# Update the instruction
system_msg = f"""Update the memory (JSON doc) to incorporate new information from the following conversation"""
 
# Invoke the extractor with the updated instruction and existing profile with the corresponding tool name (UserProfile)
result = trustcall_extractor.invoke({"messages": [SystemMessage(content=system_msg)]+updated_conversation},
                                    {"existing": {"UserProfile": schema[0].model_dump()}})
result
	
	
Copied

			

				
{'messages': [AIMessage(content=[{'id': 'toolu_01K1zTh33kXDAw1h18Yh2HBb', 'input': {'user_name': 'Maximo', 'interests': ['soccer', 'bakeries']}, 'name': 'UserProfile', 'type': 'tool_use'}], additional_kwargs={}, response_metadata={'id': 'msg_01RYUJvCdzL4b8kBYKo4BtQf', 'model': 'claude-3-7-sonnet-20250219', 'stop_reason': 'tool_use', 'stop_sequence': None, 'usage': {'cache_creation_input_tokens': 0, 'cache_read_input_tokens': 0, 'input_tokens': 538, 'output_tokens': 60}, 'model_name': 'claude-3-7-sonnet-20250219'}, id='run-06994472-5ba0-46cc-a512-5fcacce283fc-0', tool_calls=[{'name': 'UserProfile', 'args': {'user_name': 'Maximo', 'interests': ['soccer', 'bakeries']}, 'id': 'toolu_01K1zTh33kXDAw1h18Yh2HBb', 'type': 'tool_call'}], usage_metadata={'input_tokens': 538, 'output_tokens': 60, 'total_tokens': 598, 'input_token_details': {'cache_read': 0, 'cache_creation': 0}})],
'responses': [UserProfile(user_name='Maximo', interests=['soccer', 'bakeries'])],
'response_metadata': [{'id': 'toolu_01K1zTh33kXDAw1h18Yh2HBb'}],
'attempts': 1}
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
for m in result["messages"]:
    m.pretty_print()
	
	
Copied

			

				
================================== Ai Message ==================================
[{'id': 'toolu_01K1zTh33kXDAw1h18Yh2HBb', 'input': {'user_name': 'Maximo', 'interests': ['soccer', 'bakeries']}, 'name': 'UserProfile', 'type': 'tool_use'}]
Tool Calls:
  UserProfile (toolu_01K1zTh33kXDAw1h18Yh2HBb)
Call ID: toolu_01K1zTh33kXDAw1h18Yh2HBb
  Args:
    user_name: Maximo
    interests: ['soccer', 'bakeries']
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
updated_schema = result["responses"][0]
updated_schema.model_dump()
	
	
Copied

			

				
{'user_name': 'Maximo', 'interests': ['soccer', 'bakeries']}
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
from pydantic import BaseModel, Field
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.graph.message import add_messages
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langgraph.checkpoint.memory import MemorySaver # Short-term memory
from langgraph.store.base import BaseStore # Long-term memory
from langchain_core.runnables.config import RunnableConfig
from langgraph.store.memory import InMemoryStore
from IPython.display import Image, display
from pydantic import BaseModel, Field
import os
import dotenv
from trustcall import create_extractor
 
dotenv.load_dotenv()
ANTHROPIC_TOKEN = os.getenv("ANTHROPIC_LANGGRAPH_API_KEY")
 
os.environ["LANGCHAIN_TRACING_V2"] = "false" # Disable LangSmith tracing
 
# 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")
 
# Create the LLM model
llm = ChatAnthropic(model="claude-3-7-sonnet-20250219", api_key=ANTHROPIC_TOKEN)
 
# Create the extractor
trustcall_extractor = create_extractor(
    llm,
    tools=[UserProfile],
    tool_choice="UserProfile", # Enforces use of the UserProfile tool
)
 
# Chatbot instruction
MODEL_SYSTEM_MESSAGE = """You are a helpful assistant with memory that provides information about the user. 
If you have memory for this user, use it to personalize your responses.
Here is the memory (it may be empty): {memory}"""
 
# Create new memory from the chat history and any existing memory
TRUSTCALL_INSTRUCTION = """Create or update the memory (JSON doc) to incorporate information from the following conversation:"""
 
# Nodes
def call_model(state: MessagesState, config: RunnableConfig, store: BaseStore):
 
    """Load memory from the store and use it to personalize the chatbot's response."""
    
    """Load memory from the store and use it to personalize the chatbot's response."""
    
    # Get the user ID from the config
    user_id = config["configurable"]["user_id"]
 
    # Retrieve memory from the store
    namespace = ("memory", user_id)
    existing_memory = store.get(namespace, "user_memory")
 
    # Format the memories for the system prompt
    if existing_memory and existing_memory.value:
        memory_dict = existing_memory.value
        formatted_memory = (
            f"Name: {memory_dict.get('user_name', 'Unknown')}
"
            f"Location: {memory_dict.get('user_location', 'Unknown')}
"
            f"Interests: {', '.join(memory_dict.get('interests', []))}"
        )
    else:
        formatted_memory = None
    print(f"	[Call model debug] Existing memory: {formatted_memory}")
 
    # Format the memory in the system prompt
    system_msg = MODEL_SYSTEM_MESSAGE.format(memory=formatted_memory)
 
    # Respond using memory as well as the chat history
    response = llm.invoke([SystemMessage(content=system_msg)]+state["messages"])
 
    return {"messages": response}
 
def write_memory(state: MessagesState, config: RunnableConfig, store: BaseStore):
 
    """Reflect on the chat history and save a memory to the store."""
    
    # Get the user ID from the config
    user_id = config["configurable"]["user_id"]
 
    # Retrieve existing memory from the store
    namespace = ("memory", user_id)
    existing_memory = store.get(namespace, "user_memory")
        
    # Get the profile as the value from the list, and convert it to a JSON doc
    existing_profile = {"UserProfile": existing_memory.value} if existing_memory else None
    print(f"	[Write memory debug] Existing profile: {existing_profile}")
    
    # Invoke the extractor
    result = trustcall_extractor.invoke({"messages": [SystemMessage(content=TRUSTCALL_INSTRUCTION)]+state["messages"], "existing": existing_profile})
    
    # Get the updated profile as a JSON object
    updated_profile = result["responses"][0].model_dump()
    print(f"	[Write memory debug] Updated profile: {updated_profile}")
 
    # Save the updated profile
    key = "user_memory"
    store.put(namespace, key, updated_profile)
 
# Create graph builder
graph_builder = StateGraph(MessagesState)
 
# Add nodes
graph_builder.add_node("call_model", call_model)
graph_builder.add_node("write_memory", write_memory)
 
# Connect nodes
graph_builder.add_edge(START, "call_model")
graph_builder.add_edge("call_model", "write_memory")
graph_builder.add_edge("write_memory", END)
 
# Store for long-term (across-thread) memory
long_term_memory = InMemoryStore()
 
# Checkpointer for short-term (within-thread) memory
short_term_memory = MemorySaver()
 
# Compile the graph
graph = graph_builder.compile(checkpointer=short_term_memory, store=long_term_memory)
 
display(Image(graph.get_graph().draw_mermaid_png()))
	
	
Copied

			

				
<IPython.core.display.Image object>
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
# 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": "1"}}
 
# User input 
input_messages = [HumanMessage(content="Hi, my name is Maximo")]
 
# Run the graph
for chunk in graph.stream({"messages": input_messages}, config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()
	
	
Copied

			

				
================================ Human Message =================================
Hi, my name is Maximo
[Call model debug] Existing memory: None
================================== Ai Message ==================================
Hello Maximo! It's nice to meet you. How can I help you today? Whether you have questions, need information, or just want to chat, I'm here to assist you. Is there something specific you'd like to talk about?
[Write memory debug] Existing profile: None
[Write memory debug] Updated profile: {'user_name': 'Maximo', 'user_location': '<UNKNOWN>', 'interests': []}
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
# User input 
input_messages = [HumanMessage(content="I like to play soccer and I live in Madrid")]
 
# Run the graph
for chunk in graph.stream({"messages": input_messages}, config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()
	
	
Copied

			

				
================================ Human Message =================================
I like to play soccer and I live in Madrid
[Call model debug] Existing memory: Name: Maximo
Location: <UNKNOWN>
Interests:
================================== Ai Message ==================================
Hello Maximo! It's great to learn that you live in Madrid and enjoy playing soccer. Madrid is a fantastic city with a rich soccer culture, being home to world-famous clubs like Real Madrid and Atlético Madrid.
Soccer is truly a way of life in Spain, so you're in a perfect location for your interest. Do you support any particular team in Madrid? Or perhaps you enjoy playing soccer recreationally in the city's parks and facilities?
Is there anything specific about Madrid or soccer you'd like to discuss further?
[Write memory debug] Existing profile: {'UserProfile': {'user_name': 'Maximo', 'user_location': '<UNKNOWN>', 'interests': []}}
[Write memory debug] Updated profile: {'user_name': 'Maximo', 'user_location': 'Madrid', 'interests': ['soccer']}
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
# Namespace for the memory to save
user_id = "1"
namespace = ("memory", user_id)
existing_memory = long_term_memory.get(namespace, "user_memory")
existing_memory.dict()
	
	
Copied

			

				
{'namespace': ['memory', '1'],
'key': 'user_memory',
'value': {'user_name': 'Maximo',
  'user_location': 'Madrid',
  'interests': ['soccer']},
'created_at': '2025-05-12T17:35:03.583258+00:00',
'updated_at': '2025-05-12T17:35:03.583259+00:00'}
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
# The user profile saved as a JSON object
existing_memory.value
	
	
Copied

			

				
{'user_name': 'Maximo', 'user_location': 'Madrid', 'interests': ['soccer']}
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
# User input 
input_messages = [HumanMessage(content="I also like to play basketball")]
 
# Run the graph
for chunk in graph.stream({"messages": input_messages}, config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()
	
	
Copied

			

				
================================ Human Message =================================
I also like to play basketball
[Call model debug] Existing memory: Name: Maximo
Location: Madrid
Interests: soccer
================================== Ai Message ==================================
That's great to know, Maximo! It's nice that you enjoy both soccer and basketball. Basketball is also quite popular in Spain, with Liga ACB being one of the strongest basketball leagues in Europe.
In Madrid, you have the opportunity to follow Real Madrid's basketball section, which is one of the most successful basketball teams in Europe. The city offers plenty of courts and facilities where you can play basketball too.
Do you play basketball casually with friends, or are you part of any local leagues in Madrid? And how do you balance your time between soccer and basketball?
[Write memory debug] Existing profile: {'UserProfile': {'user_name': 'Maximo', 'user_location': 'Madrid', 'interests': ['soccer']}}
[Write memory debug] Updated profile: {'user_name': 'Maximo', 'user_location': 'Madrid', 'interests': ['soccer', 'basketball']}
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
# Namespace for the memory to save
user_id = "1"
namespace = ("memory", user_id)
existing_memory = long_term_memory.get(namespace, "user_memory")
existing_memory.value
	
	
Copied

			

				
{'user_name': 'Maximo',
'user_location': 'Madrid',
'interests': ['soccer', 'basketball']}
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
# 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": "1"}}
 
# User input 
input_messages = [HumanMessage(content="What soccer players do you recommend for me?")]
 
# Run the graph
for chunk in graph.stream({"messages": input_messages}, config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()
	
	
Copied

			

				
================================ Human Message =================================
What soccer players do you recommend for me?
[Call model debug] Existing memory: Name: Maximo
Location: Madrid
Interests: soccer, basketball
================================== Ai Message ==================================
Based on your interest in soccer, I can recommend some players who might appeal to you. Since you're from Madrid, you might already follow Real Madrid or Atlético Madrid players, but here are some recommendations:
From La Liga:
- Vinícius Júnior and Jude Bellingham (Real Madrid)
- Antoine Griezmann (Atlético Madrid)
- Robert Lewandowski (Barcelona)
- Lamine Yamal (Barcelona's young talent)
International stars:
- Kylian Mbappé
- Erling Haaland
- Mohamed Salah
- Kevin De Bruyne
You might also enjoy watching players with creative playing styles since you're interested in basketball as well, which is a sport that values creativity and flair - players like Rodrigo De Paul or João Félix.
Is there a particular league or playing style you prefer in soccer?
[Write memory debug] Existing profile: {'UserProfile': {'user_name': 'Maximo', 'user_location': 'Madrid', 'interests': ['soccer', 'basketball']}}
[Write memory debug] Updated profile: {'user_name': 'Maximo', 'user_location': 'Madrid', 'interests': ['soccer', 'basketball']}
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_core.messages import merge_message_runs
from langgraph.checkpoint.memory import MemorySaver # Short-term memory
from langgraph.store.base import BaseStore # Long-term memory
from langchain_core.runnables.config import RunnableConfig
from langgraph.store.memory import InMemoryStore
from IPython.display import Image, display
from trustcall import create_extractor
from pydantic import BaseModel, Field
import uuid
import os
import dotenv
 
dotenv.load_dotenv()
ANTHROPIC_TOKEN = os.getenv("ANTHROPIC_LANGGRAPH_API_KEY")
 
os.environ["LANGCHAIN_TRACING_V2"] = "false" # Disable LangSmith tracing
 
# Memory schema
class Memory(BaseModel):
    """A memory item representing a piece of information learned about the user."""
    content: str = Field(description="The main content of the memory. For example: User expressed interest in learning about French.")
 
# Create the LLM model
llm = ChatAnthropic(model="claude-3-7-sonnet-20250219", api_key=ANTHROPIC_TOKEN)
 
# Create the extractor
trustcall_extractor = create_extractor(
    llm,
    tools=[Memory],
    tool_choice="Memory",
    # This allows the extractor to insert new memories
    enable_inserts=True,
)
 
# Chatbot instruction
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}"""
 
# Create new memory from the chat history and any existing memory
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:"""
 
# Nodes
def call_model(state: MessagesState, config: RunnableConfig, store: BaseStore):
 
    """Load memory from the store and use it to personalize the chatbot's response."""
    
    # Get the user ID from the config
    user_id = config["configurable"]["user_id"]
 
    # Retrieve memory from the store
    namespace = ("memories", user_id)
    memories = store.search(namespace)
    print(f"	[Call model debug] Memories: {memories}")
 
    # Format the memories for the system prompt
    info = "
".join(f"- {mem.value['content']}" for mem in memories)
    system_msg = MODEL_SYSTEM_MESSAGE.format(memory=info)
 
    # Respond using memory as well as the chat history
    response = llm.invoke([SystemMessage(content=system_msg)]+state["messages"])
 
    return {"messages": response}
 
def write_memory(state: MessagesState, config: RunnableConfig, store: BaseStore):
 
    """Reflect on the chat history and save a memory to the store."""
    
    # Get the user ID from the config
    user_id = config["configurable"]["user_id"]
 
    # Define the namespace for the memories
    namespace = ("memories", user_id)
 
    # Retrieve the most recent memories for context
    existing_items = store.search(namespace)
 
    # Format the existing memories for the Trustcall extractor
    tool_name = "Memory"
    existing_memories = ([(existing_item.key, tool_name, existing_item.value)
                          for existing_item in existing_items]
                          if existing_items
                          else None
                        )
    print(f"	[Write memory debug] Existing memories: {existing_memories}")
 
    # Merge the chat history and the instruction
    updated_messages=list(merge_message_runs(messages=[SystemMessage(content=TRUSTCALL_INSTRUCTION)] + state["messages"]))
 
    # Invoke the extractor
    result = trustcall_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"),
            )
    print(f"	[Write memory debug] Saved memories: {result['responses']}")
 
# Create graph builder
graph_builder = StateGraph(MessagesState)
 
# Add nodes
graph_builder.add_node("call_model", call_model)
graph_builder.add_node("write_memory", write_memory)
 
# Connect nodes
graph_builder.add_edge(START, "call_model")
graph_builder.add_edge("call_model", "write_memory")
graph_builder.add_edge("write_memory", END)
 
# Store for long-term (across-thread) memory
long_term_memory = InMemoryStore()
 
# Checkpointer for short-term (within-thread) memory
short_term_memory = MemorySaver()
 
# Compile the graph
graph = graph_builder.compile(checkpointer=short_term_memory, store=long_term_memory)
 
display(Image(graph.get_graph().draw_mermaid_png()))
	
	
Copied

			

				
<IPython.core.display.Image object>
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
# 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": "1"}}
 
# User input 
input_messages = [HumanMessage(content="Hi, my name is Maximo")]
 
# Run the graph
for chunk in graph.stream({"messages": input_messages}, config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()
	
	
Copied

			

				
================================ Human Message =================================
Hi, my name is Maximo
[Call model debug] Memories: []
================================== Ai Message ==================================
Hello Maximo! It's nice to meet you. I'm your companion chatbot, here to chat, help answer questions, or just be someone to talk to.
I'll remember your name is Maximo for our future conversations. What would you like to talk about today? How are you doing?
[Write memory debug] Existing memories: None
[Write memory debug] Saved memories: [Memory(content="User's name is Maximo.")]
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
# User input 
input_messages = [HumanMessage(content="I like to play soccer")]
 
# Run the graph
for chunk in graph.stream({"messages": input_messages}, config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()
	
	
Copied

			

				
================================ Human Message =================================
I like to play soccer
[Call model debug] Memories: [Item(namespace=['memories', '1'], key='6d06c4f5-3a74-46b2-92b4-1e29ba128c90', value={'content': "User's name is Maximo."}, created_at='2025-05-12T18:32:38.070902+00:00', updated_at='2025-05-12T18:32:38.070903+00:00', score=None)]
================================== Ai Message ==================================
That's great to know, Maximo! Soccer is such a wonderful sport. Do you play on a team, or more casually with friends? I'd also be curious to know what position you typically play, or if you have a favorite professional team you follow. I'll remember that you enjoy soccer for our future conversations.
[Write memory debug] Existing memories: [('6d06c4f5-3a74-46b2-92b4-1e29ba128c90', 'Memory', {'content': "User's name is Maximo."})]
[Write memory debug] Saved memories: [Memory(content='User enjoys playing soccer.')]
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
# Namespace for the memory to save
user_id = "1"
namespace = ("memories", user_id)
memories = long_term_memory.search(namespace)
for m in memories:
    print(m.dict())
	
	
Copied

			

				
{'namespace': ['memories', '1'], 'key': '6d06c4f5-3a74-46b2-92b4-1e29ba128c90', 'value': {'content': "User's name is Maximo."}, 'created_at': '2025-05-12T18:32:38.070902+00:00', 'updated_at': '2025-05-12T18:32:38.070903+00:00', 'score': None}
{'namespace': ['memories', '1'], 'key': '25d2ee8c-5890-415b-85e0-d9fb0ea4cd43', 'value': {'content': 'User enjoys playing soccer.'}, 'created_at': '2025-05-12T18:32:42.558787+00:00', 'updated_at': '2025-05-12T18:32:42.558789+00:00', 'score': None}
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
for m in memories:
    print(m.value)
	
	
Copied

			

				
{'content': "User's name is Maximo."}
{'content': 'User enjoys playing soccer.'}
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
# User input 
input_messages = [HumanMessage(content="I also like to play basketball")]
 
# Run the graph
for chunk in graph.stream({"messages": input_messages}, config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()
	
	
Copied

			

				
================================ Human Message =================================
I also like to play basketball
[Call model debug] Memories: [Item(namespace=['memories', '1'], key='6d06c4f5-3a74-46b2-92b4-1e29ba128c90', value={'content': "User's name is Maximo."}, created_at='2025-05-12T18:32:38.070902+00:00', updated_at='2025-05-12T18:32:38.070903+00:00', score=None), Item(namespace=['memories', '1'], key='25d2ee8c-5890-415b-85e0-d9fb0ea4cd43', value={'content': 'User enjoys playing soccer.'}, created_at='2025-05-12T18:32:42.558787+00:00', updated_at='2025-05-12T18:32:42.558789+00:00', score=None)]
================================== Ai Message ==================================
That's awesome, Maximo! Both soccer and basketball are fantastic sports. I'll remember that you enjoy basketball as well. Do you find yourself playing one more than the other? And similar to soccer, do you play basketball with a team or more casually? Many people enjoy the different skills and dynamics each sport offers - soccer with its continuous flow and footwork, and basketball with its fast pace and shooting precision. Any favorite basketball teams you follow?
[Write memory debug] Existing memories: [('6d06c4f5-3a74-46b2-92b4-1e29ba128c90', 'Memory', {'content': "User's name is Maximo."}), ('25d2ee8c-5890-415b-85e0-d9fb0ea4cd43', 'Memory', {'content': 'User enjoys playing soccer.'})]
[Write memory debug] Saved memories: [Memory(content='User enjoys playing basketball.')]
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
# Namespace for the memory to save
user_id = "1"
namespace = ("memories", user_id)
memories = long_term_memory.search(namespace)
for m in memories:
    print(m.value)
	
	
Copied

			

				
{'content': "User's name is Maximo."}
{'content': 'User enjoys playing soccer.'}
{'content': 'User enjoys playing basketball.'}
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
# 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": "1"}}
 
# User input 
input_messages = [HumanMessage(content="What soccer players do you recommend for me?")]
 
# Run the graph
for chunk in graph.stream({"messages": input_messages}, config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()
	
	
Copied

			

				
================================ Human Message =================================
What soccer players do you recommend for me?
[Call model debug] Memories: [Item(namespace=['memories', '1'], key='6d06c4f5-3a74-46b2-92b4-1e29ba128c90', value={'content': "User's name is Maximo."}, created_at='2025-05-12T18:32:38.070902+00:00', updated_at='2025-05-12T18:32:38.070903+00:00', score=None), Item(namespace=['memories', '1'], key='25d2ee8c-5890-415b-85e0-d9fb0ea4cd43', value={'content': 'User enjoys playing soccer.'}, created_at='2025-05-12T18:32:42.558787+00:00', updated_at='2025-05-12T18:32:42.558789+00:00', score=None), Item(namespace=['memories', '1'], key='965f2e52-bea0-44d4-8534-4fce2bbc1c4b', value={'content': 'User enjoys playing basketball.'}, created_at='2025-05-12T18:33:38.613626+00:00', updated_at='2025-05-12T18:33:38.613629+00:00', score=None)]
================================== Ai Message ==================================
Hi Maximo! Since you enjoy soccer, I'd be happy to recommend some players you might find interesting to follow or learn from.
Based on your interests in both soccer and basketball, I might suggest players who are known for their athleticism and skill:
1. Lionel Messi - Widely considered one of the greatest players of all time
2. Cristiano Ronaldo - Known for incredible athleticism and dedication
3. Kylian Mbappé - Young talent with amazing speed and technical ability
4. Kevin De Bruyne - Master of passing and vision
5. Erling Haaland - Goal-scoring phenomenon
Is there a particular position or playing style you're most interested in? That would help me refine my recommendations further. I could also suggest players from specific leagues or teams if you have preferences!
[Write memory debug] Existing memories: [('6d06c4f5-3a74-46b2-92b4-1e29ba128c90', 'Memory', {'content': "User's name is Maximo."}), ('25d2ee8c-5890-415b-85e0-d9fb0ea4cd43', 'Memory', {'content': 'User enjoys playing soccer.'}), ('965f2e52-bea0-44d4-8534-4fce2bbc1c4b', 'Memory', {'content': 'User enjoys playing basketball.'})]
[Write memory debug] Saved memories: [Memory(content='User asked for soccer player recommendations, suggesting an active interest in following professional soccer beyond just playing it.')]
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
import os
import dotenv
 
dotenv.load_dotenv()
 
HUGGINGFACE_TOKEN = os.getenv("HUGGINGFACE_LANGGRAPH")
TAVILY_API_KEY = os.getenv("TAVILY_LANGGRAPH_API_KEY")
	
	
Copied

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
from typing import Annotated
 
from typing_extensions import TypedDict
 
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
 
class State(TypedDict):
    messages: Annotated[list, add_messages]
 
graph_builder = StateGraph(State)
	
	
Copied

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
from langchain_community.utilities.tavily_search import TavilySearchAPIWrapper
from langchain_community.tools.tavily_search import TavilySearchResults
 
wrapper = TavilySearchAPIWrapper(tavily_api_key=TAVILY_API_KEY)
search_tool = TavilySearchResults(api_wrapper=wrapper, max_results=2)
	
	
Copied

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
from langgraph.types import Command, interrupt
from langchain_core.tools import tool
 
@tool
def human_assistance(query: str) -&gt; str:
    """
    Request assistance from a human expert. Use this tool ONLY ONCE per conversation.
    After receiving the expert's response, you should provide an elaborated response to the user based on the information received
    based on the information received, without calling this tool again.
 
    Args:
        query: The query to ask the human expert.
 
    Returns:
        The response from the human expert.
    """
    human_response = interrupt({"query": query})
    return human_response["data"]
	
	
Copied

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
tools_list = [search_tool, human_assistance]
	
	
Copied

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
from langchain_huggingface import HuggingFaceEndpoint, ChatHuggingFace
from huggingface_hub import login
os.environ["LANGCHAIN_TRACING_V2"] = "false" # Disable LangSmith tracing
 
# Create the LLM
login(token=HUGGINGFACE_TOKEN)
MODEL = "Qwen/Qwen2.5-72B-Instruct"
model = HuggingFaceEndpoint(
    repo_id=MODEL,
    task="text-generation",
    max_new_tokens=512,
    do_sample=False,
    repetition_penalty=1.03,
)
# Create the chat model
llm = ChatHuggingFace(llm=model)
 
# Modification: tell the LLM which tools it can call
llm_with_tools = llm.bind_tools(tools_list)
 
# Define the chatbot function
def chatbot_function(state: State):
    message = llm_with_tools.invoke(state["messages"])
    assert len(message.tool_calls) &lt;= 1
    return {"messages": [message]}
 
# Add the chatbot node
graph_builder.add_node("chatbot_node", chatbot_function)
	
	
Copied