快速參考代碼範例

本文針對一般代理程式記憶體設定和 API 生命週期作業收集小型且重點的範例。

LLM / 內嵌設定

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

設定 LLM

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",
)

response = llm.generate("What is 2+2?")
print(response.text)

輸出：

2+2 is equal to 4

設定內嵌模型

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

設定代理程式記憶體元件

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

import oracledb

from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory

db_pool = oracledb.SessionPool(
    user="YOUR DB USER",
    password="YOUR DB PASSWORD",
    dsn="YOUR DB CONNECT STRING",
)
memory = OracleAgentMemory(
    connection=db_pool,
    embedder=embedder,
    llm=llm,  # optional: enables automatic memory extraction during add_messages()
)

設定 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="YOUR DB CONNECT STRING",
)
db_memory = OracleAgentMemory(
    connection=db_pool,
    embedder=embedder,
    llm=llm,  # optional
    schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
    memory_store_id="DEV_",
)

設定 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,  # optional
    schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
    search_strategy=SearchStrategy.HYBRID,
    search_index_sync=SearchIndexSyncMode.ON_COMMIT,
)

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

重新開啟現有的討論串

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)

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"])
print(persisted_thread.max_message_token_length)

輸出：

8000
True
8000

傳送至 get_thread() 的覆寫是暫時性的。呼叫 update_thread() 以保存繫線描述資料或持續的 Runtime-config 變更。

刪除執行緒

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

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

deleted = memory.delete_thread("thread_delete_123")
print(deleted)

輸出：

1

當您需要執行緒作用領域的連鎖清除時，請使用執行緒刪除。它會移除執行緒及其訊息、記憶體，以及由 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 時，記憶體不會繫結至特定繫線。傳回的值是記憶體 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。

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": "chat", "channel": "web"},
        ),
        {
            "id": "msg_assistant_001",
            "role": "assistant",
            "content": "Noted. I will keep that in mind.",
            "timestamp": "2026-03-27T09:00:05Z",
            "metadata": {"source": "assistant"},
        },
    ]
)

print(message_ids)

輸出：

['msg_user_001', 'msg_assistant_001']

讀取回溯繫線訊息

您可以使用 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])

print([message.content for message in all_messages])

print([message.content for message in middle_messages])

輸出：

在短時間內，繫結的預設值仍會傳回所有訊息。

['Message 1', 'Message 2', 'Message 3']

['Message 1', 'Message 2', 'Message 3']

['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

這只會移除目前執行緒中的原始訊息資料列。從該訊息建立的衍生記憶體或其他順流使用者自建物件不會自動刪除，且可能保持可搜尋或顯示在相關資訊環境卡輸出中。使用 memory.delete_thread(thread.thread_id) 可將繫線與其相關的訊息和記憶體一起刪除。訊息刪除會傳回其他執行緒所擁有 ID 的 0。

依 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

訊息更新會保留儲存的角色和時戳值。啟用自動擷取時，內容編輯會使用與 add_messages() 相同的歷史記錄視窗規則，立即重新執行已編輯訊息的擷取。之後的訊息會在立即傳遞時被忽略。在重新整理期間新增編輯內容記憶時，會保留現有的衍生記憶。

從繫線處理新增記憶體

傳回的值是記憶體 ID。

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

執行緒更新的作用領域為目前的執行緒，並發出 KeyError 以取得其他執行緒所擁有的遺漏 ID 或 ID。

依 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

執行緒刪除會作用領域至目前的執行緒，並針對另一個執行緒所擁有的 ID 傳回 0。

建立內容卡

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 搜尋

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

使用中繼資料篩選搜尋

當搜尋應該只考慮其儲存的描述資料包含要求的部分對應之記錄時，請使用 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])

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

['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])

print([result.content for result in message_results])

輸出：

['The user likes pizza.']

['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

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",
)

response = llm.generate("What is 2+2?")
print(response.text)
#2+2 is equal to 4


##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 an Oracle Memory component

import oracledb

from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory

db_pool = oracledb.SessionPool(
    user="YOUR DB USER",
    password="YOUR DB PASSWORD",
    dsn="YOUR DB CONNECT STRING",
)

memory = OracleAgentMemory(
    connection=db_pool,
    embedder=embedder,
    llm=llm,  # optional: enables automatic memory extraction during add_messages()
)


##Configure an Oracle DB component

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="YOUR DB CONNECT STRING",
)

db_memory = OracleAgentMemory(
    connection=db_pool,
    embedder=embedder,
    llm=llm,  # optional
    schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
    memory_store_id="DEV_",
)


##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,  # optional
    schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
    search_strategy=SearchStrategy.HYBRID,
    search_index_sync=SearchIndexSyncMode.ON_COMMIT,
)


##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.",
    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.",
    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": "chat", "channel": "web"},
        ),
        {
            "id": "msg_assistant_001",
            "role": "assistant",
            "content": "Noted. I will keep that in mind.",
            "timestamp": "2026-03-27T09:00:05Z",
            "metadata": {"source": "assistant"},
        },
    ]
)

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("The user likes jasmine tea.")
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.']