adk-samples 之 long-horizon-harness Live Tests 实战指南：用真实 Vertex AI 冒烟测试 ADK 长时任务 Agent

adk-samples 之 long-horizon-harness Live Tests 实战指南用真实 Vertex AI 冒烟测试 ADK 长时任务 Agent【免费下载链接】adk-samplesA collection of sample agents built with Agent Development Kit (ADK)项目地址: https://gitcode.com/GitHub_Trending/ad/adk-samples本篇指南聚焦于 adk-samples 仓库中core/python/long-horizon-harness示例Horizon——一个构建在 Google ADK Vertex AI 之上的自改进长时任务 Agent的可选真实环境测试体系tests/live/目录。该目录下的测试需要真实的 GCP 凭证与已启用的 Vertex AI / Agent Platform API默认被pyproject.toml的testpaths排除在常规pytest之外。读完本文你将掌握 live tests 的适用场景、运行前置条件、精确命令、两套测试Agent 流式运行与 FastAPI 服务端到端的源码级原理以及它与tests/unit、tests/integration确定性套件之间的分工边界。Live Tests 是什么可选择开启、命中真实 Vertex AI 的冒烟测试在core/python/long-horizon-harness/tests/live/目录下存放着一组被刻意隔离的测试。它们与常规单元/集成测试最大的区别在于不隔离任何外部服务需要真实的 GCP 凭证Application Default Credentials需要已启用的 Vertex AI / Agent Platform API测试会直接调用真实 ADK 运行时路径——例如通过Runner.run()驱动真实root_agent向 Gemini 发起推理或真正拉起一个uvicornFastAPI 服务进程并通过 HTTP 与其交互。这正是 tests/live/README.md 开篇所强调的定位这些测试是opt-in可选开启的。默认执行uv run pytest时它们绝不会被收集运行保证确定性测试套件始终全绿。为什么需要单独的目录守护 Hermetic 测试原则将 live tests 从tests/integration/中移出的根本原因写在了 README 引用的CLAUDE.md对应本仓库的维护规范 AGENTS.md中Unit tests do NOT need GCP — they useInMemoryMemoryService/InMemorySessionService. Onlyagents-cli run,agents-cli playground, andagents-cli eval runhit Vertex.即单元测试不需要 GCP它们使用内存版的服务实现只有少数 CLI 命令会命中 Vertex。然而 agents-cli 脚手架自动生成的集成测试调用Runner.run()访问真实root_agent、或拉起真实uvicorn服务违背了这条规则。将它们移出tests/integration/就能让默认测试套件保持hermetic密闭、自包含——这正是上游 agents-cli 脚手架的一个已知局限Horizon 通过目录隔离来规避它。pyproject.toml 中的排除机制排除行为由 pyproject.toml 的[tool.pytest.ini_options]声明[tool.pytest.ini_options] pythonpath . asyncio_default_fixture_loop_scope function # tests/live/ is excluded by default — those tests hit real Vertex / spawn # real uvicorn and require GCP credentials. Run them explicitly with # uv run pytest tests/live. See tests/live/README.md. testpaths [tests/unit, tests/integration]关键点testpaths只包含tests/unit与tests/integration因此不带参数执行uv run pytest时pytest 根本不会扫描tests/live/但 pytest 的testpaths只是“默认收集路径”当你显式传入目录参数如uv run pytest tests/live -v时排除即失效测试照常收集并运行文件中的注释明确说明live tests 会命中真实 Vertex、拉起真实 uvicorn需要 GCP 凭证必须显式运行。确定性套件的密闭性保障作为对照默认套件的密闭性由 tests/conftest.py 的自动 fixture 层层保证_hermetic_environment在每个测试前清除所有形如*_API_KEY、*_TOKEN、*_SECRET、*_PASSWORD、*_CREDENTIALS等凭证类环境变量防止本地密钥泄漏进测试并固定TZUTC、LANGC.UTF-8、PYTHONHASHSEED0保证确定性_scoped_environment为每个测试在tmp_path下绑定一个LocalEnvironment路径类工具都在临时目录内解析fake_memory_service/fake_session_service/runner_factory统一提供InMemoryMemoryService与InMemorySessionService构建的Runner绝不触碰真实 Vertex / Memory Bank。何时运行 Live Tests变更后验证真实运行时路径README 给出了非常具体的适用时机在一次非平凡改动之后用 live tests 冒烟验证真实 ADK 运行时路径仍然可用典型场景包括模型切换model swap例如修改 agent.py 中LHA_ROOT_MODEL默认值、或 models/registry.py 中的_MODELS注册表回调接线调整callback wiringHorizon 在root_agent上注册了before_agent_callback、before_model_callback、before_tool_callback、after_tool_callback、after_agent_callback五条有序回调链任何链的增删改都值得跑一次真实链路FastAPI 服务配置变更例如修改 fast_api_app.py 的路由挂载、会话/记忆/工件服务的 URI 解析逻辑。除此之外日常开发应优先使用tests/unit/与tests/integration/下的确定性套件它们使用InMemoryMemoryService/InMemorySessionService既快又无需云资源。如何运行前置条件与命令前置条件已安装 uv 并完成依赖同步uv syncgoogle-adk[mcp,otel-gcp,a2a]2.8.0,3.0.0等依赖来自 pyproject.toml 的[project]段拥有一个 GCP 项目且已启用 Vertex AI / Agent Platform API配置好 Application Default Credentials 与项目 ID。运行命令# 确保 GOOGLE_CLOUD_PROJECT 与凭证已配置 gcloud auth application-default login export GOOGLE_CLOUD_PROJECTyour project # 然后显式运行 live tests uv run pytest tests/live -v两点补充说明依据源码与 README若未设置GOOGLE_CLOUD_PROJECTagent.py 在导入时会尝试通过google.auth.default()探测 ADC 默认项目并给出警告日志显式设置可以精确选择计费项目-v用于展示每个测试的详细结果。由于测试会真实命中 Vertex 与本地起服务耗时通常显著高于确定性套件请预留充足超时时间。源码剖析一Agent 级流式冒烟测试tests/live/test_agent_live.py 中只有一个测试test_agent_stream它是最小化的“真实 Agent 跑通”验证from google.adk.agents.run_config import RunConfig, StreamingMode from google.adk.runners import Runner from google.adk.sessions import InMemorySessionService from google.genai import types from horizon.agent import root_agent def test_agent_stream() - None: session_service InMemorySessionService() session session_service.create_session_sync( user_idtest_user, app_nametest ) runner Runner( agentroot_agent, session_servicesession_service, app_nametest ) message types.Content( roleuser, parts[types.Part.from_text(textWhy is the sky blue?)] ) events list( runner.run( new_messagemessage, user_idtest_user, session_idsession.id, run_configRunConfig(streaming_modeStreamingMode.SSE), ) ) assert len(events) 0, Expected at least one message has_text_content False for event in events: if ( event.content and event.content.parts and any(part.text for part in event.content.parts) ): has_text_content True break assert has_text_content, Expected at least one message with text content代码要点直接导入horizon.agent的root_agent——这是 agent.py 中_build_app_object()构建的真实 Agent 单例内含 memory、ReadTool、bash、subagent、routine 等工具以及五条回调链会话服务仍使用InMemorySessionService会话容器本身不依赖云但runner.run()会真实调用 LLM——root_agent的模型由_resolve_root_model(None)解析为默认的DispatchingLlm最终路由到 Vertex Gemini指定StreamingMode.SSE验证流式事件路径断言只做两层检查至少产生一个事件、至少一个事件携带文本内容。这符合 AGENTS.md 中“pytest 不断言 LLM 输出内容只断言代码正确性类型、契约、工具 I/O”的开发规则。源码剖析二FastAPI 服务端到端冒烟测试tests/live/test_server_e2e_live.py 是更重量级的验证它真实拉起 uvicorn 服务再通过 HTTP 完成会话创建、SSE 流式对话、错误处理与反馈上报四个场景。服务启动与就绪等待def start_server() - subprocess.Popen[str]: command [ sys.executable, -m, uvicorn, horizon.fast_api_app:app, --host, 0.0.0.0, --port, 8000, ] env os.environ.copy() env[INTEGRATION_TEST] TRUE process subprocess.Popen(command, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, textTrue, bufsize1, envenv) threading.Thread(targetlog_output, args(process.stdout, logger.info), daemonTrue).start() threading.Thread(targetlog_output, args(process.stderr, logger.error), daemonTrue).start() return process以子进程方式运行uvicorn horizon.fast_api_app:app即 fast_api_app.py 中通过模块级__getattr__(app)惰性构建的 FastAPI 应用挂载 A2A JSON-RPC、/lha/*各路由、/scheduler/*端点设置INTEGRATION_TESTTRUE环境变量两个后台线程实时转发 stdout/stderr 到日志wait_for_server()以 90 秒超时、1 秒间隔轮询GET http://127.0.0.1:8000/docs直到返回 200 才继续server_fixture以 session 级 fixture 的形式在全部测试结束后通过request.addfinalizer终止子进程。会话创建与 SSE 流式对话test_chat_stream演示了完整的 HTTP 调用链创建会话POST http://127.0.0.1:8000/apps/app/users/{user_id}/sessions请求体携带state如{preferred_language: English, visit_count: 1}断言返回 200 并取出session_id发送流式消息POST /run_sse请求体包含app_name、user_id、session_id、new_message角色与文本以及streaming: True以streamTrue读取响应解析 SSE 事件逐行读取响应对data:前缀的行去掉前缀后用json.loads解析成事件对象断言至少收到一个事件且其中至少一个事件携带非空content.parts[].text。错误处理与反馈端点test_chat_stream_error_handling向/run_sse发送一个带invalid_type消息的非法请求体断言返回HTTP 422FastAPI 的请求校验失败码验证服务对畸形输入有正确的契约行为test_collect_feedback向/feedback提交{session_id: test-session-456, text: Great response!}断言返回 200。该端点由 fast_api_app.py 中的attach_feedback_routes(app, runnerrunner)挂载。测试分层总结何时用哪一层结合 AGENTS.md 的测试矩阵Horizon 的测试体系按“云依赖程度”分层层级目录云依赖用途确定性单元/集成tests/unit、tests/integration无InMemoryMemoryService/InMemorySessionService默认uv run pytest全绿验证类型、契约、持久化、工具 I/OLive 冒烟tests/live真实 Vertex AI 真实 uvicorn模型切换、回调接线、FastAPI 配置变更后的真实运行时路径验证Smoke可选tests/smokeRUN_SMOKE1门控RUN_SMOKE_LLM1才命中 LLM命中 FastAPI、不依赖 LLM 的服务级冒烟EvalLLM 行为tests/eval/evalsetsuv run adk evalLLM 行为验证本仓库不使用agents-cli eval run其 CLI 存在解析限制对于绝大多数开发场景请优先跑确定性套件只有当你的改动触及模型选择、回调链、服务配置这些真实运行时要素时才按 README 的指引显式运行uv run pytest tests/live -v结语tests/live/是 adk-samples 中 long-horizon-harness 示例“真实性兜底”的一层它用目录隔离 testpaths排除的方式把需要真实 Vertex AI 凭证的冒烟测试与默认 hermetic 套件干净地切分开同时在模型、回调与服务配置发生非平凡变更时为开发者提供一条“一键验证真实运行时路径仍可用”的可靠通道。理解这套设计与两个测试文件的实现细节你也可以在自己的 ADK Agent 项目中复刻同样的分层测试策略。【免费下载链接】adk-samplesA collection of sample agents built with Agent Development Kit (ADK)项目地址: https://gitcode.com/GitHub_Trending/ad/adk-samples创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考