๐ฏ What You'll Learn Today
Unit 1: Graph Basics & State Management
Exercise 1.2 - "Message Memory"
Try It Yourself
๐ข Joint Initiative
This tutorial is part of a collaboration between AI Product Engineer and Nebius Academy.
Objective
Extend Exercise 1.1 to create a more sophisticated conversation manager that:
- Maintains a sliding window of recent messages
- Generates summaries of longer conversations
- Includes metadata with messages
- Follows the same conversation flow (Hello -> How are you? -> Goodbye)
Requirements
-
State Management:
- Extend the State type to include summary and window_size
- Track message metadata (timestamps, roles)
- Maintain conversation history with sliding window
-
Node Implementation:
- llm_response: Generate appropriate responses with metadata
- message_windowing: Implement sliding window logic
- summary_generation: Create summaries for longer conversations
- should_end: Determine when to end the conversation
-
Graph Structure:
- Create a pipeline of nodes for message processing
- Implement proper conversation ending condition
- Connect nodes in the correct order
from typing import Annotated, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph import StateGraph
from langgraph.graph.message import add_messages
class State(TypedDict):
Extended state for message memory management.
Attributes:
messages: List of conversation messages with metadata
summary: Current conversation summary
window_size: Maximum number of messages to keep
messages: Annotated[list[BaseMessage], add_messages]
summary: str
window_size: int
def llm_response(state: State) -> State:
Generate responses while maintaining conversation context.
Args:
state: Current conversation state with message history
Returns:
Updated state with new message and preserved context
Notes:
- Handle empty initial state
- Add appropriate metadata to messages
- Follow the Hello -> How are you? -> Goodbye flow
# TODO: Implement response generation with metadata
pass
def message_windowing(state: State) -> State:
Maintain a sliding window of recent messages.
Args:
state: Current conversation state
Returns:
State with message history trimmed to window size
# TODO: Implement sliding window logic
pass
def summary_generation(state: State) -> State:
Generate summaries for longer conversations.
Args:
state: Current conversation state
Returns:
State with updated summary if needed
# TODO: Implement summary generation
pass
def should_end(state: State) -> bool:
Determine if the conversation should end.
Args:
state: Current conversation state
Returns:
True if the conversation should end, False otherwise
# TODO: Implement ending condition
pass
# Initialize the graph
graph_builder = StateGraph(State)
TODO: Add your nodes Hint: You need llm_response, message_windowing, and summary_generation
TODO: Add your edges Hint: Consider the order of operations:
- Generate response
- Apply windowing
- Update summary
- Decide whether to continue or end
TODO: Compile the graph graph = None # Replace with proper compilation
# Default input for testing
default_input = {"messages": [], "summary": "", "window_size": 3}
# Make variables available for testing
__all__ = ["default_input", "graph"]