Échantillons de code de référence rapide

Cet article collecte de petits exemples ciblés pour les opérations courantes de configuration de mémoire d'agent et de cycle de vie d'API.

LLM / Configuration de l'intégration

Les exemples suivants utilisent LiteLLM pour le LLM et le modèle d'intégration.

Configuration d'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)

Sortie :

2+2 is equal to 4

Configuration d'un modèle d'intégration

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)

Sortie :

(1, embedding_dimension)

Configuration d'API

Configurer un composant de mémoire d'agent

Cela utilise une connexion ou un pool Oracle DB avec le modèle d'intégration et un LLM facultatif pour l'extraction automatique de mémoire.

from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory

db_pool = ...  # an oracledb connection or connection pool

memory = OracleAgentMemory(
    connection=db_pool,
    embedder=embedder,
    llm=llm,  # optional: enables automatic memory extraction during add_messages()
)

Configurer un composant de mémoire Oracle DB

Cette variante utilise une connexion ou une réserve Oracle DB et montre comment définir une politique de schéma et un préfixe de nom de table.

from oracleagentmemory.core import SchemaPolicy
from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory

db_pool = ...  # an oracledb connection or connection pool

db_memory = OracleAgentMemory(
    connection=db_pool,
    embedder=embedder,
    llm=llm,  # optional
    schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
    table_name_prefix="DEV_",
)

Cycle de vie d'API

Créer un fil

Créez un fil avec un fil facultatif, un code d'utilisateur et un code d'agent.

thread = memory.create_thread(
    thread_id="thread_create_123",  # optional
    user_id="user_123",             # optional
    agent_id="agent_456",           # optional
)

print(thread.thread_id)

Sortie :

thread_create_123

Rouvrir un fil existant

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)

Sortie :

thread_reopen_123

Supprimer un fil

thread = memory.create_thread(thread_id="thread_delete_123")

deleted = memory.delete_thread("thread_delete_123")
print(deleted)

Sortie :

1

Ajouter un profil d'utilisateur

user_profile_id = memory.add_user(
    "user_123",
    "The user prefers concise answers and works mostly with Python.",
)

print(user_profile_id)

Sortie :

user_123

Ajouter un profil d'agent

agent_profile_id = memory.add_agent(
    "agent_456",
    "A coding assistant specialized in debugging and code review.",
)

print(agent_profile_id)

Sortie :

agent_456

Ajouter une mémoire globale à partir de l'API de mémoire

Lorsque thread_id est omis, la mémoire n'est pas liée à une unité d'exécution spécifique. La valeur retournée est l'identificateur de mémoire.

memory_id = memory.add_memory(
    "The user prefers short, bullet-point answers.",
    user_id="user_123",
    agent_id="agent_456",
)

print(memory_id)

Sortie :

mem:1

Ajouter une mémoire ciblée à partir de l'API de mémoire

La valeur retournée est l'identificateur de mémoire.

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)

Sortie :

mem:2

Ajouter une mémoire avec un ID personnalisé

La valeur retournée est l'identificateur de mémoire fourni par l'appelant.

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)

Sortie :

travel_pref_001

Informations de base sur le fil

Ajouter des messages à un fil

Les messages peuvent être transmis en tant que dictionnaires ou en tant qu'objets Message. Des ID de message, des horodatages et des métadonnées facultatifs peuvent être stockés avec eux.

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)

Sortie :

['msg_user_001', 'msg_assistant_001']

Messages du fil de lecture

Vous pouvez lire tous les messages stockés ou une tranche à l'aide de start et 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])

Sortie :

Sur les threads courts, la valeur par défaut limitée renvoie toujours tous les messages.

['Message 1', 'Message 2', 'Message 3']

['Message 1', 'Message 2', 'Message 3']

['Message 2', 'Message 3']

Supprimer un message de l'unité d'exécution courante par ID

La suppression d'un message supprime uniquement l'enregistrement brut du fil de discussion courant. Les mémoires dérivées ou d'autres artefacts en aval créés à partir de ce message peuvent rester consultables et peuvent toujours influencer la sortie de la carte de contexte. Si vous devez supprimer l'unité d'exécution avec les messages et les mémoires associés, utilisez plutôt delete_thread(). La transmission d'un identificateur à partir d'une unité d'exécution différente retourne toujours 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)

Sortie :

1

Cela supprime uniquement l'enregistrement de message brut de l'unité d'exécution courante. Les mémoires dérivées ou d'autres artefacts en aval créés à partir de ce message ne sont pas supprimés automatiquement et peuvent rester consultables ou apparaître dans la sortie de la carte de contexte. Utilisez memory.delete_thread(thread.thread_id) pour supprimer l'unité d'exécution ainsi que les messages et mémoires associés. Les suppressions de message retournent 0 pour les ID détenus par une autre unité d'exécution.

Ajouter une mémoire à partir d'un descripteur de fil

La valeur retournée est l'identificateur de mémoire.

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)

Sortie :

mem:3

Supprimer une mémoire du fil courant par ID

Les suppressions d'unité d'exécution sont ciblées sur l'unité d'exécution courante. La transmission d'un identificateur à partir d'une unité d'exécution différente retourne 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)

Sortie :

1

Les suppressions de threads sont ciblées sur le thread actuel et renvoient 0 pour les ID appartenant à un autre thread.

Créer une carte de contexte

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)

Sortie :

<context_card>
    The user is planning a trip to Kyoto.
</context_card>

Créer un sommaire de fil

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)

Sortie :

user (-): Hello
- assistant (-): Hi, how can I help?
- user (-): Please summarize this thread.

Créer un sommaire en excluant les N derniers 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)

Sortie :

user (-): First message
- assistant (-): Second message

Créer un sommaire avec un budget par jeton

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)

Sortie :

(truncated)
user (-): Message 1
...

Rechercher

Rechercher à partir d'un fil sans portée explicite

La recherche au niveau du fil utilise les valeurs par défaut du fil lorsque vous ne transmettez pas de portée explicite.

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

Sortie :

['The user likes pizza.']

Rechercher à partir de l'API de mémoire avec étendue

Au niveau de l'API, vous pouvez étendre l'extraction avec user_id, agent_id et thread_id au moyen de SearchScope. Pour les recherches de client de niveau supérieur, incluez user_id dans la portée.

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

Sortie :

['The user likes hiking in the Alps.']

Rechercher uniquement des mémoires ou des messages uniquement

Utilisez record_types pour limiter les résultats de recherche à des types d'enregistrement stocké spécifiques.

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

Sortie :

['The user likes pizza.']

['I mentioned pizza in a message.']

Code complet

#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

from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory

db_pool = ...  #an oracledb connection or connection pool

memory = OracleAgentMemory(
    connection=db_pool,
    embedder=embedder,
    llm=llm,  #optional: enables automatic memory extraction during add_messages()
)

##Configure an Oracle DB component

from oracleagentmemory.core import SchemaPolicy
from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory

db_pool = ...  #an oracledb connection or connection pool

db_memory = OracleAgentMemory(
    connection=db_pool,
    embedder=embedder,
    llm=llm,  #optional
    schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
    table_name_prefix="DEV_",
)

##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

##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 vector/chunk rows 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

##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.

##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:3

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