Kurzfristige Agent-Speicher-APIs mit LangGraph verwenden

LangGraph-Anwendungen müssen häufig den aktuellen Arbeitskontext beibehalten, ohne die vollständige Konversation bei jeder Kurve an das Modell zurückzugeben. Wenn Sie nur die neueste Nachricht im Diagrammstatus behalten, kann das Modell den Überblick über vorherige Aufgabendetails, den Zwischenfortschritt oder das aktive Threadthema leicht verlieren.

In diesem Artikel verwenden Sie kurzzeitige Agent Memory-APIs innerhalb eines LangGraph-Ablaufs, damit das Diagramm den aktuellen Threadkontext bei Bedarf wiederherstellen kann. Der Ablauf verwendet get_summary(), um eine kompakte Übersicht früherer Nachrichten zu übertragen, und get_context_card(), um die Datensätze anzuzeigen, die für den letzten Benutzerzug am relevantesten sind.

In diesem Artikel erfahren Sie, wie Sie:

Tipp: Informationen zum Packagesetup finden Sie unter Erste Schritte mit Agent-Speicher. Wenn Sie für dieses Beispiel eine lokale Oracle AI Database benötigen, lesen Sie Oracle AI Database lokal ausführen.

Agent-Speicher und LangGraph konfigurieren

Create an Agent Memory client with an Oracle DB connection or pool, configure a Embedder for vector search, and use ChatOpenAI for the LangGraph node.

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

Ablauf erstellen, der kurzfristigen Kontext lädt

Vor jedem Modellaufruf liest der Ablauf thread.get_summary(except_last=1) und thread.get_context_card() aus dem Agent-Speicher. Dadurch kann das Diagramm nur die letzte Benutzernachricht im Status "LangGraph" beibehalten und gleichzeitig den letzten Arbeitskontext aus dem Thread wiederherstellen.

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

Antwort in einem späteren Turn mit Übersichts- und Kontextkarte

Hängen Sie jede Benutzer- und Assistenznachricht an den Agent-Speicher-Thread an, und lassen Sie den LangGraph-Ablauf später antworten, indem Sie nur die letzte Benutzernachricht plus den geladenen Kurzzeitkontext verwenden.

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)

Ausgabe:

You're Maya, migrating your nightly invoice reconciliation workflow from cron jobs
    to LangGraph, and the ledger-enrichment step after reconciliation is currently failing.

Schlussfolgerung

In diesem Artikel haben wir gelernt, wie Sie Agent Memory-Kurzzeit-APIs innerhalb eines LangGraph-Ablaufs verwenden, get_summary(except_last=1) und get_context_card() vor jedem Modellaufruf laden und später mit dem aktuellen Threadkontext antworten, ohne das vollständige Transkript erneut zu senden.

Tipp: Nachdem Sie gelernt haben, wie Sie einem LangGraph-Ablauf Kurzzeit-Thread-Kontext hinzufügen, können Sie jetzt mit Agent-Speicher mit LangGraph integrieren fortfahren.

Vollständiger Code

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