ADK 的 Redis 集成¶

Supported in ADKPython

adk-redis 集成将你的 ADK 智能体连接到 Redis，为其提供 RedisVL 驱动的 Redis 索引搜索工具、持久会话和长期记忆，以及 LLM 响应和工具结果的语义缓存。

多种方式使用此集成：

方法	描述
RedisVL MCP	将 ADK 的原生 `McpToolset` 连接到运行中的 `rvl mcp` 服务器。
会话 + 记忆服务	`RedisSessionMemoryService` 和 `RedisLongTermMemoryService` 实现 ADK 的 `BaseSessionService` 和 `BaseMemoryService`。
记忆工具	六个 `BaseTool` 子类让 LLM 搜索、创建和管理长期记忆。
会话 + 记忆 MCP	通过 SSE 连接 ADK 的原生 `McpToolset` 到 Agent Memory Server 的 MCP 端点。
搜索工具	五个 `BaseTool` 子类（向量/混合/范围/文本/SQL 搜索）通过 RedisVL 查询绑定索引。

使用场景¶

基于你的数据做 RAG：对 Redis 索引运行向量、混合、范围、BM25 文本或 SQL 搜索。
持久的有多轮智能体：将会话和记忆服务插入任何 ADK Runner，以保留对话状态。
Schema 感知的 MCP 工具：为每个 rvl mcp 服务器建立一个 Redis 索引，并将任意数量的智能体通过 stdio、sse 或 streamable-http 连接到它。
降低延迟和成本：使用语义缓存包装 LLM 调用点。

前置条件¶

Python 3.10+
启用 RediSearch 模块的 Redis 8.4+（或 Redis Cloud）

安装¶

安装你需要的组件：

pip install 'adk-redis[memory]'      # 会话 + 长期记忆服务
pip install 'adk-redis[search]'      # RedisVL 驱动的搜索工具
pip install 'adk-redis[sql]'         # RedisSQLSearchTool (sql-redis)
pip install 'adk-redis[langcache]'   # 托管的语义缓存提供者
pip install 'adk-redis[all]'         # 以上全部

# 对于 RedisVL MCP 服务器（配合 ADK 原生 McpToolset 使用）：
pip install 'redisvl[mcp]>=0.18.2'

与智能体配合使用¶

RedisVL MCP 服务器会话 + 记忆记忆工具会话 + 记忆 MCP 服务器搜索工具

启动 RedisVL MCP 服务器（rvl mcp）并将其指向你的 Redis 索引，然后将 ADK 原生的 McpToolset 连接到它。以下示例使用 stdio 传输方式，因此无需单独的服务器进程；切换为 StreamableHTTPConnectionParams 或 SseConnectionParams 即可连接到长期运行的远程服务器。

from google.adk.agents import Agent
from google.adk.tools.mcp_tool import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StdioConnectionParams
from mcp import StdioServerParameters

root_agent = Agent(
    model="gemini-flash-latest",
    name="redis_mcp_agent",
    instruction="Use the search-records tool to answer questions.",
    tools=[
        McpToolset(
            connection_params=StdioConnectionParams(
                server_params=StdioServerParameters(
                    command="rvl",
                    args=[
                        "mcp",
                        "--config",
                        "/path/to/mcp_config.yaml",
                        "--read-only",
                    ],
                ),
                timeout=30,
            ),
            tool_filter=["search-records"],
        ),
    ],
)

Note

要从其他 ADK 语言连接到此 MCP 服务器，请参阅 MCP 工具。

将会话和记忆服务接入任何 ADK Runner。两者都通过 backend 字段选择后端："redis-agent-memory"（默认）用于托管的 Redis Agent Memory，或 "opensource-agent-memory" 用于自托管的 Agent Memory Server。工作记忆处理单会话状态；长期记忆提供跨会话搜索。

from google.adk.agents import Agent
from google.adk.runners import Runner

from adk_redis import (
    RedisLongTermMemoryService,
    RedisLongTermMemoryServiceConfig,
    RedisSessionMemoryService,
    RedisSessionMemoryServiceConfig,
)

# 托管的 Redis Agent Memory（默认后端）。
session_service = RedisSessionMemoryService(
    config=RedisSessionMemoryServiceConfig(
        backend="redis-agent-memory",
        api_base_url="https://your-endpoint.redis.io",
        api_key="...",
        store_id="...",
        default_namespace="my_app",
    ),
)
memory_service = RedisLongTermMemoryService(
    config=RedisLongTermMemoryServiceConfig(
        backend="redis-agent-memory",
        api_base_url="https://your-endpoint.redis.io",
        api_key="...",
        store_id="...",
        default_namespace="my_app",
    ),
)

root_agent = Agent(
    model="gemini-flash-latest",
    name="redis_memory_agent",
    instruction="Use long-term memory to personalize responses.",
)

runner = Runner(
    app_name="redis_memory_app",
    agent=root_agent,
    session_service=session_service,
    memory_service=memory_service,
)

自托管后端

要使用自托管的 Agent Memory Server，请设置 backend="opensource-agent-memory"，将 api_base_url 指向该服务器（例如 http://localhost:8000），并省略 api_key 和 store_id，除非你的服务器要求提供。自动摘要和近期优先搜索（recency_boost=True）在自托管后端上可用。

通过 BaseTool 子类让 LLM 直接控制长期记忆。智能体可以决定何时搜索、创建、更新或删除记忆。这些工具共享一个 MemoryToolConfig，并通过相同的 backend 字段连接到任一后端。

from google.adk.agents import Agent

from adk_redis import (
    CreateMemoryTool,
    DeleteMemoryTool,
    MemoryPromptTool,
    MemoryToolConfig,
    SearchMemoryTool,
    UpdateMemoryTool,
)

config = MemoryToolConfig(
    backend="redis-agent-memory",
    api_base_url="https://your-endpoint.redis.io",
    api_key="...",
    store_id="...",
    default_namespace="my_app",
)

root_agent = Agent(
    model="gemini-flash-latest",
    name="redis_memory_tools_agent",
    instruction="Search memory before answering. Store important facts.",
    tools=[
        SearchMemoryTool(config=config),
        CreateMemoryTool(config=config),
        UpdateMemoryTool(config=config),
        DeleteMemoryTool(config=config),
        MemoryPromptTool(config=config),
    ],
)

通过 SSE 将 ADK 原生的 McpToolset 连接到 Agent Memory Server 的 MCP 端点。这让智能体可以直接通过工具访问长期记忆操作，无需使用基于 REST 的服务。

import os

from google.adk.agents import Agent
from google.adk.tools.mcp_tool import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import SseConnectionParams

MEMORY_MCP_URL = os.getenv("MEMORY_MCP_URL", "http://localhost:9000")

root_agent = Agent(
    model="gemini-flash-latest",
    name="memory_mcp_agent",
    instruction="Use memory tools to personalize responses.",
    tools=[
        McpToolset(
            connection_params=SseConnectionParams(
                url=f"{MEMORY_MCP_URL.rstrip('/')}/sse",
            ),
            tool_filter=[
                "search_long_term_memory",
                "create_long_term_memories",
                "memory_prompt",
            ],
        ),
    ],
)

Note

Agent Memory Server 在与 REST API 不同的端口上暴露其 MCP 端点。请参阅 fitness_coach_mcp 示例了解使用 Docker Compose 的完整运行配置。

使用 RedisVL 驱动的 BaseTool 子类对 Redis 索引运行向量、混合、范围、文本或 SQL 搜索。将工具绑定到现有索引，然后直接传递给你的智能体。

from google.adk.agents import Agent
from redisvl.index import SearchIndex
from redisvl.utils.vectorize import HFTextVectorizer

from adk_redis import RedisVectorQueryConfig, RedisVectorSearchTool

vectorizer = HFTextVectorizer(model="redis/langcache-embed-v2")
index = SearchIndex.from_existing("products", redis_url="redis://localhost:6379")

search_tool = RedisVectorSearchTool(
    index=index,
    vectorizer=vectorizer,
    config=RedisVectorQueryConfig(num_results=5),
    return_fields=["title", "price", "category"],
    name="search_products",
    description="Semantic search over the product catalog.",
)

root_agent = Agent(
    model="gemini-flash-latest",
    name="redis_search_agent",
    instruction="Help users find products using semantic search.",
    tools=[search_tool],
)

语义缓存¶

使用语义缓存包装任何 LLM 调用点，这样重复或近似重复的提示词可以跳过模型。选择自托管（使用你自己的 Redis 和向量化器）或通过 Redis LangCache 托管。

语义缓存（自托管）语义缓存（LangCache）

使用 RedisVLCacheProvider 配合本地向量化器和你自己的 Redis 实例进行自托管语义缓存。

from google.adk.agents import Agent
from redisvl.utils.vectorize import HFTextVectorizer

from adk_redis import (
    LLMResponseCache,
    RedisVLCacheProvider,
    RedisVLCacheProviderConfig,
    create_llm_cache_callbacks,
)

provider = RedisVLCacheProvider(
    config=RedisVLCacheProviderConfig(
        redis_url="redis://localhost:6379",
        ttl=3600,
        distance_threshold=0.1,
    ),
    vectorizer=HFTextVectorizer(
        model="redis/langcache-embed-v2",
    ),
)

llm_cache = LLMResponseCache(provider=provider)
before_model_cb, after_model_cb = create_llm_cache_callbacks(llm_cache)

root_agent = Agent(
    model="gemini-flash-latest",
    name="cached_agent",
    instruction="You are a helpful assistant with semantic caching enabled.",
    before_model_callback=before_model_cb,
    after_model_callback=after_model_cb,
)

使用 LangCacheProvider 配合 Redis LangCache，这是一个托管的语义缓存服务。无需本地向量化器，因为嵌入在服务端处理。

import os

from google.adk.agents import Agent

from adk_redis import (
    LLMResponseCache,
    LangCacheProvider,
    LangCacheProviderConfig,
    create_llm_cache_callbacks,
)

provider = LangCacheProvider(
    config=LangCacheProviderConfig(
        cache_id=os.environ["LANGCACHE_CACHE_ID"],
        api_key=os.environ["LANGCACHE_API_KEY"],
        server_url=os.getenv(
            "LANGCACHE_SERVER_URL",
            "https://aws-us-east-1.langcache.redis.io",
        ),
        ttl=3600,
    ),
)

llm_cache = LLMResponseCache(provider=provider)
before_model_cb, after_model_cb = create_llm_cache_callbacks(llm_cache)

root_agent = Agent(
    model="gemini-flash-latest",
    name="cached_agent",
    instruction="You are a helpful assistant with semantic caching enabled.",
    before_model_callback=before_model_cb,
    after_model_callback=after_model_cb,
)

可用工具¶

搜索工具¶

工具	描述
`RedisVectorSearchTool`	通过 RedisVL `VectorQuery` 进行向量相似度（KNN）搜索。
`RedisHybridSearchTool`	向量 + BM25 混合搜索。在 Redis 8.4+ 上使用原生 `FT.HYBRID`；否则回退到客户端聚合。
`RedisRangeSearchTool`	返回向量距离阈值内的所有文档。
`RedisTextSearchTool`	BM25 关键词全文搜索。无需向量化器。
`RedisSQLSearchTool`	通过 `redisvl.query.SQLQuery` 对绑定索引执行 SQL `SELECT`。支持 `:name` 参数占位符。需要 `adk-redis[sql]`。

MCP¶

来源描述

RedisVL MCP 服务器（rvl mcp）将 ADK 原生的 McpToolset 连接到运行中的 rvl mcp 服务器。该服务器暴露 search-records（向量/全文/混合，通过 YAML 为每个服务器选择）和 upsert-records，并提供从索引派生的 schema 感知的过滤器和返回字段提示。支持 stdio、sse 和 streamable-http；HTTP 上的 bearer 认证；通过服务器端的 --read-only 或 McpToolset 端的 tool_filter=["search-records"] 来抑制写入。

会话 + 记忆 MCP 服务器通过 SSE 将 ADK 原生的 McpToolset 连接到 Agent Memory Server 的 MCP 端点。暴露 search_long_term_memory、create_long_term_memories、edit_long_term_memory、delete_long_term_memories 和 memory_prompt。在与 REST API 不同的端口上运行。

记忆工具¶

工具	描述
`MemoryPromptTool`	用相关记忆丰富智能体的提示词。
`SearchMemoryTool`	按查询搜索长期记忆。
`CreateMemoryTool`	存储新的长期记忆。
`UpdateMemoryTool`	按 ID 更新现有记忆。
`DeleteMemoryTool`	按 ID 删除记忆。
`GetMemoryTool`	按 ID 获取单条记忆。

服务¶

服务	描述
`RedisSessionMemoryService`	`BaseSessionService`，由托管的 Redis Agent Memory 或自托管的 Agent Memory Server 工作记忆支持。自托管后端在上下文窗口超出时自动摘要。
`RedisLongTermMemoryService`	`BaseMemoryService`，由托管的 Redis Agent Memory 或自托管的 Agent Memory Server 长期记忆支持。自托管后端支持近期优先的语义搜索。

缓存提供者¶

提供者	描述
`RedisVLCacheProvider`	通过 RedisVL `SemanticCache` 的自托管语义缓存。需要自带向量化器。
`LangCacheProvider`	通过 Redis LangCache 的托管语义缓存。嵌入在服务端处理。