将 Oracle Agent Memory 与 Oracle Autonomous AI Database 结合使用

本指南介绍了如何将 oracleagentmemory 连接到 Oracle Autonomous Database (ADB)。

在本指南中，您将学习如何：

为 Oracle Agent Memory 准备 ADB 方案用户。
使用无钱包 TLS 或 mTLS wallet 连接。
初始化托管内存方案并存储和搜索内存。

提示：有关软件包设置，请参见 Get Started with Agent Memory 。要改为针对容器化的本地数据库运行 Oracle 代理内存，请按照在本地运行 Oracle AI Database 。

先决条件

您需要支持 Oracle AI Vector Search（Oracle AI Database 23ai 或更高版本）的可用 Oracle Autonomous AI Database、可以创建数据库用户的 ADB 管理员以及安装了 oracleagentmemory 的 Python 环境。如果您还没有数据库，请遵循 Oracle Autonomous AI Database Provisioning Guide 。

Oracle 代理内存使用应用程序提供的连接或池；它不配置数据库传输安全本身。ADB 支持两种安全的连接选项：

TLS without a wallet 适用于大多数新应用程序。它使用数据库身份证明并对连接进行加密。在使用 IP 允许列表 (IP allowlist，ACL) 或专用端点之前，配置该列表，以便应用程序可以访问数据库。TLS 保护连接；ACL 或专用端点控制哪些网络可以访问数据库。
当 ADB 需要 mTLS 或您的组织需要客户端证书验证时，带 wallet 的相互 TLS (mTLS) 适用。

在生产环境中使用单独的数据库用户

在生产部署中使用两个 ADB 数据库用户：

方案所有者创建、升级并拥有 Oracle Agent Memory 表和索引。
应用程序用户在应用程序读取和写入记忆时连接。仅向此用户授予其对方案所有者的对象所需的权限。

请勿使用方案所有者的身份证明来运行应用程序。仅在设置或升级托管方案时使用它们。本指南首先创建方案所有者。设置后，使用 SchemaPolicy.REQUIRE_EXISTING 配置应用程序用户。故障排除指南列出了应用程序用户所需的对象授权。

在启动写入托管表的应用程序实例之前，请以单个维护操作的形式运行每个方案创建或升级。对操作使用一个模式所有者连接；不允许多个客户机同时运行 SchemaPolicy.CREATE_IF_NECESSARY。成功后，应用程序实例应使用 SchemaPolicy.REQUIRE_EXISTING。

创建 ADB 方案用户

使用专用 ADB 数据库用户拥有 Oracle 代理内存方案。在 OCI 控制台中，打开数据库操作，以 ADMIN 身份登录，然后打开 SQL 工作表。创建具有强口令的用户，授予托管方案设置使用的权限，并为用户设置有界的存储限额：

CREATE USER oam_owner IDENTIFIED BY "CHOOSE_A_STRONG_PASSWORD";
GRANT DWROLE TO oam_owner;
ALTER USER oam_owner QUOTA 1G ON DATA;

DWROLE 包括 Oracle Agent Memory 使用的常规对象创建权限，包括 CREATE TABLE、CREATE SEQUENCE、CREATE PROCEDURE 和 CREATE JOB。CREATE JOB 允许 Oracle Agent Memory 创建一个调度作业，该作业将在记录到期后永久删除记录。

1G 配额是本指南的起始限制，而不是生产规模建议。根据预期存储的内存量和保留策略选择限制。监视存储使用情况并在需要时增加配额；当用户达到其配额时写入失败。ADB 管理表空间，因此本指南不创建表空间。

无需钱包即可与 TLS 连接

在 OCI 控制台中，打开 ADB 详细信息页面。在网络下，配置允许应用出站 IP 地址或使用专用端点的 ACL。然后编辑 Mutual TLS (mTLS) Authentication ，清除 Require dual TLS (mTLS) authentication ，并等待数据库返回到 Available。

打开 Database connection（数据库连接），选择 TLS Authentication（TLS 验证）下的 TLS ，然后选择列出的连接服务之一。这些服务通常具有 myadb_low、myadb_medium 和 myadb_high 等名称；它们选择 ADB 如何在连接之间共享数据库资源。从数据库管理员为应用程序的工作量建议的服务级别开始，然后复制其连接字符串。将其和数据库身份证明存储在密钥管理系统中：

export ORACLE_MEMORY_DB_USER='oam_owner'
export ORACLE_MEMORY_DB_PASSWORD='<database-user-password>'
export ORACLE_MEMORY_DB_DSN='<copied TLS connection string>'

使用与复制完全相同的 TLS 描述符。没有 wallet，mTLS 描述符无法工作。

此示例创建一个连接池，该池是可重用的数据库连接组。如果 ORACLE_MEMORY_ADB_USE_MTLS 不是 true，则这些连接将使用 TLS。它使用 SchemaPolicy.CREATE_IF_NECESSARY 进行首次设置；在方案准备就绪后，在正常应用程序启动时使用 SchemaPolicy.REQUIRE_EXISTING。

import os

import oracledb

from oracleagentmemory.core import (
    MemoryExtractionConfig,
    OracleAgentMemory,
    SchemaPolicy,
)
from oracleagentmemory.core.embedders import Embedder


def create_adb_tls_pool() -> oracledb.ConnectionPool:
    """Create an ADB TLS pool from securely injected environment variables."""
    return oracledb.create_pool(
        user=os.environ.get("ORACLE_MEMORY_DB_USER", "YOUR DB USER"),
        password=os.environ["ORACLE_MEMORY_DB_PASSWORD"],
        #Set ORACLE_MEMORY_DB_DSN to the TLS descriptor copied from the ADB console.
        dsn=os.environ["ORACLE_MEMORY_DB_DSN"],
        min=1,
        max=5,
        increment=1,
    )


embedder = Embedder(model="YOUR_EMBEDDING_MODEL")
MEMORY_STORE_ID = "APP_MEMORY"

禁用自动内存提取只是为了使此连接示例保持焦点。配置 LLM，并在应用程序需要自动持久性内存提取时启用提取。

API 参考：OracleAgentMemory

SchemaPolicy

使用 mTLS wallet 连接

如果 ADB 网络配置需要 mTLS，请使用此选项。在 OCI 控制台中，打开数据库连接，选择实例 wallet 并下载它。将 wallet 提取到未检入源控制的受保护目录中。Oracle 文档建议对 wallet 文件使用限制性文件权限（例如，在 Linux 或 Unix 上使用 chmod 600）。

导出数据库身份证明，wallet 目录和 wallet-download 密码。ORACLE_MEMORY_WALLET_PASSWORD 是下载 wallet ZIP 文件时提供的密码；它不是数据库用户的密码。将 ORACLE_MEMORY_DB_DSN 设置为 wallet 的 tnsnames.ora 文件（例如 myadb_low）中的服务别名。请勿将 ADB 控制台中的长 TLS 连接字符串用于此 wallet 配置：

export ORACLE_MEMORY_DB_USER='oam_owner'
export ORACLE_MEMORY_DB_PASSWORD='<database-user-password>'
export ORACLE_MEMORY_DB_DSN='myadb_low'
export ORACLE_MEMORY_WALLET_DIR='/secure/path/to/wallet'
export ORACLE_MEMORY_WALLET_PASSWORD='<wallet-download-password>'
export ORACLE_MEMORY_ADB_USE_MTLS='true'

在 python-oracledb 瘦模式下，Wallet 目录需要 tnsnames.ora 和 ewallet.pem。将 ORACLE_MEMORY_ADB_USE_MTLS 设置为 true 将使示例在初始化 OracleAgentMemory 之前创建 mTLS 池。

def create_adb_mtls_pool() -> oracledb.ConnectionPool:
    """Create an ADB mTLS pool from securely injected environment variables."""
    return oracledb.create_pool(
        user=os.environ["ORACLE_MEMORY_DB_USER"],
        password=os.environ["ORACLE_MEMORY_DB_PASSWORD"],
        dsn=os.environ["ORACLE_MEMORY_DB_DSN"],
        config_dir=os.environ["ORACLE_MEMORY_WALLET_DIR"],
        wallet_location=os.environ["ORACLE_MEMORY_WALLET_DIR"],
        wallet_password=os.environ["ORACLE_MEMORY_WALLET_PASSWORD"],
        min=1,
        max=5,
        increment=1,
    )



if os.environ.get("ORACLE_MEMORY_ADB_USE_MTLS", "false").lower() == "true":
    db_pool = create_adb_mtls_pool()
else:
    db_pool = create_adb_tls_pool()

memory = OracleAgentMemory(
    connection=db_pool,
    embedder=embedder,
    memory_extraction_config=MemoryExtractionConfig(extract_memories=False),
    schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
    memory_store_id=MEMORY_STORE_ID,
)

在 ADB 中存储和搜索内存

为 Oracle Agent Memory 客户机配置了任一池后，请创建线程、添加持久内存并重新搜索。

thread = memory.create_thread(user_id="user_123")
thread.add_memory("The user prefers concise answers.")

results = memory.search(
    query="concise answers",
    user_id="user_123",
    record_types=["memory"],
    max_results=5,
)

for result in results:
    print(result.content)

API 参考：OracleAgentMemory

OracleThread

OracleSearchResult

结论

在本指南中，我们学习了如何准备 ADB 用户、选择无 wallet TLS 或 mTLS、将 Oracle 代理内存与 python-oracledb 池连接，以及验证是否可以从 ADB 存储和检索内存。

→将 Oracle 代理内存连接到 ADB 后，现在可以继续存储和搜索内存。

完整代码

本指南中包括了完整的示例，供您复制和运行。

#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 - Oracle Autonomous AI Database
#-------------------------------------------------------------

##Configure an Autonomous AI Database TLS connection pool

import os

import oracledb

from oracleagentmemory.core import (
    MemoryExtractionConfig,
    OracleAgentMemory,
    SchemaPolicy,
)
from oracleagentmemory.core.embedders import Embedder


def create_adb_tls_pool() -> oracledb.ConnectionPool:
    """Create an ADB TLS pool from securely injected environment variables."""
    return oracledb.create_pool(
        user=os.environ.get("ORACLE_MEMORY_DB_USER", "YOUR DB USER"),
        password=os.environ["ORACLE_MEMORY_DB_PASSWORD"],
        #Set ORACLE_MEMORY_DB_DSN to the TLS descriptor copied from the ADB console.
        dsn=os.environ["ORACLE_MEMORY_DB_DSN"],
        min=1,
        max=5,
        increment=1,
    )


embedder = Embedder(model="YOUR_EMBEDDING_MODEL")
MEMORY_STORE_ID = "APP_MEMORY"



##Configure an Autonomous AI Database mTLS connection pool

def create_adb_mtls_pool() -> oracledb.ConnectionPool:
    """Create an ADB mTLS pool from securely injected environment variables."""
    return oracledb.create_pool(
        user=os.environ["ORACLE_MEMORY_DB_USER"],
        password=os.environ["ORACLE_MEMORY_DB_PASSWORD"],
        dsn=os.environ["ORACLE_MEMORY_DB_DSN"],
        config_dir=os.environ["ORACLE_MEMORY_WALLET_DIR"],
        wallet_location=os.environ["ORACLE_MEMORY_WALLET_DIR"],
        wallet_password=os.environ["ORACLE_MEMORY_WALLET_PASSWORD"],
        min=1,
        max=5,
        increment=1,
    )



if os.environ.get("ORACLE_MEMORY_ADB_USE_MTLS", "false").lower() == "true":
    db_pool = create_adb_mtls_pool()
else:
    db_pool = create_adb_tls_pool()

memory = OracleAgentMemory(
    connection=db_pool,
    embedder=embedder,
    memory_extraction_config=MemoryExtractionConfig(extract_memories=False),
    schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
    memory_store_id=MEMORY_STORE_ID,
)



##Store and search memory in Autonomous AI Database

thread = memory.create_thread(user_id="user_123")
thread.add_memory("The user prefers concise answers.")

results = memory.search(
    query="concise answers",
    user_id="user_123",
    record_types=["memory"],
    max_results=5,
)

for result in results:
    print(result.content)