跳转至

Google ADK 完整指南 2026:用代码构建智能 AI Agent

什么是 Google ADK?

Google ADK (Agent Development Kit) 是 Google 推出的开源 AI Agent 开发框架,专为希望将 AI Agent 集成到实际应用程序中的开发者设计。与 LangChain、LlamaIndex 等框架不同,ADK 采用"代码优先"(code-first)的设计理念,让 Agent 的行为更像传统软件一样可预测、可测试。

核心特点

  • 代码优先:使用纯 Python 定义 Agent 行为,无需复杂的 YAML 配置
  • 类型安全:充分利用 Python 类型提示,提供 IDE 自动补全和错误检查
  • 可测试性:Agent 逻辑可以像普通函数一样进行单元测试
  • 生产就绪:内置日志、监控、错误处理等生产环境必需功能
  • 多模型支持:支持 Gemini、GPT-4、Claude 等主流大语言模型

适用场景

  • 客户服务聊天机器人
  • 自动化工作流程助手
  • 数据分析与报告生成
  • 代码审查与辅助编程
  • 企业内部知识问答系统

安装与配置

环境要求

  • Python 3.10 或更高版本
  • pip 包管理器
  • Google Cloud 账号(使用 Gemini 模型时)

安装步骤

# 创建虚拟环境(推荐)
python3 -m venv adk-env
source adk-env/bin/activate  # Linux/macOS
# adk-env\Scripts\activate  # Windows

# 安装 Google ADK
pip install google-adk

# 安装额外依赖(根据需要)
pip install google-adk[cli]  # 包含命令行工具
pip install google-adk[testing]  # 包含测试工具

API 密钥配置

# 方式 1:使用环境变量(推荐用于开发)
export GEMINI_API_KEY="your-api-key-here"

# 方式 2:使用 Google Cloud 凭据文件
export GOOGLE_APPLICATION_CREDENTIALS="/path/to/service-account.json"

# 方式 3:在代码中直接配置(不推荐用于生产)

核心概念

1. Agent(智能体)

Agent 是 ADK 的核心组件,负责接收用户输入、调用工具、生成响应。

from google.adk import Agent
from google.adk.models import Gemini

# 创建基础 Agent
agent = Agent(
    name="customer_support",
    model=Gemini(model="gemini-2.0-flash"),
    instruction="你是一个专业的客服助手,帮助用户解决产品问题。",
)

2. Tools(工具)

工具是 Agent 可以调用的外部函数,用于执行具体任务。

from google.adk import Tool
from google.adk.tools import FunctionTool

# 定义工具函数
def search_knowledge_base(query: str) -> str:
    """搜索知识库获取相关信息"""
    # 实际实现可能调用数据库或 API
    return f"搜索'{query}'的结果..."

# 创建工具
search_tool = FunctionTool(
    func=search_knowledge_base,
    name="search_kb",
    description="搜索产品知识库",
)

# 将工具添加到 Agent
agent.tools = [search_tool]

3. Sessions(会话)

会话管理用户与 Agent 的交互历史,支持多轮对话。

from google.adk import Session

# 创建会话
session = Session(
    agent=agent,
    user_id="user_123",
)

# 发送消息
response = await session.send("我的产品无法启动,怎么办?")
print(response.text)

4. Events(事件)

事件系统支持异步消息处理和流式响应。

from google.adk import Event

async def handle_event(event: Event):
    if event.type == "user_message":
        response = await session.send(event.content)
        await event.reply(response.text)

实战示例

示例 1:简单的问答机器人

from google.adk import Agent, Session, Runner
from google.adk.models import Gemini

# 创建 Agent
agent = Agent(
    name="faq_bot",
    model=Gemini(model="gemini-2.0-flash"),
    instruction="""你是一个 FAQ 机器人,回答关于产品的问题。
    - 保持回答简洁明了
    - 如果不知道答案,诚实告知
    - 提供相关帮助建议""",
)

# 运行交互式会话
async def main():
    runner = Runner(agent=agent)
    await runner.run_interactive()

# python3 faq_bot.py

示例 2:带工具的数据分析助手

from google.adk import Agent, Tool
from google.adk.tools import FunctionTool
from google.adk.models import Gemini
import pandas as pd

# 定义数据分析工具
def load_csv(file_path: str) -> dict:
    """加载 CSV 文件并返回基本信息"""
    df = pd.read_csv(file_path)
    return {
        "rows": len(df),
        "columns": list(df.columns),
        "dtypes": df.dtypes.astype(str).to_dict(),
    }

def calculate_stats(file_path: str, column: str) -> dict:
    """计算指定列的统计信息"""
    df = pd.read_csv(file_path)
    numeric_col = pd.to_numeric(df[column], errors="coerce")
    return {
        "mean": float(numeric_col.mean()),
        "median": float(numeric_col.median()),
        "std": float(numeric_col.std()),
        "min": float(numeric_col.min()),
        "max": float(numeric_col.max()),
    }

# 创建工具
tools = [
    FunctionTool(func=load_csv, name="load_csv", description="加载 CSV 文件"),
    FunctionTool(func=calculate_stats, name="calc_stats", description="计算统计信息"),
]

# 创建 Agent
agent = Agent(
    name="data_analyst",
    model=Gemini(model="gemini-2.0-flash"),
    instruction="你是一个数据分析助手,帮助用户分析 CSV 数据文件。",
    tools=tools,
)

示例 3:多 Agent 协作系统

from google.adk import Agent, Runner
from google.adk.models import Gemini

# 创建多个专业 Agent
researcher = Agent(
    name="researcher",
    model=Gemini(model="gemini-2.0-flash"),
    instruction="你负责收集和研究信息,提供详细的事实和数据。",
)

writer = Agent(
    name="writer",
    model=Gemini(model="gemini-2.0-flash"),
    instruction="你负责将研究结果整理成结构清晰、易读的报告。",
)

reviewer = Agent(
    name="reviewer",
    model=Gemini(model="gemini-2.0-flash"),
    instruction="你负责审查报告的质量,确保准确性和完整性。",
)

# 使用工作流编排多个 Agent
async def collaborative_workflow(user_query: str):
    # 研究阶段
    research_result = await researcher.run(user_query)

    # 写作阶段
    draft = await writer.run(f"基于以下研究结果撰写报告:{research_result}")

    # 审查阶段
    final_report = await reviewer.run(f"审查并改进以下报告:{draft}")

    return final_report

高级功能

流式响应

from google.adk import Agent, Session
from google.adk.models import Gemini

agent = Agent(
    name="streaming_bot",
    model=Gemini(model="gemini-2.0-flash"),
)

session = Session(agent=agent)

# 流式处理响应
async for chunk in session.send_stream("讲一个关于 AI 的故事"):
    print(chunk.text, end="", flush=True)

自定义中间件

from google.adk import Middleware

class LoggingMiddleware(Middleware):
    """日志中间件,记录所有请求和响应"""

    async def on_request(self, request):
        print(f"[REQUEST] {request.user_id}: {request.content}")
        return request

    async def on_response(self, response):
        print(f"[RESPONSE] {response.text[:100]}...")
        return response

# 添加中间件
agent.middleware = [LoggingMiddleware()]

错误处理与重试

from google.adk import Agent
from google.adk.models import Gemini
from google.adk.retry import RetryConfig

agent = Agent(
    name="robust_bot",
    model=Gemini(model="gemini-2.0-flash"),
    retry_config=RetryConfig(
        max_retries=3,
        retry_delay=1.0,
        retry_on=[TimeoutError, ConnectionError],
    ),
)

最佳实践

1. 提示词工程

# ✅ 好的做法:清晰、具体的指令
agent = Agent(
    instruction="""你是一个专业的技术支持助手。

    回答规则:
    1. 先确认理解用户问题
    2. 提供分步骤解决方案
    3. 如果问题复杂,建议联系人工支持
    4. 保持友好、专业的语气""",
)

# ❌ 避免:模糊、笼统的指令
agent = Agent(
    instruction="帮助用户解决问题",  # 太模糊
)

2. 工具设计

# ✅ 好的做法:单一职责、清晰文档
def get_order_status(order_id: str) -> dict:
    """
    查询订单状态

    Args:
        order_id: 订单编号(格式:ORD-XXXXXX)

    Returns:
        包含 status、estimated_delivery 等信息的字典
    """
    pass

# ❌ 避免:多功能、参数过多
def handle_order(action: str, order_id: str, user_id: str, 
                 items: list, address: dict, payment: dict) -> dict:
    # 太复杂,难以维护
    pass

3. 测试策略

import pytest
from google.adk import Agent, Session

@pytest.fixture
def test_agent():
    return Agent(
        name="test_bot",
        model=Gemini(model="gemini-2.0-flash"),
        instruction="你是一个测试助手",
    )

@pytest.mark.asyncio
async def test_agent_response(test_agent):
    session = Session(agent=test_agent)
    response = await session.send("你好")
    assert response.text is not None
    assert len(response.text) > 0

4. 性能优化

# 使用连接池
from google.adk.models import Gemini

model = Gemini(
    model="gemini-2.0-flash",
    max_concurrent_requests=10,
    timeout=30.0,
)

# 缓存常用响应
from functools import lru_cache

@lru_cache(maxsize=100)
def get_faq_answer(question_hash: str) -> str:
    # 缓存逻辑
    pass

部署指南

本地开发服务器

from google.adk import Agent, Runner
from google.adk.models import Gemini

agent = Agent(
    name="web_bot",
    model=Gemini(model="gemini-2.0-flash"),
)

# 启动 Web 服务器
runner = Runner(agent=agent)
runner.serve(host="0.0.0.0", port=8000)

# 访问 http://localhost:8000

Docker 部署

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["python", "main.py"]

Google Cloud Run

# cloud-run.yaml
apiVersion: serving.knative.dev/v1
kind: Service
metadata:
  name: adk-agent
spec:
  template:
    spec:
      containers:
      - image: gcr.io/your-project/adk-agent
        env:
        - name: GEMINI_API_KEY
          valueFrom:
            secretKeyRef:
              name: api-keys
              key: gemini

常见问题

Q1: ADK 与 LangChain 有什么区别?

A: ADK 更注重代码优先和类型安全,适合有编程经验的开发者;LangChain 提供更丰富的预构建组件和 YAML 配置,适合快速原型开发。

Q2: 支持哪些大语言模型?

A: 原生支持 Google Gemini 系列模型。通过自定义 Provider,也可以集成 OpenAI GPT、Anthropic Claude、本地模型等。

Q3: 如何调试 Agent 行为?

A: 使用 ADK 内置的调试模式:

from google.adk import Runner

runner = Runner(agent=agent, debug=True)
# 详细日志会输出到控制台

Q4: 生产环境需要注意什么?

A: - 使用环境变量管理 API 密钥 - 配置适当的速率限制 - 实现日志记录和监控 - 添加错误处理和重试机制 - 进行充分的负载测试

总结

Google ADK 为开发者提供了一个强大而灵活的工具来构建生产级 AI Agent 应用。其代码优先的设计理念使得 Agent 开发更接近传统软件工程实践,便于测试、维护和扩展。

下一步学习

  1. 阅读 官方文档
  2. 查看 GitHub 示例仓库
  3. 加入 开发者社区
  4. 尝试构建自己的第一个 Agent 项目

相关资源: