ADK 的 Redis 集成

Supported in ADKPython

adk-redis 集成 可将你的 ADK 智能体连接到 Redis，为其提供 基于 RedisVL 的搜索工具（通过 Redis 索引）、通过 Redis Agent Memory Server 实现的持久化会话和长期记忆，以及 LLM 响应和工具结果的语义缓存。Redis 可作为托管服务或自托管运行（需 Redis 8.4+ 及 RediSearch 模块）。

该集成有几种使用方式：

方式	说明
RedisVL MCP	将 ADK 原生的 McpToolset 连接到正在运行的 rvl mcp 服务器。公开 search-records（向量/全文/混合）和 upsert-records，并提供 schema 感知的过滤器和返回字段提示。
会话 + 记忆服务	RedisWorkingMemorySessionService 和 RedisLongTermMemoryService 实现了 ADK 的 BaseSessionService 和 BaseMemoryService，由 Agent Memory Server 支持。
会话 + 记忆 MCP	将 ADK 原生的 McpToolset 通过 SSE 连接到 Agent Memory Server 的 MCP 端点。使智能体能够直接通过工具访问 search_long_term_memory、create_long_term_memories 和 memory_prompt。
搜索工具	五个 BaseTool 子类（RedisVectorSearchTool、RedisHybridSearchTool、RedisRangeSearchTool、RedisTextSearchTool、RedisSQLSearchTool），通过 RedisVL 查询绑定的索引。

使用场景

• 基于数据构建 RAG：对 Redis 索引运行向量、混合、范围、BM25 文本或 SQL 搜索。混合搜索在 Redis 8.4+ 上使用原生 FT.HYBRID，在其他环境中回退到客户端聚合。
• 持久化多轮对话智能体：将会话和记忆服务插入任何 ADK Runner，以保留对话状态，当上下文窗口填满时自动汇总，并将持久性事实提升到长期记忆。
• Schema 感知的 MCP 工具：为每个 rvl mcp 服务器部署一个 Redis 索引，并通过 stdio、sse 或 streamable-http 连接任意数量的智能体。MCP 工具描述包含从索引 schema 派生出的过滤器和返回字段提示。
• 降低延迟和成本：用语义缓存包装 LLM 调用，使得重复或近似重复的提示跳过模型调用。

前提条件

• Python 3.10+
• 启用 RediSearch 模块的 Redis 8.4+（或 Redis Cloud）
• 对于会话和记忆服务：需要在本地或环境中运行 Redis Agent Memory Server
• 对于 LangCache 缓存提供程序：需要 Redis LangCache 缓存和 API 密钥

安装

安装你所需的组件：

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 服务器搜索工具

启动指向 Redis 索引的 RedisVL MCP 服务器（rvl mcp），然后将 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="使用 search-records 工具回答问题。",
    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 Tools。

通过基于 REST 的会话和记忆服务将 Agent Memory Server 插入任何 ADK Runner。工作记忆处理每个会话的状态并自动汇总；长期记忆提供跨会话的混合搜索。

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

from adk_redis import (
    RedisLongTermMemoryService,
    RedisLongTermMemoryServiceConfig,
    RedisWorkingMemorySessionService,
    RedisWorkingMemorySessionServiceConfig,
)

session_service = RedisWorkingMemorySessionService(
    config=RedisWorkingMemorySessionServiceConfig(
        api_base_url="http://localhost:8000",
    ),
)
memory_service = RedisLongTermMemoryService(
    config=RedisLongTermMemoryServiceConfig(
        api_base_url="http://localhost:8000",
        recency_boost=True,
    ),
)

root_agent = Agent(
    model="gemini-flash-latest",
    name="redis_memory_agent",
    instruction="使用长期记忆来个性化回复。",
)

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

将 ADK 原生的 McpToolset 通过 SSE 连接到 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="使用记忆工具来个性化回复。",
    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="对产品目录进行语义搜索。",
)

root_agent = Agent(
    model="gemini-flash-latest",
    name="redis_search_agent",
    instruction="帮助用户使用语义搜索查找产品。",
    tools=[search_tool],
)

语义缓存

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

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

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

```python 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="你是一个启用语义缓存的得力助手。", 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="你是一个启用语义缓存的得力助手。",
    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 服务器	将 ADK 原生的 McpToolset 通过 SSE 连接到 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 获取单个记忆。

服务

服务	说明
RedisWorkingMemorySessionService	由 Agent Memory Server 工作记忆支持的 BaseSessionService。当超出上下文窗口时自动汇总。
RedisLongTermMemoryService	由 Agent Memory Server 长期记忆支持的 BaseMemoryService，具有最近性增强的语义搜索。

缓存提供程序

提供程序	说明
RedisVLCacheProvider	通过 RedisVL SemanticCache 自托管语义缓存。自带向量化器。
LangCacheProvider	通过 Redis LangCache 托管的语义缓存。在服务端处理嵌入向量。

其他资源

• adk-redis on GitHub
• adk-redis on PyPI
• adk-redis documentation
• ADK + Redis on redis.io
• Runnable examples
• Redis Agent Memory Server
• RedisVL documentation
• Redis LangCache