Uso de API a corto plazo de memoria de agente con LangGraph
Los agentes a menudo necesitan preservar el contexto de trabajo reciente sin devolver la conversación completa al modelo en cada turno. Si solo mantiene el último mensaje en estado, el modelo puede perder fácilmente el seguimiento de los detalles de tareas anteriores, el progreso intermedio o el tema de thread activo.
La memoria del agente de Oracle expone dos ayudantes diferentes a corto plazo para este problema:
get_summary()devuelve un objeto de resumen cuyo valorcontentcomprime la transcripción del thread. Se prefiere cuando la compactación solo necesita compresión de transcripción.get_context_card()devuelve un objeto de tarjeta de contexto cuyo valorcontentes un bloque de contexto listo para petición de datos con resumen de thread, temas de recuperación, registros relevantes y, cuando se solicita, mensajes de conversación relevantes o recientes. Preferir cuando la compactación debe mantener el contexto consciente de la recuperación para el turno actual.
Cuando la aplicación mantiene los últimos mensajes N como mensajes originales, puede excluir esa misma cola del contenido derivado de la tarjeta de la siguiente manera:
#External-tail pattern:
#LLM Prompt = [context card, last N raw messages]
thread.get_context_card(except_last_messages=N, max_recent_messages=0)La petición de datos resultante es una lista que contiene un mensaje de tarjeta de contexto seguido de los mensajes de cola originales:
prompt_messages = [
{"role": "user", "content": "<context_card>...</context_card>"},
{"role": "user", "content": "The latest user message"},
{"role": "assistant", "content": "The latest assistant reply"},
]Es preferible aprovechar el almacenamiento en caché de peticiones de datos, ya que mantiene la misma tarjeta de contexto para varias vueltas de conversación y agrega nuevos mensajes a la petición de datos del LLM.
Alternativamente, puede hacer que la tarjeta sea autocontenida y mantener la cola cruda dentro de ella de la siguiente manera:
#Self-contained-card pattern:
#LLM Prompt = [context card only]
thread.get_context_card(
except_last_messages=N,
max_recent_messages=N,
)La petición de datos resultante es una lista con un solo mensaje: el mensaje de tarjeta de contexto, cuya sección <recent_messages> contiene la cola sin formato:
prompt_messages = [
{
"role": "user",
"content": (
"<context_card>..."
"<recent_messages>...</recent_messages>"
"</context_card>"
),
},
]Tenga en cuenta que cuando except_last_messages es distinto de cero, max_recent_messages debe ser cero o el mismo valor. Esto evita que se suministre la misma cola sin procesar tanto dentro como fuera de la tarjeta de contexto.
En esta guía, utilizará el middleware de LangGraph en torno a un agente predefinido para que la memoria del agente de Oracle pueda mantener las curvas automáticamente e inyectar una tarjeta de contexto de Oracle cuando la petición de datos en ejecución crezca demasiado. En otras palabras, el middleware compactará el indicador una vez que supere un umbral configurado. En este ejemplo, se selecciona get_context_card() porque la compactación debe conservar el contexto que tiene en cuenta la recuperación, no solo una recapitulación de transcripción.
Para ajustar qué tipos de registro aparecen en la información relevante de la tarjeta de contexto, incluidos los mensajes buscados del thread actual, consulte Personalización del contenido de la tarjeta de contexto.
Importante: los resúmenes, las tarjetas de contexto, los registros recuperados y las memorias extraídas automáticamente son texto derivado del modelo o recuperado y se deben tratar como no de confianza. Cuando la extracción o el resumen automáticos están activados, el SDK también puede reutilizar ese texto en peticiones de datos posteriores, como extracción de memoria, resumen, tarjeta de contexto o peticiones de datos de agente, antes de que la aplicación tenga la oportunidad de revisar el valor intermedio específico. Revise las salidas que consume la aplicación, evite permitir que el texto derivado de la memoria autorice acciones con privilegios y utilice memory_extraction_config=MemoryExtractionConfig(extract_memories=False) o escrituras de memoria explícitas cuando el flujo de trabajo requiera revisión antes de que el texto derivado pueda influir en la extracción posterior o la construcción del contexto.
En este tutorial, aprenderá a:
- Configurar la memoria del agente de Oracle con un
Embedder, un LLM de memoria de Oracle y, a continuación, emparejarlo con un modelo LangGraphChatOpenAI - encapsular un agente LangGraph predefinido con middleware que persiste en nuevas vueltas e inyecta la salida
get_context_card()cuando aumenta la presión del token y se activa la compactación de petición de datos - la respuesta se convierte posteriormente con contexto a corto plazo desde el thread de memoria del agente de Oracle en lugar de volver a enviar la transcripción completa
Indicación: Para la configuración del paquete, consulte Get Started with Agent Memory. Si necesita una instancia local de Oracle AI Database para este ejemplo, siga Ejecución local de Oracle AI Database.
Configurar la memoria del agente y los modelos LangGraph
Cree un cliente de memoria de agente de Oracle con una conexión o un pool de Oracle DB, configure Embedder para la búsqueda vectorial, proporcione un LLM de memoria de Oracle para la resolución de tarjetas de contexto y utilice ChatOpenAI para el agente LangGraph.
from typing import Any
import oracledb
from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, RemoveMessage
from langchain_core.messages.utils import count_tokens_approximately
from langchain_openai import ChatOpenAI
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.runtime import Runtime
from oracleagentmemory.core import MemoryExtractionConfig, SchemaPolicy
from oracleagentmemory.core.embedders.embedder import Embedder
from oracleagentmemory.core.llms.llm import Llm
from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory
embedder = Embedder(
model="YOUR_EMBEDDING_MODEL",
api_base="YOUR_EMBEDDING_BASE_URL",
api_key="YOUR_EMBEDDING_API_KEY",
)
memory_llm = Llm(
model="YOUR_MEMORY_LLM_MODEL",
api_base="YOUR_MEMORY_LLM_BASE_URL",
api_key="YOUR_MEMORY_LLM_API_KEY",
temperature=0,
)
langgraph_llm = ChatOpenAI(
model="YOUR_CHAT_MODEL",
base_url="YOUR_CHAT_BASE_URL",
api_key="YOUR_CHAT_API_KEY",
temperature=0,
)
db_pool = oracledb.SessionPool(
user="YOUR DB USER",
password="YOUR DB PASSWORD",
dsn="localhost:1521/...",
)
memory_store_id = "T_ST_MEMORY"
agent_memory = OracleAgentMemory(
connection=db_pool,
embedder=embedder,
llm=memory_llm,
schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
memory_store_id=memory_store_id,
)
thread_id = "langgraph_short_term_demo"
user_id = "user_123"
agent_id = "assistant_456"| Referencia de API: OracleAgentMemory | OracleThread |
Configuración de Middleware y un Agente Predefinido
El middleware mantiene al nuevo usuario y el asistente se convierte en memoria del agente de Oracle. Una vez que la petición de datos en ejecución cruza un umbral de token, compacta el estado sustituyendo la lista de mensajes completa por un mensaje oracle_context_card sintético más una pequeña cola de los últimos giros raw. Esto mantiene el estado de LangGraph compacto y, al mismo tiempo, proporciona un contexto a corto plazo compatible con la recuperación de agentes incorporada.
Esta guía utiliza la compactación basada en token, pero el mismo patrón se puede adaptar a otras políticas, como compactar cada pocos giros o después de algún otro disparador específico de la aplicación.
def _message_text(message: BaseMessage | Any) -> str:
content = getattr(message, "content", "")
if isinstance(content, str):
return content
return str(content)
def _is_context_card_message(message: BaseMessage) -> bool:
return isinstance(message, HumanMessage) and (
getattr(message, "name", None) == "memory_context_card"
)
class OracleShortTermMemoryMiddleware(AgentMiddleware):
"""Persist LangGraph turns and compact prompts with an OracleAgentMemory context card.
Notes
-----
- ``before_model()`` receives the current LangGraph message state for this turn.
After compaction, that state already includes the synthetic ``memory_context_card``
message returned by a previous ``before_model()`` call.
- The middleware strips that synthetic message back out before persisting or
measuring token usage so OracleAgentMemory only stores real user/assistant turns
and the compaction threshold is based on the organic conversation.
- When compaction triggers, the middleware replaces the message history with one
context-card message plus the most recent raw turns. On the next turn, that
same injected message is seen again and filtered out before recomputing the
next compacted prompt.
"""
def __init__(
self,
memory: OracleAgentMemory,
thread_id: str,
user_id: str,
agent_id: str,
compaction_token_trigger: int,
kept_message_count: int,
) -> None:
self._thread = memory.create_thread(
thread_id=thread_id,
user_id=user_id,
agent_id=agent_id,
memory_extraction_config=MemoryExtractionConfig(
context_summary_update_frequency=4
),
)
self._compaction_token_trigger = int(compaction_token_trigger)
self._kept_message_count = int(kept_message_count)
self._persisted_message_ids: set[str] = set()
def before_model(
self,
state: dict[str, Any],
runtime: Runtime[Any],
) -> dict[str, Any] | None:
del runtime
messages = list(state["messages"])
#^ This will contain the context card message once the compaction occurs
raw_messages = [message for message in messages if not _is_context_card_message(message)]
self._persist_new_messages(raw_messages)
#we exclude the context card from the token counting
if count_tokens_approximately(raw_messages) < self._compaction_token_trigger:
return None
#External-tail pattern: exclude the raw tail from derived card
#content, then append the original messages separately.
context_card = self._thread.get_context_card(
except_last_messages=self._kept_message_count,
max_recent_messages=0,
).content
#Self-contained-card alternative: use the matching count for
#max_recent_messages and omit the raw-message tail from the return
#value.
#context_card = self._thread.get_context_card(
#except_last_messages=self._kept_message_count,
#max_recent_messages=self._kept_message_count,
#).content
if not context_card:
context_card = "<context_card>\n No relevant short-term context yet.\n</context_card>"
return {
"messages": [
RemoveMessage(id=REMOVE_ALL_MESSAGES), # Clear existing message state.
HumanMessage(content=context_card, name="memory_context_card"),
*raw_messages[-self._kept_message_count :],
]
}
def _persist_new_messages(self, messages: list[BaseMessage]) -> None:
persisted: list[dict[str, str]] = []
for message in messages:
#Persist only the conversational roles that map directly to short-
#term memory turns. Tool/system/synthetic messages are skipped here.
role = (
"user"
if isinstance(message, HumanMessage)
else "assistant" if isinstance(message, AIMessage) else None
)
if role is None:
continue
content = _message_text(message).strip()
if not content:
continue
#LangGraph messages usually have stable IDs. When they do not, fall back
#to a content-derived key so the same turn is not persisted repeatedly if
#the caller reuses the returned message list across later invocations.
message_id = str(getattr(message, "id", "") or f"{role}:{hash(content)}")
if message_id in self._persisted_message_ids:
continue
#Track what this middleware instance has already written so each real turn
#is added to Oracle once even though later turns may still carry the same
#messages in the LangGraph state.
self._persisted_message_ids.add(message_id)
persisted.append({"role": role, "content": content})
if persisted:
self._thread.add_messages(persisted)
short_term_middleware = OracleShortTermMemoryMiddleware(
memory=agent_memory,
thread_id=thread_id,
user_id=user_id,
agent_id=agent_id,
compaction_token_trigger=6000,
kept_message_count=3,
)
agent = create_agent(
model=langgraph_llm,
tools=[],
middleware=[short_term_middleware],
)Respuesta posteriormente con contexto inyectado en middleware
El usuario Append recurre a la lista de mensajes en ejecución del agente predefinido y permite que el middleware decida cuándo inyectar una tarjeta de contexto. En el momento en que llegue el turno posterior, el agente puede responder desde un estado compacto que aún contenga el contexto a corto plazo de la memoria del agente de Oracle. El ejemplo imprime la tarjeta de contexto inyectada e incluye una muestra recortada para que pueda inspeccionar qué compactación se insertó en la petición de datos sin volcar el bloque completo en línea.
messages: list[BaseMessage] = []
def print_current_context_card(messages: list[BaseMessage]) -> None:
for message in messages:
if _is_context_card_message(message):
print(_message_text(message))
return
print("<context_card>\n No injected context card yet.\n</context_card>")
def run_turn(user_text: str) -> str:
messages.append(HumanMessage(content=user_text))
result = agent.invoke({"messages": messages})
messages[:] = list(result["messages"])
assistant_message = next(
message for message in reversed(messages) if isinstance(message, AIMessage)
)
return _message_text(assistant_message)
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_current_context_card(messages)
#<context_card>
#<topics>
#<topic>invoice reconciliation migration</topic>
#<topic>ledger enrichment failure</topic>
#...
#</topics>
#<summary>
#Maya is migrating the nightly invoice reconciliation workflow from cron jobs
#to LangGraph. The failing step is ledger enrichment after reconciliation.
#</summary>
#...
#</context_card>
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.Conclusión
En esta guía hemos aprendido a distinguir get_summary() de get_context_card(), a configurar el contexto a corto plazo de la memoria del agente de Oracle en torno a un agente LangGraph predefinido y a permitir que el middleware compacte la petición de datos con una tarjeta de contexto cuando la conversación sea demasiado grande para mantenerla literal.
→ Después de haber aprendido a agregar contexto de thread a corto plazo a un flujo de LangGraph, ahora puede continuar con Use Oracle Agent Memory with LangGraph.
Código Completo
Copie el código completo que aparece a continuación.
#Copyright © 2026 Oracle and/or its affiliates.
#This software is under the Apache License 2.0
#(LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0) or Universal Permissive License
#(UPL) 1.0 (LICENSE-UPL or https://oss.oracle.com/licenses/upl), at your option.
#Oracle Agent Memory Code Example - LangGraph Short-Term Memory
#--------------------------------------------------------------
##Configure Oracle Agent Memory and LangGraph models for short term context
from typing import Any
import oracledb
from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, RemoveMessage
from langchain_core.messages.utils import count_tokens_approximately
from langchain_openai import ChatOpenAI
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.runtime import Runtime
from oracleagentmemory.core import MemoryExtractionConfig, SchemaPolicy
from oracleagentmemory.core.embedders.embedder import Embedder
from oracleagentmemory.core.llms.llm import Llm
from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory
embedder = Embedder(
model="YOUR_EMBEDDING_MODEL",
api_base="YOUR_EMBEDDING_BASE_URL",
api_key="YOUR_EMBEDDING_API_KEY",
)
memory_llm = Llm(
model="YOUR_MEMORY_LLM_MODEL",
api_base="YOUR_MEMORY_LLM_BASE_URL",
api_key="YOUR_MEMORY_LLM_API_KEY",
temperature=0,
)
langgraph_llm = ChatOpenAI(
model="YOUR_CHAT_MODEL",
base_url="YOUR_CHAT_BASE_URL",
api_key="YOUR_CHAT_API_KEY",
temperature=0,
)
db_pool = oracledb.SessionPool(
user="YOUR DB USER",
password="YOUR DB PASSWORD",
dsn="localhost:1521/...",
)
memory_store_id = "T_ST_MEMORY"
agent_memory = OracleAgentMemory(
connection=db_pool,
embedder=embedder,
llm=memory_llm,
schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
memory_store_id=memory_store_id,
)
thread_id = "langgraph_short_term_demo"
user_id = "user_123"
agent_id = "assistant_456"
##Configure short term memory middleware and a prebuilt LangGraph agent
def _message_text(message: BaseMessage | Any) -> str:
content = getattr(message, "content", "")
if isinstance(content, str):
return content
return str(content)
def _is_context_card_message(message: BaseMessage) -> bool:
return isinstance(message, HumanMessage) and (
getattr(message, "name", None) == "memory_context_card"
)
class OracleShortTermMemoryMiddleware(AgentMiddleware):
"""Persist LangGraph turns and compact prompts with an OracleAgentMemory context card.
Notes
-----
- ``before_model()`` receives the current LangGraph message state for this turn.
After compaction, that state already includes the synthetic ``memory_context_card``
message returned by a previous ``before_model()`` call.
- The middleware strips that synthetic message back out before persisting or
measuring token usage so OracleAgentMemory only stores real user/assistant turns
and the compaction threshold is based on the organic conversation.
- When compaction triggers, the middleware replaces the message history with one
context-card message plus the most recent raw turns. On the next turn, that
same injected message is seen again and filtered out before recomputing the
next compacted prompt.
"""
def __init__(
self,
memory: OracleAgentMemory,
thread_id: str,
user_id: str,
agent_id: str,
compaction_token_trigger: int,
kept_message_count: int,
) -> None:
self._thread = memory.create_thread(
thread_id=thread_id,
user_id=user_id,
agent_id=agent_id,
memory_extraction_config=MemoryExtractionConfig(
context_summary_update_frequency=4
),
)
self._compaction_token_trigger = int(compaction_token_trigger)
self._kept_message_count = int(kept_message_count)
self._persisted_message_ids: set[str] = set()
def before_model(
self,
state: dict[str, Any],
runtime: Runtime[Any],
) -> dict[str, Any] | None:
del runtime
messages = list(state["messages"])
#^ This will contain the context card message once the compaction occurs
raw_messages = [message for message in messages if not _is_context_card_message(message)]
self._persist_new_messages(raw_messages)
#we exclude the context card from the token counting
if count_tokens_approximately(raw_messages) < self._compaction_token_trigger:
return None
#External-tail pattern: exclude the raw tail from derived card
#content, then append the original messages separately.
context_card = self._thread.get_context_card(
except_last_messages=self._kept_message_count,
max_recent_messages=0,
).content
#Self-contained-card alternative: use the matching count for
#max_recent_messages and omit the raw-message tail from the return
#value.
#context_card = self._thread.get_context_card(
#except_last_messages=self._kept_message_count,
#max_recent_messages=self._kept_message_count,
#).content
if not context_card:
context_card = "<context_card>\n No relevant short-term context yet.\n</context_card>"
return {
"messages": [
RemoveMessage(id=REMOVE_ALL_MESSAGES), # Clear existing message state.
HumanMessage(content=context_card, name="memory_context_card"),
*raw_messages[-self._kept_message_count :],
]
}
def _persist_new_messages(self, messages: list[BaseMessage]) -> None:
persisted: list[dict[str, str]] = []
for message in messages:
#Persist only the conversational roles that map directly to short-
#term memory turns. Tool/system/synthetic messages are skipped here.
role = (
"user"
if isinstance(message, HumanMessage)
else "assistant" if isinstance(message, AIMessage) else None
)
if role is None:
continue
content = _message_text(message).strip()
if not content:
continue
#LangGraph messages usually have stable IDs. When they do not, fall back
#to a content-derived key so the same turn is not persisted repeatedly if
#the caller reuses the returned message list across later invocations.
message_id = str(getattr(message, "id", "") or f"{role}:{hash(content)}")
if message_id in self._persisted_message_ids:
continue
#Track what this middleware instance has already written so each real turn
#is added to Oracle once even though later turns may still carry the same
#messages in the LangGraph state.
self._persisted_message_ids.add(message_id)
persisted.append({"role": role, "content": content})
if persisted:
self._thread.add_messages(persisted)
short_term_middleware = OracleShortTermMemoryMiddleware(
memory=agent_memory,
thread_id=thread_id,
user_id=user_id,
agent_id=agent_id,
compaction_token_trigger=6000,
kept_message_count=3,
)
agent = create_agent(
model=langgraph_llm,
tools=[],
middleware=[short_term_middleware],
)
##Answer later turns with the middleware backed agent
messages: list[BaseMessage] = []
def print_current_context_card(messages: list[BaseMessage]) -> None:
for message in messages:
if _is_context_card_message(message):
print(_message_text(message))
return
print("<context_card>\n No injected context card yet.\n</context_card>")
def run_turn(user_text: str) -> str:
messages.append(HumanMessage(content=user_text))
result = agent.invoke({"messages": messages})
messages[:] = list(result["messages"])
assistant_message = next(
message for message in reversed(messages) if isinstance(message, AIMessage)
)
return _message_text(assistant_message)
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_current_context_card(messages)
#<context_card>
#<topics>
#<topic>invoice reconciliation migration</topic>
#<topic>ledger enrichment failure</topic>
#...
#</topics>
#<summary>
#Maya is migrating the nightly invoice reconciliation workflow from cron jobs
#to LangGraph. The failing step is ledger enrichment after reconciliation.
#</summary>
#...
#</context_card>
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.