Ejemplos de código de referencia rápida
Este artículo recopila ejemplos pequeños y específicos para la configuración común de la memoria del agente y las operaciones del ciclo de vida de la API.
Configuración de LLM/incrustación
Los siguientes ejemplos utilizan LiteLLM tanto para el modelo LLM como para el modelo de incrustación.
Configuración de un 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)
Salida:
2+2 is equal to 4
Configuración de un modelo embebido
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)
Salida:
(1, embedding_dimension)
Configuración de API
Configuración de un Componente de Memoria de Agente
Utiliza una conexión o un pool de Oracle DB junto con el modelo de embebido y un LLM opcional para la extracción automática de memoria.
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()
)
Configuración de un Componente de Memoria de Oracle DB
Esta variante utiliza una conexión o un pool de Oracle DB y muestra cómo definir una política de esquema y un prefijo de nombre de tabla.
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_",
)
Configuración de un Componente de Memoria de BD Híbrida de Oracle
Esta variante permite la búsqueda híbrida gestionada por Oracle sobre el texto de búsqueda almacenado y muestra cómo seleccionar el modo de sincronización del índice de búsqueda gestionado.
SearchStrategy.HYBRID crea o valida el índice vectorial híbrido gestionado de Oracle y necesita que el embebido principal sea un OracleDBEmbedder para que el índice gestionado utilice el modelo de base de datos del embebido. SearchStrategy.KEYWORD es de solo texto: se clasifica por texto de búsqueda almacenado y no necesita un embebido. Los esquemas de palabras clave se pueden crear sin almacenamiento de vectores local, por lo que no vuelva a abrir los esquemas de palabras clave con SearchStrategy.VECTOR a menos que vuelva a crear primero el esquema o las incrustaciones de relleno. Aún se pueden actualizar a la búsqueda híbrida con OracleDBEmbedder porque el índice híbrido gestionado de Oracle se crea a partir del texto de búsqueda almacenado.
Advertencia: cuando se crea por primera vez un índice híbrido a través de datos existentes, Oracle explora el texto de búsqueda almacenado y crea el estado del índice gestionado durante la configuración del esquema. SchemaPolicy.CREATE_IF_NECESSARY puede llevar tiempo y se debe planificar como una migración de base de datos 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 de API
Crear un tema
Cree un thread con un ID de thread, un ID de usuario y un ID de agente opcionales.
thread = memory.create_thread(
thread_id="thread_create_123", # optional
user_id="user_123", # optional
agent_id="agent_456", # optional
)
print(thread.thread_id)
Salida:
thread_create_123
Volver a abrir un thread 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)
Salida:
thread_reopen_123
Actualización de un thread existente
Utilice update_thread() para mantener los metadatos del thread o los cambios duraderos en la configuración en tiempo de ejecución. Las sustituciones transferidas a get_thread() solo afectan al manejador reabierto hasta que se mantienen explícitamente.
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)
Salida:
8000
True
8000
Las sustituciones transferidas a get_thread() son temporales. Llame a update_thread() para mantener los metadatos del thread o los cambios duraderos en la configuración en tiempo de ejecución.
Borrar un tema
Utilice esta operación cuando necesite una limpieza en cascada con ámbito de thread. Elimina el thread junto con los mensajes asociados, los recuerdos duraderos y los datos de recuperación de copia de seguridad gestionados por el SDK.
thread = memory.create_thread(thread_id="thread_delete_123")
deleted = memory.delete_thread("thread_delete_123")
print(deleted)
Salida:
1
Utilice la supresión de threads cuando necesite una limpieza en cascada de ámbito de thread. Elimina el thread junto con sus mensajes, memorias y datos de recuperación de copia de seguridad gestionados por el SDK.
Agregar un perfil de usuario
user_profile_id = memory.add_user(
"user_123",
"The user prefers concise answers and works mostly with Python.",
)
print(user_profile_id)
Salida:
user_123
Agregar un perfil de agente
agent_profile_id = memory.add_agent(
"agent_456",
"A coding assistant specialized in debugging and code review.",
)
print(agent_profile_id)
Salida:
agent_456
Adición de una memoria global desde la API de memoria
Cuando se omite thread_id, la memoria no está vinculada a un thread específico. El valor devuelto es el identificador de memoria.
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)
Salida:
mem:1
Adición de una memoria con ámbito desde la API de memoria
El valor devuelto es el identificador de memoria.
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)
Salida:
mem:2
Actualización de una memoria desde la API de memoria
Utilice update_memory() para sustituir el contenido almacenado o los metadatos de un registro similar a la memoria 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)
Salida:
mem:3
Adición de una Memoria con un ID Personalizado
El valor devuelto es el identificador de memoria proporcionado por el emisor de llamada.
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)
Salida:
travel_pref_001
Conceptos Básicos de los Temas
Agregar mensajes a un tema
Los mensajes se pueden transferir como diccionarios o como objetos Message. Los ID de mensajes opcionales, los registros de hora y los metadatos se pueden almacenar con ellos.
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)
Salida:
['msg_user_001', 'msg_assistant_001']
Mensajes de thread de readmisión
Puede leer todos los mensajes almacenados o un segmento mediante start y 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])
Salida:
En threads cortos, el valor por defecto enlazado sigue devolviendo todos los mensajes.
['Message 1', 'Message 2', 'Message 3']
['Message 1', 'Message 2', 'Message 3']
['Message 2', 'Message 3']
Eliminar un mensaje del thread actual por ID
Al suprimir un mensaje, solo se elimina la fila de mensaje sin formato del thread actual. Las memorias derivadas u otros artefactos descendentes creados a partir de ese mensaje pueden seguir siendo aptas para la búsqueda y pueden seguir influyendo en la salida de la tarjeta de contexto. Si necesita suprimir el thread junto con sus mensajes y memorias asociados, utilice delete_thread() en su lugar. Al transferir un identificador de un thread diferente, se devuelve 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)
Salida:
1
Esto elimina solo la fila de mensaje sin formato del thread actual. Las memorias derivadas u otros artefactos descendentes creados a partir de ese mensaje no se suprimen automáticamente y pueden seguir siendo aptas para búsqueda o aparecer en la salida de la tarjeta de contexto. Utilice memory.delete_thread(thread.thread_id) para suprimir el thread junto con sus mensajes y memorias asociados. Las supresiones de mensajes devuelven 0 para los ID que son propiedad de otro thread.
Actualización de un mensaje del thread actual por ID
Las actualizaciones de mensajes de ámbito de thread solo afectan a los mensajes raw que son propiedad del thread actual. Los valores de rol y registro de hora almacenados se conservan y, cuando se activa la extracción automática, la edición del contenido del mensaje vuelve a ejecutar inmediatamente la extracción para el mensaje editado utilizando las mismas reglas de ventana de historial que add_messages(). Solo se puede utilizar el historial de threads anterior como contexto de soporte. Los mensajes posteriores se ignoran durante ese pase inmediato, y el mismo refrescamiento mantiene las memorias derivadas existentes en su lugar mientras se agregan nuevas memorias del contenido 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)
Salida:
msg_update_001
Las actualizaciones de mensajes conservan los valores de registro de hora y rol almacenados. Cuando la extracción automática está activada, el contenido vuelve a ejecutar inmediatamente la extracción del mensaje editado utilizando las mismas reglas de ventana de historial que add_messages(). Los mensajes posteriores se ignoran durante ese pase inmediato. Las memorias derivadas existentes permanecen en su lugar mientras se agregan nuevas memorias de contenido editado durante esa actualización.
Adición de una memoria desde un manejador de threads
El valor devuelto es el identificador de memoria.
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)
Salida:
mem:4
Actualización de una Memoria del Thread Actual por ID
Las actualizaciones de ámbito de thread solo afectan a los registros similares a la memoria que son propiedad del thread actual. Al transferir un identificador de otro thread, se emite 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)
Salida:
mem:5
Las actualizaciones de thread tienen un ámbito en el thread actual y generan KeyError para los ID faltantes o los ID que son propiedad de otro thread.
Supresión de una Memoria del Thread Actual por ID
Las supresiones de subprocesos se limitan al subproceso actual. Al transferir un identificador de un thread diferente, se devuelve 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)
Salida:
1
Las supresiones de subprocesos se limitan al subproceso actual y devuelven 0 para los ID que son propiedad de otro subproceso.
Creación de una tarjeta contextual
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)
Salida:
<context_card>
The user is planning a trip to Kyoto.
</context_card>
Crear un resumen de tema
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)
Salida:
user (-): Hello
- assistant (-): Hi, how can I help?
- user (-): Please summarize this thread.
Creación de un resumen que excluya los últimos N mensajes
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)
Salida:
user (-): First message
- assistant (-): Second message
Crear un resumen con un presupuesto 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)
Salida:
(truncated)
user (-): Message 1
...
Búsqueda
Búsqueda desde un thread sin ámbito explícito
La búsqueda de nivel de thread utiliza los valores por defecto del thread cuando no se transfiere un ámbito explícito.
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])
Salida:
['The user likes pizza.']
Buscar desde la API de memoria con ámbito
En el nivel de API, puede restringir la recuperación con user_id, agent_id y thread_id mediante SearchScope. Para las búsquedas de cliente de nivel superior, proporcione un ámbito de usuario explícito. Utilice user_id=None solo cuando desee intencionalmente registros sin ámbito. Para obtener un resumen de cómo se resuelven los valores omitidos, los indicadores explícitos None y los indicadores de coincidencia exacta en cada capa de API, consulte 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])
Salida:
['The user likes hiking in the Alps.']
Buscar con filtrado de metadatos
Utilice metadata_filter cuando la búsqueda solo debe tener en cuenta los registros cuyos metadatos almacenados contienen una asignación parcial solicitada. Se combinan varias claves de filtro con la semántica AND, los diccionarios anidados coinciden con los campos de metadatos anidados y los valores de lista con hardware dedicado deben coincidir exactamente. Para probar la pertenencia a una matriz, utilice un diccionario de operador de nivel de campo, como {"tags": {"$array_contains": "outdoor"}}. "$array_contains" con una lista necesita todos los valores mostrados, "$array_contains_any" necesita al menos un valor mostrado y "$not" niega otra expresión de nivel de campo en el mismo campo, incluido un diccionario de operador o un valor de coincidencia exacta raw. Una expresión negada también coincide cuando falla la expresión positiva, incluidos los campos que faltan. La pertenencia a la matriz negada también coincide con los campos que no son 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])
Salida:
['The user likes alpine hiking.']
['The user likes alpine hiking.']
Buscar solo recuerdos o solo mensajes
Utilice record_types para restringir los resultados de búsqueda a tipos de registro almacenados 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])
Salida:
['The user likes pizza.']
['I mentioned pizza in a message.']
Código Completo
#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.']