Esempi di codici di riferimento rapido
Questo articolo raccoglie esempi piccoli e mirati per le operazioni comuni di impostazione della memoria agente e del ciclo di vita delle API.
LLM/Impostazione incorporamento
Gli esempi seguenti utilizzano LiteLLM sia per il modello LLM che per il modello di incorporamento.
Configurazione di 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)
Output:
2+2 is equal to 4
Configurazione di un modello di incorporamento
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)
Output:
(1, embedding_dimension)
Impostazione API
Configurazione di un componente di memoria agente
In questo modo vengono utilizzati una connessione o un pool Oracle DB insieme al modello di incorporamento e un LLM facoltativo per l'estrazione automatica della 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()
)
Configurazione di un componente di memoria di Oracle DB
Questa variante utilizza una connessione o un pool Oracle DB e mostra come impostare un criterio di schema e un prefisso del nome di tabella.
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_",
)
Configurazione di un componente di memoria Oracle Hybrid DB
Questa variante consente la ricerca ibrida gestita da Oracle rispetto al testo di ricerca memorizzato e mostra come scegliere la modalità di sincronizzazione gestita dell'indice di ricerca.
SearchStrategy.HYBRID crea o convalida l'indice vettoriale ibrido gestito di Oracle e richiede che l'incorporatore principale sia un OracleDBEmbedder in modo che l'indice gestito utilizzi il modello in-database dell'incorporatore. SearchStrategy.KEYWORD è solo testuale: si classifica in base al testo di ricerca memorizzato e non richiede un incorporamento. Gli schemi delle parole chiave possono essere creati senza memoria vettoriale locale, quindi non riaprire gli schemi delle parole chiave con SearchStrategy.VECTOR a meno che non si crei lo schema o si esegua prima il backfill delle integrazioni. È comunque possibile eseguire l'upgrade alla ricerca ibrida con un OracleDBEmbedder perché l'indice ibrido gestito di Oracle si basa sul testo di ricerca memorizzato.
Avvertenza: quando un indice ibrido viene creato per la prima volta su dati esistenti, Oracle analizza il testo di ricerca memorizzato e crea lo stato dell'indice gestito durante l'impostazione dello schema. SchemaPolicy.CREATE_IF_NECESSARY può richiedere tempo e deve essere pianificato come una migrazione del database per schemi di grandi dimensioni.
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 di vita API
Crea discussione
Creare un thread con un ID thread, un ID utente e un ID agente facoltativi.
thread = memory.create_thread(
thread_id="thread_create_123", # optional
user_id="user_123", # optional
agent_id="agent_456", # optional
)
print(thread.thread_id)
Output:
thread_create_123
Riapri un thread esistente
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)
Output:
thread_reopen_123
Aggiorna un thread esistente
Utilizzare update_thread() per rendere persistenti i metadati del thread o le modifiche durature della configurazione runtime. Gli override passati a get_thread() influiscono solo sull'handle riaperto finché non vengono resi persistenti in modo esplicito.
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)
Output:
8000
True
8000
Le sostituzioni passate a get_thread() sono temporanee. Chiamare update_thread() per rendere persistenti i metadati del thread o le modifiche durature della configurazione runtime.
Elimina un thread
Utilizzare questa operazione quando è necessario eseguire il cleanup a cascata con ambito thread. Rimuove il thread insieme a messaggi associati, memorie durevoli e dati di recupero di backup gestiti dall'SDK.
thread = memory.create_thread(thread_id="thread_delete_123")
deleted = memory.delete_thread("thread_delete_123")
print(deleted)
Output:
1
Utilizzare l'eliminazione dei thread quando è necessario eseguire la pulizia a cascata con ambito thread. Rimuove il thread insieme ai suoi messaggi, memorie e dati di recupero di supporto gestiti dall'SDK.
Aggiungi un profilo utente
user_profile_id = memory.add_user(
"user_123",
"The user prefers concise answers and works mostly with Python.",
)
print(user_profile_id)
Output:
user_123
Aggiungi un profilo agente
agent_profile_id = memory.add_agent(
"agent_456",
"A coding assistant specialized in debugging and code review.",
)
print(agent_profile_id)
Output:
agent_456
Aggiungere una memoria globale dall'API di memoria
Quando thread_id viene omesso, la memoria non è legata a un thread specifico. Il valore restituito è l'identificativo di 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)
Output:
mem:1
Aggiungere una memoria con ambito dall'API di memoria
Il valore restituito è l'identificativo di 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)
Output:
mem:2
Aggiorna una memoria dall'API di memoria
Utilizzare update_memory() per sostituire il contenuto o i metadati memorizzati per un record simile alla memoria esistente in base all'identificativo.
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)
Output:
mem:3
Aggiungere una memoria con un ID personalizzato
Il valore restituito è l'identificativo di memoria fornito dal chiamante.
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)
Output:
travel_pref_001
Informazioni di base thread
Aggiungi messaggi a una discussione
I messaggi possono essere passati come dizionari o come oggetti Message. È possibile memorizzare con loro ID messaggio, indicatori orari e metadati facoltativi.
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)
Output:
['msg_user_001', 'msg_assistant_001']
Messaggi contenuti nella discussione
È possibile leggere tutti i messaggi memorizzati o una slice utilizzando 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])
Output:
Nei thread brevi, il valore predefinito associato restituisce comunque tutti i messaggi.
['Message 1', 'Message 2', 'Message 3']
['Message 1', 'Message 2', 'Message 3']
['Message 2', 'Message 3']
Elimina un messaggio dal thread corrente per ID
L'eliminazione di un messaggio rimuove solo la riga del messaggio raw dal thread corrente. Le memorie derivate o altri artifact a valle creati da quel messaggio possono rimanere ricercabili e possono comunque influenzare l'output della scheda contesto. Se è necessario eliminare il thread insieme ai messaggi e alle memorie associati, utilizzare delete_thread(). Se si passa un identificativo da un thread diverso, viene comunque restituito 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)
Output:
1
In questo modo viene rimossa solo la riga del messaggio raw dal thread corrente. Le memorie derivate o altri artifact a valle creati da quel messaggio non vengono eliminati automaticamente e potrebbero rimanere ricercabili o apparire nell'output della scheda contesto. Utilizzare memory.delete_thread(thread.thread_id) per eliminare il thread insieme ai messaggi e alle memorie associati. Il messaggio elimina il valore restituito 0 per gli ID di proprietà di un altro thread.
Aggiorna un messaggio dal thread corrente per ID
Gli aggiornamenti con ambito thread interessano solo i messaggi raw di proprietà del thread corrente. I valori del ruolo memorizzato e dell'indicatore orario vengono conservati e, quando l'estrazione automatica è abilitata, la modifica del contenuto del messaggio esegue immediatamente l'estrazione per il messaggio modificato utilizzando le stesse regole della finestra della cronologia di add_messages(). Solo la cronologia thread precedente può essere utilizzata come contesto di supporto. I messaggi successivi vengono ignorati durante quel passaggio immediato e lo stesso aggiornamento mantiene in posizione le memorie derivate esistenti mentre si aggiungono nuove memorie dal contenuto modificato.
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)
Output:
msg_update_001
Gli aggiornamenti dei messaggi conservano i valori memorizzati del ruolo e dell'indicatore orario. Quando l'estrazione automatica è abilitata, le modifiche al contenuto eseguono immediatamente di nuovo l'estrazione per il messaggio modificato utilizzando le stesse regole della finestra della cronologia di add_messages(). I messaggi successivi vengono ignorati durante quel passaggio immediato. Le memorie derivate esistenti rimangono in vigore mentre durante tale aggiornamento vengono aggiunte nuove memorie a contenuto modificato.
Aggiungere una memoria da una maniglia thread
Il valore restituito è l'identificativo di 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)
Output:
mem:4
Aggiorna una memoria dal thread corrente per ID
Gli aggiornamenti con ambito thread interessano solo i record simili alla memoria di proprietà del thread corrente. Se si passa un identificativo da un altro thread, viene generato 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)
Output:
mem:5
Gli aggiornamenti dei thread vengono applicati al thread corrente e generano KeyError per gli ID o gli ID mancanti di proprietà di un altro thread.
Eliminare una memoria dal thread corrente per ID
Le eliminazioni thread sono definite nel thread corrente. Se si passa un identificativo da un thread diverso, viene restituito 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)
Output:
1
Le eliminazioni thread sono definite nel thread corrente e restituiscono 0 per gli ID di proprietà di un altro thread.
Creare una scheda contesto
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)
Output:
<context_card>
The user is planning a trip to Kyoto.
</context_card>
Crea un riepilogo 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)
Output:
user (-): Hello
- assistant (-): Hi, how can I help?
- user (-): Please summarize this thread.
Creare un riepilogo che escluda gli ultimi N messaggi
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)
Output:
user (-): First message
- assistant (-): Second message
Crea un riepilogo con un budget 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)
Output:
(truncated)
user (-): Message 1
...
Cerca
Cerca da un thread senza ambito esplicito
La ricerca a livello di thread utilizza le impostazioni predefinite del thread quando non si passa un ambito esplicito.
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])
Output:
['The user likes pizza.']
Cerca dall'API di memoria con ambito
A livello di API, è possibile recuperare l'ambito con user_id, agent_id e thread_id tramite SearchScope. Per le ricerche client di livello superiore, fornire un ambito utente esplicito. Utilizzare user_id=None solo se si desidera intenzionalmente record senza ambito. Per un riepilogo della risoluzione dei valori omessi, dei flag espliciti None e della corrispondenza esatta in ogni livello API, vedere Risoluzione ambito.
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])
Output:
['The user likes hiking in the Alps.']
Cerca con filtro metadati
Utilizzare metadata_filter quando la ricerca deve considerare solo i record i cui metadati memorizzati contengono un mapping parziale richiesto. Più chiavi di filtro vengono combinate con la semantica AND, i dizionari nidificati corrispondono ai campi di metadati nidificati e i valori della lista Bare devono corrispondere esattamente. Per eseguire il test dell'appartenenza all'array, utilizzare un dizionario operatore a livello di campo, ad esempio {"tags": {"$array_contains": "outdoor"}}. "$array_contains" con un elenco richiede tutti i valori elencati, "$array_contains_any" richiede almeno un valore elencato e "$not" nega un'altra espressione a livello di campo nello stesso campo, incluso un dizionario operatore o un valore raw di corrispondenza esatta. Un'espressione negata corrisponde anche quando l'espressione positiva non riesce, inclusi i campi mancanti. L'appartenenza all'array negativa corrisponde anche ai campi non array.
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])
Output:
['The user likes alpine hiking.']
['The user likes alpine hiking.']
Cerca solo memorie o solo messaggi
Utilizzare record_types per limitare i risultati della ricerca a tipi di record memorizzati specifici.
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])
Output:
['The user likes pizza.']
['I mentioned pizza in a message.']
Codice 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.']