使用代理程式記憶體短期 API 與 LangGraph
LangGraph 應用程式通常需要保留最近的工作環境，而無需每次將完整對話傳回模型。如果您只將最新訊息保持在圖表狀態，則模型可以輕易地追蹤先前的工作詳細資訊、中間進度或作用中繫線主題。
在本文中，您將在 LangGraph 流程中使用「代理程式記憶體」短期 API，以便圖表可視需要復原最近的執行緒環境定義。流程會使用 get_summary() 來結轉較早訊息的精簡摘要，並使用 get_context_card() 來顯示與最新使用者回合最相關的記錄。
在本文中，您將學習如何：
- 使用內嵌程式設定代理程式記憶體，並使用 ChatOpenAI 設定 LangGraph 流程
- 在每次模型呼叫前載入 get_summary (except_last=1) 和 get_context_card()
- 稍後回答會轉成「代理程式記憶體」繫線的短期相關資訊環境，而非完整記錄
秘訣：如需設定套裝程式，請參閱開始使用代理程式記憶體。如果此範例需要本機 Oracle AI Database，請參閱在本機執行 Oracle AI Database 。
設定代理程式記憶體和語言圖表
建立具有 Oracle DB 連線或集區的「代理程式記憶體」從屬端、設定 Embedder 進行向量搜尋，以及使用 ChatOpenAI 作為 LangGraph 節點。
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.