Kurzfristige Agent-Speicher-APIs mit LangGraph verwenden
Agents müssen häufig den aktuellen Arbeitskontext beibehalten, ohne die vollständige Konversation bei jeder Kurve an das Modell zurückzugeben. Wenn Sie nur die neueste Nachricht im Status halten, kann das Modell den Überblick über vorherige Aufgabendetails, den Zwischenfortschritt oder das aktive Thema verlieren.
Oracle Agent Memory stellt zwei verschiedene kurzfristige Helfer für dieses Problem bereit:
get_summary()gibt ein Übersichtsobjekt zurück, dessencontentdas Threadtranskript komprimiert. Bevorzugt, wenn die Komprimierung nur eine Transkriptkomprimierung erfordert.get_context_card()gibt ein Kontextkartenobjekt zurück, dessencontentein sofort einsatzbereiter Kontextblock mit Threadübersicht, Abrufthemen, relevanten Datensätzen und, wenn angefordert, relevanten oder aktuellen Unterhaltungsnachrichten ist. Bevorzugen Sie es, wenn die Verdichtung den Retrieval-Aware-Kontext für den aktuellen Zug beibehalten soll.
Wenn die Anwendung die letzten N-Nachrichten als Originalnachrichten speichert, können Sie diesen Tail wie folgt aus dem abgeleiteten Inhalt der Karte ausschließen:
#External-tail pattern:
#LLM Prompt = [context card, last N raw messages]
thread.get_context_card(except_last_messages=N, max_recent_messages=0)Die resultierende Eingabeaufforderung ist eine Liste, die eine Kontextkartennachricht gefolgt von den ursprünglichen Nachrichten enthält:
prompt_messages = [
{"role": "user", "content": "<context_card>...</context_card>"},
{"role": "user", "content": "The latest user message"},
{"role": "assistant", "content": "The latest assistant reply"},
]Dies wird bevorzugt, wenn Sie das Prompt-Caching nutzen, da Sie dieselbe Kontextkarte für mehrere Konversationsumschläge beibehalten und neue Nachrichten an den LLM-Prompt anhängen.
Alternativ können Sie die Karte als eigenständig festlegen und den rohen Schwanz wie folgt darin halten:
#Self-contained-card pattern:
#LLM Prompt = [context card only]
thread.get_context_card(
except_last_messages=N,
max_recent_messages=N,
)Der resultierende Prompt ist eine Liste mit nur einer Meldung: der Kontextkartennachricht, deren Abschnitt <recent_messages> den Raw-Tail enthält:
prompt_messages = [
{
"role": "user",
"content": (
"<context_card>..."
"<recent_messages>...</recent_messages>"
"</context_card>"
),
},
]Wenn except_last_messages ungleich null ist, muss max_recent_messages null oder derselbe Wert sein. Dadurch wird verhindert, dass der gleiche Raw-Tail sowohl innerhalb als auch außerhalb der Kontextkarte bereitgestellt wird.
In diesem Handbuch verwenden Sie die LangGraph-Middleware um einen vordefinierten Agent, sodass Oracle Agent Memory automatische Umdrehungen beibehalten und eine Oracle-Kontextkarte injizieren kann, wenn der ausgeführte Prompt zu groß wird. Das heißt, die Middleware kompakt die Eingabeaufforderung, sobald sie einen konfigurierten Schwellenwert überschritten hat. In diesem Beispiel wird get_context_card() ausgewählt, da die Komprimierung den Retrieval-bezogenen Kontext beibehalten soll, nicht nur eine Transkript-Zusammenfassung.
Informationen zum Optimieren der Datensatztypen, die in den relevanten Informationen der Kontextkarte angezeigt werden, einschließlich durchsuchter Nachrichten aus dem aktuellen Thread, finden Sie unter Kontextkarteninhalt anpassen.
Wichtig: Zusammenfassungen, Kontextkarten, abgerufene Datensätze und automatisch extrahierte Speicher sind vom Modell abgeleiteter oder abgerufener Text und müssen als nicht vertrauenswürdig behandelt werden. Wenn die automatische Extraktion oder Zusammenfassung aktiviert ist, kann dieser Text vom SDK auch in späteren Eingabeaufforderungen wie Speicherextraktion, Zusammenfassung, Kontextkarte oder Agent-Eingabeaufforderungen wiederverwendet werden, bevor die Anwendung die Möglichkeit hat, den spezifischen Zwischenwert zu prüfen. Prüfen Sie die Ausgaben, die Ihre Anwendung verbraucht, vermeiden Sie, dass vom Speicher abgeleiteter Text privilegierte Aktionen autorisiert, und verwenden Sie memory_extraction_config=MemoryExtractionConfig(extract_memories=False) oder explizite Speicherschreibvorgänge, wenn Ihr Workflow geprüft werden muss, bevor abgeleiteter Text die spätere Extraktion oder Kontexterstellung beeinflussen kann.
In diesem Tutorial erfahren Sie, wie Sie folgende Aktionen durchführen:
- Oracle Agent-Speicher mit einem
Embedder, einem Oracle-Speicher-LLM konfigurieren und dann mit einem LangGraphChatOpenAI-Modell koppeln - einen vorgefertigten LangGraph-Agent mit Middleware umwickeln, die neue Umdrehungen andauert, und
get_context_card()-Ausgabe injizieren, wenn der Tokendruck steigt und die schnelle Verdichtung eintritt - Antwort später mit kurzfristigem Kontext aus dem Oracle Agent-Speicher-Thread, anstatt das vollständige Transkript erneut zu senden
Hinweis: Informationen zum Packagesetup finden Sie unter Erste Schritte mit Agent-Speicher. Wenn Sie für dieses Beispiel eine lokale Oracle AI Database benötigen, folgen Sie Oracle AI Database lokal ausführen.
Agent-Speicher und LangGraph-Modelle konfigurieren
Erstellen Sie einen Oracle Agent-Speicherclient mit einer Oracle DB-Verbindung oder einem Oracle DB-Pool, konfigurieren Sie eine Embedder für die Vektorsuche, stellen Sie ein Oracle-Speicher-LLM zur Auflösung von Kontextkarten bereit, und verwenden Sie ChatOpenAI für den LangGraph-Agent.
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-Referenz: OracleAgentMemory | OracleThread |
Middleware und einen vordefinierten Agent konfigurieren
Der neue Benutzer und Assistent wird in Oracle Agent Memory umgewandelt. Sobald die ausgeführte Eingabeaufforderung einen Tokenschwellenwert überschreitet, wird der Status komprimiert, indem die vollständige Nachrichtenliste durch eine synthetische oracle_context_card-Nachricht plus einen kleinen Schwanz der letzten Raw-Wendungen ersetzt wird. Dadurch bleibt der LangGraph-Zustand kompakt, während der vordefinierte Agent-Abruf-konformer Kurzzeitkontext erhalten bleibt.
In diesem Handbuch wird tokenbasierte Komprimierung verwendet. Das gleiche Muster kann jedoch an andere Policys angepasst werden, wie z. B. die Komprimierung alle paar Kurven oder nach einem anderen anwendungsspezifischen Trigger.
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],
)Spätere Antwort mit Middleware-injiziertem Kontext
"Benutzer anhängen" wechselt zur aktiven Nachrichtenliste des vordefinierten Agents und lässt die Middleware entscheiden, wann eine Kontextkarte injiziert werden soll. Wenn der spätere Turn eintrifft, kann der Agent aus einem kompakten Zustand antworten, der noch den Kurzzeitkontext des Oracle Agent-Speichers enthält. Das Beispiel druckt die eingespritzte Kontextkarte und enthält ein getrimmtes Beispiel, sodass Sie prüfen können, welche Verdichtung in den Prompt eingefügt wurde, ohne den vollständigen Block inline abzugeben.
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.Abschluss
In diesem Handbuch haben wir gelernt, wie Sie get_summary() von get_context_card() unterscheiden, den Kurzzeitkontext des Oracle Agent-Speichers um einen vordefinierten LangGraph-Agent konfigurieren und Middleware die Eingabeaufforderung mit einer Kontextkarte komprimieren lassen, wenn die Unterhaltung zu groß wird, um sie wörtlich zu halten.
→ Nachdem Sie gelernt haben, wie Sie einem LangGraph-Ablauf Kurzzeit-Thread-Kontext hinzufügen, können Sie jetzt mit Oracle Agent Memory mit LangGraph verwenden fortfahren.
Vollständiger Code
Kopieren Sie den folgenden vollständigen Code.
#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.