快速參考代碼範例

此頁面收集用於一般 Oracle 代理程式記憶體設定和 API 生命週期作業的小型重點範例。

LLM / 內嵌設定

下列範例將 LiteLLM 用於 LLM 和嵌入模型。

設定 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.

API 參照：Llm

Llm 回應

設定內嵌模型

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 參考：嵌入器

設定自訂 TLS 憑證

當 OpenAI 相容的端點使用專用憑證授權機構時，請將路徑傳送至其 PEM 編碼 CA 憑證或組合作為 ca_file。如果端點需要相互 TLS，請同時透過 cert_file 和 key_file 傳送從屬端憑證鏈及其私密金鑰。用戶端憑證與私密金鑰必須同時提供。

路徑為本機檔案系統路徑。當省略這些參數時，會使用預設 TLS 信任組態。TLS 組態會套用至提供者支援的 Llm 和 Embedder 從屬端；它不會設定 OracleDBEmbedder 或 Oracle AI Database 連線。

提供方要求預設使用代理主機和 TLS 相關環境變數。使用 proxy 來選取明確的代理主機；其優先順序高於代理主機環境變數。設定 trust_env=False 以忽略這些環境設定值。trust_env=False 時，仍會使用明確設定的 proxy。

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 參照：Llm

嵌入器

API 設定

設定代理程式記憶體元件

這會使用 Oracle DB 連線或集區以及內嵌模型和選擇性的 LLM，自動擷取記憶體。

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 參考： OracleAgentMemory

嵌入器

設定 Oracle DB 記憶體元件

此變體使用 Oracle DB 連線或集區，並顯示如何設定綱要原則和表格名稱前置碼。

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 參考： SchemaPolicy

記憶體擷取組態

設定 Oracle Hybrid DB 記憶體元件

此變體可讓 Oracle 管理的混合式搜尋取代預存搜尋文字，並顯示如何選取受管理搜尋索引同步模式。

SearchStrategy.HYBRID 會建立或驗證 Oracle 的受管理混合向量索引，而且主要內嵌器必須是 OracleDBEmbedder，因此受管理索引會使用內嵌器的資料庫內模型。SearchStrategy.KEYWORD 是僅限文字的：它會依預存的搜尋文字排列，不需要內嵌程式。關鍵字綱要可以在沒有本機向量儲存的情況下建立，因此除非您先重新建立綱要或重新填入內嵌項目，否則請勿使用 SearchStrategy.VECTOR 重新開啟關鍵字綱要。Oracle 管理的混合索引是從預存搜尋文字建立，因此仍然可以使用 OracleDBEmbedder 升級成混合搜尋。

警告：第一次透過現有資料建立混合索引時，Oracle 會掃描預存搜尋文字，並在綱要設定期間建立受管理索引狀態。SchemaPolicy.CREATE_IF_NECESSARY 可能需要一些時間，且應該像大型綱要的資料庫移轉一樣進行規劃。

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 參考： OracleAgentMemory

OracleDB 內嵌程式

綱要原則

搜尋策略

搜尋索引同步模式

API 生命週期

建立討論串

建立一個含有選擇性繫線 ID、使用者 ID 以及代理程式 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 參照：OracleThread

重新開啟現有的討論串

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_thread() 來保存繫線描述資料或持續的 Runtime-config 變更。傳送至 get_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.

刪除執行緒

當您需要執行緒作用領域的連鎖清除時，請使用此作業。它會移除執行緒以及由 SDK 管理的相關訊息、持久記憶體及備份擷取資料。

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.

新增使用者設定檔

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_profile_id = memory.add_agent(
    "agent_456",
    "A coding assistant specialized in debugging and code review.",
)

print(agent_profile_id)
#agent_456

從記憶體 API 新增全域記憶體

省略 thread_id 時，記憶體不會繫結至特定繫線。使用 memory_type 來儲存一般 "memory" (預設值)、"fact"、"guideline" 或 "preference"。傳回的值是記憶體 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

從記憶體 API 新增作用領域記憶體

傳回的值是記憶體 ID。記錄的類型會在建立時選取，且無法使用 update_memory() 變更。

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

從記憶體 API 更新記憶體

使用 update_memory() 可依 ID 取代現有類似記憶體之記錄的已儲存內容或描述資料。

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

新增具有自訂 ID 的記憶體

傳回的值是呼叫程式提供的記憶體 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

執行緒基本知識

新增訊息到討論串

訊息可以用字典或 Message 物件的形式傳送。選用訊息 ID、時間戳記和中繼資料可以與它們一起儲存。

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 參照：訊息

讀取回溯繫線訊息

您可以使用 start 和 end 讀取所有儲存的訊息或片段。

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_thread()。從不同的繫線傳送 ID 仍然會傳回 0。

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.

依 ID 更新目前執行緒的訊息

執行緒作用領域訊息更新只會影響目前執行緒所擁有的原始訊息。系統會保留儲存的角色和時戳值，當啟用自動擷取時，編輯訊息內容會使用與 add_messages() 相同的歷史記錄視窗規則，立即重新擷取編輯的訊息。只有較早的執行緒歷史記錄才能用來作為支援相關資訊環境。之後的訊息會在立即傳遞時被忽略，而相同的重新整理會保留現有的衍生記憶體，同時從編輯的內容新增任何新的記憶體。

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.

從繫線處理新增記憶體

傳回的值是記憶體 ID。此範例會儲存可重複使用的準則；請針對這些類別使用 memory_type="fact" 或 memory_type="preference"，或針對一般記憶體加以省略。

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

依 ID 更新目前繫線的記憶體

執行緒作用領域更新只會影響目前執行緒所擁有的類似記憶體記錄。從另一個繫線傳送 ID 會產生 KeyError。

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.

依 ID 刪除目前繫線的記憶體

執行緒刪除會作用領域至目前的執行緒。從不同的繫線傳送 ID 會傳回 0。

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.

建立內容卡

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 = 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.

建置摘要 (不包括最後 N 則訊息)

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

使用權杖預算建立摘要

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
#...

搜尋

從沒有明確範圍的繫線搜尋

當您未通過明確的範圍時，執行緒層次搜尋會使用執行緒預設值。

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 參照：OracleSearchResult

從具有範圍的記憶體 API 搜尋

在 API 層次，您可以透過 user_id、agent_id 和 thread_id 到 SearchScope 進行範圍擷取。對於最上層用戶端搜尋，請提供明確的使用者範圍。只有當您刻意想要取消範圍的記錄時，才使用 user_id=None。如需省略值、明確 None 和完全相符旗標在每個 API 層解析的摘要，請參閱範圍解析。

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 參照：SearchScope

使用中繼資料篩選搜尋

當搜尋應該只考慮其儲存的描述資料包含要求的部分對應之記錄時，請使用 metadata_filter。多個篩選索引鍵與 AND 語意結合、巢狀字典符合巢狀描述資料欄位，而裸機清單值必須完全相符。若要測試陣列成員身分，請使用欄位層次運算子字典，例如 {"tags": {"$array_contains": "outdoor"}}。含有清單的 "$array_contains" 需要所有列出的值，"$array_contains_any" 至少需要一個列出的值，而 "$not" 會否定相同欄位的另一個欄位層次表示式，包括運算子字典或原始完全相符值。否定表示式也會在正數表示式失敗時 (包括遺漏的欄位) 比對。否定陣列成員身分也與非陣列欄位相符。

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.']

僅搜尋備忘錄或僅搜尋訊息

使用 record_types 可將搜尋結果限制為特定儲存的記錄類型。

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.']

完整代碼

複製後續的完整代碼。

#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.']