实用笔记：使用 Google ADK 入门智能体 AI
《实用笔记：使用 Google ADK 掌握代理型 AI 入门》的操作指南：为采用该架构的团队提供的契约、校验规则以及可直接插入的代码模块。
本指南将逐步构建从原始材料到可运行系统的完整流程，内容为《使用 Google ADK 掌握智能体 AI 入门》。重点在于可操作的步骤、明确的检查点，以及可直接放入代码仓库的代码，无需猜测其用途。 在概览阶段，应在修改代码之前明确输入参数、各步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行相应步骤，而无需推测隐藏状态。 可将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功标准，杜绝默许的半完成状态。
从聊天机器人到智能体的转变
在完成“从聊天机器人到智能体”的阶段时，首先写下相关契约：所需的输入参数、成功标志以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次收取相同的LLM调用费用。
理解智能体AI背后的核心理念
在“理解核心理念”阶段工作时，首先写下相关约定：所需输入、成功标志以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将配置信息置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放，这样操作人员无需查看整个系统结构即可进行审计。 在耗时较高的步骤之后设置检查点。当操作人员重新执行后续节点时，恢复流程不应再次调用相同的大型语言模型。
from google.adk.agents import Agent
root_agent = Agent(
name="assistant",
model="gemini-2.5-flash",
instruction="You are a helpful assistant"
)
通过工具赋予智能体真正的能力
在“赋予智能体实际功能”阶段，首先需写下契约内容：所需的输入参数、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 同时记录正常流程与异常恢复路径。重试机制、人工审核环节以及死信处理都是产品功能的一部分，而非后续需要补充的内容。 需为每次调用记录工具名称、参数哈希值、延迟时间以及最终结果。没有这些记录，调试智能体循环将会耗费大量时间。 在“赋予智能体实际功能”阶段，首先需写下契约内容：所需的输入参数、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 应将此阶段视为输入与验证后输出之间的契约。为相关产物命名，明确成功判定标准，杜绝无声的半完成状态。
from google.adk.agents import Agent
def calculator(a: float, b: float, operation: str) -> float:
if operation == "add":
return a + b
if operation == "subtract":
return a - b
if operation == "multiply":
return a * b
if operation == "divide":
if b == 0:
raise Exception("Cannot divide by zero")
return a / b
raise Exception("Unsupported operation")
root_agent = Agent(
name="assistant",
model="gemini-2.5-flash",
instruction=(
"You are a helpful assistant with calculator capabilities. "
"Use the calculator tool for arithmetic. "
"Supported operations are add, subtract, multiply, divide."
),
tools=[calculator]
)
构建多工具智能体
将“建筑多工具代理”阶段视为可度量的对象来处理时，其效果最佳。在扩大范围之前，先记录一份理想的执行结果、一个失败案例以及回滚说明。在功能结果旁同时记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。
from google.adk.agents import Agent
def calculator(a: float, b: float, operation: str) -> float:
if operation == "add":
return a + b
if operation == "subtract":
return a - b
if operation == "multiply":
return a * b
if operation == "divide":
if b == 0:
raise ValueError("Cannot divide by zero.")
return a / b
raise ValueError("Unsupported operation.")
def convert_units(value: float, from_unit: str, to_unit: str) -> float:
from_unit = from_unit.lower()
to_unit = to_unit.lower()
if from_unit == "km" and to_unit == "miles":
return value * 0.621371
if from_unit == "miles" and to_unit == "km":
return value / 0.621371
if from_unit == "celsius" and to_unit == "fahrenheit":
return value * 9 / 5 + 32
if from_unit == "fahrenheit" and to_unit == "celsius":
return (value - 32) * 5 / 9
raise ValueError("Unsupported unit conversion.")
def get_weather_mock(city: str) -> dict:
weather_data = {
"bucharest": {
"temperature_celsius": 23,
"condition": "sunny",
"wind_speed_kmh": 10,
},
"london": {
"temperature_celsius": 16,
"condition": "rain",
"wind_speed_kmh": 18,
},
}
key = city.lower()
if key not in weather_data:
return {
"city": city,
"error": "Weather data not available."
}
return {
"city": city,
**weather_data[key],
}
root_agent = Agent(
name="multi_tool_agent",
model="gemini-2.5-flash",
instruction=(
"You are a practical assistant. "
"Use the available tools when the user asks for calculations, "
"unit conversions, or weather information."
),
tools=[
calculator,
convert_units,
get_weather_mock,
],
)
多代理系统：利用其他代理的代理
将“多智能体系统智能体使用阶段”视为可度量的对象时，其效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 将配置置于应用程序代码之外。环境文件、密钥存储和功能标志应集中存放于一个位置，这样操作人员无需查看整个结构即可进行审计。 保持图结构的扁平化与类型化。嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，且在中断后会导致无法继续执行。
from google.adk.agents import LlmAgent
from google.adk.tools import google_search
from google.adk.tools import google_maps_grounding
from google.adk.tools.agent_tool import AgentTool
routing_agent = LlmAgent(
name="routing_agent",
model="gemini-2.5-pro",
instruction="""
You are a routing agent.
Use google_maps_grounding to estimate routes and travel times.
""",
tools=[google_maps_grounding],
)
discovery_agent = LlmAgent(
name="discovery_agent",
model="gemini-2.5-pro",
instruction="""
You are a travel discovery agent.
Use Google Search to find interesting places.
""",
tools=[google_search]
)
composer_agent = LlmAgent(
name="composer_agent",
model="gemini-2.5-pro",
instruction="""
Write a friendly travel itinerary based on the collected information.
""",
tools=[]
)
root_agent = LlmAgent(
name="travel_agent",
model="gemini-2.5-pro",
instruction="""
You are a travel assistant.
Coordinate discovery, routing, and itinerary composition.
""",
tools=[
AgentTool(discovery_agent),
AgentTool(routing_agent),
AgentTool(composer_agent)
]
)
顺序工作流与确定性编排
将“顺序工作流与确定性阶段”视为可度量的对象来处理时，其效果最佳。在扩大范围之前，需记录一份理想状态下的流程记录、一个故障案例以及回滚说明。 同时记录正常流程与恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的一部分，而非后续需要补充的内容。 保持图结构的状态简洁且类型明确。嵌套的数据块会掩盖哪个节点编写了哪个字段的信息，还会在中断后导致流程无法继续。 将“顺序工作流与确定性阶段”视为可度量的对象来处理时，其效果最佳。在扩大范围之前，需记录一份理想状态下的流程记录、一个故障案例以及回滚说明。 应将这一阶段视为输入与已验证输出之间的契约。为相关成果命名，明确成功标准，绝不允许出现无声无息的半完成状态。
from google.adk.agents import Agent, SequentialAgent
from google.adk.tools import AgentTool
planner_agent = Agent(
name="planner_agent",
model="gemini-2.5-flash",
instruction="""
Read the user request and create a short execution plan.
"""
)
executor_agent = Agent(
name="executor_agent",
model="gemini-2.5-flash",
instruction="""
Execute the plan and delegate specialist work.
""",
tools=[]
)
report_agent = Agent(
name="report_agent",
model="gemini-2.5-flash",
instruction="""
Produce the final report based on execution results.
"""
)
root_agent = SequentialAgent(
name="planner_executor_report_workflow",
sub_agents=[
planner_agent,
executor_agent,
report_agent,
],
)
在本地运行 ADK Agent
在“运行ADK代理”阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。除了功能结果外，还需记录执行时间以及令牌或查询成本。提前了解这些成本可以避免在从演示环境过渡到共享环境时出现意外费用。对于会消耗资金或修改生产数据的操作，必须经过人工审批。编译时的配置并不等同于业务功能的完整性。
GOOGLE_CLOUD_PROJECT=PROJECT_ID
GOOGLE_CLOUD_LOCATION=us-central1
GOOGLE_GENAI_USE_VERTEXAI=True
adk web
将代理部署到Google Cloud Run
在将代理部署到测试环境之前，需先确定输入参数、该步骤的负责人以及结束标准，然后再进行代码修改。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测其中的隐藏状态。 配置信息应与应用程序代码分开存放。环境文件、密钥存储以及功能开关应集中于一个位置，这样操作人员无需查看整个系统结构即可进行审核。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务功能的完整性。
FROM python:3.11-slim
WORKDIR /app
COPY . .
RUN pip install --no-cache-dir -r requirements.txt
CMD ["adk", "web", "--host", "0.0.0.0", "--port", "8080"]
gcloud run deploy simple-agent \
--source . \
--region us-central1 \
--allow-unauthenticated \
--set-env-vars GOOGLE_GENAI_USE_VERTEXAI=TRUE \
--set-env-vars GOOGLE_CLOUD_PROJECT=PROJECT_ID \
--set-env-vars GOOGLE_CLOUD_LOCATION=us-central1
gcloud run services describe simple-agent \
--region us-central1 \
--format='value(status.url)'
通过 FastAPI 暴露代理
在“通过某阶段暴露代理”这一环节中，修改代码之前需明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程与异常恢复路径。重试机制、人工审核环节以及错误处理都是产品功能的一部分，而非后续需要补充的内容。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接配置并不等同于业务功能的完整性。 在“通过某阶段暴露代理”这一环节中，修改代码之前需明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 应将这一阶段视为输入参数与经过验证的输出结果之间的契约。需为相关输出文件命名，明确成功判定标准，并杜绝无声的半完成状态。
import uuid
from fastapi import FastAPI
from pydantic import BaseModel
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
from agent import root_agent
app = FastAPI()
session_service = InMemorySessionService()
runner = Runner(
agent=root_agent,
app_name="weather_agent_service",
session_service=session_service,
)
class QueryRequest(BaseModel):
message: str
@app.post("/weather")
async def weather(request: QueryRequest):
user_id = "api_user"
session_id = str(uuid.uuid4())
await session_service.create_session(
app_name="weather_agent_service",
user_id=user_id,
session_id=session_id,
)
content = types.Content(
role="user",
parts=[
types.Part(text=request.message)
],
)
final_answer = ""
async for event in runner.run_async(
user_id=user_id,
session_id=session_id,
new_message=content,
):
if event.is_final_response():
final_answer = event.content.parts[0].text
return {
"response": final_answer
}
FROM python:3.11-slim
WORKDIR /app
COPY . .
RUN pip install --no-cache-dir -r requirements.txt
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8080"]
部署到 Agent Engine
在处理“部署到 Agent Engine”这一阶段时，首先需记录下相关要求：所需的输入参数、成功标志，以及部分失败时的处理方式。这样的清单能确保后续的代码修改不会出错。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次收取相同的 LLM 调用费用。
gcloud services enable \
aiplatform.googleapis.com \
storage.googleapis.com
export STAGING_BUCKET="gs://${PROJECT_ID}-agent-staging"
gsutil mb -l us-central1 $STAGING_BUCKET
adk deploy agent_engine \
--project=$PROJECT_ID \
--region=us-central1 \
--staging_bucket=$STAGING_BUCKET \
basic_agent
总结
在完成“最终思考”阶段时，首先写下相关契约：所需的输入参数、成功标志以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将配置信息与应用程序代码分开存放。环境文件、密钥存储以及功能开关应集中于一个位置，这样操作人员无需查看整个系统结构即可进行审计。 在耗时较高的步骤之后设置检查点。当操作人员重新执行后续节点时，恢复流程不应再次调用相同的大型语言模型。
操作检查清单
若将“操作检查清单”阶段视为可度量的工作面，其效果会更好。在扩大范围之前，先记录一份标准操作示例、一个故障案例以及回滚说明。 相比复杂的脚本，应优先使用小型且可测试的单元。当某个步骤出现故障时，故障原因应能明确指向某个具体功能模块，而非整个复杂的流程体系。
保持图结构的状态扁平且具有类型约束。嵌套的数据块会隐藏是哪个节点修改了哪个字段，还会在中断后导致流程无法继续。
只要预算允许，就在持续集成过程中使用测试用例而非真实的付费 API 来执行关键路径的冒烟测试。
将此阶段视为输入与经过验证的输出之间的契约。为相关产物命名，明确成功标准，绝不允许出现无声的半完成状态。
保持图结构的状态扁平且具有类型约束。嵌套的数据块会隐藏是哪个节点修改了哪个字段，还会在中断后导致流程无法继续。
在升级技术栈之前，先冻结版本，为关键路径生成标准化的操作记录，并确认回滚步骤。共享环境需要设置速率限制、进行租户身份验证，同时要明确负责密钥轮换的人员。与其追求华丽的临时演示，不如注重扎实的可靠性。
关于18b8374abe5a的批处理说明：不要将提供者密钥放入仓库中，设定每会话的令牌上限，并将转录内容存储在评估测试用例旁边，以便后续更换模型时仍能保持可比性。