源码级别解析 · 源码解析 · 2026最新版
2026-05-16 | 每日技术深度解读
ADK 应用软件工程原则于AI智能体开发，提供从简单任务到复杂系统的完整解决方案
ADK 旨在让智能体开发更像软件开发，提供可测试、可版本化的开发体验
分层架构确保了灵活性和可扩展性，每个层次都有明确的职责边界
分层架构支持从简单到复杂的智能体系统开发
每种智能体类型针对不同的使用场景和业务需求
from google.adk import Agent
from google.adk.tools import google_search
# 基础LLM智能体
research_agent = Agent(
name="research_assistant",
model="gemini-2.5-flash",
instruction="You are a helpful research assistant. Use search when needed.",
tools=[google_search]
)
# 配置更复杂的智能体
complex_agent = Agent(
name="complex_reasoning",
model="gemini-2.5-flash",
instruction="""
You are an advanced reasoning agent. Break down complex problems
into smaller tasks and solve them systematically.
""",
tools=[google_search, analysis_tool]
)LLM Agent是ADK中最基础的智能体类型，支持工具调用和多轮对话
from google.adk.agents import LlmAgent
# 定义子智能体
greeter = LlmAgent(
name="greeter",
model="gemini-2.5-flash",
instruction="Greet users warmly and introduce the system."
)
task_executor = LlmAgent(
name="task_executor",
model="gemini-2.5-flash",
instruction="Execute user tasks efficiently and provide detailed results."
)
# 创建协调智能体
coordinator = LlmAgent(
name="coordinator",
model="gemini-2.5-flash",
description="Coordinate between greeter and task executor agents.",
sub_agents=[
greeter,
task_executor
]
)多智能体系统支持分层协调和任务分工，实现复杂业务逻辑
ADK提供了完整的工具生态系统，支持企业级API集成
原生Google Cloud服务集成，提供企业级可靠性
from google.adk.tools.bigquery import BigQueryTool
# 配置BigQuery工具
bigquery_tool = BigQueryTool(
project_id="your-project-id",
dataset_id="your_dataset",
credentials_path="path/to/credentials.json"
)
# 在智能体中使用
analysis_agent = Agent(
name="data_analyst",
model="gemini-2.5-flash",
instruction="Analyze data and provide insights using BigQuery.",
tools=[bigquery_tool]
)BigQuery工具支持SQL查询、数据分析和结果可视化
from google.adk.tools.openapi_tool import OpenApiTool
# 从OpenAPI规范创建工具
weather_api_tool = OpenApiTool(
api_spec_url="https://api.openweathermap.org/data/2.5/swagger.json",
base_url="https://api.openweathermap.org/data/2.5"
)
# 自定义OpenAPI工具
custom_api_tool = OpenApiTool(
api_spec_file="path/to/api-spec.yaml",
authentication={
"type": "bearer_token",
"token": "your-access-token"
}
)OpenAPI工具支持自动工具生成，无需手动定义函数签名
MCP提供统一的工具接口标准，支持跨平台工具互操作
from google.adk.tools.mcp_tool import McpTool
# 配置MCP服务器连接
mcp_tool = McpTool(
server_url="ws://localhost:8080",
server_id="filesystem-server",
capabilities=["read", "write", "list"]
)
# 在智能体中使用
file_system_agent = Agent(
name="file_manager",
model="gemini-2.5-flash",
instruction="Manage files using MCP filesystem tools.",
tools=[mcp_tool]
)MCP工具支持远程工具调用，提供安全的工具访问机制
ADK提供了完整的自定义工具开发框架
from google.adk.tools.base_tool import BaseTool
from typing import Dict, Any
class SentimentAnalysisTool(BaseTool):
def __init__(self):
super().__init__(
name="sentiment_analysis",
description="Analyze sentiment of text input"
)
def _execute(self, text: str) -> Dict[str, Any]:
# 实现情感分析逻辑
sentiment = self._analyze_sentiment(text)
return {
"text": text,
"sentiment": sentiment["label"],
"confidence": sentiment["score"],
"details": sentiment["details"]
}
def _analyze_sentiment(self, text: str) -> Dict[str, Any]:
# 实际的情感分析实现
# 这里可以是调用外部API或本地模型
pass
# 使用自定义工具
sentiment_tool = SentimentAnalysisTool()
analysis_agent = Agent(
name="sentiment_analyzer",
model="gemini-2.5-flash",
tools=[sentiment_tool]
)自定义工具支持完整的类型检查、错误处理和验证机制
ADK的上下文管理系统将上下文视为源代码，提供高效的上下文管理
from google.adk.agents.context import Context
from google.adk.memory import MemoryManager
# 上下文配置
context_config = {
"max_tokens": 100000,
"compression_threshold": 0.8,
"summarization_enabled": True,
"memory_enabled": True
}
# 创建上下文管理器
context = Context(config=context_config)
# 会话管理
session = context.create_session(
user_id="user123",
session_id="session_456"
)
# 记忆系统
memory_manager = MemoryManager()
memory_manager.add_memory(
session_id="session_456",
content="User prefers concise answers",
metadata={"type": "preference", "timestamp": "2026-05-16"}
)上下文管理支持智能压缩和总结，确保长期对话的连贯性
会话管理系统支持复杂的交互模式和状态管理
from google.adk.sessions import SessionManager
from google.adk.agents import Agent
# 创建会话管理器
session_manager = SessionManager()
# 创建智能体
agent = Agent(
name="customer_service",
model="gemini-2.5-flash",
instruction="Provide excellent customer service support."
)
# 创建新会话
session = session_manager.create_session(
agent=agent,
user_id="customer_789",
session_config={
"max_turns": 50,
"timeout": 3600,
"memory_enabled": True
}
)
# 处理用户输入
user_input = "I need help with my order"
response = session.process_input(user_input)
# 会话恢复
recovered_session = session_manager.load_session(session_id="session_123")
recovered_response = recovered_session.process_input("Continue our conversation")会话管理支持状态持久化，可以中断和恢复对话
记忆系统支持智能体的长期学习和知识积累
from google.adk.memory import MemoryManager, MemoryEntry
# 创建记忆管理器
memory_manager = MemoryManager()
# 添加记忆条目
memory_manager.add_memory(
session_id="session_456",
user_id="user123",
content="Customer John Doe has premium account status",
metadata={
"type": "customer_info",
"importance": "high",
"timestamp": "2026-05-16T10:30:00Z"
}
)
# 搜索相关记忆
relevant_memories = memory_manager.search_memory(
query="John Doe premium support",
session_id="session_456",
limit=10
)
# 获取上下文相关记忆
context_memories = memory_manager.get_context_memory(
session_id="session_456",
max_tokens=2000
)记忆系统支持智能搜索和上下文相关记忆提取
执行引擎是ADK的核心，确保智能体可靠执行
from google.adk.engine import Engine
from google.adk.models import GeminiModelAdapter
# 配置执行引擎
engine_config = {
"model_adapter": GeminiModelAdapter(),
"max_retries": 3,
"timeout": 30,
"temperature": 0.7,
"max_tokens": 8000
}
# 创建执行引擎
engine = Engine(config=engine_config)
# 执行智能体
result = engine.execute_agent(
agent=agent,
input="What's the weather today?",
session_id="session_456",
context=execution_context
)
# 处理执行结果
if result.success:
print(f"Response: {result.response}")
print(f"Tools used: {result.tools_used}")
print(f"Execution time: {result.execution_time}s")
else:
print(f"Error: {result.error}")
print(f"Retry attempts: {result.retry_count}")执行引擎提供了完整的执行控制和错误处理机制
ADK支持复杂的多智能体工作流编排
from google.adk.agents import ParallelAgent, Agent
# 创建并行智能体
research_agent = Agent(
name="researcher",
model="gemini-2.5-flash",
instruction="Research the topic and provide comprehensive analysis.",
tools=[google_search]
)
analysis_agent = Agent(
name="analyst",
model="gemini-2.5-flash",
instruction="Analyze the research findings and provide insights.",
tools=[analysis_tool]
)
# 创建并行智能体系统
parallel_agent = ParallelAgent(
name="parallel_team",
agents=[
research_agent,
analysis_agent
],
aggregation_strategy="merge"
)
# 执行并行任务
parallel_result = parallel_agent.execute("Analyze the impact of AI on healthcare")
print(f"Research: {parallel_result.research}")
print(f"Analysis: {parallel_result.analysis}")并行智能体可以同时处理不同方面的任务，然后聚合结果
from google.adk.agents import SequentialAgent, Agent
# 创建智能体链
planning_agent = Agent(
name="planner",
model="gemini-2.5-flash",
instruction="Create a detailed plan for the project."
)
execution_agent = Agent(
name="executor",
model="gemini-2.5-flash",
instruction="Execute the plan step by step and provide progress updates."
)
review_agent = Agent(
name="reviewer",
model="gemini-2.5-flash",
instruction="Review the execution results and provide quality assessment."
)
# 创建顺序智能体
sequential_agent = SequentialAgent(
name="project_manager",
agents=[
planning_agent,
execution_agent,
review_agent
]
)
# 执行顺序任务
project_result = sequential_agent.execute(
"Create and execute a marketing campaign for product launch"
)
print(f"Plan: {project_result.plan}")
print(f"Execution: {project_result.execution}")
print(f"Review: {project_result.review}")顺序智能体确保任务按预定顺序执行，适合流程化工作流
ADK提供了完善的错误处理和恢复机制
from google.adk.errors import ErrorHandler, RetryConfig
# 配置重试策略
retry_config = RetryConfig(
max_retries=3,
backoff_factor=2,
retry_exceptions=["TimeoutError", "APIError"],
exclude_exceptions=["ValidationError"]
)
# 配置错误处理器
error_handler = ErrorHandler(
retry_config=retry_config,
log_errors=True,
notify_on_failure=True,
recovery_strategies=["retry", "fallback", "escalate"]
)
# 使用错误处理器
try:
result = agent.execute("Complex task")
except Exception as e:
handled_result = error_handler.handle_error(
error=e,
context={"task": "Complex task", "agent": agent.name}
)
if handled_result.recovered:
print(f"Recovered result: {handled_result.result}")
else:
print(f"Failed to recover: {handled_result.error}")错误处理器支持多种恢复策略，确保系统可靠性
ADK内置了完整的监控和遥测系统
from google.adk.telemetry import TelemetryConfig, TelemetryClient
# 配置监控
telemetry_config = TelemetryConfig(
enable_metrics=True,
enable_tracing=True,
log_level="INFO",
metrics_endpoint="https://monitoring.googleapis.com"
)
# 创建监控客户端
telemetry_client = TelemetryClient(config=telemetry_config)
# 监控智能体执行
with telemetry_client.trace_span("agent_execution"):
metrics = telemetry_client.measure_execution(
agent_name="research_assistant",
input="Research topic",
start_time="2026-05-16T10:00:00Z"
)
try:
result = agent.execute("Research topic")
metrics.success = True
metrics.output_tokens = len(result.response.split())
except Exception as e:
metrics.success = False
metrics.error = str(e)
telemetry_client.record_metrics(metrics)监控系统支持分布式追踪和性能指标收集
ADK提供了完整的智能体评估系统
from google.adk.evaluation import Evaluator, EvaluationDataset
# 创建评估数据集
eval_dataset = EvaluationDataset(
name="customer_service_benchmark",
test_cases=[
{
"input": "I want to cancel my order",
"expected_response_type": "cancellation_process",
"expected_keywords": ["cancel", "refund", "confirmation"]
},
{
"input": "What's my order status?",
"expected_response_type": "status_inquiry",
"expected_actions": ["check_order", "provide_status"]
}
]
)
# 创建评估器
evaluator = Evaluator(
metrics=["accuracy", "relevance", "completeness", "helpfulness"],
dataset=eval_dataset,
comparison_agents=["baseline_model", "improved_model"]
)
# 执行评估
evaluation_results = evaluator.evaluate_agent(agent)
print(f"Accuracy: {evaluation_results.accuracy}")
print(f"Relevance: {evaluation_results.relevance}")
print(f"Overall Score: {evaluation_results.overall_score}")评估系统支持多种评估指标和自动化测试
ADK支持多种部署选项，适应不同的业务需求
from google.adk.deployment import DockerDeployment
# 配置Docker部署
docker_config = {
"image": "gcr.io/your-project/adk-agent:latest",
"port": 8080,
"env_vars": {
"GOOGLE_APPLICATION_CREDENTIALS": "/app/credentials.json",
"MODEL_NAME": "gemini-2.5-flash"
},
"health_check": {
"path": "/health",
"interval": 30
}
}
# 创建Docker部署
docker_deployment = DockerDeployment(config=docker_config)
# 部署智能体
deployment_result = docker_deployment.deploy(agent)
print(f"Container ID: {deployment_result.container_id}")
print(f"Port: {deployment_result.port}")
print(f"Health Check URL: {deployment_result.health_check_url}")Docker部署提供了轻量级和可移植的部署方案
from google.adk.deployment import CloudRunDeployment
# 配置Cloud Run部署
cloud_run_config = {
"project_id": "your-project",
"service_name": "adk-agent-service",
"region": "us-central1",
"memory": "2Gi",
"cpu": "1000m",
"max_instances": 10,
"env_vars": {
"GOOGLE_APPLICATION_CREDENTIALS": "/var/secrets/credentials.json"
},
"labels": {
"env": "production",
"version": "1.0"
}
}
# 创建Cloud Run部署
cloud_run_deployment = CloudRunDeployment(config=cloud_run_config)
# 部署到Cloud Run
deployment_result = cloud_run_deployment.deploy(agent)
print(f"Service URL: {deployment_result.service_url}")
print(f"Service ID: {deployment_result.service_id}")
print(f"Traffic Split: {deployment_result.traffic_split}")Cloud Run提供了自动扩展的无服务器部署方案
from google.adk.deployment import KubernetesDeployment
# 配置Kubernetes部署
k8s_config = {
"namespace": "production",
"replicas": 3,
"image": "gcr.io/your-project/adk-agent:1.0",
"resources": {
"requests": {"cpu": "500m", "memory": "1Gi"},
"limits": {"cpu": "1000m", "memory": "2Gi"}
},
"env_vars": {
"GOOGLE_APPLICATION_CREDENTIALS": "/etc/secret/credentials.json"
},
"liveness_probe": {
"path": "/health",
"initial_delay": 30,
"period": 10
},
"readiness_probe": {
"path": "/ready",
"initial_delay": 5,
"period": 5
}
}
# 创建Kubernetes部署
k8s_deployment = KubernetesDeployment(config=k8s_config)
# 部署到Kubernetes
deployment_result = k8s_deployment.deploy(agent)
print(f"Deployment Name: {deployment_result.deployment_name}")
print(f"Service URL: {deployment_result.service_url}")
print(f"Replicas: {deployment_result.replicas}")Kubernetes部署提供了企业级的容器编排能力
Vertex AI提供企业级的AI模型部署和扩展能力
ADK提供了完整的生产环境部署和运维指南
ADK内置了完整的安全机制，保障企业级应用安全
from google.adk.security import SecurityConfig, AuthenticationConfig
# 配置安全设置
security_config = SecurityConfig(
authentication=AuthenticationConfig(
type="oauth2",
provider="google-cloud",
scopes=["https://www.googleapis.com/auth/cloud-platform"]
),
authorization={
"roles": {
"admin": ["read", "write", "delete"],
"user": ["read", "write"],
"viewer": ["read"]
}
},
encryption={
"data_encryption": "aes-256",
"transport_encryption": "tls-1.3"
},
audit_logging=True
)
# 应用安全配置
agent.configure_security(security_config)安全配置支持多种认证和授权机制
ADK提供了多种性能优化策略
from google.adk.optimization import OptimizationConfig, CacheConfig
# 配置缓存
cache_config = CacheConfig(
enabled=True,
max_size=1000,
ttl=3600, # 1 hour
storage="redis"
)
# 配置优化
optimization_config = OptimizationConfig(
cache_config=cache_config,
concurrency=4,
batch_size=10,
timeout=30,
retry_policy={
"max_retries": 3,
"backoff_factor": 2
}
)
# 应用优化配置
agent.configure_optimization(optimization_config)性能优化支持缓存、并发和批量处理
ADK提供了完整的开发工具链
开发UI提供了直观的智能体开发体验
from google.adk.ui import DevelopmentUI
# 创建开发UI
dev_ui = DevelopmentUI(
agent=agent,
port=8080,
debug=True
)
# 启动开发UI
dev_ui.start()
# 在浏览器中访问 http://localhost:8080
# 可以进行智能体配置、测试和调试
# 停止开发UI
dev_ui.stop()开发UI提供了完整的智能体开发和调试环境
ADK提供了完整的测试框架支持
from google.adk.testing import TestSuite, TestCase
# 创建测试用例
test_case = TestCase(
name="customer_service_test",
input="I want to cancel my order",
expected_response_type="confirmation",
expected_actions=["cancel_order", "send_notification"],
timeout=30
)
# 创建测试套件
test_suite = TestSuite(
name="customer_service_suite",
test_cases=[
test_case,
TestCase(
name="order_status_test",
input="What's my order status?",
expected_response_type="status_info"
)
]
)
# 运行测试
test_results = test_suite.run(agent=agent)
print(f"Test Results: {test_results.passed}/{test_results.total}")
print(f"Success Rate: {test_results.success_rate}%")测试框架支持多种测试类型和自动化测试
ADK支持完整的版本控制管理
ADK已经在多个领域得到成功应用
ADK构建的智能客服系统可以处理复杂的客户需求
from google.adk import Agent
from google.adk.tools import knowledge_base_tool
# 创建客服智能体
customer_service_agent = Agent(
name="customer_service_agent",
model="gemini-2.5-flash",
instruction="""
You are an expert customer service representative.
Help customers with their inquiries, provide support,
and resolve issues efficiently.
Steps:
1. Understand the customer's issue
2. Check knowledge base for solutions
3. Provide clear instructions
4. Follow up if needed
""",
tools=[
knowledge_base_tool,
order_status_tool,
cancellation_tool
]
)客服智能体集成了多种工具，提供全面的客户支持
ADK支持复杂的数据分析和商业智能应用
ADK可以生成高质量的内容，支持多种创作场景
ADK已经在多个大型企业中得到成功应用
ADK提供了丰富的集成选项，支持现有工具和系统
from google.adk.tools.langchain_tool import LangChainTool
# 集成LangChain工具
langchain_tool = LangChainTool(
tool_name="document_qa",
tool_config={
"retriever": "vector_store",
"llm": "gemini-2.5-flash"
}
)
# 创建文档分析智能体
document_agent = Agent(
name="document_analyzer",
model="gemini-2.5-flash",
instruction="Analyze documents and provide insights using LangChain.",
tools=[langchain_tool]
)ADK与LangChain无缝集成，支持现有工具生态
ADK与CrewAI结合，实现复杂的多智能体协作
from google.adk.agents import Agent
from crewai import Task
# 创建协作智能体
researcher = Agent(
name="researcher",
model="gemini-2.5-flash",
instruction="Research the topic thoroughly."
)
analyst = Agent(
name="analyst",
model="gemini-2.5-flash",
instruction="Analyze research findings and provide insights."
)
writer = Agent(
name="writer",
model="gemini-2.5-flash",
instruction="Create comprehensive documentation based on research and analysis."
)
# 创建协作流程
research_task = Task(
description="Research the impact of AI on healthcare",
agent=researcher
)
analysis_task = Task(
description="Analyze research findings",
agent=analyst,
context=[research_task]
)
writing_task = Task(
description="Create comprehensive report",
agent=writer,
context=[research_task, analysis_task]
)多智能体协作可以实现复杂的知识处理任务
ADK正在不断发展和完善，为企业提供更好的AI开发工具
ADK提供了完整的快速开始指南，帮助用户快速上手
# 安装ADK
# pip install google-adk
from google.adk import Agent
# 创建简单智能体
hello_agent = Agent(
name="hello_assistant",
model="gemini-2.5-flash",
instruction="Greet users and provide basic information."
)
# 使用智能体
response = hello_agent.execute("Hello, can you help me?")
print(response)
# 输出：
# "Hello! I'd be happy to help you. What would you like assistance with today?"ADK提供了简单的API，让开发者快速上手智能体开发
ADK提供了丰富的学习资源，帮助用户快速掌握
Google ADK是构建企业级AI智能体的理想选择
感谢阅读！
访问 https://atcfu.com/ai-articles/google-adk/ 回顾本文