Amostras de código de referência rápida
Este artigo coleta exemplos pequenos e focados para configuração comum de Memória do Agente e operações de ciclo de vida da API.
Configuração do LLM / Incorporação
Os exemplos a seguir usam o LiteLLM para o modelo de LLM e incorporação.
Configurando um 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)
Saída:
2+2 is equal to 4
Configurando um Modelo Incorporado
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)
Saída:
(1, embedding_dimension)
Configuração da API
Configurando um Componente de Memória do Agente
Isso usa uma conexão ou um pool do Oracle DB junto com o modelo de incorporação e um LLM opcional para extração automática de memória.
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()
)
Configurando um Componente de Memória do Oracle DB
Essa variante usa uma conexão ou um pool do Oracle DB e mostra como definir uma política de esquema e um prefixo de nome de tabela.
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_",
)
Configurando um Componente de Memória do Oracle Hybrid DB
Esta variante permite a pesquisa híbrida gerenciada pela Oracle sobre o texto de pesquisa armazenado e mostra como escolher o modo de sincronização de índice de pesquisa gerenciado.
O SearchStrategy.HYBRID cria ou valida o índice vetorial híbrido gerenciado da Oracle e requer que o incorporador principal seja um OracleDBEmbedder para que o índice gerenciado use o modelo no banco de dados do incorporador. SearchStrategy.KEYWORD é somente texto: ele é classificado pelo texto de pesquisa armazenado e não requer um incorporador. Os esquemas de palavra-chave podem ser criados sem armazenamento vetorial local; portanto, não reabra os esquemas de palavra-chave com SearchStrategy.VECTOR, a menos que você recrie o esquema ou preencha as incorporações primeiro. Eles ainda podem ser atualizados para pesquisa híbrida com um OracleDBEmbedder porque o índice híbrido gerenciado da Oracle é criado com base no texto de pesquisa armazenado.
Advertência: Quando um índice híbrido é criado pela primeira vez sobre dados existentes, o sistema Oracle verifica o texto de pesquisa armazenado e cria o estado do índice gerenciado durante a configuração do esquema. O SchemaPolicy.CREATE_IF_NECESSARY pode levar tempo e deve ser planejado como uma migração de banco de dados para esquemas grandes.
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,
)
Ciclo de Vida da API
Criar um tópico
Crie um thread com um ID de thread, ID de usuário e ID de agente opcionais.
thread = memory.create_thread(
thread_id="thread_create_123", # optional
user_id="user_123", # optional
agent_id="agent_456", # optional
)
print(thread.thread_id)
Saída:
thread_create_123
Reabrir um tópico existente
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)
Saída:
thread_reopen_123
Atualizar um tópico existente
Use update_thread() para persistir metadados de thread ou alterações duráveis de configuração de runtime. As substituições passadas para get_thread() afetam somente o handle reaberto até que sejam explicitamente persistidas.
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)
Saída:
8000
True
8000
As substituições passadas para get_thread() são temporárias. Chame update_thread() para persistir metadados de thread ou alterações duráveis de configuração de runtime.
Excluir um tópico
Use esta operação quando precisar de limpeza em cascata com escopo de thread. Ele remove o thread junto com mensagens associadas, memórias duráveis e dados de recuperação de backup gerenciados pelo SDK.
thread = memory.create_thread(thread_id="thread_delete_123")
deleted = memory.delete_thread("thread_delete_123")
print(deleted)
Saída:
1
Use a exclusão de thread quando precisar de limpeza em cascata com escopo de thread. Ele remove o thread junto com suas mensagens, memórias e dados de recuperação de backup gerenciados pelo SDK.
Adicionar um Perfil de Usuário
user_profile_id = memory.add_user(
"user_123",
"The user prefers concise answers and works mostly with Python.",
)
print(user_profile_id)
Saída:
user_123
Adicionar um Perfil de Agente
agent_profile_id = memory.add_agent(
"agent_456",
"A coding assistant specialized in debugging and code review.",
)
print(agent_profile_id)
Saída:
agent_456
Adicionar uma Memória Global da API de Memória
Quando thread_id é omitido, a memória não é vinculada a um thread específico. O valor retornado é o identificador de memória.
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)
Saída:
mem:1
Adicionar uma Memória com Escopo da API de Memória
O valor retornado é o identificador de memória.
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)
Saída:
mem:2
Atualizar uma Memória da API de Memória
Use update_memory() para substituir conteúdo armazenado ou metadados de um registro semelhante à memória existente por identificador.
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)
Saída:
mem:3
Adicionar uma Memória com um ID Personalizado
O valor retornado é o identificador de memória fornecido pelo chamador.
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)
Saída:
travel_pref_001
Noções Básicas de Thread
Adicionar mensagens a um tópico
As mensagens podem ser passadas como dicionários ou como objetos Message. IDs de mensagem, timestamps e metadados opcionais podem ser armazenados com eles.
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)
Saída:
['msg_user_001', 'msg_assistant_001']
Mensagens de Read Back Thread
Você pode ler todas as mensagens armazenadas ou um segmento usando start e 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])
Saída:
Em threads curtos, o padrão delimitado ainda retorna todas as mensagens.
['Message 1', 'Message 2', 'Message 3']
['Message 1', 'Message 2', 'Message 3']
['Message 2', 'Message 3']
Excluir uma Mensagem do Thread Atual por ID
A exclusão de uma mensagem remove somente a linha de mensagem bruta do thread atual. Memórias derivadas ou outros artefatos downstream criados a partir dessa mensagem podem permanecer pesquisáveis e ainda podem influenciar a saída do cartão de contexto. Se você precisar excluir o thread junto com suas mensagens e memórias associadas, use delete_thread(). A transmissão de um identificador de outro thread ainda retorna 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)
Saída:
1
Isso remove apenas a linha de mensagem bruta do thread atual. Memórias derivadas ou outros artefatos downstream criados a partir dessa mensagem não são excluídos automaticamente e podem permanecer pesquisáveis ou aparecer na saída do cartão de contexto. Use memory.delete_thread(thread.thread_id) para excluir o thread junto com suas mensagens e memórias associadas. A mensagem exclui o retorno 0 para IDs pertencentes a outro thread.
Atualizar uma Mensagem do Thread Atual por ID
As atualizações de mensagem no escopo do thread afetam somente as mensagens brutas pertencentes ao thread atual. Os valores de atribuição e timestamp armazenados são preservados e, quando a extração automática é ativada, a edição do conteúdo da mensagem reexecuta imediatamente a extração da mensagem editada usando as mesmas regras da janela de histórico que add_messages(). Somente o histórico de thread anterior pode ser usado como contexto de suporte. Mensagens posteriores são ignoradas durante esse passe imediato, e a mesma atualização mantém memórias derivadas existentes no lugar enquanto adiciona novas memórias do conteúdo editado.
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)
Saída:
msg_update_001
As atualizações de mensagem preservam valores de atribuição e timestamp armazenados. Quando a extração automática está ativada, o conteúdo edita imediatamente a extração para a mensagem editada usando as mesmas regras da janela de histórico que add_messages(). Mensagens posteriores são ignoradas durante esse passe imediato. As memórias derivadas existentes permanecem no lugar enquanto novas memórias de conteúdo editado são adicionadas durante essa atualização.
Adicionar Memória de um Identificador de Thread
O valor retornado é o identificador de memória.
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)
Saída:
mem:4
Atualizar uma Memória do Thread Atual por ID
As atualizações no escopo do thread afetam somente os registros semelhantes à memória pertencentes ao thread atual. A transmissão de um identificador de outro thread gera 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)
Saída:
mem:5
As atualizações de thread têm como escopo o thread atual e geram KeyError para IDs ausentes ou IDs pertencentes a outro thread.
Excluir uma Memória do Thread Atual por ID
As exclusões de thread têm escopo para o thread atual. A transmissão de um identificador de outro thread retorna 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)
Saída:
1
As exclusões de thread têm como escopo o thread atual e retornam 0 para IDs pertencentes a outro thread.
Criar um Cartão de Contexto
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)
Saída:
<context_card>
The user is planning a trip to Kyoto.
</context_card>
Criar um Resumo de Thread
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)
Saída:
user (-): Hello
- assistant (-): Hi, how can I help?
- user (-): Please summarize this thread.
Criar um Resumo Excluindo as N Últimas Mensagens
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)
Saída:
user (-): First message
- assistant (-): Second message
Criar um Resumo com um Orçamento de Token
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)
Saída:
(truncated)
user (-): Message 1
...
Pesquisar
Pesquisar em um Thread sem Escopo Explícito
A pesquisa em nível de thread usa os padrões de thread quando você não especifica um escopo.
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])
Saída:
['The user likes pizza.']
Pesquisar na API de Memória com Escopo
No nível da API, você pode definir o escopo da recuperação com user_id, agent_id e thread_id até SearchScope. Para pesquisas de cliente de nível superior, forneça um escopo de usuário explícito. Use user_id=None somente quando quiser intencionalmente registros sem escopo. Para obter um resumo de como os valores omitidos, None explícito e sinalizadores de correspondência exata são resolvidos em cada camada da API, consulte Resolução de Escopo.
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])
Saída:
['The user likes hiking in the Alps.']
Pesquisar com Filtragem de Metadados
Use metadata_filter quando a pesquisa só deve considerar registros cujos metadados armazenados contenham um mapeamento parcial solicitado. Várias chaves de filtro são combinadas com a semântica AND, os dicionários aninhados correspondem a campos de metadados aninhados e os valores de lista nua devem corresponder exatamente. Para testar a associação de array, use um dicionário de operador de nível de campo, como {"tags": {"$array_contains": "outdoor"}}. "$array_contains" com uma lista requer todos os valores listados, "$array_contains_any" requer pelo menos um valor listado e "$not" nega outra expressão de nível de campo no mesmo campo, incluindo um dicionário de operador ou valor de correspondência exata bruto. Uma expressão negada também corresponde quando a expressão positiva falharia, incluindo campos ausentes. A associação de matriz negativa também corresponde a campos que não são de matriz.
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])
Saída:
['The user likes alpine hiking.']
['The user likes alpine hiking.']
Pesquisar Somente Memórias ou Somente Mensagens
Use record_types para restringir os resultados da pesquisa a tipos de registro armazenados específicos.
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])
Saída:
['The user likes pizza.']
['I mentioned pizza in a message.']
Código Inteiro
#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.']