快速参考代码示例

本文收集了常见代理内存设置和 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 混合数据库内存组件

此变量支持 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,  # 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() 可持久保存线程元数据或持久的运行时配置更改。传递到 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() 以持久保存线程元数据或持久的运行时配置更改。

删除线程

需要线程范围级联清除时使用此操作。它将删除线程以及由 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 时，内存不会绑定到特定线程。返回的值是内存标识符。

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

返回的值是内存标识符。

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

按 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

这仅从当前线程中删除原始消息行。从该消息创建的派生存储器或其他下游构件不会自动删除，并且可能会保持可搜索状态或显示在上下文卡输出中。使用 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() 相同的历史记录窗口规则重新运行已编辑消息的提取。在该即时传递期间将忽略后面的消息。在刷新期间添加新的编辑内容存储器时，现有派生存储器将保持原样。

从线程句柄添加内存

返回的值是内存标识符。

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

线程更新将限定到当前线程，并针对其他线程拥有的缺失 ID 或 ID 引发 KeyError。

按 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 层如何解析的摘要，请参见 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.']

使用元数据筛选进行搜索

当搜索应仅考虑其存储元数据包含请求的部分映射的记录时，请使用 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.']