LangGraphでのエージェント・メモリー短期APIの使用

LangGraphアプリケーションは、多くの場合、すべてのターンで完全な会話をモデルに戻さずに、最近の作業コンテキストを保持する必要があります。グラフ状態の最新のメッセージのみを保持すると、以前のタスクの詳細、中間進行状況またはアクティブなスレッド・トピックを簡単に追跡できます。

この記事では、LangGraphフロー内でエージェント・メモリーの短期APIを使用して、グラフが最近のスレッド・コンテキストをオンデマンドでリカバリできるようにします。このフローでは、get_summary()を使用して以前のメッセージの圧縮サマリーを繰り越し、get_context_card()を使用して、最新のユーザーのターンに最も関連性の高いレコードを表示します。

この記事では、次の方法について学習します。

ヒント:パッケージの設定については、エージェント・メモリーの開始を参照してください。この例にローカルのOracle AI Databaseが必要な場合は、Run Oracle AI Database Locallyを参照してください。

エージェント・メモリーおよびLangGraphの構成

Oracle DB接続またはプールを使用してエージェント・メモリー・クライアントを作成し、ベクトル検索用にEmbedderを構成し、LangGraphノードにChatOpenAIを使用します。

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

短期コンテキストをロードするフローの構築

各モデル・コールの前に、フローはエージェント・メモリーからthread.get_summary(except_last=1)およびthread.get_context_card()を読み取ります。これにより、最新の作業コンテキストをスレッドから回復したまま、グラフは最新のユーザーメッセージのみをLangGraph状態に保つことができます。

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

サマリー・カードとコンテキスト・カードで後から回答

各ユーザーおよびアシスタント・メッセージをエージェント・メモリー・スレッドに追加し、最新のユーザー・メッセージとロードされた短期コンテキストのみを使用して、LangGraphフローが後で応答できるようにします。

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.

まとめ

この記事では、LangGraphフロー内でエージェント・メモリーの短期APIを使用する方法、各モデル・コールの前にget_summary(except_last=1)およびget_context_card()をロードする方法、および完全なトランスクリプトを再送信せずに最近のスレッド・コンテキストで応答する方法を学習しました。

ヒント:短期スレッド・コンテキストをLangGraphフローに追加する方法を学習した後、エージェント・メモリーとLangGraphの統合に進むことができます。

完全コード

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