Usa memoria agente con LangGraph

In questo articolo, imparerai come connettere la memoria agente a LangGraph in due modi:

• un agente ReAct predefinito che chiama uno strumento di ricerca della memoria quando necessario;
• un flusso personalizzato creato con StateGraph(MessagesState).

Suggerimento: per l'impostazione dei package, vedere Introduzione alla memoria agente. Se hai bisogno di un Oracle AI Database locale per questo esempio, vedere Esegui Oracle AI Database localmente.

Impostazione memoria agente

Iniziare configurando il client di memoria agente, un modello di chat LangGraph e uno strumento search_memory riutilizzabile. Il client Agent Memory utilizza anche il proprio LLM per estrarre periodicamente memorie durevoli dai messaggi di thread recenti, mentre il modello LangGraph gestisce le risposte degli agenti e l'uso degli strumenti.

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.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 = ...  #an oracledb connection or connection pool
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,
)


@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."

Agente sostitutivo predefinito

LangChain fornisce un agente in stile ReAct predefinito in aggiunta al runtime LangGraph. È possibile esporre search_memory come uno dei suoi strumenti e lasciare che l'agente decida quando eseguire query sulla memoria permanente.

Configurare l'agente precostituito

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,
)

Rendi persistente il contesto utente dopo una sessione di recupero predefinita

Al termine della prima esecuzione, aggiungere i messaggi scambiati alla memoria dell'agente e memorizzare i dati fact permanenti che devono essere riutilizzati in un secondo momento.

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.")

Riutilizza memoria in una nuova sessione di recupero predefinita

Quando viene avviata un'esecuzione successiva, riaprire lo stesso thread di memoria dell'agente e lasciare che l'agente predefinito chiami search_memory prima di rispondere.

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)

Output:

The user is John, a Python developer.

Flusso personalizzato

Se hai bisogno di un controllo più stretto sull'orchestrazione, crea un flusso LangGraph personalizzato e inserisci i risultati della memoria dell'agente direttamente nel nodo del modello.

Configurare il flusso personalizzato

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 below 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,
)

Rendi persistente il contesto utente dopo una sessione di flusso

Al termine della prima esecuzione del flusso, aggiungere i messaggi scambiati alla memoria dell'agente e memorizzare i dati fact permanenti che devono essere riutilizzati in un secondo momento.

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.")

Riutilizza memoria in una nuova sessione flusso

Quando viene avviata un'esecuzione di flusso successiva, riaprire lo stesso thread di memoria agente e consentire al grafico di cercare la memoria permanente per rispondere con il contesto utente precedente.

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)

Output:

The user is John, a Python developer.

Disabilita estrazione automatica

Se si desidera rendere persistenti solo i messaggi e aggiungere manualmente memorie permanenti, creare il client di memoria agente con extract_memories=False e scrivere le righe di memoria manualmente.

manual_agent_memory = OracleAgentMemory(
    connection=db_pool,
    embedder=embedder,
    extract_memories=False,
)
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.")

Conclusione

In questo articolo abbiamo imparato come connettere la memoria agente a LangGraph con un agente ReAct predefinito o un flusso StateGraph(MessagesState) personalizzato, rendere persistenti i messaggi thread dopo ogni sessione e riutilizzare la memoria permanente nelle esecuzioni successive.

Suggerimento: dopo aver appreso come integrare la memoria dell'agente con LangGraph, potrebbe essere interessante anche Integrare la memoria dell'agente con WayFlow.

Codice completo

#Copyright © 2026 Oracle and/or its affiliates.
#isort:skip_file
#fmt: off
#Agent Memory Code Example - Integration with LangGraph
#-------------------------------------------------------

#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 integration_with_langgraph.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 Memory and LangGraph setup

#%%
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.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 = ...  #an oracledb connection or connection pool
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,
)


@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?

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,
        },
    ]
)
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 below 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,
    extract_memories=False,
)
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.")