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

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

Oracle Agent Memoryは、この問題に対する2つの異なる短期ヘルパーを公開します。

アプリケーションが最新のNメッセージを元のメッセージとして保持する場合、カードの派生コンテンツから同じ尾を次のように除外できます。

#External-tail pattern:
#LLM Prompt = [context card, last N raw messages]
thread.get_context_card(except_last_messages=N, max_recent_messages=0)

結果のプロンプトは、1つのコンテキスト・カード・メッセージとそれに続く元のテール・メッセージを含むリストです。

prompt_messages = [
    {"role": "user", "content": "<context_card>...</context_card>"},
    {"role": "user", "content": "The latest user message"},
    {"role": "assistant", "content": "The latest assistant reply"},
]

これは、プロンプト・キャッシュを利用することをお薦めします。これは、複数の会話の交代時に同じコンテキスト・カードを保持し、LLMプロンプトに新しいメッセージを追加するためです。

または、カードを自己完結型にし、その中に生の尾を次のように保持することもできます。

#Self-contained-card pattern:
#LLM Prompt = [context card only]
thread.get_context_card(
    except_last_messages=N,
    max_recent_messages=N,
)

結果のプロンプトは、コンテキスト・カード・メッセージ(<recent_messages>セクションにRAWテールが含まれる)という1つのメッセージのみを含むリストです。

prompt_messages = [
    {
        "role": "user",
        "content": (
            "<context_card>..."
            "<recent_messages>...</recent_messages>"
            "</context_card>"
        ),
    },
]

except_last_messagesがゼロ以外の場合、max_recent_messagesはゼロまたは同じ値である必要があります。これにより、同じ生尾がコンテキストカードの内側と外側の両方に供給されるのを防ぎます。

このガイドでは、デフォルトのエージェントを中心にLangGraphミドルウェアを使用して、Oracle Agent Memoryが自動的に回転し続けるようにし、実行中のプロンプトが大きくなりすぎるとOracleコンテキスト・カードを注入します。つまり、ミドルウェアは、構成されたしきい値を渡すとプロンプトを圧縮します。この例では、get_context_card()が選択されています。これは、圧縮によって、トランスクリプト・リキャップのみでなく、検索対応のコンテキストが保持されるためです。

現在のスレッドからの検索メッセージなど、コンテキスト・カードの関連情報に表示されるレコード・タイプをチューニングするには、「コンテキスト・カード・コンテンツのカスタマイズ」を参照してください。

重要:要約、コンテキスト・カード、取得されたレコードおよび自動抽出されたメモリーは、モデル導出または取得されたテキストであり、信頼できないものとして扱われる必要があります。自動抽出または要約が有効な場合、アプリケーションが特定の中間値を確認する前に、そのテキストをメモリー抽出、要約、コンテキスト・カードまたはエージェント・プロンプトなどの後のプロンプトでSDKで再利用できます。アプリケーションが消費する出力を確認し、メモリー由来のテキストによる特権アクションの許可を回避し、導出されたテキストが後の抽出またはコンテキスト構成に影響を与える前にワークフローで確認が必要な場合に、memory_extraction_config=MemoryExtractionConfig(extract_memories=False)または明示的なメモリー書込みを使用します。

このチュートリアルでは、次の操作について学習します。

ヒント:パッケージの設定については、エージェント・メモリーのスタート・ガイドを参照してください。この例にローカルのOracle AI Databaseが必要な場合は、「Oracle AI Databaseをローカルで実行」に従います。

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

Oracle DB接続またはプールを使用してOracle Agent Memoryクライアントを作成し、ベクトル検索用にEmbedderを構成し、コンテキスト・カード解決用にOracleメモリーLLMを提供し、LangGraphエージェントにChatOpenAIを使用します。

from typing import Any

import oracledb
from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, RemoveMessage
from langchain_core.messages.utils import count_tokens_approximately
from langchain_openai import ChatOpenAI
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.runtime import Runtime

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_BASE_URL",
    api_key="YOUR_EMBEDDING_API_KEY",
)
memory_llm = Llm(
    model="YOUR_MEMORY_LLM_MODEL",
    api_base="YOUR_MEMORY_LLM_BASE_URL",
    api_key="YOUR_MEMORY_LLM_API_KEY",
    temperature=0,
)
langgraph_llm = ChatOpenAI(
    model="YOUR_CHAT_MODEL",
    base_url="YOUR_CHAT_BASE_URL",
    api_key="YOUR_CHAT_API_KEY",
    temperature=0,
)
db_pool = oracledb.SessionPool(
    user="YOUR DB USER",
    password="YOUR DB PASSWORD",
    dsn="localhost:1521/...",
)
memory_store_id = "T_ST_MEMORY"

agent_memory = OracleAgentMemory(
    connection=db_pool,
    embedder=embedder,
    llm=memory_llm,
    schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
    memory_store_id=memory_store_id,
)
thread_id = "langgraph_short_term_demo"
user_id = "user_123"
agent_id = "assistant_456"
APIリファレンス: OracleAgentMemory OracleThread

ミドルウェアおよび事前作成済エージェントの構成

ミドルウェアによって新しいユーザーが保持され、アシスタントはOracle Agent Memoryになります。実行中のプロンプトがトークンのしきい値を超えると、完全なメッセージ・リストを合成oracle_context_cardメッセージと最新のRAW回転の小さな尾に置き換えることで、状態が圧縮されます。これにより、LangGraph状態をコンパクトに保ちながら、組み込みのエージェント検索対応の短期的なコンテキストを実現できます。

このガイドではトークンベースの圧縮を使用しますが、数回転ごとに圧縮する、または他のアプリケーション固有のトリガーの後に圧縮するなど、他のポリシーにも同じパターンを適用できます。

def _message_text(message: BaseMessage | Any) -> str:
    content = getattr(message, "content", "")
    if isinstance(content, str):
        return content
    return str(content)

def _is_context_card_message(message: BaseMessage) -> bool:
    return isinstance(message, HumanMessage) and (
        getattr(message, "name", None) == "memory_context_card"
    )

class OracleShortTermMemoryMiddleware(AgentMiddleware):
    """Persist LangGraph turns and compact prompts with an OracleAgentMemory context card.

    Notes
    -----
    - ``before_model()`` receives the current LangGraph message state for this turn.
      After compaction, that state already includes the synthetic ``memory_context_card``
      message returned by a previous ``before_model()`` call.
    - The middleware strips that synthetic message back out before persisting or
      measuring token usage so OracleAgentMemory only stores real user/assistant turns
      and the compaction threshold is based on the organic conversation.
    - When compaction triggers, the middleware replaces the message history with one
      context-card message plus the most recent raw turns. On the next turn, that
      same injected message is seen again and filtered out before recomputing the
      next compacted prompt.
    """

    def __init__(
        self,
        memory: OracleAgentMemory,
        thread_id: str,
        user_id: str,
        agent_id: str,
        compaction_token_trigger: int,
        kept_message_count: int,
    ) -> None:
        self._thread = memory.create_thread(
            thread_id=thread_id,
            user_id=user_id,
            agent_id=agent_id,
            memory_extraction_config=MemoryExtractionConfig(
                context_summary_update_frequency=4
            ),
        )
        self._compaction_token_trigger = int(compaction_token_trigger)
        self._kept_message_count = int(kept_message_count)
        self._persisted_message_ids: set[str] = set()

    def before_model(
        self,
        state: dict[str, Any],
        runtime: Runtime[Any],
    ) -> dict[str, Any] | None:
        del runtime
        messages = list(state["messages"])
        #^ This will contain the context card message once the compaction occurs
        raw_messages = [message for message in messages if not _is_context_card_message(message)]
        self._persist_new_messages(raw_messages)

        #we exclude the context card from the token counting
        if count_tokens_approximately(raw_messages) < self._compaction_token_trigger:
            return None

        #External-tail pattern: exclude the raw tail from derived card
        #content, then append the original messages separately.
        context_card = self._thread.get_context_card(
            except_last_messages=self._kept_message_count,
            max_recent_messages=0,
        ).content
        #Self-contained-card alternative: use the matching count for
        #max_recent_messages and omit the raw-message tail from the return
        #value.
        #context_card = self._thread.get_context_card(
        #except_last_messages=self._kept_message_count,
        #max_recent_messages=self._kept_message_count,
        #).content
        if not context_card:
            context_card = "<context_card>\n  No relevant short-term context yet.\n</context_card>"
        return {
            "messages": [
                RemoveMessage(id=REMOVE_ALL_MESSAGES),  # Clear existing message state.
                HumanMessage(content=context_card, name="memory_context_card"),
                *raw_messages[-self._kept_message_count :],
            ]
        }

    def _persist_new_messages(self, messages: list[BaseMessage]) -> None:
        persisted: list[dict[str, str]] = []
        for message in messages:
            #Persist only the conversational roles that map directly to short-
            #term memory turns. Tool/system/synthetic messages are skipped here.
            role = (
                "user"
                if isinstance(message, HumanMessage)
                else "assistant" if isinstance(message, AIMessage) else None
            )
            if role is None:
                continue

            content = _message_text(message).strip()
            if not content:
                continue

            #LangGraph messages usually have stable IDs. When they do not, fall back
            #to a content-derived key so the same turn is not persisted repeatedly if
            #the caller reuses the returned message list across later invocations.
            message_id = str(getattr(message, "id", "") or f"{role}:{hash(content)}")
            if message_id in self._persisted_message_ids:
                continue

            #Track what this middleware instance has already written so each real turn
            #is added to Oracle once even though later turns may still carry the same
            #messages in the LangGraph state.
            self._persisted_message_ids.add(message_id)
            persisted.append({"role": role, "content": content})

        if persisted:
            self._thread.add_messages(persisted)

short_term_middleware = OracleShortTermMemoryMiddleware(
    memory=agent_memory,
    thread_id=thread_id,
    user_id=user_id,
    agent_id=agent_id,
    compaction_token_trigger=6000,
    kept_message_count=3,
)
agent = create_agent(
    model=langgraph_llm,
    tools=[],
    middleware=[short_term_middleware],
)

ミドルウェア・インジェクトされたコンテキストで後から回答

「追加」ユーザーは、事前作成エージェントの実行中のメッセージ・リストに移動し、ミドルウェアがコンテキスト・カードを注入するタイミングを決定できるようにします。後でターンが到着するまでに、エージェントはOracle Agent Memoryの短期コンテキストを含むコンパクトな状態から応答できます。この例では、注入されたコンテキスト・カードが出力され、トリミングされたサンプルが含まれているため、全ブロックをインラインでダンプせずにプロンプトに挿入された圧縮を検査できます。

messages: list[BaseMessage] = []

def print_current_context_card(messages: list[BaseMessage]) -> None:
    for message in messages:
        if _is_context_card_message(message):
            print(_message_text(message))
            return
    print("<context_card>\n  No injected context card yet.\n</context_card>")

def run_turn(user_text: str) -> str:
    messages.append(HumanMessage(content=user_text))
    result = agent.invoke({"messages": messages})
    messages[:] = list(result["messages"])
    assistant_message = next(
        message for message in reversed(messages) if isinstance(message, AIMessage)
    )
    return _message_text(assistant_message)

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_current_context_card(messages)
#<context_card>
#<topics>
#<topic>invoice reconciliation migration</topic>
#<topic>ledger enrichment failure</topic>
#...
#</topics>
#<summary>
#Maya is migrating the nightly invoice reconciliation workflow from cron jobs
#to LangGraph. The failing step is ledger enrichment after reconciliation.
#</summary>
#...
#</context_card>
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.

まとめ

このガイドでは、get_summary()とget_context_card()を区別する方法を学習し、事前構築済のLangGraphエージェントを中心にOracle Agent Memoryの短期コンテキストを構成し、会話が大きくなりすぎて冗長性を維持できない場合に、ミドルウェアがコンテキスト・カードでプロンプトを圧縮する方法を学習しました。

→LangGraphフローに短期スレッド・コンテキストを追加する方法を学習した後、「LangGraphでのOracle Agent Memoryの使用」に進むことができます。

完全コード

次の完全なコードをコピーします。

#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 - LangGraph Short-Term Memory
#--------------------------------------------------------------

##Configure Oracle Agent Memory and LangGraph models for short term context

from typing import Any

import oracledb
from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, RemoveMessage
from langchain_core.messages.utils import count_tokens_approximately
from langchain_openai import ChatOpenAI
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.runtime import Runtime

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_BASE_URL",
    api_key="YOUR_EMBEDDING_API_KEY",
)
memory_llm = Llm(
    model="YOUR_MEMORY_LLM_MODEL",
    api_base="YOUR_MEMORY_LLM_BASE_URL",
    api_key="YOUR_MEMORY_LLM_API_KEY",
    temperature=0,
)
langgraph_llm = ChatOpenAI(
    model="YOUR_CHAT_MODEL",
    base_url="YOUR_CHAT_BASE_URL",
    api_key="YOUR_CHAT_API_KEY",
    temperature=0,
)
db_pool = oracledb.SessionPool(
    user="YOUR DB USER",
    password="YOUR DB PASSWORD",
    dsn="localhost:1521/...",
)
memory_store_id = "T_ST_MEMORY"

agent_memory = OracleAgentMemory(
    connection=db_pool,
    embedder=embedder,
    llm=memory_llm,
    schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
    memory_store_id=memory_store_id,
)
thread_id = "langgraph_short_term_demo"
user_id = "user_123"
agent_id = "assistant_456"

##Configure short term memory middleware and a prebuilt LangGraph agent

def _message_text(message: BaseMessage | Any) -> str:
    content = getattr(message, "content", "")
    if isinstance(content, str):
        return content
    return str(content)

def _is_context_card_message(message: BaseMessage) -> bool:
    return isinstance(message, HumanMessage) and (
        getattr(message, "name", None) == "memory_context_card"
    )

class OracleShortTermMemoryMiddleware(AgentMiddleware):
    """Persist LangGraph turns and compact prompts with an OracleAgentMemory context card.

    Notes
    -----
    - ``before_model()`` receives the current LangGraph message state for this turn.
      After compaction, that state already includes the synthetic ``memory_context_card``
      message returned by a previous ``before_model()`` call.
    - The middleware strips that synthetic message back out before persisting or
      measuring token usage so OracleAgentMemory only stores real user/assistant turns
      and the compaction threshold is based on the organic conversation.
    - When compaction triggers, the middleware replaces the message history with one
      context-card message plus the most recent raw turns. On the next turn, that
      same injected message is seen again and filtered out before recomputing the
      next compacted prompt.
    """

    def __init__(
        self,
        memory: OracleAgentMemory,
        thread_id: str,
        user_id: str,
        agent_id: str,
        compaction_token_trigger: int,
        kept_message_count: int,
    ) -> None:
        self._thread = memory.create_thread(
            thread_id=thread_id,
            user_id=user_id,
            agent_id=agent_id,
            memory_extraction_config=MemoryExtractionConfig(
                context_summary_update_frequency=4
            ),
        )
        self._compaction_token_trigger = int(compaction_token_trigger)
        self._kept_message_count = int(kept_message_count)
        self._persisted_message_ids: set[str] = set()

    def before_model(
        self,
        state: dict[str, Any],
        runtime: Runtime[Any],
    ) -> dict[str, Any] | None:
        del runtime
        messages = list(state["messages"])
        #^ This will contain the context card message once the compaction occurs
        raw_messages = [message for message in messages if not _is_context_card_message(message)]
        self._persist_new_messages(raw_messages)

        #we exclude the context card from the token counting
        if count_tokens_approximately(raw_messages) < self._compaction_token_trigger:
            return None

        #External-tail pattern: exclude the raw tail from derived card
        #content, then append the original messages separately.
        context_card = self._thread.get_context_card(
            except_last_messages=self._kept_message_count,
            max_recent_messages=0,
        ).content
        #Self-contained-card alternative: use the matching count for
        #max_recent_messages and omit the raw-message tail from the return
        #value.
        #context_card = self._thread.get_context_card(
        #except_last_messages=self._kept_message_count,
        #max_recent_messages=self._kept_message_count,
        #).content
        if not context_card:
            context_card = "<context_card>\n  No relevant short-term context yet.\n</context_card>"
        return {
            "messages": [
                RemoveMessage(id=REMOVE_ALL_MESSAGES),  # Clear existing message state.
                HumanMessage(content=context_card, name="memory_context_card"),
                *raw_messages[-self._kept_message_count :],
            ]
        }

    def _persist_new_messages(self, messages: list[BaseMessage]) -> None:
        persisted: list[dict[str, str]] = []
        for message in messages:
            #Persist only the conversational roles that map directly to short-
            #term memory turns. Tool/system/synthetic messages are skipped here.
            role = (
                "user"
                if isinstance(message, HumanMessage)
                else "assistant" if isinstance(message, AIMessage) else None
            )
            if role is None:
                continue

            content = _message_text(message).strip()
            if not content:
                continue

            #LangGraph messages usually have stable IDs. When they do not, fall back
            #to a content-derived key so the same turn is not persisted repeatedly if
            #the caller reuses the returned message list across later invocations.
            message_id = str(getattr(message, "id", "") or f"{role}:{hash(content)}")
            if message_id in self._persisted_message_ids:
                continue

            #Track what this middleware instance has already written so each real turn
            #is added to Oracle once even though later turns may still carry the same
            #messages in the LangGraph state.
            self._persisted_message_ids.add(message_id)
            persisted.append({"role": role, "content": content})

        if persisted:
            self._thread.add_messages(persisted)

short_term_middleware = OracleShortTermMemoryMiddleware(
    memory=agent_memory,
    thread_id=thread_id,
    user_id=user_id,
    agent_id=agent_id,
    compaction_token_trigger=6000,
    kept_message_count=3,
)
agent = create_agent(
    model=langgraph_llm,
    tools=[],
    middleware=[short_term_middleware],
)

##Answer later turns with the middleware backed agent

messages: list[BaseMessage] = []

def print_current_context_card(messages: list[BaseMessage]) -> None:
    for message in messages:
        if _is_context_card_message(message):
            print(_message_text(message))
            return
    print("<context_card>\n  No injected context card yet.\n</context_card>")

def run_turn(user_text: str) -> str:
    messages.append(HumanMessage(content=user_text))
    result = agent.invoke({"messages": messages})
    messages[:] = list(result["messages"])
    assistant_message = next(
        message for message in reversed(messages) if isinstance(message, AIMessage)
    )
    return _message_text(assistant_message)

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_current_context_card(messages)
#<context_card>
#<topics>
#<topic>invoice reconciliation migration</topic>
#<topic>ledger enrichment failure</topic>
#...
#</topics>
#<summary>
#Maya is migrating the nightly invoice reconciliation workflow from cron jobs
#to LangGraph. The failing step is ledger enrichment after reconciliation.
#</summary>
#...
#</context_card>
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.