Memória de Resumo

Implemente sumarização de conversação para agentes de longa duração, nós de memória de resumo e estratégias de truncamento para gerenciar janelas de contexto.

Quando as conversas crescem além da janela de contexto do LLM, o truncamento simples descarta contexto valioso. A Sumarização preserva as informações-chave em uma forma comprimida, mantendo a conversa gerenciável.

O Problema com Conversas Longas

LLMs têm limites de tokens. Quando uma conversa excede o limite:

Truncamento: Descarta as mensagens iniciais: perde contexto
Erro: Janela de contexto excedida: a solicitação falha

A sumarização resolve isso comprimindo o histórico em um resumo e mantendo os detalhes importantes.

Arquitetura de Memória de Resumo

Conversation Flow:
[Message 1] → [Message 2] → ... → [Message 10] → [Summarize]
                                                       ↓
[Summary] ← [Message 11] → [Message 12] → ... → [Message 20] → [Update Summary]
                                                                    ↓
[Updated Summary] ← ...

Um nó de resumo é acionado periodicamente para condensar a conversa.

Nó de Resumo Básico

python

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langgraph.graph import StateGraph, START, END, add_messages
from langgraph.checkpoint.memory import MemorySaver
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

def summarize_conversation(state: State) -> dict:
    current_summary = state.get("summary", "No summary yet.")

    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are a conversation summarizer. Given the current "
                   "summary and the new messages, create an updated summary. "
                   "Keep it concise but include all key information.\n\n"
                   "Current summary:\n{summary}"),
        ("human", "New messages to incorporate:\n{messages}")
    ])

    chain = prompt | llm | StrOutputParser()
    recent = state["messages"][-4:]  # Last 4 messages
    messages_text = "\n".join(f"{m.type}: {m.content}" for m in recent)

    new_summary = chain.invoke({
        "summary": current_summary,
        "messages": messages_text
    })

    return {"summary": new_summary}

ℹ️Nota

O nó de resumo lê o resumo atual e as mensagens recentes e produz um resumo atualizado. Essa abordagem incremental é mais eficiente do que resumir a conversa inteira novamente.

Quando Sumarizar: Estratégias de Ativação

Estratégia 1: Limiar de Contagem de Mensagens

python

def should_summarize(state: State) -> str:
    if len(state["messages"]) > 10:  # Summarize every 10 messages
        return "summarize"
    return "respond"

# In graph:
builder.add_conditional_edges(
    "check",
    should_summarize,
    {
        "summarize": "summarize_node",
        "respond": "chat_node"
    }
)

Estratégia 2: Limiar de Contagem de Tokens

python

import tiktoken

def should_summarize_by_tokens(state: State) -> str:
    enc = tiktoken.encoding_for_model("gpt-4o")
    total = sum(len(enc.encode(m.content)) for m in state["messages"])

    if total > 3000:  # Summarize if over 3000 tokens
        return "summarize"
    return "respond"

Estratégia 3: Periódico (A Cada N Turnos)

python

def should_summarize_periodic(state: State) -> str:
    # Summarize every 5 turns
    if len(state["messages"]) >= 5 and len(state["messages"]) % 5 == 0:
        return "summarize"
    return "respond"

✅Sucesso

Escolha sua estratégia de acionamento com base no tamanho esperado da conversa e das mensagens. A baseada em tokens é a mais precisa; a baseada em contagem é a mais simples.

Agente de Sumarização Completo

python

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 langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from typing_extensions import TypedDict, Annotated
from typing import List, Any

llm = ChatOpenAI(model="gpt-4o-mini")

class AgentState(TypedDict):
    messages: Annotated[List[Any], add_messages]
    summary: str

def chat_node(state: AgentState) -> dict:
    system_msg = "You are a helpful assistant."
    if state.get("summary"):
        system_msg += f"\n\nSummary of conversation:\n{state['summary']}"

    messages = [SystemMessage(system_msg)] + state["messages"]
    response = llm.invoke(messages)
    return {"messages": [response]}

def summarize_node(state: AgentState) -> dict:
    prompt = ChatPromptTemplate.from_messages([
        ("system", "Update the conversation summary. Include all important "
                   "information from the new messages.\n\n"
                   "Current summary: {summary}"),
        ("human", "New messages:\n{messages}")
    ])
    chain = prompt | llm | StrOutputParser()
    recent = state["messages"][-6:]
    msgs = "\n".join(f"{m.type}: {m.content}" for m in recent)
    return {"summary": chain.invoke({"summary": state.get("summary", ""), "messages": msgs})}

def router(state: AgentState) -> str:
    if len(state["messages"]) > 10:
        return "summarize"
    return "respond"

# Build graph
builder = StateGraph(AgentState)
builder.add_node("chat", chat_node)
builder.add_node("summarize", summarize_node)

builder.add_edge(START, "chat")
builder.add_conditional_edges("chat", router, {
    "summarize": "summarize",
    "respond": END
})
builder.add_edge("summarize", END)

app = builder.compile(checkpointer=MemorySaver())

# Usage
session = {"configurable": {"thread_id": "summary-test"}}
for i in range(15):
    app.invoke({"messages": [HumanMessage(f"Message number {i}")]}, session)

# Check the summary
state = app.get_state(session)
print(state.values.get("summary"))

💡Dica

No agente completo, o resumo é injetado no prompt de sistema a cada turno. Isso significa que o LLM sempre tem acesso ao contexto comprimido das partes iniciais da conversa.

Truncando Após a Sumarização

Depois de resumir, você pode truncar a lista de mensagens para economizar tokens:

python

def summarize_and_truncate(state: AgentState) -> dict:
    # 1. Generate updated summary
    prompt = ChatPromptTemplate.from_messages([
        ("system", "Summarize:\nCurrent: {summary}\nNew: {messages}"),
        ("human", "Create updated summary.")
    ])
    chain = prompt | llm | StrOutputParser()
    recent = state["messages"][-6:]
    msgs = "\n".join(f"{m.type}: {m.content}" for m in recent)
    new_summary = chain.invoke({"summary": state.get("summary", ""), "messages": msgs})

    # 2. Truncate — keep only last 4 messages
    truncated = state["messages"][-4:]

    return {
        "summary": new_summary,
        "messages": truncated  # Replace messages with short history
    }

⚠️Aviso

Quando você substitui messages por uma lista truncada, as mensagens antigas desaparecem permanentemente (exceto o que o resumo capturou). Garanta que seu resumo seja suficientemente abrangente.

Resumo Contínuo (Incremental)

Em vez de resumir tudo de uma vez, mantenha um resumo contínuo:

python

def update_running_summary(state: AgentState) -> dict:
    existing = state.get("summary", "")
    last_exchange = state["messages"][-2:]  # Last user + assistant turn

    prompt = ChatPromptTemplate.from_messages([
        ("system", "Incorporate this exchange into the running summary. "
                   "Keep the summary under 200 words.\n\n"
                   "Current summary: {existing}"),
        ("human", "Exchange:\n{exchange}")
    ])

    chain = prompt | llm | StrOutputParser()
    exchange_text = "\n".join(f"{m.type}: {m.content}" for m in last_exchange)
    new_summary = chain.invoke({"existing": existing, "exchange": exchange_text})

    return {"summary": new_summary}

Resumo com Extração de Entidades

Vá além do resumo simples: extraia entidades estruturadas:

python

from pydantic import BaseModel, Field

class ConversationMemory(BaseModel):
    summary: str = Field(description="Conversation summary")
    user_name: str = Field(default="")
    preferences: List[str] = Field(default_factory=list)
    facts: List[str] = Field(default_factory=list)

def extract_memory(state: AgentState) -> dict:
    parser = PydanticOutputParser(pydantic_object=ConversationMemory)

    prompt = ChatPromptTemplate.from_messages([
        ("system", "Extract conversation memory. {format_instructions}"),
        ("human", "History:\n{messages}")
    ])

    chain = prompt | llm | parser
    msgs_text = "\n".join(f"{m.type}: {m.content}" for m in state["messages"])
    memory = chain.invoke({"messages": msgs_text, "format_instructions": parser.get_format_instructions()})

    return {"summary": memory.summary}

Comparação: Sumarização vs Truncamento

Aspecto	Truncamento	Sumarização
O que faz	Descarta mensagens antigas	Comprime o histórico em texto
Retenção de informação	Perde tudo o que foi descartado	Retém as informações-chave no resumo
Eficiência de tokens	Excelente	Boa (o resumo consome alguns tokens)
Conhecimento do LLM	Vê apenas o histórico recente	Vê o histórico completo comprimido
Implementação	Simples (slicing)	Moderada (chamada de LLM necessária)
Custo	Gratuita	Chamada de LLM por sumarização