快速参考代码示例
本文收集了常见代理内存设置和 API 生命周期操作的小型重点示例。
LLM/嵌入设置
下面的示例将 LiteLLM 用于 LLM 和嵌入模型。
配置 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:3按 ID 从当前线程中删除内存
线程删除的范围将限定到当前线程。从其他线程传递标识符将返回 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.']