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()를 로드하고, 전체 transcript를 재전송하지 않고 최근 스레드 컨텍스트로 나중에 응답하는 방법을 배웠습니다.

참고: LangGraph 플로우에 단기 스레드 컨텍스트를 추가하는 방법을 배웠으면 이제 Integrate Agent Memory with 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.