Usar APIs de Curto Prazo de Memória do Agente com LangGraph
Os aplicativos LangGraph geralmente precisam preservar o contexto de trabalho recente sem passar a conversa completa de volta ao modelo em cada turno. Se você mantiver apenas a mensagem mais recente no estado do gráfico, o modelo poderá facilmente perder o controle dos detalhes da tarefa anterior, do progresso intermediário ou do tópico do thread ativo.
Neste artigo, você usará APIs de curto prazo da Memória do Agente dentro de um fluxo LangGraph para que o gráfico possa recuperar o contexto de thread recente sob demanda. O fluxo usa get_summary() para transportar um resumo compacto de mensagens anteriores e get_context_card() para exibir os registros mais relevantes para o último turno do usuário.
Neste artigo, você aprenderá a:
- configurar a Memória do Agente com um Embedder e um fluxo LangGraph com ChatOpenAI
- carregar get_summary(except_last=1) e get_context_card() antes de cada chamada de modelo
- responda mais tarde com o contexto de curto prazo do thread de Memória do Agente em vez da transcrição completa
Dica: Para configurar o pacote, consulte Conceitos Básicos da Memória do Agente. Se você precisar de um Oracle AI Database local para este exemplo, consulte Executar o Oracle AI Database Localmente.
Configurar Memória do Agente e LangGraph
Crie um cliente de Memória do Agente com uma conexão ou pool do Oracle DB, configure um Embedder para pesquisa vetorial e use ChatOpenAI para o nó LangGraph.
from typing import Any
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, MessagesState, StateGraph
from oracleagentmemory.core.embedders.embedder import Embedder
from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory
embedder = Embedder(
model="YOUR_EMBEDDING_MODEL",
api_base="YOUR_EMBEDDING_BASE_URL",
api_key="YOUR_EMBEDDING_API_KEY",
)
langgraph_llm = ChatOpenAI(
model="YOUR_CHAT_MODEL",
base_url="YOUR_CHAT_BASE_URL",
api_key="YOUR_CHAT_API_KEY",
temperature=0,
)
db_pool = ... #an oracledb connection or connection pool
class ShortTermState(MessagesState):
"""LangGraph state extended with Oracle Agent Memory short-term context."""
thread_summary: str
context_card: str
agent_memory = OracleAgentMemory(
connection=db_pool,
embedder=embedder,
extract_memories=False,
)
thread = agent_memory.create_thread(
thread_id="langgraph_short_term_demo",
user_id="user_123",
agent_id="assistant_456",
)Criar um Fluxo que Carregue o Contexto de Curto Prazo
Antes de cada chamada de modelo, o fluxo lê thread.get_summary(except_last=1) e thread.get_context_card() na Memória do Agente. Isso permite que o gráfico mantenha apenas a mensagem do usuário mais recente no estado LangGraph enquanto ainda recupera o contexto de trabalho recente do thread.
def _message_text(message: Any) -> str:
content = getattr(message, "content", "")
if isinstance(content, str):
return content
return str(content)
def load_short_term_context(_: ShortTermState) -> dict[str, str]:
summary_messages = thread.get_summary(except_last=1, token_budget=250)
summary_text = (
summary_messages[0].content if summary_messages else "No prior thread summary."
)
context_card = thread.get_context_card()
if not context_card:
context_card = "<context_card>\n No relevant short-term context yet.\n</context_card>"
return {
"thread_summary": summary_text,
"context_card": context_card,
}
def call_model(state: ShortTermState) -> dict[str, list[Any]]:
response = langgraph_llm.invoke(
[
SystemMessage(
content=(
"You are a helpful engineering assistant. "
"Answer in at most two short sentences. "
"Use the Oracle Agent Memory short-term context below.\n\n"
f"Thread summary:\n{state['thread_summary']}\n\n"
f"Context card:\n{state['context_card']}"
)
),
HumanMessage(content=_message_text(state["messages"][-1])),
]
)
return {"messages": [response]}
builder = StateGraph(ShortTermState)
builder.add_node("load_short_term_context", load_short_term_context)
builder.add_node("call_model", call_model)
builder.add_edge(START, "load_short_term_context")
builder.add_edge("load_short_term_context", "call_model")
builder.add_edge("call_model", END)
graph = builder.compile()Resposta Posteriormente com Resumo e Cartão de Contexto
Anexe cada mensagem do usuário e do assistente ao thread da Memória do Agente e deixe que o fluxo LangGraph responda mais tarde usando apenas a mensagem do usuário mais recente e o contexto de curto prazo carregado.
def run_turn(user_text: str) -> str:
thread.add_messages([{"role": "user", "content": user_text}])
result = graph.invoke({"messages": [HumanMessage(content=user_text)]})
assistant_text = _message_text(result["messages"][-1])
thread.add_messages([{"role": "assistant", "content": assistant_text}])
print("Thread summary:")
print(result["thread_summary"])
print("Context card:")
print(result["context_card"])
print("Assistant:")
print(assistant_text)
return assistant_text
run_turn(
"I'm Maya. I'm migrating our nightly invoice reconciliation workflow "
"from cron jobs to LangGraph."
)
run_turn("The failing step right now is ledger enrichment after reconciliation.")
final_answer = run_turn("What workflow am I migrating, which step is failing, and who am I?")
print(final_answer)Saída:
You're Maya, migrating your nightly invoice reconciliation workflow from cron jobs
to LangGraph, and the ledger-enrichment step after reconciliation is currently failing.Conclusão
Neste artigo, aprendemos a usar APIs de curto prazo da Memória do Agente dentro de um fluxo LangGraph, carregar get_summary(except_last=1) e get_context_card() antes de cada chamada de modelo e responder posteriormente com contexto de thread recente sem reenviar a transcrição completa.
Dica: Depois de aprender a adicionar contexto de thread de curto prazo a um fluxo do LangGraph, agora você pode prosseguir para Integrar Memória do Agente com o LangGraph.
Código Inteiro
#Copyright © 2026 Oracle and/or its affiliates.
#isort:skip_file
#fmt: off
#Agent Memory Code Example - LangGraph Short-Term Memory
#--------------------------------------------------------
#How to use:
#Create a new Python virtual environment and install the latest oracleagentmemory version.
#You can now run the script
#1. As a Python file:
#```bash
#python howto_shorttermmemory.py
#```
#2. As a Notebook (in VSCode):
#When viewing the file,
#- press the keys Ctrl + Enter to run the selected cell
#- or Shift + Enter to run the selected cell and move to the cell below
##Configure Oracle Agent Memory and LangGraph for short term context
#%%
from typing import Any
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, MessagesState, StateGraph
from oracleagentmemory.core.embedders.embedder import Embedder
from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory
embedder = Embedder(
model="YOUR_EMBEDDING_MODEL",
api_base="YOUR_EMBEDDING_BASE_URL",
api_key="YOUR_EMBEDDING_API_KEY",
)
langgraph_llm = ChatOpenAI(
model="YOUR_CHAT_MODEL",
base_url="YOUR_CHAT_BASE_URL",
api_key="YOUR_CHAT_API_KEY",
temperature=0,
)
db_pool = ... #an oracledb connection or connection pool
class ShortTermState(MessagesState):
"""LangGraph state extended with Oracle Agent Memory short-term context."""
thread_summary: str
context_card: str
agent_memory = OracleAgentMemory(
connection=db_pool,
embedder=embedder,
extract_memories=False,
)
thread = agent_memory.create_thread(
thread_id="langgraph_short_term_demo",
user_id="user_123",
agent_id="assistant_456",
)
##Build a LangGraph flow that loads short term context
#%%
def _message_text(message: Any) -> str:
content = getattr(message, "content", "")
if isinstance(content, str):
return content
return str(content)
def load_short_term_context(_: ShortTermState) -> dict[str, str]:
summary_messages = thread.get_summary(except_last=1, token_budget=250)
summary_text = (
summary_messages[0].content if summary_messages else "No prior thread summary."
)
context_card = thread.get_context_card()
if not context_card:
context_card = "<context_card>\n No relevant short-term context yet.\n</context_card>"
return {
"thread_summary": summary_text,
"context_card": context_card,
}
def call_model(state: ShortTermState) -> dict[str, list[Any]]:
response = langgraph_llm.invoke(
[
SystemMessage(
content=(
"You are a helpful engineering assistant. "
"Answer in at most two short sentences. "
"Use the Oracle Agent Memory short-term context below.\n\n"
f"Thread summary:\n{state['thread_summary']}\n\n"
f"Context card:\n{state['context_card']}"
)
),
HumanMessage(content=_message_text(state["messages"][-1])),
]
)
return {"messages": [response]}
builder = StateGraph(ShortTermState)
builder.add_node("load_short_term_context", load_short_term_context)
builder.add_node("call_model", call_model)
builder.add_edge(START, "load_short_term_context")
builder.add_edge("load_short_term_context", "call_model")
builder.add_edge("call_model", END)
graph = builder.compile()
##Answer a new turn with summary and context card
#%%
def run_turn(user_text: str) -> str:
thread.add_messages([{"role": "user", "content": user_text}])
result = graph.invoke({"messages": [HumanMessage(content=user_text)]})
assistant_text = _message_text(result["messages"][-1])
thread.add_messages([{"role": "assistant", "content": assistant_text}])
print("Thread summary:")
print(result["thread_summary"])
print("Context card:")
print(result["context_card"])
print("Assistant:")
print(assistant_text)
return assistant_text
run_turn(
"I'm Maya. I'm migrating our nightly invoice reconciliation workflow "
"from cron jobs to LangGraph."
)
run_turn("The failing step right now is ledger enrichment after reconciliation.")
final_answer = run_turn("What workflow am I migrating, which step is failing, and who am I?")
print(final_answer)
#You're Maya, migrating your nightly invoice reconciliation workflow from cron jobs
#to LangGraph, and the ledger-enrichment step after reconciliation is currently failing.