快速参考代码示例

此页收集常见 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 参考：Embedder

配置定制 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 混合数据库内存组件

此变体允许 Oracle 管理的混合搜索基于存储的搜索文本，并显示如何选择托管的搜索索引同步模式。

SearchStrategy.HYBRID 创建或验证 Oracle 的托管混合向量索引，并要求主嵌入程序为 OracleDBEmbedder，以便托管索引使用嵌入程序的数据库内模型。SearchStrategy.KEYWORD 是纯文本的：它按存储的搜索文本排列，不需要嵌入。可以在没有本地向量存储的情况下创建关键字方案，因此，除非先重新创建方案或回填嵌入，否则不要使用 SearchStrategy.VECTOR 重新打开关键字方案。它们仍然可以通过 OracleDBEmbedder 升级到混合搜索，因为 Oracle 托管的混合索引基于存储的搜索文本构建。

警告：首次基于现有数据创建混合索引时，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

OracleDBEmbedder

SchemaPolicy

搜索策略

搜索索引同步模式

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() 可持久保存线程元数据或持久的运行时配置更改。传递到 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"。返回的值是内存标识符。

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 添加范围内存

返回的值是内存标识符。创建记录时会选择记录类型，并且无法使用 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() 按标识符替换现有类似内存的记录所存储的内容或元数据。

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 的内存

返回的值是调用方提供的内存标识符。

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

按 ID 从当前线程中删除消息

删除消息只会从当前线程中删除原始消息行。根据该消息创建的派生记忆或其他下游构件可以保持可搜索状态，并且仍可能影响上下文卡输出。如果需要删除线程及其关联的消息和内存，请改用 delete_thread()。从其他线程传递标识符仍然返回 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.

从线程句柄添加内存

返回的值是内存标识符。此示例存储可重用的准则；对这些类别使用 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 更新当前线程中的内存

线程范围的更新仅影响当前线程拥有的类似内存的记录。从另一个线程传递标识符会引发 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 从当前线程中删除内存

线程删除的范围将限定到当前线程。从其他线程传递标识符将返回 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 层如何解析的摘要，请参见 Scope Resolution 。

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