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 开发更接近传统软件工程实践,便于测试、维护和扩展。
下一步学习
- 阅读 官方文档
- 查看 GitHub 示例仓库
- 加入 开发者社区
- 尝试构建自己的第一个 Agent 项目
相关资源: