ADK 的 Aerospike 集成

Supported in ADKPython

adk-aerospike 集成将你的 ADK 智能体连接到 Aerospike，一个分布式实时键值数据库。它在单个集群上实现了所有三个 ADK Python 存储接口，使用应用程序进程中的原生 Aerospike 客户端。注册一次 aerospike:// URI 方案，adk CLI 即可将 Aerospike 用于会话、制品和记忆。

有多种方式使用此集成：

方法	描述
会话服务	AerospikeSessionService：范围化状态（app:、user:、session）、带分块存储的事件历史、原子 append_event。
记忆服务	AerospikeMemoryService：通过每令牌倒排列表键的词法词重叠搜索；与 InMemoryMemoryService 语义相同。
制品服务	AerospikeArtifactService：每个会话或 user: 命名空间的版本化 blob。
完整栈	将所有三个服务连接到一个 Runner，或将匹配的 aerospike:// URI 传递给 adk web / adk run。

使用场景

• 生产级智能体持久化：在重启和副本之间保持对话状态、工具输出和用户范围化数据，无需运行单独的记忆服务。
• 高吞吐量智能体：聊天、语音和实时编排的亚毫秒级读写，适用于会话追加延迟敏感的场景。
• 词法长期记忆：在写入时对文本进行分词；在倒排列表键（app:user:kw:<token>）上使用点读搜索并恢复记忆行，无需嵌入模型。
• 多模态制品：存储图像、文件和生成输出，支持版本历史；user: 文件名跨会话可见（ADK 约定）。
• 自托管和多租户：一个命名空间，复合二级索引用于租户范围的制品和记忆操作；社区版或企业版，可本地或云端部署。

前置条件

• Python 3.11 或更高版本
• ADK for Python（google-adk）
• Aerospike Database 7.x 或 8.x（社区版或企业版）
• 可达的集群（下方有本地 Docker 示例）

本地开发用 Aerospike：

docker run --rm -d --name aerospike -p 3000-3003:3000-3003 aerospike/aerospike-server:latest

如需在可运行示例中进行模型调用，请设置 GOOGLE_API_KEY（或你的模型提供者凭据）。

安装

pip install google-adk adk-aerospike

与智能体配合使用

Session + RunnerSession APIMemory serviceArtifact serviceAll three servicesadk web 和 adk run

将 AerospikeSessionService 插入任何 ADK Runner，获得具有持久会话的完整多轮智能体。

import asyncio

from adk_aerospike import AerospikeSessionService
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.genai import types

async def main() -> None:
    session_service = AerospikeSessionService.from_uri(
        "aerospike://localhost:3000/adk"
    )
    agent = LlmAgent(
        name="assistant",
        model="gemini-flash-latest",
        instruction="Be helpful. Keep replies under 30 words.",
    )
    runner = Runner(
        agent=agent,
        app_name="myapp",
        session_service=session_service,
    )

    session = await session_service.create_session(
        app_name="myapp", user_id="user-1"
    )
    async for event in runner.run_async(
        user_id="user-1",
        session_id=session.id,
        new_message=types.Content(
            role="user", parts=[types.Part(text="Hello")]
        ),
    ):
        if event.content:
            for part in event.content.parts or []:
                if part.text:
                    print(part.text)

    session_service.close()

asyncio.run(main())

直接使用会话服务进行状态、事件和列出操作。范围化键遵循 ADK 约定（app:、user:、temp:）。

import asyncio

from adk_aerospike import AerospikeSessionService
from google.adk.events import Event, EventActions
from google.genai import types

async def main() -> None:
    svc = AerospikeSessionService.from_uri("aerospike://localhost:3000/adk")

    session = await svc.create_session(
        app_name="support_bot",
        user_id="alice",
        state={
            "topic": "billing",
            "app:tenant": "acme-corp",
            "user:nickname": "Allie",
            "temp:scratch": "throwaway",
        },
    )

    await svc.append_event(
        session,
        Event(
            invocation_id="i1",
            author="user",
            content=types.Content(
                role="user",
                parts=[types.Part(text="Where is my invoice?")],
            ),
            actions=EventActions(state_delta={"turn": 1}),
        ),
    )

    fetched = await svc.get_session(
        app_name="support_bot",
        user_id="alice",
        session_id=session.id,
    )
    print(fetched.state)
    # topic, turn, app:tenant, user:nickname — temp: keys are not persisted

    svc.close()

asyncio.run(main())

持久化带文本的会话事件，然后使用词重叠搜索（无向量索引）。

import asyncio

from adk_aerospike import AerospikeMemoryService
from google.adk.events import Event, EventActions
from google.adk.sessions import Session
from google.genai import types

async def main() -> None:
    memory = AerospikeMemoryService.from_uri(
        "aerospike://localhost:3000/adk", top_k=10
    )

    session = Session(
        id="s-1",
        app_name="support_bot",
        user_id="alice",
        events=[
            Event(
                invocation_id="i",
                author="user",
                content=types.Content(
                    role="user",
                    parts=[types.Part(text="Python uses duck typing.")],
                ),
                actions=EventActions(),
            ),
        ],
    )
    await memory.add_session_to_memory(session)

    resp = await memory.search_memory(
        app_name="support_bot",
        user_id="alice",
        query="python duck typing",
    )
    for m in resp.memories:
        print(m.content.parts[0].text)

    memory.close()

asyncio.run(main())

按会话保存版本化制品；使用 user: 文件名前缀实现跨会话可见性。

import asyncio

from adk_aerospike import AerospikeArtifactService
from google.genai import types

async def main() -> None:
    svc = AerospikeArtifactService.from_uri(
        "aerospike://localhost:3000/adk"
    )

    await svc.save_artifact(
        app_name="support_bot",
        user_id="alice",
        session_id="s-1",
        filename="report.pdf",
        artifact=types.Part(
            inline_data=types.Blob(
                mime_type="application/pdf", data=b"%PDF-1.4..."
            ),
        ),
    )

    latest = await svc.load_artifact(
        app_name="support_bot",
        user_id="alice",
        session_id="s-1",
        filename="report.pdf",
    )
    print(latest.inline_data.mime_type)

    svc.close()

asyncio.run(main())

from adk_aerospike import (
    AerospikeArtifactService,
    AerospikeMemoryService,
    AerospikeSessionService,
)
from google.adk.agents import LlmAgent
from google.adk.runners import Runner

uri = "aerospike://localhost:3000/adk"

session_service = AerospikeSessionService.from_uri(uri)
artifact_service = AerospikeArtifactService.from_uri(uri)
memory_service = AerospikeMemoryService.from_uri(uri)

agent = LlmAgent(name="assistant", model="gemini-flash-latest")
runner = Runner(
    agent=agent,
    app_name="myapp",
    session_service=session_service,
    artifact_service=artifact_service,
    memory_service=memory_service,
)

注册一次 URI 方案（例如在智能体旁边的 services.py 中）：

import adk_aerospike

adk_aerospike.register()

然后为每个存储角色将 CLI 指向相同的命名空间：

adk web \
  --session_service_uri=aerospike://localhost:3000/adk \
  --artifact_service_uri=aerospike://localhost:3000/adk \
  --memory_service_uri=aerospike://localhost:3000/adk

Note

register() 将 aerospike:// 连接到 ADK 的服务注册表，以便开发 UI 和 CLI 无需自定义工厂代码即可解析这些 URL。

配置

连接 URI

三个服务共享一个 URI 格式：

aerospike://[user:pass@]host[:port][,host2[:port],…]/<namespace>[?option=value]

示例：

aerospike://localhost:3000/adk
aerospike://user:pass@node1:3000,node2:3000/prod?set_prefix=prod_&tls=true

查询参数	描述
set_prefix	Aerospike 集合名称的前缀（默认 adk_）。多个应用可共享一个命名空间。
tls=true	启用 TLS。向 from_uri 传递 tls_config={...} 以配置 mTLS 详情。
auth_mode	INTERNAL（默认）、EXTERNAL、EXTERNAL_INSECURE 或 PKI。

你也可以使用现有的 aerospike.Client 和 Schema 构造服务，以在多个服务之间共享连接池。

状态范围化

会话 state 使用键前缀（与 google.adk.sessions.state.State 相同）：

前缀	存储位置	可见性
app:foo	adk_app_state	应用的所有用户
user:foo	adk_user_state	此用户跨会话
temp:foo	不持久化	仅当前调用
(无前缀)	会话记录	仅此会话

get_session 将所有范围合并为一个字典，并为 ADK 兼容性恢复前缀。

可用服务

服务

服务	ADK 接口	描述
AerospikeSessionService	BaseSessionService	会话、事件、范围化状态。会话记录上的热事件尾部；256 KiB 时密封分块。大多数追加是单个原子 operate()；get_session 使用 batch_read 在一次 RTT 中获取会话 + 应用 + 用户状态。
AerospikeArtifactService	BaseArtifactService	每个 (app, user, session, filename) 的版本化制品。内联负载每个版本最大 8 MiB。user: 文件名使用 ADK 用户命名空间哨兵。
AerospikeMemoryService	BaseMemoryService	每个带文本事件一行记忆；每令牌的倒排列表主键。search_memory 按查询令牌重叠排序。

URI 注册

函数	描述
adk_aerospike.register()	向 ADK 的服务注册表注册 aerospike://，用于 CLI 和 adk web。

存储布局

命名空间中默认集合前缀 adk_：

集合	键模式	用途
adk_sessions	app:user:session	会话记录（状态 + 热事件尾部）
adk_sessions	app:user:session:c:NNNNNNNN	密封事件分块
adk_sessions	app:user:sl	list_sessions 的会话列表清单
adk_app_state	app	应用范围化状态
adk_user_state	app:user	用户范围化状态
adk_artifacts	app:user:session:fname:ver	制品版本
adk_memory	app:user:session:event_id	记忆行
adk_memory	app:user:kw:token	词法搜索的倒排列表

有关索引、分块不变量和运维说明，请参阅仓库中的数据模型。

附加资源

• GitHub 上的 adk-aerospike
• PyPI 上的 adk-aerospike
• 可运行示例
• Aerospike 文档
• ADK 会话和记忆