Agno 集成 Valkey 存储的完整实测指南：Agent、Team 与 Workflow 的会话持久化验证

导读：本文以 cookbook/06_storage/valkey 目录下三份测试脚本的 TEST_LOG 验证记录为主体，完整解读 Agno 将 Valkey 作为数据库后端、为 Agent（单代理）、Team（多代理团队）与 Workflow（工作流）持久化会话与 Run 记录的全过程。读完你可以复现三组端到端验证，理解 ValkeyDb 的连接与键设计、get_sessions 校验方式，以及实测中暴露出的外部依赖波动现象，为生产环境选用 Valkey 做存储提供依据。

Valkey 是 Redis 的分支开源数据存储，Agno 通过 agno.db.valkey.ValkeyDb 提供一套完整的会话/运行记录存储适配层。与关系型数据库不同，Valkey 没有"表"的概念，Agno 借助统一前缀的键空间、辅助索引集合与 GLIDE（Valkey 官方通用客户端库）Pipeline 批量读写，实现了接近 SQL 适配层的增删改查语义。测试日志表明，这套适配层在 Agent、Team、Workflow 三种执行体上均已通过端到端验证。

一、测试范围总览：三份脚本、三种执行体、同一个 Valkey 后端

TEST_LOG.md 记录了三组用例，覆盖 Agno 三种主要运行载体：

测试用例脚本	被测对象	核心断言	记录状态
valkey_for_agent.py	Agent（带联网搜索工具）	多轮对话借助历史上下文正确指代，会话经 get_sessions 确认已持久化	PASS
valkey_for_team.py	Team（HackerNews + Web 搜索双成员）	返回符合 Article schema 的结构化结果，团队/成员会话索引入库	PASS
valkey_for_workflow.py	Workflow（两阶段：研究团队 → 内容规划 Agent）	产出 4 周内容计划，工作流会话以 workflow_id:content-creation-workflow 索引入库	PASS

三份脚本位于 cookbook/06_storage/valkey 目录下，配套的 README.md 给出了安装与启动说明。更广义的存储选型目录见 cookbook/06_storage。

二、运行环境准备：依赖安装与 Valkey 容器启动

2.1 安装 GLIDE 同步客户端

ValkeyDb 底层依赖 valkey-glide-sync。若未安装，valkey.py 会直接抛出 ImportError，提示先执行：

uv pip install valkey-glide-sync

2.2 启动本地 Valkey

README 与三份脚本的 docstring 都给出了统一的容器启动方式：

docker run --name my-valkey -p 6379:6379 -d valkey/valkey-bundle

随后可用 docker ps 确认容器运行状态，再执行示例（以 agent 为例）：

python cookbook/06_storage/valkey/valkey_for_agent.py

默认配置下 ValkeyDb() 连接 localhost:6379，无需额外参数；若使用远程、集群或带鉴权的实例，则需要显式传参（见下文第三部分）。

三、ValkeyDb 源码级速览：连接参数、键空间与批量读写

测试通过的关键在于 ValkeyDb 适配层的设计。其完整实现位于 libs/agno/agno/db/valkey/valkey.py，构造函数（约第 68–166 行）支持以下参数：

参数	默认值	说明
host / port	"localhost" / 6379	Valkey 服务地址与端口
valkey_client	None	现成的 GlideClient/GlideClusterClient；提供则优先复用，否则按下方参数自建客户端
database_id	None	逻辑库索引（如 0–15）；不设置则使用服务端默认库 0
username / password	None	鉴权信息；设置 username 而未设 password 会抛 ValueError
use_tls	False	是否启用 TLS 加密连接
request_timeout	None	单请求超时（毫秒）；不设置则采用客户端默认值
db_prefix	"agno"	所有键的统一前缀，用于命名空间隔离与多租户分隔
client_name	"agno_db_client"	通过 CLIENT SETNAME 设置的连接名，便于在 CLIENT LIST 中识别
expire	None	键的 TTL（秒），用于自动过期清理；不设置则永久保存
session_table	None	会话"表"名，最终落到 BaseDb 中作为逻辑表名的一部分
runs_table / memory_table / metrics_table / eval_table / knowledge_table / traces_table / spans_table / learnings_table	None	其余各类记录的逻辑表名

关于"表"，需要注意一个关键实现事实：Valkey 本身没有表结构。源码 valkey.py 中的 table_exists() 恒返回 True，注释明确写道"Valkey 没有表，键在首次写入时创建"。表名实际体现为键的组成片段，见 utils.py：

def generate_valkey_key(prefix: str, table_type: str, key_id: str) -> str:
    return f"{prefix}:{table_type}:{key_id}"

def generate_index_key(prefix: str, table_type: str, index_field: str, index_value: str) -> str:
    return f"{prefix}:{table_type}:index:{index_field}:{index_value}"

因此默认前缀 agno 下，一个 agent 会话键形如 agno:sessions:<session_id>，字段级索引键形如 agno:sessions:index:session_type:agent。在 BaseDb（libs/agno/agno/db/base.py 附近）中，若调用方未自定义表名，逻辑表默认名分别为 agno_sessions、agno_runs、agno_memories、agno_learnings 等；若自定义了 session_table（如 workfolow 用例中的 workflow_session）而未指定 runs 表，runs 表会自动命名为 ${session_table}_runs。

读写层面，适配层用 GLIDE 的 Batch/ClusterBatch（非原子 Pipeline）将同一会话的多条 Run 记录合并为一次网络往返（源码 _create_pipeline/_exec_pipeline，约第 170–182 行）；Run 记录在 v3+ 版本中被拆分为独立键，并通过有序集合 {prefix}:runs:by_session:{session_id} 维护会话内的 run 顺序索引（约第 389–393 行），实现 O(1) 级更新。

此外，ValkeyDb 以带前缀的版本戳键记录表结构版本（valkey.py），当键不存在时 get_latest_schema_version 默认返回 "2.0.0"，以便 MigrationManager 正常执行迁移，对应单元测试可见 test_valkey.py。

四、用例一：Agent 会话持久化与多轮上下文验证

4.1 示例代码解析

valkey_for_agent.py 的核心逻辑非常精简：

from agno.agent import Agent
from agno.db.base import SessionType
from agno.db.valkey import ValkeyDb
from agno.tools.websearch import WebSearchTools

db = ValkeyDb()

agent = Agent(
    db=db,
    tools=[WebSearchTools()],
    add_history_to_context=True,
)

if __name__ == "__main__":
    agent.print_response("How many people live in Canada?")
    agent.print_response("What is their national anthem called?")

    print("\nVerifying db contents...")
    all_sessions = db.get_sessions(session_type=SessionType.AGENT)
    print(f"Total sessions in Valkey: {len(all_sessions)}")

两个关键配置值得展开：

• db=db：将 ValkeyDb 实例注入 Agent。此后该 Agent 的会话（session）与每次运行（run）都会被写入 Valkey。
• add_history_to_context=True：让历史会话在每次请求时注入上下文，保证多轮指代可解析。这正是测试日志中"Agent answered both turns (correctly resolved 'their' from prior turn via history)"的技术前提——第二轮提问"What is their national anthem called?"中的 their 依赖第一轮的 Canada 答案。

SessionType 枚举定义于 libs/agno/agno/db/base.py，取值为 agent、team、workflow。db.get_sessions(session_type=SessionType.AGENT) 即"只统计 agent 类型的会话"。

4.2 测试结果与复测差异解读

TEST_LOG 记录该用例 PASS，核心观测：

• 两个回合均正确作答，"their" 经历史上下文正确解析；
• 首次运行时 get_sessions 报告 Valkey 中持久化了 3 个 agent 会话；
• 在后续"learnings/isolation（学习与隔离）"特性合入后的复测中，同一脚本在全新 server 上持久化了 2 个会话，两个回合依然正常作答。

会话数量的差异源于被测代码库随版本演进而变化（学习记录、用户隔离等新特性改变了会话的创建与归并逻辑），属于测试基线演进，而非存储功能失效；两次验证均以"会话被成功持久化"为通过标准。配套的集成测试目录 libs/agno/tests/integration/db/valkey 中就包含 test_learnings.py 与 test_user_isolation.py，它们验证的正是这些新特性与 Valkey 存储的兼容性。

五、用例二：Team 结构化输出与团队会话入库

5.1 示例代码解析

valkey_for_team.py 演示了"存储 + 多成员协作 + 结构化输出"的组合：

from agno.agent import Agent
from agno.db.valkey import ValkeyDb
from agno.models.openai import OpenAIResponses
from agno.team import Team
from agno.tools.hackernews import HackerNewsTools
from agno.tools.websearch import WebSearchTools
from pydantic import BaseModel

db = ValkeyDb()

class Article(BaseModel):
    title: str
    summary: str
    reference_links: List[str]

hn_researcher = Agent(
    name="HackerNews Researcher",
    model=OpenAIResponses(id="gpt-5.5"),
    role="Gets top stories from hackernews.",
    tools=[HackerNewsTools()],
)

web_searcher = Agent(
    name="Web Searcher",
    model=OpenAIResponses(id="gpt-5.5"),
    role="Searches the web for information on a topic",
    tools=[WebSearchTools()],
    add_datetime_to_context=True,
)

hn_team = Team(
    name="HackerNews Team",
    model=OpenAIResponses(id="gpt-5.5"),
    members=[hn_researcher, web_searcher],
    db=db,
    instructions=[
        "First, search hackernews for what the user is asking about.",
        "Then, ask the web searcher to search for each story to get more information.",
        "Finally, provide a thoughtful and engaging summary.",
    ],
    output_schema=Article,
    markdown=True,
    show_members_responses=True,
)

需要注意两点：

• output_schema=Article 让团队以 Pydantic 模型约束最终输出。测试日志确认团队返回了包含 title、summary、reference_links 字段的合法结构化 Article——这意味着不仅会话与运行记录入库，结构化结果本身也经受了校验。
• 团队成员会话与团队会话共同入库：db 同时挂在 Team 上，因此 HackerNews Researcher、Web Searcher 两个成员 Agent 的会话以及团队自身的会话都会被持久化。测试日志特别提到"team session index present in Valkey"，即 agno:sessions:index:session_type:team 这类辅助索引集合存在记录。

5.2 上游 Web 搜索波动的观测记录

该用例完整运行约 13 分钟，日志给出了明确归因：外部 DuckDuckGo（ddgs）Web 搜索偶发 RemoteProtocolError/TimeoutException，脚本内部做了重试后成功。TEST_LOG 特别强调"这是上游 web 搜索的波动，不是 Valkey 或 cookbook 的问题"。在 learnings/isolation 合入后的复测中，团队同样完成了结构化 Article 输出。

这一观测对读者有实操价值：当多成员团队叠加外部联网工具时，端到端耗时不仅取决于存储层，更取决于各工具上游的可用性。测试基础设施在 libs/agno/tests/integration/db/valkey/conftest.py 中用一个 valkey_db fixture 连接 localhost:6379、使用 agno_test 前缀隔离测试键，并在每个用例后自动清理 sessions/memories/metrics/evals/... 各表键与索引键，便于反复跑这类联网测试。

六、用例三：Workflow 两阶段编排与会话表自定义

6.1 示例代码解析

valkey_for_workflow.py 演示了 Workflow 场景下自定义会话表名的用法：

from agno.agent import Agent
from agno.db.valkey import ValkeyDb
from agno.team import Team
from agno.workflow.step import Step
from agno.workflow.workflow import Workflow

hackernews_agent = Agent(..., tools=[HackerNewsTools()], role="Extract key insights...")
web_agent = Agent(..., tools=[WebSearchTools()], role="Search the web for the latest news and trends")

research_team = Team(name="Research Team", members=[hackernews_agent, web_agent], ...)
content_planner = Agent(..., instructions=[...])

research_step = Step(name="Research Step", team=research_team)
content_planning_step = Step(name="Content Planning Step", agent=content_planner)

if __name__ == "__main__":
    content_creation_workflow = Workflow(
        name="Content Creation Workflow",
        description="Automated content creation from blog posts to social media",
        db=ValkeyDb(
            session_table="workflow_session",
        ),
        steps=[research_step, content_planning_step],
    )
    content_creation_workflow.print_response(input="AI trends in 2024", markdown=True)

工作流被编排为两个 Step：

1. Research Step：执行 research_team（Hackernews Agent + Web Agent），负责从 Hackernews 与全网挖掘研究素材；
2. Content Planning Step：执行 content_planner Agent，基于研究内容规划 4 周、每周 3 篇的内容排期。

与前面两个用例不同，这里构造 ValkeyDb 时显式传入 session_table="workflow_session"。依据 BaseDb 的表名派生规则，若不额外设置 runs 表，运行记录表会被自动命名为 workflow_session_runs，从而与 agent/team 用例的默认会话表区分开——这是同一个 Valkey 实例中隔离不同业务数据的一种轻量手段。

6.2 测试结果解读

TEST_LOG 记录该用例 PASS：

• 首次运行约 444 秒完成，产出 4 周内容计划；
• 工作流会话索引（workflow_id:content-creation-workflow）在 Valkey 中确认存在；
• learnings/isolation 合入后的复测约 223 秒完成，同样产出 4 周内容计划。

workflow_id 正是 valkey.py 中 get_sessions 的筛选维度之一：当传入 session_type == SessionType.WORKFLOW 时，记录会按 workflow_id 匹配 component_id 过滤；结合 utils.py 的排序/分页逻辑，可以实现"查询某个工作流最近 N 个会话"这类运维需求。

七、验证手段总结：如何确认数据真的写进了 Valkey

三组用例的"验证"套路一致，可归纳为两种可复现手段：

手段一：应用层调用 get_sessions 统计。 agent 用例脚本本身就以 db.get_sessions(session_type=SessionType.AGENT) 打印会话总数作为结束断言；team 与 workflow 用例则通过会话索引键的存在性间接确认。get_sessions 支持 session_type、user_id、component_id（agent_id/team_id/workflow_id）、会话名子串、created_at 时间上下界、排序与分页等多维过滤，其默认实现是"全量取键 + 内存过滤"（源码约第 837–925 行，含一处 TODO 注释计划改用索引集合避免全量扫描）。

手段二：仓库级测试守护。 除 cookbook 三份脚本外，Valkey 存储行为还有两层测试保护：

• 单元测试 libs/agno/tests/unit/db/test_valkey.py：用 mock 客户端验证 get_session/delete_session/过滤表达式求值/内存主题提取/版本戳读写等纯逻辑；
• 集成测试 libs/agno/tests/integration/db/valkey：针对真实 localhost:6379 实例验证批量操作、用户隔离、学习记录、trace/span 统计等跨表功能。

八、结语与关键参数速查

综合三份脚本、源码与测试记录可以得出：Valkey 在 Agno 中已具备成熟的会话存储能力，同一套 ValkeyDb 可以无差别地支撑 Agent、Team 与 Workflow；自定义 session_table 可以细分业务数据空间，db_prefix 提供多环境/多租户隔离，expire 可让记录按 TTL 自动过期。对于生产接入，建议按如下顺序核对配置：

1. uv pip install valkey-glide-sync 并确认导入无误（缺失会抛 ImportError）；
2. 用 docker run --name my-valkey -p 6379:6379 -d valkey/valkey-bundle 或等价方式启动服务端；
3. 本地开发直接用 ValkeyDb()；远程/集群实例按需传入 host、port、username、password、use_tls、database_id，或直接注入自建的 GlideClient/GlideClusterClient；
4. 将实例通过 db= 注入 Agent/Team/Workflow；需要多轮记忆时给 Agent 开启 add_history_to_context=True；
5. 需要结构化结果约束时使用 output_schema；需要区分业务时自定义 session_table；
6. 用 db.get_sessions(session_type=...) 配合 SessionType.AGENT/TEAM/WORKFLOW 枚举校验入库情况。

最后提醒：凡涉及外部联网工具（如 ddgs 搜索）的端到端用例，运行时长可能受上游服务稳定性影响，评测时宜把存储层与工具层问题分开归因，这也是 TEST_LOG 记录本身给出的方法论。