mcpField
G
google-adk

google-adk

github.com/vladm3105Agent 工作流17119 次安装1 天前更新
代码仓库

关于 google-adk

google-adk 是使用 Google's Agent Development Kit 构建 Agent 应用的技能,涵盖代码优先的 Agent 设计、工具、多 Agent 工作流、状态、评估和部署,并介绍 MCP 工具接入。

主要功能

  1. 01LlmAgent 与工作流 Agent 指导
  2. 02可组合的多 Agent 编排模式
  3. 03内置、自定义、OpenAPI 和 MCP 工具
  4. 04会话与持久化状态管理
  5. 05Agent 测试与评估指导
  6. 06部署与容器化选项

适用场景

  1. 01构建使用自定义工具或 API 的 Agent
  2. 02编排顺序或并行的 Agent 工作流
  3. 03跨用户会话持久保存上下文
  4. 04评估并部署 Agent 应用

MCP 相关

该技能介绍连接 MCP 服务并将其工具导入 ADK Agent,以便在不同框架间共享工具。

安装

复制下面的命令,在终端中运行。

$npx skills add https://github.com/vladm3105/aidoc-flow-framework --skill google-adk

安装后在 Claude Code 中输入/google-adk确认已加载。

包含的文件

除 SKILL.md 外,安装时一起下载的参考资料和脚本。

examples/
README.md
google_adk_agent_implementation.py
google_adk_deployment.py
google_adk_multi_agent.py
google_adk_tools_example.py

详细介绍

Google Agent Development Kit (ADK) 技能

用途

提供使用 Google Agent Development Kit (ADK) 开发 Agent 应用和多 Agent 系统的专业指导。帮助 AI 助手设计 Agent、构建工具、编排多 Agent 工作流、实现记忆/状态管理,并按照代码优先的开发模式部署基于 Agent 的应用。

适用场景

在以下情况下调用此技能:

  • 构建集成工具的对话式 AI Agent
  • 创建多 Agent 编排系统
  • 开发工作流 Agent(顺序、并行、迭代)
  • 为 Agent 实现自定义工具
  • 为复杂任务设计 Agent 架构
  • 将 Agent 应用部署到生产环境
  • 评估 Agent 的性能和行为
  • 实现人在回路模式

请勿在以下情况下使用此技能:

  • 通用 Python 开发(使用 Python 专用技能)
  • 简单的 REST API 开发(ADK 面向 Agent 系统)
  • 前端开发(ADK 是后端 Agent 框架)
  • 不涉及 Agent 编排、直接使用 LLM API(使用 LLM 提供商 SDK)
  • 非 Python Agent 框架(LangChain、CrewAI、AutoGPT——模式不同)

ADK 核心概念

平台架构

框架理念:

  • 代码优先方法——在 Python 代码中定义 Agent(而非使用 YAML/JSON 配置)
  • 与模型无关——针对 Gemini 优化,同时支持其他 LLM
  • 可组合——使用简单的 Agent 基元构建复杂系统
  • 可观测——内置与跟踪和监控工具的集成

支持的语言:

  • Python(主要语言,成熟度最高)——google-adk 包
  • Go(可用)——adk-go 仓库
  • Java(可用)——adk-java 仓库

运行环境:

  • 需要 Python 3.9+
  • 使用 Agent Engine 部署(容器化执行)
  • 用于开发/测试的 Web UI(Angular + FastAPI)
  • 用于评估和部署操作的命令行
Agent 类型和层级结构

1. LlmAgent(动态、由模型驱动)

适用于:

  • 对话界面
  • 存在不确定性的决策
  • 自然语言理解
  • 创意任务(内容生成)
  • 上下文推理

特点:

  • 使用 LLM 进行决策
  • 执行过程具有非确定性
  • 工具选择由模型驱动
  • 处理模糊输入

2. 工作流 Agent(确定性、以程序方式执行)

顺序 Agent:

  • 按固定顺序执行工具
  • 适用于:存在依赖关系的多步骤流程
  • 示例:数据管道(获取 → 转换 → 加载)

并行 Agent:

  • 并发执行多个工具
  • 适用于:需要聚合结果的独立操作
  • 示例:从多个数据源收集数据

循环 Agent:

  • 重复执行,直到满足条件
  • 适用于:迭代优化、收敛任务
  • 示例:生成器-评判器模式

3. 自定义 Agent(用户定义的逻辑)

适用于:

  • 特定领域的编排
  • 复杂状态机
  • 与现有系统集成
  • 专用执行模式

Agent 组合:

  • Agent 可以包含子 Agent(层级结构)
  • 父 Agent 协调子 Agent
  • 支持多层嵌套
工具生态系统

工具类别:

  1. 内置工具:

    • Search - 通过 Google Search API 进行网络搜索
    • Code Execution - Python 代码解释器(沙箱环境)
    • Google Cloud 工具 - Vertex AI、BigQuery、Cloud Storage
  2. 自定义函数工具:

    • 将 Python 函数封装为工具
    • 根据类型提示自动生成架构
    • 支持异步函数
  3. OpenAPI 工具:

    • 根据 OpenAPI/Swagger 规范自动生成
    • 基于 HTTP 的服务集成
  4. MCP(Model Context Protocol)工具:

    • 与 MCP 服务集成
    • 跨框架共享工具

工具属性:

  • 名称 - 唯一标识符
  • 描述 - 面向大模型的自然语言说明
  • 参数 - 定义输入的 JSON 架构
  • 函数 - 执行逻辑
  • 确认 - 可选的人工参与审批
内存与状态管理

会话管理:

  • Agent 维护对话历史
  • 自动管理上下文窗口
  • 可配置的历史记录保留策略

状态持久化:

  • 每个 Agent 的自定义状态对象
  • 支持序列化(JSON、pickle)
  • 与数据库集成,实现长期存储

上下文缓存:

  • 减少重复上下文的令牌用量
  • 自动使缓存失效
  • 可配置的缓存 TTL

Agent 开发方法

规划阶段

步骤 1:定义 Agent 目的

  • 主要目标(单一职责)
  • 输入/输出格式
  • 成功标准
  • 失败模式

步骤 2:确定所需工具

选择标准:

  • 有可用的内置工具时使用(Search、Code Execution)
  • 对于简单操作(少于 100 行),创建自定义函数
  • 对于现有 REST API,使用 OpenAPI 工具
  • 对于跨框架兼容性,使用 MCP 工具

步骤 3:选择 Agent 类型

START: What's the agent's decision pattern?
  │
  ├─> Requires natural language reasoning? ─Yes─> LlmAgent ★
  │
  ├─> Fixed sequence of steps?
  │   └─> Sequential Workflow Agent ★
  │
  ├─> Independent parallel operations?
  │   └─> Parallel Workflow Agent ★
  │
  ├─> Iterative refinement needed?
  │   └─> Loop Workflow Agent ★
  │
  └─> Custom orchestration logic?
      └─> Custom Agent ★

步骤 4:设计多 Agent 架构(如有需要)

模式:

  • 协调器/分发器 - 中央 Agent 将任务路由给专用 Agent
  • 顺序流水线 - Agent A 的输出 → Agent B 的输入
  • 并行分发/汇总 - 分配工作并汇总结果
  • 分层分解 - 将复杂任务拆分为子任务
实现阶段

Agent 实现示例:

[查看代码示例:examples/google_adk_agent_implementation.py]

演示的关键 Agent 模式:

  • LlmAgent - create_weather_assistant() - 配备自定义工具的对话式 Agent
  • SequentialAgent - create_data_pipeline() - 按顺序执行(获取 → 转换 → 保存)
  • ParallelAgent - create_market_researcher() - 并发执行工具
  • LoopAgent - create_content_generator() - 在满足中止条件前迭代优化
  • 会话管理 - create_stateful_session() - 带历史记录的多轮对话
测试阶段

Web UI 测试:

# Start API server
adk api_server --port 8000

# Start web UI (separate terminal)
cd adk-web
npm install
npm start
# Access: http://localhost:4200

程序化测试:

# Unit test for agent
def test_weather_agent():
    agent = create_weather_agent()
    response = agent.run("Weather in NYC?")
    assert "weather" in response.content.lower()
    assert response.success is True

# Integration test with mock tools
def test_pipeline_agent():
    agent = create_pipeline_agent(mock_tools=True)
    result = agent.run({"input": "test_data"})
    assert result["status"] == "completed"

评估框架:

from google.adk.evaluation import evaluate_agent

# Criteria-based evaluation
results = evaluate_agent(
    agent=my_agent,
    test_cases=[
        {"input": "What's 2+2?", "expected_output": "4"},
        {"input": "Explain quantum computing", "criteria": "mentions_qubits"}
    ],
    evaluator_model="gemini-2.0-flash"
)

print(f"Pass rate: {results.pass_rate}")
print(f"Average score: {results.avg_score}")

工具开发

[查看代码示例:examples/google_adk_tools_example.py]

自定义函数工具

演示示例:

  • 基本函数工具 - calculate_tax() - 带类型提示的简单工具
  • 异步工具 - fetch_user_data() - 异步 API 调用
  • HITL 确认 - send_email()、delete_user_account() - 需要人工审批
  • 输入验证 - send_email_tool() - 验证并清理电子邮件格式
  • 重试逻辑 - fetch_external_data() - 使用指数退避机制自动重试
  • 速率限制 - call_external_api() - 基于装饰器的速率限制
  • 错误处理 - fetch_stock_price() - 发生故障时优雅降级
OpenAPI 工具

集成模式:

  • 从 OpenAPI 规范 URL 加载工具
  • 可选:筛选特定操作的工具
  • 根据规范自动生成架构 [参见 examples/google_adk_tools_example.py 中的 create_api_agent()]
MCP 工具集成

集成模式:

  • 连接到 MCP 服务端点
  • 导入工具以实现跨框架兼容性 [参见:examples/google_adk_tools_example.py 中的 create_mcp_agent()]

多 Agent 编排

[参见代码示例:examples/google_adk_multi_agent.py]

多 Agent 模式

模式 1:协调器/调度器(复杂度:4)

  • 用例: 将用户请求路由到专用 Agent
  • 函数: create_coordinator_system() - 天气 + 金融领域专家

模式 2:顺序流水线(复杂度:3)

  • 用例: 具有依赖关系的多阶段处理
  • 函数: create_content_pipeline() - 研究 → 撰写 → 编辑

模式 3:并行扇出/汇总(复杂度:4)

  • 用例: 汇总多个来源的结果
  • 函数: create_market_analysis_system() - 技术面 + 基本面 + 情绪分析

模式 4:分层分解(复杂度:5)

  • 用例: 将复杂任务拆分为易于管理的子任务
  • 函数: create_project_management_system() - 多层级 Agent 层次结构

模式 5:生成器-批评者循环(复杂度:4)

  • 用例: 通过反馈进行迭代改进
  • 函数: create_quality_content_system() - 生成 → 评审 → 改进

模式 6:人在回路中(HITL)(复杂度:3)

  • 用例: 关键决策需要人工批准
  • 函数: create_account_management_agent() - 删除前确认

模式 7:状态管理

  • 用例: 在不同会话间持久保存用户上下文
  • 类: StatefulAgent - 带历史记录的内存状态存储

模式 8:数据库持久化

  • 用例: 长期状态存储
  • 函数: save_state()、load_state() - 基于 PostgreSQL 的持久化

记忆与状态管理

[参见代码示例:examples/google_adk_multi_agent.py - 状态管理部分]

会话管理

基本会话模式:

  • 保留历史记录的多轮对话
  • 自动管理上下文窗口
  • 可配置的历史记录限制

[参见:examples/google_adk_agent_implementation.py 中的 create_stateful_session()]

状态持久化

自定义状态对象:

  • 每位用户的内存状态存储
  • 基于数据类的状态建模
  • 对话历史跟踪

[参见:examples/google_adk_multi_agent.py 中的 StatefulAgent 类]

数据库持久化:

  • 使用 SQLAlchemy 进行长期状态存储
  • JSON 序列化的状态数据
  • 支持 PostgreSQL/MySQL

[参见:examples/google_adk_multi_agent.py 中的 save_state()、load_state() 函数]

部署选项

[参见代码示例:examples/google_adk_deployment.py]

Agent Engine(托管服务)

部署命令:

pip install google-adk[cli]
adk auth login
adk deploy --agent-file agent.py --agent-name my_agent --project-id my-gcp-project --region us-central1

[参见 create_production_agent() 的配置示例]

Cloud Run 部署

组件:

  • 带有 Agent 端点的 FastAPI 服务
  • 用于容器化的 Dockerfile
  • 健康检查和错误处理
  • 环境配置

[参见 examples/google_adk_deployment.py 中的 FastAPI 应用实现、Dockerfile 参考和部署命令]

Docker 容器化

自托管选项:

  • 使用 Redis 的 Docker Compose
  • 单容器部署
  • 环境变量配置

[参见 examples/google_adk_deployment.py 中的 docker-compose.yml 参考和部署命令]

资源需求
Agent 复杂度CPURAM并发请求数
简单 LlmAgent1 核512MB10
工作流 Agent2 核1GB20
多 Agent(3-5 个 Agent)4 核2GB10
复杂多 Agent(>5 个)8 核4GB5

评估与测试

[参见代码示例:examples/google_adk_deployment.py - 评估端点]

基于标准的评估

模式:

  • 定义自定义评估标准(准确性、实用性等)
  • 针对 Agent 运行测试用例
  • 分析通过率和评分

[参见:evaluate_agent_endpoint(),位于 examples/google_adk_deployment.py]

用户模拟评估

模式:

  • 根据既定目标模拟用户交互
  • 跟踪目标完成率
  • 衡量完成目标所需的平均轮数

[参见 UserSimulator 示例的文档]

最佳实践

[参见代码示例:examples/google_adk_tools_example.py 中的工具设计最佳实践部分]

编写 Agent 指令

有效模式:

  • 明确的角色和职责
  • 结构化格式并包含约束条件
  • 具体的工具使用指导
  • 交互示例

[参见 examples/google_adk_tools_example.py 末尾的 GOOD 与 BAD 示例]

工具设计原则

核心原则:

  1. 单一职责——每个工具只负责一个明确的用途
  2. 描述性命名——清晰地命名操作和对象
  3. 类型提示——为所有参数提供完整的类型注解

[参见 examples/google_adk_tools_example.py 末尾的工具设计示例]

错误处理

策略:

  • 平稳降级——返回错误消息,而不是引发异常
  • 重试逻辑——使用指数退避自动重试
  • 输入验证——验证并清理所有输入

[参见 examples/google_adk_tools_example.py 中的 fetch_stock_price()、fetch_external_data() 和 send_email_tool()]

安全与防护

实现方式:

  • 输入验证——验证电子邮件格式并限制长度
  • 速率限制——基于装饰器的请求限流
  • 清理——移除危险的 HTML/脚本内容

[参见 examples/google_adk_tools_example.py 中的 send_email_tool() 和 rate_limit() 装饰器]

性能优化

异步工具:

  • 自动并行执行异步函数
  • 提高 I/O 密集型操作的吞吐量

[参见 examples/google_adk_tools_example.py 中的 fetch_price() 和 create_portfolio_analyzer()]

质量门槛

完成定义:Agent

满足以下条件时,Agent 即可用于生产环境:

  1. 功能:

    • ✓ Agent 能在测试用例中完成主要目标
    • ✓ 使用有效输入时,工具执行成功
    • ✓ 错误处理涵盖预期的失败模式
    • ✓ 多轮对话能够保持上下文
  2. 性能:

    • ✓ 简单查询的响应时间 <5 秒
    • ✓ 复杂工作流的响应时间 <30 秒
    • ✓ 按标准评估的通过率 ≥80%
    • ✓ 资源使用量在部署限制范围内
  3. 安全:

    • ✓ 所有工具都进行输入验证
    • ✓ 高风险操作需要确认(HITL)
    • ✓ 不存在硬编码的凭据或 API 密钥
    • ✓ 对外部 API 调用实施速率限制
  4. 可观测性:

    • ✓ 已配置日志记录以便调试
    • ✓ 已为多 Agent 工作流启用追踪
    • ✓ 已跟踪评估指标
    • ✓ 已配置错误告警
  5. 文档:

    • ✓ 已记录 Agent 的用途和功能
    • ✓ 工具描述清晰准确
    • ✓ 提供了使用示例
    • ✓ 已记录已知限制
完成定义:工具

满足以下条件时,工具即可用于生产环境:

  1. 接口:

    • ✓ 函数的所有参数均有类型提示
    • ✓ 文档字符串说明用途、参数和返回值
    • ✓ 参数描述可指导大模型进行选择
    • ✓ 返回值可序列化为 JSON
  2. 可靠性:

    • ✓ 错误处理会提供有用的消息
    • ✓ 输入验证可防止无效操作
    • ✓ 为长时间运行的操作配置了超时
    • ✓ 为暂时性故障配置了重试逻辑
  3. 测试:

    • ✓ 单元测试涵盖成功用例
    • ✓ 单元测试涵盖错误用例
    • ✓ 使用 Agent 执行进行集成测试
    • ✓ 对高开销操作进行性能基准测试

错误处理指南

常见问题与解决方案

问题:Agent 未调用正确的工具

  • **原因:**工具描述不清晰或含糊
  • 解决方案:
    # BAD: Vague description
    def process(data):
        """Process data."""  # Too generic
    
    # GOOD: Specific description
    def validate_email_format(email: str) -> bool:
        """Check if email address matches valid format ([email protected]).
    
        Use this tool ONLY to validate email syntax, not to verify
        if email exists or is deliverable."""
    

问题:Agent 无限循环

  • **原因:**Loop Agent 中没有终止条件
  • 解决方案:
    # Add max_iterations and explicit break condition
    agent = LoopAgent(
        tools=[...],
        max_iterations=10,  # Hard limit
        break_condition=lambda result: result.get("completed", False)
    )
    

问题:“Tool execution failed”错误

  • **原因:**工具引发了未处理的异常
  • 解决方案:
    def robust_tool(param: str) -> str:
        try:
            result = risky_operation(param)
            return f"Success: {result}"
        except SpecificError as e:
            return f"Operation failed: {e.message}"
        except Exception as e:
            logger.error(f"Unexpected error in robust_tool: {e}")
            return "Temporary service error, please try again"
    

问题:Agent 响应太慢

  • **原因:**本可并行执行的工具调用却按顺序执行
  • 解决方案:
    # Use Parallel Agent or async tools
    agent = ParallelAgent(
        tools=[tool1, tool2, tool3]  # Execute concurrently
    )
    

问题:超出上下文限制

  • **原因:**对话历史过长
  • 解决方案:
    session = Session(
        agent=my_agent,
        max_history_turns=10,  # Limit history
        context_window_tokens=30000  # Set explicit limit
    )
    

问题:在 Cloud Run 上部署失败

  • **原因:**缺少依赖项或环境变量
  • 解决方案:
    # Ensure requirements.txt is complete
    pip freeze > requirements.txt
    
    # Set required environment variables
    gcloud run deploy my-agent \
      --set-env-vars GOOGLE_API_KEY=your_key,AGENT_CONFIG=prod
    
调试策略

1. 启用详细日志记录:

import logging

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("google.adk")
logger.setLevel(logging.DEBUG)

2. 独立测试工具:

# Test tool without agent
tool = Tool.from_function(my_function)
result = my_function("test_input")
print(f"Tool output: {result}")

3. 使用 Web UI 进行交互式调试:

adk api_server --debug
# Access UI at http://localhost:4200
# View tool calls, agent reasoning, response generation

4. 检查 Agent 执行跟踪:

response = agent.run("test query", return_trace=True)
print(response.trace)  # Shows all tool calls and decisions

复杂度评级

任务评级描述
使用工具的简单 LlmAgent2基本对话式 Agent
顺序工作流 Agent2按固定顺序执行工具
并行工作流 Agent3并发操作
循环工作流 Agent3迭代优化
自定义 Agent3用户定义的编排
协调器/分发器(2-3 个 Agent)4多 Agent 路由
顺序流水线(3 个以上 Agent)4链式 Agent 执行
分层多 Agent(超过 5 个 Agent)5复杂的嵌套架构
自定义工具开发2Python 函数封装
OpenAPI 工具集成2根据规范自动生成
MCP 工具集成3跨框架工具
部署到 Agent Engine2托管式部署
可自托管的 Docker 部署3容器编排
高级评估框架4自定义标准与模拟

参考资料

官方文档
Python SDK 资源
其他语言
社区资源
相关技能
  • 用于工作流自动化:使用 n8n 技能
  • 用于 API 设计:使用 api-design-architect 技能
  • 用于云部署:使用 cloud-devops-expert 技能
  • 用于 Python 开发:使用 Python 专用技能
  • 用于大模型集成:使用模型提供商 SDK(OpenAI、Anthropic 等)

版本: 1.0.0 最后更新: 2025-11-13 复杂度评级: 3(中等——需要了解 Agent 架构) 预计学习时间: 熟练掌握需要 10-15 小时

评论

支持 Markdown