Memória de Conversação
Implemente memória de conversação em LangGraph usando o reducer add_messages, armazenando e gerenciando histórico de conversação entre turnos.
Memória de Conversação
Conversation memory is what enables a chatbot to remember what was said earlier in the conversation. LangGraph handles this naturally through persistent state and the add_messages reducer.
O Redutor add_messages
LangGraph provides a built-in add_messages reducer specifically designed for conversation message lists:
from langgraph.graph import add_messages
from typing_extensions import TypedDict, Annotated
from typing import List, Any
class ConversationState(TypedDict):
messages: Annotated[List[Any], add_messages]
context: strUnlike operator.add (which blindly concatenates), add_messages is smart:
- Appends new messages to the list
- Updates existing messages if a message with the same ID is added
- Preserves message ordering
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
# add_messages handles these intelligently:
updates = [
HumanMessage(content="Hello", id="1"),
AIMessage(content="Hi!", id="2"),
# If we add a message with the same ID, it updates in place:
AIMessage(content="Hello there!", id="2") # Replaces previous AIMessage
][!NOTA]
The add_messages reducer is preferred over operator.add for message lists because it handles deduplication and in-place updates correctly.
Adicionando Mensagens ao Estado
There are two patterns for updating messages in state:
Padrão 1: Anexar à lista existente
def chat_node(state: ConversationState) -> dict:
response = llm.invoke(state["messages"])
# Returns the new message(s) — add_messages handles appending
return {"messages": [response]}Padrão 2: Retornar todas as mensagens (também funciona)
def chat_node(state: ConversationState) -> dict:
response = llm.invoke(state["messages"])
# The reducer concatenates these with existing messages
return {"messages": state["messages"] + [response]}
# Both patterns produce the same result with add_messages[!DICA] Pattern 1 is more efficient — you only return the new messages rather than the entire list. The reducer handles the concatenation.
Exemplo Completo de Memória de Conversação
from langgraph.graph import StateGraph, START, END, add_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from typing_extensions import TypedDict, Annotated
from typing import List, Any
llm = ChatOpenAI(model="gpt-4o-mini")
class ChatState(TypedDict):
messages: Annotated[List[Any], add_messages]
user_name: str
def chatbot(state: ChatState) -> dict:
response = llm.invoke(state["messages"])
return {"messages": [response]}
# Build with persistence
checkpointer = MemorySaver()
builder = StateGraph(ChatState)
builder.add_node("chat", chatbot)
builder.add_edge(START, "chat")
builder.add_edge("chat", END)
app = builder.compile(checkpointer=checkpointer)
# Multi-turn conversation
session = {"configurable": {"thread_id": "user-1"}}
app.invoke({"messages": [HumanMessage("Hi, my name is Alice")]}, session)
app.invoke({"messages": [HumanMessage("What's my name?")]}, session)
# The bot remembers: "Your name is Alice"[!SUCESSO] With persistence + add_messages, the conversation history grows across invocations. The LLM receives the full history and maintains context.
Truncando Histórico de Conversação
Conversation history can grow beyond the LLM's context window. Trim messages before sending to the LLM:
from langchain_core.messages import trim_messages
def chat_with_trimming(state: ChatState) -> dict:
# Keep last 10 messages or 4000 tokens
trimmed = trim_messages(
state["messages"],
strategy="last",
max_tokens=4000,
token_counter=len, # Use char count as proxy
start_on="human", # Always start with a human message
include_system=True
)
response = llm.invoke(trimmed)
return {"messages": [response]}Estratégia de Truncamento Personalizada
def trim_to_token_limit(messages: list, max_tokens: int = 4000) -> list:
"""Keep messages within token budget, preserving the most recent."""
import tiktoken
enc = tiktoken.encoding_for_model("gpt-4o")
total_tokens = 0
keep = []
for msg in reversed(messages):
msg_tokens = len(enc.encode(msg.content))
if total_tokens + msg_tokens > max_tokens:
break
total_tokens += msg_tokens
keep.insert(0, msg)
# Always include the system message if present
if messages and messages[0].type == "system":
keep.insert(0, messages[0])
return keep[!AVISO] Trimming removes old context. The LLM will forget about early parts of the conversation. For very long conversations, use summary-based memory (Lesson 4).
Persistência de Mensagem do Sistema
Keep a system message at the start of the conversation:
class ChatState(TypedDict):
messages: Annotated[List[Any], add_messages]
def ensure_system_message(state: ChatState) -> dict:
"""Ensure the system message is always first."""
messages = state["messages"]
system_msg = SystemMessage(
"You are a helpful assistant. The current date is 2024."
)
if not messages or messages[0].type != "system":
return {"messages": [system_msg] + messages}
return {} # System message already presentExtração Estruturada de Memória
Extract and store structured information from conversations:
from pydantic import BaseModel, Field
class UserProfile(BaseModel):
name: str = Field(default="")
preferences: list = Field(default_factory=list)
pain_points: list = Field(default_factory=list)
class MemoryState(TypedDict):
messages: Annotated[List[Any], add_messages]
profile: UserProfile
def extract_profile(state: MemoryState) -> dict:
prompt = ChatPromptTemplate.from_messages([
("system", "Extract user info from the conversation. Update the profile:\n"
"{current_profile}"),
("human", "{last_message}")
])
parser = PydanticOutputParser(pydantic_object=UserProfile)
chain = prompt | llm | parser
last_msg = state["messages"][-1].content
profile = chain.invoke({
"current_profile": state["profile"].dict(),
"last_message": last_msg
})
return {"profile": profile}Exemplo Completo: Chatbot com Memória Aprimorada
from langgraph.graph import StateGraph, START, END, add_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from typing_extensions import TypedDict, Annotated
from typing import List, Any
llm = ChatOpenAI(model="gpt-4o-mini")
class State(TypedDict):
messages: Annotated[List[Any], add_messages]
summary: str # For long-term memory
def call_model(state: State) -> dict:
system_prompt = "You are a helpful assistant."
if state.get("summary"):
system_prompt += f"\nConversation summary: {state['summary']}"
messages = [SystemMessage(system_prompt)] + state["messages"]
response = llm.invoke(messages)
return {"messages": [response]}
checkpointer = MemorySaver()
builder = StateGraph(State)
builder.add_node("model", call_model)
builder.add_edge(START, "model")
builder.add_edge("model", END)
app = builder.compile(checkpointer=checkpointer)
session = {"configurable": {"thread_id": "memory-demo"}}
app.invoke({"messages": [HumanMessage("I love Python!")]}, session)
app.invoke({"messages": [HumanMessage("What do I love?")]}, session)
# → "You said you love Python!"Perguntas Práticas
What reducer is specifically designed for conversation message lists?
What advantage does add_messages have over operator.add for message lists?
What happens to conversation state across invocations with persistence enabled?
Why is message trimming necessary for long conversations?
What strategy does trim_messages(strategy='last') use?
When returning messages from a node with add_messages reducer, what is the most efficient pattern?
What field in a message does add_messages use for deduplication?
What is the limitation of simple message trimming?
How do you ensure a system message stays at the beginning of the message list?
What is the 'start_on' parameter in trim_messages used for?
[!SUCESSO]
Principais Conclusões
add_messagesis the smart reducer for conversation history — append, dedupe, update- Persistence + add_messages enables multi-turn conversation memory
- Trim messages to stay within LLM context windows
- Use
trim_messages()with strategy "last" for simple truncation - Efficient pattern: return only new messages, let the reducer append
- Message IDs enable in-place updates via add_messages
- System message preservation requires explicit logic
- For very long conversations, consider summary-based memory (Lesson 4)