Usar Memória do Agente com LangGraph

Neste artigo, você aprenderá a conectar a Memória do Agente ao LangGraph de duas maneiras:

• um agente ReAct pré-criado que chama uma ferramenta de pesquisa de memória quando necessário;
• um fluxo personalizado criado com StateGraph(MessagesState).

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.

Configuração da Memória do Agente

Comece configurando o cliente de Memória do Agente, um modelo de chat LangGraph e uma ferramenta search_memory reutilizável. O cliente Agent Memory também usa seu próprio LLM para extrair periodicamente memórias duráveis de mensagens de thread recentes, enquanto o modelo LangGraph trata das respostas do agente e do uso da ferramenta.

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 ReAct Pré-Criado

LangChain fornece um agente pré-criado no estilo ReAct sobre o tempo de execução LangGraph. Você pode expor search_memory como uma de suas ferramentas e permitir que o agente decida quando a memória durável deve ser consultada.

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

Persistir contexto do usuário após uma sessão ReAct pré-criada

Após a conclusão da primeira execução, anexe as mensagens trocadas à Memória do Agente e armazene qualquer fato durável que deva ser reutilizado posteriormente.

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

Reutilizar memória em uma nova sessão ReAct pré-construída

Quando uma execução posterior for iniciada, reabra o mesmo thread de Memória do Agente e deixe o agente pré-criado chamar 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)

Saída:

The user is John, a Python developer.

Fluxo Personalizado

Se você precisar de um controle mais rígido sobre a orquestração, crie um fluxo LangGraph personalizado e injete os resultados da Memória do Agente diretamente no nó do modelo.

Configurar o Fluxo 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 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,
)

Persistir Contexto do Usuário Após uma Sessão de Fluxo

Após a conclusão da primeira execução do fluxo, anexe as mensagens trocadas à Memória do Agente e armazene qualquer fato durável que deva ser reutilizado posteriormente.

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

Reutilizar Memória em uma Nova Sessão de Fluxo

Quando uma execução de fluxo posterior for iniciada, reabra o mesmo thread de Memória do Agente e deixe que o gráfico pesquise memória durável para responder com contexto de usuário 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)

Saída:

The user is John, a Python developer.

Desativar extração automática

Se você quiser apenas persistir mensagens e adicionar memórias duráveis manualmente, crie o cliente de Memória do Agente com extract_memories=False e grave você mesmo as linhas de memória.

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

Conclusão

Neste artigo, aprendemos a conectar a Memória do Agente ao LangGraph com um agente ReAct pré-criado ou um fluxo StateGraph(MessagesState) personalizado, persistir mensagens de thread após cada sessão e reutilizar a memória durável em execuções posteriores.

Dica: Depois de aprender a integrar a Memória do Agente ao LangGraph, você também pode se interessar em Integrar Memória do Agente ao WayFlow.

Código Inteiro

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