Uso de la memoria del agente con LangGraph
LangGraph le ayuda a crear flujos de trabajo de agentes que transfieren mensajes y estados entre pasos en una sola ejecución. Sin embargo, muchas aplicaciones también necesitan contexto de usuario para persistir después de que termine esa ejecución o conversación. Oracle AI Agent Memory almacena este contexto de larga duración para que una ejecución posterior de LangGraph pueda recuperarlo y utilizarlo.
En esta guía, aprenderá a conectar la memoria del agente a LangGraph de dos formas:
- un agente ReAct predefinido que llama a una herramienta de búsqueda de memoria cuando es necesario;
- un flujo personalizado creado con
StateGraph(MessagesState).
Nota: Utilice esta integración cuando un agente necesite recordar información más allá de una única ejecución de LangGraph, como preferencias de usuario, detalles de soporte o instrucciones permanentes. Utilice el estado LangGraph solo cuando el contexto solo importe durante la ejecución o conversación actual.
Indicación: Para la configuración del paquete, consulte Get Started with Agent Memory. Para aprender a configurar una instancia local de Oracle AI Database para este ejemplo, siga Ejecución local de Oracle AI Database.
Configuración de memoria de agente
Comience por configurar el cliente de memoria del agente de Oracle, un modelo de chat LangGraph y una herramienta search_memory reutilizable. El cliente de memoria del agente de Oracle también utiliza su propio LLM para extraer periódicamente memorias duraderas de mensajes de thread recientes, mientras que el modelo LangGraph maneja las respuestas del agente y el uso de herramientas.
import oracledb
from typing import Annotated
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, MessagesState, StateGraph
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_API_BASE",
api_key="YOUR_EMBEDDING_API_KEY",
)
llm = Llm(
model="gpt-4.1-mini",
api_key="YOUR_OPENAI_API_KEY",
)
db_pool = oracledb.SessionPool(
user="YOUR DB USER",
password="YOUR DB PASSWORD",
dsn="localhost:1521/...",
)
memory_store_id = "T_LANGGRAPH"
langgraph_llm = ChatOpenAI(
model="gpt-4.1-mini",
api_key="YOUR_OPENAI_API_KEY",
)
#Keep these identifiers stable for the same assistant and end user so memory
#is scoped consistently across threads and sessions.
agent_id = "support_agent"
user_id = "user_123"
agent_memory = OracleAgentMemory(
connection=db_pool,
embedder=embedder,
llm=llm,
schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
memory_store_id=memory_store_id,
)
@tool
def search_memory(
query: Annotated[str, "Question to search in Oracle Agent Memory"],
) -> Annotated[str, "Top matching durable memory content"]:
"""Search Oracle Agent Memory for durable user facts relevant to the current request."""
results = agent_memory.search(
query=query,
user_id=user_id,
agent_id=agent_id,
max_results=3,
record_types=["memory"],
)
if not results:
return "No relevant memory found."
return "\n".join(result.content for result in results)
def _latest_user_message(state: MessagesState) -> str:
for message in reversed(state["messages"]):
if getattr(message, "type", None) == "human":
return str(message.content)
if getattr(message, "role", None) == "user":
return str(message.content)
return ""
def _build_memory_context(query: str) -> str:
results = agent_memory.search(
query=query,
user_id=user_id,
agent_id=agent_id,
max_results=3,
record_types=["memory"],
)
memory_context = "\n".join(f"- {result.content}" for result in results)
return memory_context or "- No relevant memory found."| Referencia de API: OracleAgentMemory | OracleThread |
Agente de React predefinido
LangChain proporciona un agente de estilo ReAct predefinido sobre el tiempo de ejecución de LangGraph. Puede exponer search_memory como una de sus herramientas y permitir que el agente decida cuándo se debe consultar la memoria duradera.
Configuración del agente predefinido
react_agent = create_agent(
model=langgraph_llm,
tools=[search_memory],
system_prompt=(
"You are a support agent. When the user asks about durable facts from "
"prior sessions, call the search_memory tool before answering."
),
)
react_memory_thread = agent_memory.create_thread(
thread_id="langgraph_react_memory_demo",
user_id=user_id,
agent_id=agent_id,
)Conservar contexto de usuario después de una sesión de modificación predefinida
Una vez finalizada la primera ejecución, agregue los mensajes intercambiados a la memoria del agente de Oracle y almacene cualquier hecho duradero que se deba volver a utilizar más adelante.
react_session_1 = react_agent.invoke(
{
"messages": [
HumanMessage(
content="I am John, a Python developer and I need help debugging a payment service."
)
]
}
)
react_assistant_reply = react_session_1["messages"][-1].content
print(react_assistant_reply)
#I can help with that. What error are you seeing?
#add_messages will add messages to the DB and extract memories automatically
react_memory_thread.add_messages(
[
{
"role": "user",
"content": "I am John, a Python developer and I need help debugging a payment service.",
},
{
"role": "assistant",
"content": react_assistant_reply,
},
]
)
#add_memory adds memory to the DB
react_memory_thread.add_memory("The user is John, a Python developer.")Reutilización de Memoria en una Nueva Sesión de Reacción Predefinida
Cuando se inicie una ejecución posterior, vuelva a abrir el mismo thread de memoria del agente de Oracle y deje que el agente predefinido llame a search_memory antes de responder.
react_memory_thread = agent_memory.get_thread("langgraph_react_memory_demo")
react_session_2 = react_agent.invoke(
{
"messages": [
HumanMessage(content="Who am I?")
]
}
)
react_remembered_reply = react_session_2["messages"][-1].content
print(react_remembered_reply)
#The user is John, a Python developer.Flujo personalizado
Si necesita un control más estricto sobre la orquestación, cree un flujo LangGraph personalizado e inyecte los resultados de la memoria del agente de Oracle directamente en el nodo de modelo.
Configuración del flujo personalizado
def call_model(state: MessagesState):
from langchain_core.messages import SystemMessage
query = _latest_user_message(state)
memory_context = _build_memory_context(query)
response = langgraph_llm.invoke(
[
SystemMessage(
content=(
"You are a support agent. Use the durable memory provided when it is "
"relevant to the current user request.\n\n"
f"Durable memory:\n{memory_context}"
)
),
*state["messages"],
]
)
return {"messages": [response]}
builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model)
builder.add_edge(START, "call_model")
builder.add_edge("call_model", END)
flow_graph = builder.compile()
flow_memory_thread = agent_memory.create_thread(
thread_id="langgraph_flow_memory_demo",
user_id=user_id,
agent_id=agent_id,
)Conservación del Contexto del Usuario después de una Sesión de Flujo
Una vez finalizada la primera ejecución del flujo, agregue los mensajes intercambiados a la memoria del agente de Oracle y almacene cualquier hecho duradero que se deba volver a utilizar más adelante.
flow_session_1 = flow_graph.invoke(
{
"messages": [
HumanMessage(
content="I am John, a Python developer and I need help debugging a payment service."
)
]
}
)
flow_assistant_reply = flow_session_1["messages"][-1].content
print(flow_assistant_reply)
#I can help with that. What error are you seeing?
flow_memory_thread.add_messages(
[
{
"role": "user",
"content": "I am John, a Python developer and I need help debugging a payment service.",
},
{
"role": "assistant",
"content": flow_assistant_reply,
},
]
)
flow_memory_thread.add_memory("The user is John, a Python developer.")Reutilización de Memoria en una Nueva Sesión de Flujo
Cuando se inicie una ejecución de flujo posterior, vuelva a abrir el mismo thread de memoria de Oracle Agent y permita que la memoria duradera de búsqueda de gráficos responda con el contexto de usuario anterior.
flow_memory_thread = agent_memory.get_thread("langgraph_flow_memory_demo")
flow_session_2 = flow_graph.invoke(
{
"messages": [
HumanMessage(content="Who am I?")
]
}
)
flow_remembered_reply = flow_session_2["messages"][-1].content
print(flow_remembered_reply)
#The user is John, a Python developer.Desactivar extracción automática
Si solo desea mantener mensajes y agregar memorias duraderas manualmente, cree el cliente de memoria del agente de Oracle con memory_extraction_config=MemoryExtractionConfig(extract_memories=False) y escriba las filas de memoria usted mismo.
manual_agent_memory = OracleAgentMemory(
connection=db_pool,
embedder=embedder,
memory_extraction_config=MemoryExtractionConfig(extract_memories=False),
schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
memory_store_id=memory_store_id,
)
manual_memory_thread = manual_agent_memory.create_thread(
thread_id="langgraph_manual_memory_demo",
user_id=user_id,
agent_id=agent_id,
)
manual_memory_thread.add_messages(
[
{
"role": "user",
"content": "Please remember that I prefer concise code reviews.",
},
{
"role": "assistant",
"content": "Understood. I will keep responses concise.",
},
]
)
manual_memory_thread.add_memory("The user prefers concise code reviews.")Conclusión
En esta guía hemos aprendido a conectar la memoria del agente de Oracle a LangGraph con un agente de ReAct predefinido o un flujo StateGraph(MessagesState) personalizado, a mantener los mensajes de thread después de cada sesión y a reutilizar la memoria duradera en ejecuciones posteriores.
→ Después de haber aprendido a utilizar la memoria del agente de Oracle con LangGraph, también puede que le interese Usar la memoria del agente de Oracle con WayFlow.
Código Completo
#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 - Integration with LangGraph
#-------------------------------------------------------------
##Configure shared Oracle Memory and LangGraph setup
import oracledb
from typing import Annotated
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, MessagesState, StateGraph
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_API_BASE",
api_key="YOUR_EMBEDDING_API_KEY",
)
llm = Llm(
model="gpt-4.1-mini",
api_key="YOUR_OPENAI_API_KEY",
)
db_pool = oracledb.SessionPool(
user="YOUR DB USER",
password="YOUR DB PASSWORD",
dsn="localhost:1521/...",
)
memory_store_id = "T_LANGGRAPH"
langgraph_llm = ChatOpenAI(
model="gpt-4.1-mini",
api_key="YOUR_OPENAI_API_KEY",
)
#Keep these identifiers stable for the same assistant and end user so memory
#is scoped consistently across threads and sessions.
agent_id = "support_agent"
user_id = "user_123"
agent_memory = OracleAgentMemory(
connection=db_pool,
embedder=embedder,
llm=llm,
schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
memory_store_id=memory_store_id,
)
@tool
def search_memory(
query: Annotated[str, "Question to search in Oracle Agent Memory"],
) -> Annotated[str, "Top matching durable memory content"]:
"""Search Oracle Agent Memory for durable user facts relevant to the current request."""
results = agent_memory.search(
query=query,
user_id=user_id,
agent_id=agent_id,
max_results=3,
record_types=["memory"],
)
if not results:
return "No relevant memory found."
return "\n".join(result.content for result in results)
def _latest_user_message(state: MessagesState) -> str:
for message in reversed(state["messages"]):
if getattr(message, "type", None) == "human":
return str(message.content)
if getattr(message, "role", None) == "user":
return str(message.content)
return ""
def _build_memory_context(query: str) -> str:
results = agent_memory.search(
query=query,
user_id=user_id,
agent_id=agent_id,
max_results=3,
record_types=["memory"],
)
memory_context = "\n".join(f"- {result.content}" for result in results)
return memory_context or "- No relevant memory found."
##Configure a prebuilt LangGraph ReAct agent
react_agent = create_agent(
model=langgraph_llm,
tools=[search_memory],
system_prompt=(
"You are a support agent. When the user asks about durable facts from "
"prior sessions, call the search_memory tool before answering."
),
)
react_memory_thread = agent_memory.create_thread(
thread_id="langgraph_react_memory_demo",
user_id=user_id,
agent_id=agent_id,
)
##Persist user context after a prebuilt ReAct session
react_session_1 = react_agent.invoke(
{
"messages": [
HumanMessage(
content="I am John, a Python developer and I need help debugging a payment service."
)
]
}
)
react_assistant_reply = react_session_1["messages"][-1].content
print(react_assistant_reply)
#I can help with that. What error are you seeing?
#add_messages will add messages to the DB and extract memories automatically
react_memory_thread.add_messages(
[
{
"role": "user",
"content": "I am John, a Python developer and I need help debugging a payment service.",
},
{
"role": "assistant",
"content": react_assistant_reply,
},
]
)
#add_memory adds memory to the DB
react_memory_thread.add_memory("The user is John, a Python developer.")
##Reuse memory in a new prebuilt ReAct session
react_memory_thread = agent_memory.get_thread("langgraph_react_memory_demo")
react_session_2 = react_agent.invoke(
{
"messages": [
HumanMessage(content="Who am I?")
]
}
)
react_remembered_reply = react_session_2["messages"][-1].content
print(react_remembered_reply)
#The user is John, a Python developer.
##Configure a custom LangGraph flow
def call_model(state: MessagesState):
from langchain_core.messages import SystemMessage
query = _latest_user_message(state)
memory_context = _build_memory_context(query)
response = langgraph_llm.invoke(
[
SystemMessage(
content=(
"You are a support agent. Use the durable memory provided when it is "
"relevant to the current user request.\n\n"
f"Durable memory:\n{memory_context}"
)
),
*state["messages"],
]
)
return {"messages": [response]}
builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model)
builder.add_edge(START, "call_model")
builder.add_edge("call_model", END)
flow_graph = builder.compile()
flow_memory_thread = agent_memory.create_thread(
thread_id="langgraph_flow_memory_demo",
user_id=user_id,
agent_id=agent_id,
)
##Persist user context after a flow session
flow_session_1 = flow_graph.invoke(
{
"messages": [
HumanMessage(
content="I am John, a Python developer and I need help debugging a payment service."
)
]
}
)
flow_assistant_reply = flow_session_1["messages"][-1].content
print(flow_assistant_reply)
#I can help with that. What error are you seeing?
flow_memory_thread.add_messages(
[
{
"role": "user",
"content": "I am John, a Python developer and I need help debugging a payment service.",
},
{
"role": "assistant",
"content": flow_assistant_reply,
},
]
)
flow_memory_thread.add_memory("The user is John, a Python developer.")
##Reuse memory in a new flow session
flow_memory_thread = agent_memory.get_thread("langgraph_flow_memory_demo")
flow_session_2 = flow_graph.invoke(
{
"messages": [
HumanMessage(content="Who am I?")
]
}
)
flow_remembered_reply = flow_session_2["messages"][-1].content
print(flow_remembered_reply)
#The user is John, a Python developer.
##Disable automatic memory extraction
manual_agent_memory = OracleAgentMemory(
connection=db_pool,
embedder=embedder,
memory_extraction_config=MemoryExtractionConfig(extract_memories=False),
schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
memory_store_id=memory_store_id,
)
manual_memory_thread = manual_agent_memory.create_thread(
thread_id="langgraph_manual_memory_demo",
user_id=user_id,
agent_id=agent_id,
)
manual_memory_thread.add_messages(
[
{
"role": "user",
"content": "Please remember that I prefer concise code reviews.",
},
{
"role": "assistant",
"content": "Understood. I will keep responses concise.",
},
]
)
manual_memory_thread.add_memory("The user prefers concise code reviews.")