LangGraph에서 에이전트 메모리 단기 API 사용

LangGraph 응용 프로그램은 종종 모든 차례에 모델로 전체 대화를 다시 전달하지 않고 최근 작업 컨텍스트를 보존해야합니다.

에이전트 메모리는 이 문제에 대해 두 개의 서로 다른 단기 도우미를 노출합니다.

이 문서에서는 사전 구축된 에이전트 주위에 LangGraph 미들웨어를 사용하므로 실행 중인 프롬프트가 너무 커지면 에이전트 메모리가 자동으로 회전을 지속하고 Oracle 컨텍스트 카드를 삽입할 수 있습니다. 미들웨어는 구성된 임계값을 통과한 후 프롬프트를 압축합니다. 이 예제에서는 압축을 통해 검색 인식 컨텍스트뿐만 아니라 transcript 요약도 보존하므로 get_context_card()를 선택합니다.

경고: 요약, 컨텍스트 카드, 검색된 레코드 및 자동으로 추출된 메모리는 모델 파생 또는 검색된 텍스트이므로 신뢰할 수 없는 것으로 처리되어야 합니다. 자동 추출 또는 요약이 사용으로 설정된 경우 응용 프로그램이 특정 중간 값을 검토할 수 있기 전에 나중에 메모리 추출, 요약, 컨텍스트 카드 또는 에이전트 프롬프트와 같은 프롬프트에서 SDK에서 해당 텍스트를 재사용할 수도 있습니다. 애플리케이션이 소비하는 출력을 검토하고, 메모리에서 파생된 텍스트가 권한 있는 작업을 승인하지 않도록 하고, 파생된 텍스트가 나중에 추출 또는 컨텍스트 구성에 영향을 줄 수 있기 전에 워크플로우에 검토가 필요한 경우 extract_memories=False 또는 명시적 메모리 쓰기를 사용합니다.

이 문서에서는 다음 방법을 배우게 됩니다.

참고: 패키지 설정은 에이전트 메모리 시작하기를 참조하십시오. 이 예에 대한 로컬 Oracle AI Database가 필요한 경우 Run Oracle AI Database Locally를 참조하십시오.

에이전트 메모리 및 LangGraph 모델 구성

Oracle DB 접속 또는 풀을 사용하여 에이전트 메모리 클라이언트를 생성하고, 벡터 검색을 위해 Embedder를 구성하고, 컨텍스트 카드 분석을 위해 Oracle 메모리 LLM을 제공하고, LangGraph 에이전트에 ChatOpenAI를 사용합니다.

from typing import Any

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


agent_memory = OracleAgentMemory(
    connection=db_pool,
    embedder=embedder,
    llm=memory_llm,
)
thread_id = "langgraph_short_term_demo"
user_id = "user_123"
agent_id = "assistant_456"

미들웨어 및 사전 구축된 에이전트 구성

미들웨어는 새 사용자를 유지하며 도우미는 에이전트 메모리로 바뀝니다. 실행 중인 프롬프트가 토큰 임계값을 초과하면 전체 메시지 목록을 합성 memory_context_card 메시지와 최신 원시 회전의 작은 테일로 바꾸어 상태를 압축합니다. 이렇게 하면 사전 구축된 에이전트 검색 인식 단기 컨텍스트를 제공하면서 LangGraph 상태가 압축 상태로 유지됩니다.

이 문서에서는 토큰 기반 압축을 사용하지만 몇 번씩 압축하거나 응용 프로그램 특정 트리거 후에 압축하는 것과 같은 다른 정책에 동일한 패턴을 적용할 수 있습니다. transcript 전용 압축을 구현하는 경우 summary = thread.get_summary(...)로 전화하여 summary.content를 읽습니다. get_summary()를 메시지 목록으로 취급하지 마십시오.

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

        context_card = self._thread.get_context_card().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=120,
    kept_message_count=3,
)
agent = create_agent(
    model=langgraph_llm,
    tools=[],
    middleware=[short_term_middleware],
)

미들웨어 주입 컨텍스트를 사용하여 나중에 정답 제시

사용자 추가는 사전 구축된 에이전트의 실행 중인 메시지 목록으로 전환하고 미들웨어가 컨텍스트 카드를 삽입할 시기를 결정하도록 합니다. 나중에 차례가 오면 에이전트는 에이전트 메모리 단기 컨텍스트를 포함하는 컴팩트 상태에서 응답할 수 있습니다. 예제에서는 삽입된 컨텍스트 카드를 인쇄하고 잘린 샘플을 포함하므로 전체 블록 인라인을 덤프하지 않고 프롬프트에 삽입된 압축을 검사할 수 있습니다.

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().content와 get_context_card().content를 구분하고, 사전 구축된 LangGraph 에이전트를 중심으로 에이전트 메모리 단기 컨텍스트를 구성하고, 대화가 너무 커서 동사를 유지할 수 없을 때 미들웨어에서 컨텍스트 카드로 프롬프트를 압축하는 방법을 배웠습니다.

참고: LangGraph 플로우에 단기 스레드 컨텍스트를 추가하는 방법을 배웠으므로 이제 LangGraph에서 에이전트 메모리 사용을 진행할 수 있습니다.

전체 코드

#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

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

agent_memory = OracleAgentMemory(
    connection=db_pool,
    embedder=embedder,
    llm=memory_llm,
)
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,
            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

        context_card = self._thread.get_context_card().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=120,
    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.