Google ADK Agent:从单个 Gemini Agent 到 Cloud Run 部署
将Google ADK从某个Gemini代理部署到生产环境:相关工具、作为工具的代理、顺序流程、本地测试,以及向Cloud Run和Vertex AI Agent Engine的部署。
大型语言模型改变了人们使用软件的方式,但下一步的发展重点不再是更智能的对话功能,而是那些能够自主推理、调用工具、协同工作并完成整个工作流程的系统。这正是智能体人工智能的领域。传统应用只需发送一个提示并等待一个回复,而智能体则能自行决定下一步该做什么:它可能会调用API、获取信息、将子任务交给专家处理,或返回结构化的结果。
谷歌的智能体开发工具包(ADK)是学习这一技术的实用框架。它允许你在Gemini模型上构建智能体,将Python函数作为工具使用,协调多个智能体的工作,并将它们部署到谷歌云上。以下内容将从单个智能体逐步讲解到云端部署。
从提示到智能体
传统的大型语言模型应用采用的是提示与响应的模式:你提出问题,模型给出答案,仅此而已。而在代理式架构中,模型位于更大的执行系统之中,它能够设定目标、选择工具、与API交互以及与其他代理协作。需要记住的是,代理不仅仅是一个提示,它是由语言模型驱动的软件组件。
ADK应用以root_agent为核心,最简单的版本结构如下:
from google.adk.agents import Agent
root_agent = Agent(
name="assistant",
model="gemini-2.5-flash",
instruction="You are a helpful assistant"
)
即便如此,它也是一个具备名称、行为、模型以及运行能力的实际智能体。它可以进行聊天,但在获得工具之前无法触碰其外部的一切。
工具
定义智能体型人工智能的关键在于工具的使用。没有工具时,模型只能生成文本;而有了工具,智能体就能执行真正的逻辑运算。在ADK中,一旦将普通的Python函数传递到智能体的配置中,它就会变成一种工具。计算器智能体便展示了这一过程:
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]
)
在这里,模型理解了请求,但它并不亲自进行计算,而是将具体的运算任务交给相应的函数。大语言模型负责推理,而工具则负责执行操作。这种分工是成熟企业架构的基础。
实际系统很少依赖单一工具。一个助手可能会调用API、转换数据、查询数据库、获取文档、检查规则并总结结果。下图所示的多工具代理则集成了算术运算、单位转换和天气查询功能:
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,
],
)
该智能体现在充当协调者,它会根据请求即时选择合适的工具,用户无需指定使用哪一个。
作为其他智能体工具的智能体
ADK允许一个智能体作为另一个智能体的工具,从而促进专业化发展。无需编写试图涵盖所有内容的庞大提示语,只需创建具有特定功能的小型智能体即可。由多个智能体组合而成的旅行规划器便是个很好的例子:
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)
]
)
核心代理充当协调者,没有固定的工作流程,而每个子代理都是专家,可以按任意顺序被调用。这样的系统更易于维护、扩展和理解,这也符合许多企业级人工智能设计的趋势。
结构化流程
并非所有流程都需要完全自主运行,因为业务系统通常需要结构化。常见的流程是规划、执行,然后汇报。ADK通过顺序式代理来实现这一点,例如由规划器、执行器和汇报器组成的链条:
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的本地开发流程十分快捷。首先,通过一个.env文件来指定Google Cloud项目、位置以及Vertex AI的使用配置:
GOOGLE_CLOUD_PROJECT=PROJECT_ID
GOOGLE_CLOUD_LOCATION=us-central1
GOOGLE_GENAI_USE_VERTEXAI=True
接着,一条命令会启动一个本地网页界面,以便交互式地测试这些智能体:
adk web
如此快速的循环非常适合进行实验。
部署到 Cloud Run
仅靠本地测试远远不够,而 Cloud Run 是最简便的生产环境部署方式之一。对于上述第一个代理,一个简短的 Dockerfile 就足够了:
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 从源代码部署:
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
之后,查询服务URL:
gcloud run services describe simple-agent \
--region us-central1 \
--format='value(status.url)'
打开该URL即可直接从Cloud Run获取ADK接口,从而得到一个可部署的AI服务,而非仅用于演示的笔记本程序。
使用FastAPI暴露API
有时HTTP API比浏览器界面更适用,而ADK与FastAPI搭配得非常好。用于封装天气工具代理的最简版main.py如下所示:
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
}
部署它需要一个略有不同的 Dockerfile:
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
ADK还可以部署到Google的Agent Engine上,这是一种专为企业级代理程序执行而构建的托管运行时环境。首先需要启用相应的API:
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
这会在 Google Cloud 中生成一个受管理的推理引擎资源,尽管背后涉及诸多复杂流程,但整体操作相当简洁。
为何这很重要
智能体 AI 是一种基于语言模型构建软件的新方式,而不仅仅是更优的提示词。智能体会进行推理、使用工具、协调工作流程,并与外部系统交互,而非仅生成孤立的回复。
ADK 是一个很好的学习工具,因为它能逐步引入这些概念。你可以从一个 Gemini 智能体开始,逐步添加工具、连接多个智能体,最终实现包含 RAG、MCP 集成、评估流程以及 GCP 部署的企业级工作流程。从计算器示例到多智能体企业级架构,其核心原则始终如一:推理、任务分配、协调与执行。软件正在从单纯回答问题转向在真实应用中运行。