クイック参照コード・サンプル
この記事では、一般的なエージェント・メモリー設定およびAPIライフサイクル操作について、焦点を絞った小さな例を収集します。
LLM/埋込み設定
次の例では、LLMと埋込みモデルの両方に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埋込みモデルの構成
from oracleagentmemory.core.embedders.embedder import Embedder
embedder = Embedder(
model="YOUR_EMBEDDING_MODEL",
api_base="YOUR_EMBEDDING_API_BASE",
api_key="YOUR_EMBEDDING_API_KEY",
)
embedding_matrix = embedder.embed(["The quick brown fox jumps over the lazy dog"])
print(embedding_matrix.shape)出力:
(1, embedding_dimension)API設定
エージェント・メモリー・コンポーネントの構成
これにより、Oracle DB接続またはプールが埋込みモデルおよびオプションのLLMとともに使用され、自動メモリー抽出が行われます。
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()
)Oracle DBメモリー・コンポーネントの構成
このバリアントは、Oracle DB接続またはプールを使用し、スキーマ・ポリシーおよび表名の接頭辞を設定する方法を示します。
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_",
)APIのライフサイクル
スレッドの作成
オプションのスレッドID、ユーザーID、およびエージェントIDを持つスレッドを作成します。
thread = memory.create_thread(
thread_id="thread_create_123", # optional
user_id="user_123", # optional
agent_id="agent_456", # optional
)
print(thread.thread_id)出力:
thread_create_123既存のスレッドの再オープン
thread = memory.create_thread(
thread_id="thread_reopen_123",
user_id="user_123",
agent_id="agent_456",
)
same_thread = memory.get_thread("thread_reopen_123")
print(same_thread.thread_id)出力:
thread_reopen_123スレッドの削除
thread = memory.create_thread(thread_id="thread_delete_123")
deleted = memory.delete_thread("thread_delete_123")
print(deleted)出力:
1ユーザー・プロファイルの追加
user_profile_id = memory.add_user(
"user_123",
"The user prefers concise answers and works mostly with Python.",
)
print(user_profile_id)出力:
user_123エージェント・プロファイルの追加
agent_profile_id = memory.add_agent(
"agent_456",
"A coding assistant specialized in debugging and code review.",
)
print(agent_profile_id)出力:
agent_456メモリーAPIからのグローバル・メモリーの追加
thread_idを省略すると、メモリーは特定のスレッドに関連付けられません。戻り値はメモリー識別子です。
memory_id = memory.add_memory(
"The user prefers short, bullet-point answers.",
user_id="user_123",
agent_id="agent_456",
)
print(memory_id)出力:
mem:1メモリーAPIからのスコープ付きメモリーの追加
戻り値はメモリー識別子です。
thread = memory.create_thread(
thread_id="thread_scoped_123",
user_id="user_123",
agent_id="agent_456",
)
memory_id = memory.add_memory(
"The user is planning a trip to Kyoto next month.",
user_id="user_123",
agent_id="agent_456",
thread_id=thread.thread_id,
)
print(memory_id)出力:
mem:2カスタムIDを使用したメモリーの追加
戻り値は、コール元提供のメモリー識別子です。
memory_id = memory.add_memory(
"The user prefers aisle seats on flights.",
user_id="user_123",
agent_id="agent_456",
memory_id="travel_pref_001",
)
print(memory_id)出力:
travel_pref_001スレッドの基本
スレッドへのメッセージの追加
メッセージは、ディクショナリとして、またはMessageオブジェクトとして渡すことができます。オプションのメッセージID、タイムスタンプおよびメタデータをそれらとともに格納できます。
from oracleagentmemory.apis import Message
thread = memory.create_thread(
thread_id="thread_messages_123",
user_id="user_123",
agent_id="agent_456",
)
message_ids = thread.add_messages(
[
Message(
id="msg_user_001",
role="user",
content="I prefer window seats on flights.",
timestamp="2026-03-27T09:00:00Z",
metadata={"source": "chat", "channel": "web"},
),
{
"id": "msg_assistant_001",
"role": "assistant",
"content": "Noted. I will keep that in mind.",
"timestamp": "2026-03-27T09:00:05Z",
"metadata": {"source": "assistant"},
},
]
)
print(message_ids)出力:
['msg_user_001', 'msg_assistant_001']スレッド・メッセージの読取り
格納されているすべてのメッセージまたはスライスは、startおよびendを使用して読み取ることができます。
thread = memory.create_thread(thread_id="thread_read_messages_123")
thread.add_messages(
[
{"role": "user", "content": "Message 1"},
{"role": "assistant", "content": "Message 2"},
{"role": "user", "content": "Message 3"},
]
)
default_messages = thread.get_messages()
all_messages = thread.get_messages(end=None)
middle_messages = thread.get_messages(start=1, end=3)
print([message.content for message in default_messages])
print([message.content for message in all_messages])
print([message.content for message in middle_messages])出力:
短いスレッドでは、バインドされたデフォルトが引き続きすべてのメッセージを返します。
['Message 1', 'Message 2', 'Message 3']
['Message 1', 'Message 2', 'Message 3']
['Message 2', 'Message 3']IDによる現在のスレッドからのメッセージの削除
スレッド削除は、現在のスレッドにスコープ指定されます。別のスレッドから識別子を渡すと、0が返されます。
thread = memory.create_thread(thread_id="thread_delete_message_123")
message_ids = thread.add_messages(
[
{"role": "user", "content": "Message to delete"},
]
)
deleted = thread.delete_message(message_ids[0])
print(deleted)出力:
1スレッド削除は現在のスレッドにスコープ指定され、別のスレッドが所有するIDの場合は0を返します。
スレッド・ハンドルからのメモリーの追加
戻り値はメモリー識別子です。
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:3IDによる現在のスレッドからのメモリーの削除
スレッド削除は、現在のスレッドにスコープ指定されます。別のスレッドから識別子を渡すと、0が返されます。
thread = memory.create_thread(thread_id="thread_delete_memory_123")
memory_id = thread.add_memory("Temporary memory to delete.")
deleted = thread.delete_memory(memory_id)
print(deleted)出力:
1スレッド削除は現在のスレッドにスコープ指定され、別のスレッドが所有するIDの場合は0を返します。
コンテキスト・カードの作成
thread = memory.create_thread(thread_id="thread_context_card_123")
thread.add_messages(
[
{"role": "user", "content": "I am planning a trip to Kyoto next spring."},
]
)
thread.add_memory("The user is planning a trip to Kyoto.")
context_card = thread.get_context_card()
print(context_card)出力:
<context_card>
The user is planning a trip to Kyoto.
</context_card>スレッド・サマリーの作成
thread = memory.create_thread(thread_id="thread_summary_123")
thread.add_messages(
[
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "Hi, how can I help?"},
{"role": "user", "content": "Please summarize this thread."},
]
)
summary = thread.get_summary()
print(summary[0].content)出力:
user (-): Hello
- assistant (-): Hi, how can I help?
- user (-): Please summarize this thread.最後のNメッセージを除くサマリーの作成
thread = memory.create_thread(thread_id="thread_summary_except_last_123")
thread.add_messages(
[
{"role": "user", "content": "First message"},
{"role": "assistant", "content": "Second message"},
{"role": "user", "content": "Third message"},
]
)
summary = thread.get_summary(except_last=1)
print(summary[0].content)出力:
user (-): First message
- assistant (-): Second messageトークン予算を使用した要約の作成
thread = memory.create_thread(thread_id="thread_summary_budget_123")
thread.add_messages(
[
{"role": "user", "content": "Message 1"},
{"role": "assistant", "content": "Message 2"},
{"role": "user", "content": "Message 3"},
{"role": "assistant", "content": "Message 4"},
]
)
summary = thread.get_summary(token_budget=20)
print(summary[0].content)出力:
(truncated)
user (-): Message 1
...検索
明示的な範囲指定のないスレッドからの検索
スレッド・レベルの検索では、明示的なスコープを渡さない場合、スレッドのデフォルトが使用されます。
thread = memory.create_thread(
thread_id="thread_search_default_123",
user_id="user_123",
agent_id="agent_456",
)
thread.add_memory("The user likes pizza.")
thread.add_memory("The user likes cats.")
results = thread.search("pizza", max_results=5)
print([result.content for result in results])出力:
['The user likes pizza.']スコーピングを使用したメモリーAPIからの検索
APIレベルでは、SearchScopeを介してuser_id、agent_idおよびthread_idを使用して取得の範囲を指定できます。最上位のクライアント検索の場合は、スコープにuser_idを含めます。
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.']メモのみまたはメッセージのみを検索
record_typesを使用して、検索結果を特定のストアド・レコード・タイプに制限します。
thread = memory.create_thread(thread_id="thread_entity_type_search_123")
thread.add_messages(
[
{"role": "user", "content": "I mentioned pizza in a message."},
]
)
thread.add_memory("The user likes pizza.")
memory_results = thread.search("pizza", max_results=5, record_types=["memory"])
message_results = thread.search("pizza", max_results=5, record_types=["message"])
print([result.content for result in memory_results])
print([result.content for result in message_results])出力:
['The user likes pizza.']
['I mentioned pizza in a message.']完全コード
#Copyright © 2026 Oracle and/or its affiliates.
#isort:skip_file
#fmt: off
#Agent Memory Code Example - Reference Sheet
#-------------------------------------------
#How to use:
#Create a new Python virtual environment and install the latest oracleagentmemory version.
#You can now run the script
#1. As a Python file:
#bash
#python reference_sheet.py
#2. As a Notebook (in VSCode):
#When viewing the file,
#- press the keys Ctrl + Enter to run the selected cell
#- or Shift + Enter to run the selected cell and move to the cell below
##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
##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
#Thread deletes are scoped to the current thread and 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)
#<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[0].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[0].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[0].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.']