Kurzreferenzcodebeispiele

Auf dieser Seite werden kleine, fokussierte Beispiele für häufige Oracle Agent-Speichersetup- und API-Lebenszyklusvorgänge gesammelt.

LLM/Einbettung einrichten

In den folgenden Beispielen wird LiteLLM für das LLM- und das Einbettungsmodell verwendet.

LLM konfigurieren

import oracledb

from oracleagentmemory.core.llms.llm import Llm

llm = Llm(
    model="YOUR_LLM_MODEL",
    api_base="YOUR_LLM_API_BASE",
    api_key="YOUR_LLM_API_KEY",
)

#Call ``llm.generate("What is 2+2?")`` when the application needs a response.

API-Referenz: Llm

LLM-Antwort

Einbettungsmodell konfigurieren

from oracleagentmemory.core.embedders.embedder import Embedder

embedder = Embedder(
    model="YOUR_EMBEDDING_MODEL",
    api_base="YOUR_EMBEDDING_API_BASE",
    api_key="YOUR_EMBEDDING_API_KEY",
)

embedding_matrix = embedder.embed(["The quick brown fox jumps over the lazy dog"])
print(embedding_matrix.shape)
#(1, embedding_dimension)

API-Referenz: Embedder

Benutzerdefinierte TLS-Zertifikate konfigurieren

Wenn ein OpenAI-kompatibler Endpunkt eine private Certificate Authority verwendet, übergeben Sie den Pfad zu seinem PEM-codierten CA-Zertifikat oder -Bundle als ca_file. Wenn der Endpunkt eine gegenseitige TLS erfordert, übergeben Sie auch die Clientzertifikatskette und den zugehörigen Private Key an cert_file und key_file. Das Clientzertifikat und der Private Key müssen zusammen bereitgestellt werden.

Die Pfade sind lokale Dateisystempfade. Wenn diese Parameter ausgelassen werden, wird die TLS-Standardvertrauenskonfiguration verwendet. Die TLS-Konfiguration gilt für die vom Provider gesicherten Clients Llm und Embedder. Sie konfiguriert keine OracleDBEmbedder- oder Oracle AI Database-Verbindung.

Der Anbieter verlangt standardmäßig die Anerkennung von Proxy- und TLS-bezogenen Umgebungsvariablen. Verwenden Sie proxy, um einen expliziten Proxy auszuwählen. Er hat Vorrang vor Proxyumgebungsvariablen. Legen Sie trust_env=False fest, um diese Umgebungseinstellungen zu ignorieren. Eine explizit konfigurierte proxy wird weiterhin verwendet, wenn trust_env=False.

from pathlib import Path

ca_file = Path("/path/to/provider-ca.pem")
if ca_file.is_file():
    secure_llm = Llm(
        model="openai/YOUR_LLM_MODEL",
        api_base="https://YOUR_LLM_API_BASE/v1",
        api_key="YOUR_LLM_API_KEY",
        ca_file=str(ca_file),
        proxy="http://proxy.example.com:8080",
        trust_env=True,
    )

client_cert_file = Path("/path/to/client-cert.pem")
client_key_file = Path("/path/to/client-key.pem")
if client_cert_file.is_file() and client_key_file.is_file():
    mtls_embedder = Embedder(
        model="hosted_vllm/YOUR_EMBEDDING_MODEL",
        api_base="https://YOUR_EMBEDDING_API_BASE/v1",
        api_key="YOUR_EMBEDDING_API_KEY",
        ca_file=str(ca_file),
        cert_file=str(client_cert_file),
        key_file=str(client_key_file),
    )

#Replace the example paths with files available in your deployment.
#``ca_file`` configures the CA used to verify the server. For mutual TLS,
#provide both ``cert_file`` and ``key_file`` as well.

API-Referenz: Llm

Einbettung

API-Setup

Agent-Speicherkomponente konfigurieren

Dabei wird eine Oracle DB-Verbindung oder ein Oracle DB-Pool zusammen mit dem Einbettungsmodell und einem optionalen LLM für die automatische Speicherextraktion verwendet.

import oracledb

from oracleagentmemory.core import SchemaPolicy
from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory

db_pool = oracledb.SessionPool(
    user="YOUR DB USER",
    password="YOUR DB PASSWORD",
    dsn="localhost:1521/...",
)
memory_store_id = "T_REF_SHEET"

memory_client = OracleAgentMemory(
    connection=db_pool,
    embedder=embedder,
    llm=llm,  # optional: enables automatic memory extraction during add_messages()
    schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
    memory_store_id=memory_store_id,
)

API-Referenz: OracleAgentMemory

Einbettung

Oracle DB-Speicherkomponente konfigurieren

Diese Variante verwendet eine Oracle DB-Verbindung oder einen Oracle DB-Pool und zeigt, wie eine Schema-Policy und ein Tabellennamenspräfix festgelegt werden.

import oracledb

from oracleagentmemory.core import SchemaPolicy
from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory

db_pool = oracledb.SessionPool(
    user="YOUR DB USER",
    password="YOUR DB PASSWORD",
    dsn="localhost:1521/...",
)
memory_store_id = "T_REF_SHEET"

memory = OracleAgentMemory(
    connection=db_pool,
    embedder=embedder,
    llm=llm,
    schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
    memory_store_id=memory_store_id,
)

API-Referenz: SchemaPolicy

MemoryExtractionConfig

Oracle Hybrid DB-Speicherkomponente konfigurieren

Diese Variante ermöglicht die von Oracle verwaltete hybride Suche über gespeicherten Suchtext und zeigt, wie der verwaltete Suchindex-Synchronisierungsmodus ausgewählt wird.

SearchStrategy.HYBRID erstellt oder validiert den verwalteten Hybridvektorindex von Oracle und erfordert, dass der Haupteinbettungsordner ein OracleDBEmbedder ist, sodass der verwaltete Index das datenbankinterne Modell des Einbettungsmoduls verwendet. SearchStrategy.KEYWORD ist Nur-Text: Es wird nach gespeichertem Suchtext sortiert und erfordert keine Einbettung. Schlüsselwortschemas können ohne lokalen Vektorspeicher erstellt werden. Öffnen Sie Schlüsselwortschemas daher nur dann erneut mit SearchStrategy.VECTOR, wenn Sie das Schema oder die Backfill-Einbettungen zuerst neu erstellen. Sie können weiterhin mit einer OracleDBEmbedder auf eine hybride Suche upgegradet werden, da der verwaltete Hybridindex von Oracle aus gespeichertem Suchtext erstellt wird.

Warnung: Wenn ein Hybridindex zum ersten Mal über vorhandene Daten erstellt wird, scannt Oracle gespeicherten Suchtext und erstellt den Status des verwalteten Index während des Schemasetups. SchemaPolicy.CREATE_IF_NECESSARY kann Zeit in Anspruch nehmen und sollte wie eine Datenbankmigration für große Schemas geplant werden.

from oracleagentmemory.core import SchemaPolicy, SearchIndexSyncMode, SearchStrategy
from oracleagentmemory.core.embedders import OracleDBEmbedder
from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory

db_embedder = OracleDBEmbedder(
    connection=db_pool,
    model="YOUR_DB_EMBEDDING_MODEL",
    embedding_dimension=384,
)

hybrid_db_memory = OracleAgentMemory(
    connection=db_pool,
    embedder=db_embedder,
    llm=llm,
    schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
    search_strategy=SearchStrategy.HYBRID,
    search_index_sync=SearchIndexSyncMode.ON_COMMIT,
    memory_store_id=memory_store_id,
)

API-Referenz: OracleAgentMemory

OracleDBEmbedder

Schema-Policy

Strategie suchen

SuchindexSyncModus

API-Leben

Thread erstellen

Erstellen Sie einen Thread mit einer optionalen Thread-ID, Benutzer-ID und Agent-ID.

thread = memory.create_thread(
    thread_id="thread_create_123",  # optional
    user_id="user_123",          # optional
    agent_id="agent_456",        # optional
)

print(thread.thread_id)
#thread_create_123

API-Referenz: OracleThread

Vorhandenen Thread erneut öffnen

thread = memory.create_thread(
    thread_id="thread_reopen_123",
    user_id="user_123",
    agent_id="agent_456",
)

same_thread = memory.get_thread("thread_reopen_123")
print(same_thread.thread_id)
#thread_reopen_123

Vorhandenen Thread aktualisieren

Verwenden Sie update_thread(), um Threadmetadaten oder dauerhafte Laufzeitkonfigurationsänderungen zu persistieren. Überschreibungen, die an get_thread() übergeben werden, wirken sich nur auf das erneut geöffnete Handle aus, bis sie explizit beibehalten werden.

thread = memory.create_thread(
    thread_id="thread_update_123",
    user_id="user_123",
    agent_id="agent_456",
)

loaded_thread = memory.get_thread(
    "thread_update_123",
    max_message_token_length=8_000,
)
print(loaded_thread.max_message_token_length)
#8000

updated_thread = memory.update_thread(
    "thread_update_123",
    metadata={"source": "support", "flags": {"vip": True}},
    max_message_token_length=8_000,
)
persisted_thread = memory.get_thread("thread_update_123")

print(updated_thread.metadata["flags"]["vip"])
#True
print(persisted_thread.max_message_token_length)
#8000
#Overrides passed to get_thread() are temporary. Call update_thread()
#to persist thread metadata or durable runtime-config changes.

Diskussionsthema löschen

Verwenden Sie diesen Vorgang, wenn Sie eine Thread-bezogene kaskadierende Bereinigung benötigen. Es entfernt den Thread zusammen mit zugehörigen Nachrichten, dauerhaften Speichern und Backup-Abrufdaten, die vom SDK verwaltet werden.

thread = memory.create_thread(thread_id="thread_delete_123")

deleted = memory.delete_thread("thread_delete_123")
print(deleted)
#1
#Use thread deletion when you need thread-scoped cascading cleanup.
#It removes the thread together with its messages, memories,
#and backing retrieval data managed by the SDK.

Benutzerprofil hinzufügen

user_profile_id = memory.add_user(
    "user_123",
    "The user prefers concise answers and works mostly with Python.",
)

print(user_profile_id)
#user_123

Agent-Profil hinzufügen

agent_profile_id = memory.add_agent(
    "agent_456",
    "A coding assistant specialized in debugging and code review.",
)

print(agent_profile_id)
#agent_456

Globalen Speicher aus der Speicher-API hinzufügen

Wenn thread_id ausgelassen wird, ist der Speicher nicht an einen bestimmten Thread gebunden. Verwenden Sie memory_type, um eine allgemeine "memory" (Standard), "fact", "guideline" oder "preference" zu speichern. Der zurückgegebene Wert ist die Speicher-ID.

memory_id = memory.add_memory(
    "The user prefers short, bullet-point answers.",
    memory_type="preference",
    user_id="user_123",
    agent_id="agent_456",
)

print(memory_id)
#mem:1

Bereichsspeicher aus der Speicher-API hinzufügen

Der zurückgegebene Wert ist die Speicher-ID. Der Typ eines Datensatzes wird beim Erstellen ausgewählt und kann nicht mit update_memory() geändert werden.

thread = memory.create_thread(
    thread_id="thread_scoped_123",
    user_id="user_123",
    agent_id="agent_456",
)

memory_id = memory.add_memory(
    "The user is planning a trip to Kyoto next month.",
    memory_type="fact",
    user_id="user_123",
    agent_id="agent_456",
    thread_id=thread.thread_id,
)

print(memory_id)
#mem:2

Speicher aus der Speicher-API aktualisieren

Verwenden Sie update_memory(), um gespeicherten Inhalt oder Metadaten für einen vorhandenen speicherähnlichen Datensatz nach ID zu ersetzen.

thread = memory.create_thread(
    thread_id="thread_update_memory_api_123",
    user_id="user_123",
    agent_id="agent_456",
)
memory_id = memory.add_memory(
    "The user likes short status updates.",
    user_id=thread.user_id,
    agent_id=thread.agent_id,
    thread_id=thread.thread_id,
    metadata={"source": "chat"},
)

updated_memory_id = memory.update_memory(
    memory_id,
    content="The user prefers short status updates.",
    metadata={"source": "support"},
)

print(updated_memory_id)
#mem:3

Speicher mit einer benutzerdefinierten ID hinzufügen

Der zurückgegebene Wert ist die vom Aufrufer bereitgestellte Speicher-ID.

memory_id = memory.add_memory(
    "The user prefers aisle seats on flights.",
    user_id="user_123",
    agent_id="agent_456",
    memory_id="travel_pref_001",
)

print(memory_id)
#travel_pref_001

Grundlagen zu Threads

Nachrichten zu einem Thread hinzufügen

Nachrichten können entweder als Wörterbücher oder als Message-Objekte übergeben werden. Optionale Nachrichten-IDs, Zeitstempel und Metadaten können mit ihnen gespeichert werden.

from oracleagentmemory.apis import Message

thread = memory.create_thread(
    thread_id="thread_messages_123",
    user_id="user_123",
    agent_id="agent_456",
)

message_ids = thread.add_messages(
    [
        Message(
            id="msg_user_001",
            role="user",
            content="I prefer window seats on flights.",
            timestamp="2026-03-27T09:00:00Z",
            metadata={"source": "assistant"},
        ),
        {
            "id": "msg_assistant_001",
            "role": "assistant",
            "content": "Noted. I will keep that in mind.",
            "timestamp": "2026-03-27T09:00:05Z",
            "metadata": {"source": "assistant"},
            #message metadata must be identical for one `add_messages` call
        },
    ]
)

print(message_ids)
#['msg_user_001', 'msg_assistant_001']

API-Referenz: Nachricht

Thread-Nachrichten zurücklesen

Sie können alle gespeicherten Nachrichten oder einen Bereich mit start und end lesen.

thread = memory.create_thread(thread_id="thread_read_messages_123")
thread.add_messages(
    [
        {"role": "user", "content": "Message 1"},
        {"role": "assistant", "content": "Message 2"},
        {"role": "user", "content": "Message 3"},
    ]
)

default_messages = thread.get_messages()
all_messages = thread.get_messages(end=None)
middle_messages = thread.get_messages(start=1, end=3)

print([message.content for message in default_messages])
#On short threads, the bounded default still returns all messages.
#['Message 1', 'Message 2', 'Message 3']
print([message.content for message in all_messages])
#['Message 1', 'Message 2', 'Message 3']
print([message.content for message in middle_messages])
#['Message 2', 'Message 3']

Nachricht aus aktuellem Thread nach ID löschen

Beim Löschen einer Nachricht wird nur die Zeile mit Raw-Nachrichten aus dem aktuellen Thread entfernt. Abgeleitete Speicher oder andere nachgeschaltete Artefakte, die aus dieser Nachricht erstellt wurden, können weiterhin durchsuchbar bleiben und die Ausgabe von Kontextkarten beeinflussen. Wenn Sie den Thread zusammen mit den zugehörigen Nachrichten und Speichern löschen müssen, verwenden Sie stattdessen delete_thread(). Wenn Sie eine ID aus einem anderen Thread übergeben, wird weiterhin 0 zurückgegeben.

thread = memory.create_thread(thread_id="thread_delete_message_123")
message_ids = thread.add_messages(
    [
        {"role": "user", "content": "Message to delete"},
    ]
)

deleted = thread.delete_message(message_ids[0])
print(deleted)
#1
#This removes only the raw message row from the current thread.
#Derived memories or other downstream artifacts created from that message
#are not deleted automatically and may remain searchable or appear in
#context-card output. Use memory.delete_thread(thread.thread_id) to delete
#the thread together with its associated messages and memories.
#Message deletes return 0 for IDs owned by another thread.

Nachricht aus aktuellem Thread nach ID aktualisieren

Threadbezogene Nachrichtenaktualisierungen wirken sich nur auf Raw-Nachrichten aus, die dem aktuellen Thread gehören. Gespeicherte Rollen- und Zeitstempelwerte werden beibehalten. Wenn die automatische Extraktion aktiviert ist, wird beim Bearbeiten des Nachrichteninhalts die Extraktion für die bearbeitete Nachricht sofort mit denselben Historienfensterregeln wie add_messages() erneut ausgeführt. Nur eine frühere Threadhistorie kann als unterstützender Kontext verwendet werden. Spätere Nachrichten werden während dieses unmittelbaren Durchlaufs ignoriert, und die gleiche Aktualisierung hält vorhandene abgeleitete Speicher an Ort und Stelle, während neue Speicher aus dem bearbeiteten Inhalt hinzugefügt werden.

thread = memory.create_thread(thread_id="thread_update_message_123")
thread.add_messages(
    [
        {
            "id": "msg_update_001",
            "role": "user",
            "content": "Original message text.",
            "timestamp": "2026-03-27T10:00:00Z",
            "metadata": {"source": "chat"},
        }
    ]
)

updated_message_id = thread.update_message(
    "msg_update_001",
    content="Edited message text.",
    metadata={"source": "support"},
)
print(updated_message_id)
#msg_update_001
#Message updates preserve stored role and timestamp values.
#When automatic extraction is enabled, content edits immediately rerun
#extraction for the edited message using the same history-window
#rules as add_messages().
#Later messages are ignored during that immediate pass.
#Existing derived memories stay in place while new edited-content
#memories are added during that refresh.

Speicher aus einem Thread-Handle hinzufügen

Der zurückgegebene Wert ist die Speicher-ID. In diesem Beispiel wird eine wiederverwendbare Richtlinie gespeichert. Verwenden Sie memory_type="fact" oder memory_type="preference" für diese Kategorien, oder weglassen Sie sie für einen allgemeinen Speicher.

thread = memory.create_thread(
    thread_id="thread_add_memory_123",
    user_id="user_123",
    agent_id="agent_456",
)

memory_id = thread.add_memory(
    "Use pytest for this repository's test suite.",
    memory_type="guideline",
)
print(memory_id)
#mem:4

Speicher aus aktuellem Thread nach ID aktualisieren

Threadbezogene Updates wirken sich nur auf speicherähnliche Datensätze aus, die dem aktuellen Thread gehören. Wenn Sie eine ID von einem anderen Thread übergeben, wird KeyError ausgelöst.

thread = memory.create_thread(
    thread_id="thread_update_memory_123",
    user_id="user_123",
    agent_id="agent_456",
)
memory_id = thread.add_memory(
    "The user likes jasmine tea.",
    metadata={"source": "survey"},
)

updated_memory_id = thread.update_memory(
    memory_id,
    content="The user likes jasmine tea in the afternoon.",
    metadata={"source": "support"},
)
print(updated_memory_id)
#mem:5
#Thread updates are scoped to the current thread and raise KeyError
#for missing IDs or IDs owned by another thread.

Speicher aus aktuellem Thread nach ID löschen

Threadlöschungen gelten für den aktuellen Thread. Wenn Sie eine ID von einem anderen Thread übergeben, wird 0 zurückgegeben.

thread = memory.create_thread(thread_id="thread_delete_memory_123")
memory_id = thread.add_memory("Temporary memory to delete.")

deleted = thread.delete_memory(memory_id)
print(deleted)
#1
#Thread deletes are scoped to the current thread and return 0 for IDs owned by another thread.

Kontextkarte erstellen

thread = memory.create_thread(thread_id="thread_context_card_123")
thread.add_messages(
    [
        {"role": "user", "content": "I am planning a trip to Kyoto next spring."},
    ]
)
thread.add_memory("The user is planning a trip to Kyoto.")

context_card = thread.get_context_card()
print(context_card.content)
#<context_card>
#The user is planning a trip to Kyoto.
#</context_card>

Threadübersicht erstellen

thread = memory.create_thread(thread_id="thread_summary_123")
thread.add_messages(
    [
        {"role": "user", "content": "Hello"},
        {"role": "assistant", "content": "Hi, how can I help?"},
        {"role": "user", "content": "Please summarize this thread."},
    ]
)

summary = thread.get_summary()
print(summary.content)
#user (-): Hello
#- assistant (-): Hi, how can I help?
#- user (-): Please summarize this thread.

Übersicht mit Ausnahme der letzten n Nachrichten erstellen

thread = memory.create_thread(thread_id="thread_summary_except_last_123")
thread.add_messages(
    [
        {"role": "user", "content": "First message"},
        {"role": "assistant", "content": "Second message"},
        {"role": "user", "content": "Third message"},
    ]
)

summary = thread.get_summary(except_last=1)
print(summary.content)
#user (-): First message
#- assistant (-): Second message

Übersicht mit einem Tokenbudget erstellen

thread = memory.create_thread(thread_id="thread_summary_budget_123")
thread.add_messages(
    [
        {"role": "user", "content": "Message 1"},
        {"role": "assistant", "content": "Message 2"},
        {"role": "user", "content": "Message 3"},
        {"role": "assistant", "content": "Message 4"},
    ]
)

summary = thread.get_summary(token_budget=20)
print(summary.content)
#(truncated)
#user (-): Message 1
#...

Suchen

Aus einem Thread ohne expliziten Geltungsbereich suchen

Bei der Suche auf Thread-Ebene werden die Thread-Standardwerte verwendet, wenn Sie keinen expliziten Geltungsbereich übergeben.

thread = memory.create_thread(
    thread_id="thread_search_default_123",
    user_id="user_123",
    agent_id="agent_456",
)
thread.add_memory("The user likes pizza.")
thread.add_memory("The user likes cats.")

results = thread.search("pizza", max_results=5)
print([result.content for result in results])
#['The user likes pizza.']

API-Referenz: OracleSearchResult

Über die Speicher-API mit Scoping suchen

Auf API-Ebene können Sie den Abruf mit user_id, agent_id und thread_id bis SearchScope festlegen. Geben Sie für Clientsuchen der obersten Ebene einen expliziten Benutzergeltungsbereich an. Verwenden Sie user_id=None nur, wenn Sie absichtlich nicht kopierte Datensätze verwenden möchten. Eine Übersicht darüber, wie ausgelassene Werte, explizite None- und Flags mit genauer Übereinstimmung auf den einzelnen API-Layern aufgelöst werden, finden Sie unter Geltungsauflösung.

from oracleagentmemory.apis.searchscope import SearchScope

thread = memory.create_thread(
    thread_id="thread_memory_search_123",
    user_id="user_123",
    agent_id="agent_456",
)
thread.add_memory("The user likes hiking in the Alps.")

results = memory.search(
    "hiking",
    scope=SearchScope(
        user_id="user_123",
        agent_id="agent_456",
        thread_id="thread_memory_search_123",
        exact_thread_match=True,
    ),
    max_results=5,
)

print([result.content for result in results])
#['The user likes hiking in the Alps.']

API-Referenz: SearchScope

Mit Metadatenfilterung suchen

Verwenden Sie metadata_filter, wenn bei der Suche nur Datensätze berücksichtigt werden sollen, deren gespeicherte Metadaten eine angeforderte teilweise Zuordnung enthalten. Mehrere Filterschlüssel werden mit AND-Semantik kombiniert, verschachtelte Wörterbücher stimmen mit verschachtelten Metadatenfeldern überein, und Bare-Listenwerte müssen genau übereinstimmen. Um die Arraymitgliedschaft zu testen, verwenden Sie ein Operatordictionary auf Feldebene, wie {"tags": {"$array_contains": "outdoor"}}. "$array_contains" mit einer Liste erfordert alle aufgelisteten Werte, "$array_contains_any" erfordert mindestens einen aufgelisteten Wert, und "$not" negiert einen anderen Ausdruck auf Feldebene im selben Feld, einschließlich eines Operator-Dictionarys oder eines Raw-Exact-Match-Werts. Ein negierter Ausdruck stimmt auch überein, wenn der positive Ausdruck fehlschlagen würde, einschließlich fehlender Felder. Negierte Array-Mitgliedschaft stimmt auch mit Nicht-Array-Feldern überein.

from oracleagentmemory.apis.searchscope import SearchScope

thread = memory.create_thread(
    thread_id="thread_metadata_filter_123",
    user_id="user_123",
    agent_id="agent_456",
)
thread.add_memory(
    "The user likes alpine hiking.",
    metadata={"source": "survey", "category": {"kind": "travel"}, "tags": ["outdoor"]},
)
thread.add_memory(
    "The user likes indoor climbing.",
    metadata={"source": "chat", "category": {"kind": "sports"}, "tags": ["indoor"]},
)

results = memory.search(
    "hiking",
    scope=SearchScope(user_id="user_123"),
    max_results=5,
    record_types=["memory"],
    metadata_filter={"source": "survey"},
)

print([result.content for result in results])
#['The user likes alpine hiking.']

outdoor_results = memory.search(
    "hiking",
    scope=SearchScope(user_id="user_123"),
    max_results=5,
    record_types=["memory"],
    metadata_filter={
        "source": "survey",
        "tags": {"$array_contains": "outdoor"},
    },
)

print([result.content for result in outdoor_results])
#['The user likes alpine hiking.']

Nur Erinnerungen oder nur Nachrichten suchen

Verwenden Sie record_types, um die Suchergebnisse auf bestimmte gespeicherte Datensatztypen zu beschränken.

thread = memory.create_thread(thread_id="thread_entity_type_search_123")
thread.add_messages(
    [
        {"role": "user", "content": "I mentioned pizza in a message."},
    ]
)
thread.add_memory("The user likes pizza.")

memory_results = thread.search("pizza", max_results=5, record_types=["memory"])
message_results = thread.search("pizza", max_results=5, record_types=["message"])

print([result.content for result in memory_results])
#['The user likes pizza.']
print([result.content for result in message_results])
#['I mentioned pizza in a message.']

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 - Reference Sheet
#--------------------------------------------------

##Configure a LiteLLM LLM

import oracledb

from oracleagentmemory.core.llms.llm import Llm

llm = Llm(
    model="YOUR_LLM_MODEL",
    api_base="YOUR_LLM_API_BASE",
    api_key="YOUR_LLM_API_KEY",
)

#Call ``llm.generate("What is 2+2?")`` when the application needs a response.

##Configure a LiteLLM embedding model

from oracleagentmemory.core.embedders.embedder import Embedder

embedder = Embedder(
    model="YOUR_EMBEDDING_MODEL",
    api_base="YOUR_EMBEDDING_API_BASE",
    api_key="YOUR_EMBEDDING_API_KEY",
)

embedding_matrix = embedder.embed(["The quick brown fox jumps over the lazy dog"])
print(embedding_matrix.shape)
#(1, embedding_dimension)

##Configure custom TLS for OpenAI-compatible providers

from pathlib import Path

ca_file = Path("/path/to/provider-ca.pem")
if ca_file.is_file():
    secure_llm = Llm(
        model="openai/YOUR_LLM_MODEL",
        api_base="https://YOUR_LLM_API_BASE/v1",
        api_key="YOUR_LLM_API_KEY",
        ca_file=str(ca_file),
        proxy="http://proxy.example.com:8080",
        trust_env=True,
    )

client_cert_file = Path("/path/to/client-cert.pem")
client_key_file = Path("/path/to/client-key.pem")
if client_cert_file.is_file() and client_key_file.is_file():
    mtls_embedder = Embedder(
        model="hosted_vllm/YOUR_EMBEDDING_MODEL",
        api_base="https://YOUR_EMBEDDING_API_BASE/v1",
        api_key="YOUR_EMBEDDING_API_KEY",
        ca_file=str(ca_file),
        cert_file=str(client_cert_file),
        key_file=str(client_key_file),
    )

#Replace the example paths with files available in your deployment.
#``ca_file`` configures the CA used to verify the server. For mutual TLS,
#provide both ``cert_file`` and ``key_file`` as well.

##Configure an Oracle Memory component

from oracleagentmemory.core import SchemaPolicy
from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory

db_pool = oracledb.SessionPool(
    user="YOUR DB USER",
    password="YOUR DB PASSWORD",
    dsn="localhost:1521/...",
)
memory_store_id = "T_REF_SHEET"

memory_client = OracleAgentMemory(
    connection=db_pool,
    embedder=embedder,
    llm=llm,  # optional: enables automatic memory extraction during add_messages()
    schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
    memory_store_id=memory_store_id,
)

##Configure an Oracle DB component

from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory

db_pool = oracledb.SessionPool(
    user="YOUR DB USER",
    password="YOUR DB PASSWORD",
    dsn="localhost:1521/...",
)
memory_store_id = "T_REF_SHEET"

memory = OracleAgentMemory(
    connection=db_pool,
    embedder=embedder,
    llm=llm,
    schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
    memory_store_id=memory_store_id,
)

##Configure an Oracle Hybrid DB component

from oracleagentmemory.core import SchemaPolicy, SearchIndexSyncMode, SearchStrategy
from oracleagentmemory.core.embedders import OracleDBEmbedder
from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory

db_embedder = OracleDBEmbedder(
    connection=db_pool,
    model="YOUR_DB_EMBEDDING_MODEL",
    embedding_dimension=384,
)

hybrid_db_memory = OracleAgentMemory(
    connection=db_pool,
    embedder=db_embedder,
    llm=llm,
    schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
    search_strategy=SearchStrategy.HYBRID,
    search_index_sync=SearchIndexSyncMode.ON_COMMIT,
    memory_store_id=memory_store_id,
)

##Create a thread

thread = memory.create_thread(
    thread_id="thread_create_123",  # optional
    user_id="user_123",          # optional
    agent_id="agent_456",        # optional
)

print(thread.thread_id)
#thread_create_123

##Re open an existing thread

thread = memory.create_thread(
    thread_id="thread_reopen_123",
    user_id="user_123",
    agent_id="agent_456",
)

same_thread = memory.get_thread("thread_reopen_123")
print(same_thread.thread_id)
#thread_reopen_123

##Update an existing thread

thread = memory.create_thread(
    thread_id="thread_update_123",
    user_id="user_123",
    agent_id="agent_456",
)

loaded_thread = memory.get_thread(
    "thread_update_123",
    max_message_token_length=8_000,
)
print(loaded_thread.max_message_token_length)
#8000

updated_thread = memory.update_thread(
    "thread_update_123",
    metadata={"source": "support", "flags": {"vip": True}},
    max_message_token_length=8_000,
)
persisted_thread = memory.get_thread("thread_update_123")

print(updated_thread.metadata["flags"]["vip"])
#True
print(persisted_thread.max_message_token_length)
#8000
#Overrides passed to get_thread() are temporary. Call update_thread()
#to persist thread metadata or durable runtime-config changes.

##Delete a thread

thread = memory.create_thread(thread_id="thread_delete_123")

deleted = memory.delete_thread("thread_delete_123")
print(deleted)
#1
#Use thread deletion when you need thread-scoped cascading cleanup.
#It removes the thread together with its messages, memories,
#and backing retrieval data managed by the SDK.

##Add a user profile

user_profile_id = memory.add_user(
    "user_123",
    "The user prefers concise answers and works mostly with Python.",
)

print(user_profile_id)
#user_123

##Add an agent profile

agent_profile_id = memory.add_agent(
    "agent_456",
    "A coding assistant specialized in debugging and code review.",
)

print(agent_profile_id)
#agent_456

##Add a global memory from the memory API

memory_id = memory.add_memory(
    "The user prefers short, bullet-point answers.",
    memory_type="preference",
    user_id="user_123",
    agent_id="agent_456",
)

print(memory_id)
#mem:1

##Add a scoped memory from the memory API

thread = memory.create_thread(
    thread_id="thread_scoped_123",
    user_id="user_123",
    agent_id="agent_456",
)

memory_id = memory.add_memory(
    "The user is planning a trip to Kyoto next month.",
    memory_type="fact",
    user_id="user_123",
    agent_id="agent_456",
    thread_id=thread.thread_id,
)

print(memory_id)
#mem:2

##Update a memory from the memory API

thread = memory.create_thread(
    thread_id="thread_update_memory_api_123",
    user_id="user_123",
    agent_id="agent_456",
)
memory_id = memory.add_memory(
    "The user likes short status updates.",
    user_id=thread.user_id,
    agent_id=thread.agent_id,
    thread_id=thread.thread_id,
    metadata={"source": "chat"},
)

updated_memory_id = memory.update_memory(
    memory_id,
    content="The user prefers short status updates.",
    metadata={"source": "support"},
)

print(updated_memory_id)
#mem:3

##Add a memory with a custom ID

memory_id = memory.add_memory(
    "The user prefers aisle seats on flights.",
    user_id="user_123",
    agent_id="agent_456",
    memory_id="travel_pref_001",
)

print(memory_id)
#travel_pref_001

##Add messages to a thread

from oracleagentmemory.apis import Message

thread = memory.create_thread(
    thread_id="thread_messages_123",
    user_id="user_123",
    agent_id="agent_456",
)

message_ids = thread.add_messages(
    [
        Message(
            id="msg_user_001",
            role="user",
            content="I prefer window seats on flights.",
            timestamp="2026-03-27T09:00:00Z",
            metadata={"source": "assistant"},
        ),
        {
            "id": "msg_assistant_001",
            "role": "assistant",
            "content": "Noted. I will keep that in mind.",
            "timestamp": "2026-03-27T09:00:05Z",
            "metadata": {"source": "assistant"},
            #message metadata must be identical for one `add_messages` call
        },
    ]
)

print(message_ids)
#['msg_user_001', 'msg_assistant_001']

##Read back thread messages

thread = memory.create_thread(thread_id="thread_read_messages_123")
thread.add_messages(
    [
        {"role": "user", "content": "Message 1"},
        {"role": "assistant", "content": "Message 2"},
        {"role": "user", "content": "Message 3"},
    ]
)

default_messages = thread.get_messages()
all_messages = thread.get_messages(end=None)
middle_messages = thread.get_messages(start=1, end=3)

print([message.content for message in default_messages])
#On short threads, the bounded default still returns all messages.
#['Message 1', 'Message 2', 'Message 3']
print([message.content for message in all_messages])
#['Message 1', 'Message 2', 'Message 3']
print([message.content for message in middle_messages])
#['Message 2', 'Message 3']

##Delete a message from the current thread by ID

thread = memory.create_thread(thread_id="thread_delete_message_123")
message_ids = thread.add_messages(
    [
        {"role": "user", "content": "Message to delete"},
    ]
)

deleted = thread.delete_message(message_ids[0])
print(deleted)
#1
#This removes only the raw message row from the current thread.
#Derived memories or other downstream artifacts created from that message
#are not deleted automatically and may remain searchable or appear in
#context-card output. Use memory.delete_thread(thread.thread_id) to delete
#the thread together with its associated messages and memories.
#Message deletes return 0 for IDs owned by another thread.

##Update a message from the current thread by ID

thread = memory.create_thread(thread_id="thread_update_message_123")
thread.add_messages(
    [
        {
            "id": "msg_update_001",
            "role": "user",
            "content": "Original message text.",
            "timestamp": "2026-03-27T10:00:00Z",
            "metadata": {"source": "chat"},
        }
    ]
)

updated_message_id = thread.update_message(
    "msg_update_001",
    content="Edited message text.",
    metadata={"source": "support"},
)
print(updated_message_id)
#msg_update_001
#Message updates preserve stored role and timestamp values.
#When automatic extraction is enabled, content edits immediately rerun
#extraction for the edited message using the same history-window
#rules as add_messages().
#Later messages are ignored during that immediate pass.
#Existing derived memories stay in place while new edited-content
#memories are added during that refresh.

##Add a memory from a thread handle

thread = memory.create_thread(
    thread_id="thread_add_memory_123",
    user_id="user_123",
    agent_id="agent_456",
)

memory_id = thread.add_memory(
    "Use pytest for this repository's test suite.",
    memory_type="guideline",
)
print(memory_id)
#mem:4

##Update a memory from the current thread by ID

thread = memory.create_thread(
    thread_id="thread_update_memory_123",
    user_id="user_123",
    agent_id="agent_456",
)
memory_id = thread.add_memory(
    "The user likes jasmine tea.",
    metadata={"source": "survey"},
)

updated_memory_id = thread.update_memory(
    memory_id,
    content="The user likes jasmine tea in the afternoon.",
    metadata={"source": "support"},
)
print(updated_memory_id)
#mem:5
#Thread updates are scoped to the current thread and raise KeyError
#for missing IDs or IDs owned by another thread.

##Delete a memory from the current thread by ID

thread = memory.create_thread(thread_id="thread_delete_memory_123")
memory_id = thread.add_memory("Temporary memory to delete.")

deleted = thread.delete_memory(memory_id)
print(deleted)
#1
#Thread deletes are scoped to the current thread and return 0 for IDs owned by another thread.

##Build a context card

thread = memory.create_thread(thread_id="thread_context_card_123")
thread.add_messages(
    [
        {"role": "user", "content": "I am planning a trip to Kyoto next spring."},
    ]
)
thread.add_memory("The user is planning a trip to Kyoto.")

context_card = thread.get_context_card()
print(context_card.content)
#<context_card>
#The user is planning a trip to Kyoto.
#</context_card>

##Build a thread summary

thread = memory.create_thread(thread_id="thread_summary_123")
thread.add_messages(
    [
        {"role": "user", "content": "Hello"},
        {"role": "assistant", "content": "Hi, how can I help?"},
        {"role": "user", "content": "Please summarize this thread."},
    ]
)

summary = thread.get_summary()
print(summary.content)
#user (-): Hello
#- assistant (-): Hi, how can I help?
#- user (-): Please summarize this thread.

##Build a summary excluding the last N messages

thread = memory.create_thread(thread_id="thread_summary_except_last_123")
thread.add_messages(
    [
        {"role": "user", "content": "First message"},
        {"role": "assistant", "content": "Second message"},
        {"role": "user", "content": "Third message"},
    ]
)

summary = thread.get_summary(except_last=1)
print(summary.content)
#user (-): First message
#- assistant (-): Second message

##Build a summary with a token budget

thread = memory.create_thread(thread_id="thread_summary_budget_123")
thread.add_messages(
    [
        {"role": "user", "content": "Message 1"},
        {"role": "assistant", "content": "Message 2"},
        {"role": "user", "content": "Message 3"},
        {"role": "assistant", "content": "Message 4"},
    ]
)

summary = thread.get_summary(token_budget=20)
print(summary.content)
#(truncated)
#user (-): Message 1
#...

##Search from a thread with no explicit scoping

thread = memory.create_thread(
    thread_id="thread_search_default_123",
    user_id="user_123",
    agent_id="agent_456",
)
thread.add_memory("The user likes pizza.")
thread.add_memory("The user likes cats.")

results = thread.search("pizza", max_results=5)
print([result.content for result in results])
#['The user likes pizza.']

##Search from the memory API with scoping

from oracleagentmemory.apis.searchscope import SearchScope

thread = memory.create_thread(
    thread_id="thread_memory_search_123",
    user_id="user_123",
    agent_id="agent_456",
)
thread.add_memory("The user likes hiking in the Alps.")

results = memory.search(
    "hiking",
    scope=SearchScope(
        user_id="user_123",
        agent_id="agent_456",
        thread_id="thread_memory_search_123",
        exact_thread_match=True,
    ),
    max_results=5,
)

print([result.content for result in results])
#['The user likes hiking in the Alps.']

##Search with metadata filtering

from oracleagentmemory.apis.searchscope import SearchScope

thread = memory.create_thread(
    thread_id="thread_metadata_filter_123",
    user_id="user_123",
    agent_id="agent_456",
)
thread.add_memory(
    "The user likes alpine hiking.",
    metadata={"source": "survey", "category": {"kind": "travel"}, "tags": ["outdoor"]},
)
thread.add_memory(
    "The user likes indoor climbing.",
    metadata={"source": "chat", "category": {"kind": "sports"}, "tags": ["indoor"]},
)

results = memory.search(
    "hiking",
    scope=SearchScope(user_id="user_123"),
    max_results=5,
    record_types=["memory"],
    metadata_filter={"source": "survey"},
)

print([result.content for result in results])
#['The user likes alpine hiking.']

outdoor_results = memory.search(
    "hiking",
    scope=SearchScope(user_id="user_123"),
    max_results=5,
    record_types=["memory"],
    metadata_filter={
        "source": "survey",
        "tags": {"$array_contains": "outdoor"},
    },
)

print([result.content for result in outdoor_results])
#['The user likes alpine hiking.']

##Search only memories or messages

thread = memory.create_thread(thread_id="thread_entity_type_search_123")
thread.add_messages(
    [
        {"role": "user", "content": "I mentioned pizza in a message."},
    ]
)
thread.add_memory("The user likes pizza.")

memory_results = thread.search("pizza", max_results=5, record_types=["memory"])
message_results = thread.search("pizza", max_results=5, record_types=["message"])

print([result.content for result in memory_results])
#['The user likes pizza.']
print([result.content for result in message_results])
#['I mentioned pizza in a message.']