Google 的 Agent Development Kit（ADK）：指南与演示项目

项目概览：基于 ADK 与 A2A 的 AI 旅行规划器

该项目将构建一个旅行规划器，功能包括：

• 接收目的地、日期与预算作为输入。
• 调用三个独立智能体：
• flight_agent：推荐航班选项。
• stay_agent：在预算内寻找酒店。
• activities_agent：推荐有趣活动。
• 随后使用中央 host_agent 对所有请求进行编排。
• 最后使用 Streamlit UI 进行用户交互。

所有智能体均作为独立的 FastAPI 服务器托管，并暴露 /run 端点。通信通过共享的、兼容 A2A 的客户端完成。

注意：为简化起见，本项目完全在本地运行，但您也可以轻松将各智能体和 UI 部署到 Render、Railway 或 Google Cloud Run 等云平台，以实现可扩展访问。

步骤 1：先决条件

先安装以下库：

pip install google-adk litellm fastapi uvicorn httpx pydantic openai streamlit技术支持

然后设置您的 OpenAI API 密钥——也可以选择其他模型提供商。要了解如何设置 OpenAI 的 API 密钥，建议参考这篇 GPT-4o API 入门教程。

export OPENAI_API_KEY="your_key_here"技术支持

步骤 2：共享模式与工具

在构建智能体之前，我们需要为它们定义一种彼此交流的通用语言。在本项目中，通过以下方式实现：

• 用于输入的共享模式（通过 Pydantic 定义）
• 调用智能体的 REST 客户端工具
• 标准化所有智能体 /run 端点的 REST 服务器封装

这些文件放置在 shared/ 与 common/ 文件夹中，以保持代码模块化。我们逐一来看。

创建 shared/schemas.py 文件

我们使用 Pydantic 定义 TravelRequest 模式。这确保所有智能体在传入请求的结构上达成一致，其中包括目的地、出行日期和用户预算。

from pydantic import BaseModel
class TravelRequest(BaseModel):
    destination: str
    start_date: str
    end_date: str
    budget: float技术支持

该类可帮助：

• 让所有智能体的输入保持一致。
• 通过 FastAPI 自动进行校验。
• 简化代码复用。

创建 common/a2a_client.py 文件

这个轻量级异步工具允许任意智能体（尤其是 host）通过调用 /run 端点，使用 A2A 协议来触发另一个智能体。

import httpx
async def call_agent(url, payload):
    async with httpx.AsyncClient() as client:
        response = await client.post(url, json=payload, timeout=60.0)
        response.raise_for_status()
        return response.json()技术支持

该工具通过 httpx 异步向其他智能体的 /run 端点发送 POST 请求，返回解析后的 JSON 响应，并在请求失败时抛出错误。

我们将在 host 智能体中使用它来调用 flight_agent、stay_agent 和 activities_agent。

创建 common/a2a_server.py 文件

我们不为每个智能体单独编写 FastAPI 路由，而是通过 create_app(agent) 函数进行泛化，它负责：

• 在 /run 上提供智能体服务
• 接收旅行请求
• 返回结构化响应

from fastapi import FastAPI
import uvicorn
def create_app(agent):
    app = FastAPI()
    @app.post("/run")
    async def run(payload: dict):
        return await agent.execute(payload)
    return app技术支持

该工具创建了一个包含标准 /run 路由的 FastAPI 应用，将执行委托给传入的智能体对象。这确保了所有服务都使用结构化 JSON 输入，具备一致的智能体到智能体（A2A）接口。

这些共享组件共同提升了我们多智能体系统的可维护性与复用性，并契合 Google 的 A2A 理念：简洁、结构化的智能体间消息传递。

步骤 3：使用 ADK 与 A2A 构建多智能体系统

既然我们已准备好共享约定与工具，接下来开始构建各个智能体。为了将其真正打造成模块化的多智能体系统，我们将使用 Google 的 A2A 协议——基于 HTTP 的简单接口，使智能体能够以一致、可互操作的方式通信。

A2A（Agent-to-Agent）让本地 Python 函数或跨网络托管的智能体都能即插即用地协调。每个智能体都以通用模式暴露一个 /run 端点，并作为一个服务运行。

在本演示中，我们有四个智能体：

• host_agent：编排所有其他智能体
• flight_agent：查找合适航班
• stay_agent：推荐住宿
• activities_agent：推荐当地活动

所有智能体结构相似，包含 3 个文件与一个子文件夹：

agents/
├── host_agent/
│   │   ├── agent.py              # Optional if host logic is minimal
│   │   ├── task_manager.py       # Calls other agents and aggregates responses
│   │   ├── __main__.py           # Starts FastAPI app via common/a2a_server.py
│   │   └── .well-known/
│   │       └── agent.json        # A2A Agent Card metadata
├── flight_agent/
├── stay_agent/
└── activities_agent/技术支持

每个智能体都使用 google.adk.agents.Agent、LiteLlm 模型封装器，以及用于执行的 Runner。先在 activities_agent 文件夹内创建以下文件，再为 flight_agent 与 stay_agent 以同样方式重复。

创建 agent.py 文件

我们来定义 activities_agent 的逻辑，它将根据用户行程生成有吸引力的本地体验建议。 

步骤 1：导入

首先导入配置与运行智能体所需的基础模块。

from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
import json技术支持

该智能体使用 Google ADK 的组件，如 Agent、Runner 与 LiteLlm，并通过 InMemorySessionService 进行状态管理。types 库用于构建结构化提示词。

步骤 2：活动智能体

接下来，使用 ADK 的 Agent 类来实例化智能体本身。

activities_agent = Agent(
    name="activities_agent",
    model=LiteLlm("openai/gpt-4o"),
    description="Suggests interesting activities for the user at a destination.",
    instruction=(
        "Given a destination, dates, and budget, suggest 2-3 engaging tourist or cultural activities. "
        "For each activity, provide a name, a short description, price estimate, and duration in hours. "
        "Respond in plain English. Keep it concise and well-formatted."
    )
)技术支持

instruction 参数定义了引导 LLM 行为的系统提示。尽管此示例使用英文，也可以调整 instruction 以返回更易解析的结构化 JSON。

步骤 3：会话管理

接下来，为了跟踪用户交互，我们配置 Runner 以及会话信息。

session_service = InMemorySessionService()
runner = Runner(
    agent=activities_agent,
    app_name="activities_app",
    session_service=session_service
)
USER_ID = "user_activities"
SESSION_ID = "session_activities"技术支持

Runner 负责管理特定应用会话中的智能体执行，InMemorySessionService 将上下文存储在内存中。随后我们定义用户与会话 ID。在生产中，这些可以是动态或与用户绑定的。这样可确保在向 LLM 智能体发送任何提示之前，ADK 中存在一个全新的会话。

步骤 4：执行智能体逻辑

execute() 函数处理传入请求、构建提示、调用模型并解析输出。

async def execute(request):
    session_service.create_session(
        app_name="activities_app",
        user_id=USER_ID,
        session_id=SESSION_ID
    )
    prompt = (
        f"User is flying to {request['destination']} from {request['start_date']} to {request['end_date']}, "
        f"with a budget of {request['budget']}. Suggest 2-3 activities, each with name, description, price estimate, and duration. "
        f"Respond in JSON format using the key 'activities' with a list of activity objects."
    )
    message = types.Content(role="user", parts=[types.Part(text=prompt)])
    async for event in runner.run_async(user_id=USER_ID, session_id=SESSION_ID, new_message=message):
        if event.is_final_response():
            response_text = event.content.parts[0].text
            try:
                parsed = json.loads(response_text)
                if "activities" in parsed and isinstance(parsed["activities"], list):
                    return {"activities": parsed["activities"]}
                else:
                    print("'activities' key missing or not a list in response JSON")
                    return {"activities": response_text}  # fallback to raw text
            except json.JSONDecodeError as e:
                print("JSON parsing failed:", e)
                print("Response content:", response_text)
                return {"activities": response_text}  # fallback to raw text技术支持

execute() 函数会根据传入的目的地、日期与预算动态构建提示。其核心流程如下：

• 提示指示模型返回包含活动列表的结构化 JSON 对象。
• 构建 Content 对象并传给 ADK Runner，Runner 使用流式生成器异步等待模型的最终响应。
• 接收最终响应后，智能体提取原始文本输出并尝试将其解析为 JSON。
• 若解析成功且存在预期的 activities 键，则返回结构化数据。
• 若键缺失或格式不正确，则回退为返回原始文本，以便 UI 仍可用。
• 这种双轨处理方式可在 LLM 返回纯文本而非结构化 JSON 时优雅降级。

该策略提高了鲁棒性与用户体验，尤其当模型输出受温度或提示理解略有波动时。

创建 task_manager.py 文件

在 agent.py 中定义好 execute() 逻辑后，我们通过 task_manager.py 将其接入符合 ADK 的服务器设置。

from .agent import execute
async def run(payload):
    return await execute(payload)技术支持

该文件是对先前 execute() 函数的轻薄封装。它让 run() 方法可被外部模块（尤其是 __main__.py 中的服务器脚本）调用。

创建 __main__.py 文件

__main__.py 文件在 8003 端口启动 FastAPI 服务器，并在 /run 端点提供服务。

from common.a2a_server import create_app
from .task_manager import run
app = create_app(agent=type("Agent", (), {"execute": run}))
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, port=8003)技术支持

具体如下：

• create_app()（来自 common/a2a_server.py）将我们的智能体包装为标准的、兼容 A2A 的 FastAPI 接口。
• 我们动态构造了一个包含 execute() 方法的对象，以便 ADK 正确调用 run()。
• 这种分离让我们在复用核心智能体逻辑的同时，保持无状态的 API 接口。

创建 .well-known/agent.json 文件

我们使用该 JSON 文件，根据 A2A（Agent-to-Agent）协议描述智能体的标识与用途。

{
    "name": "activity_agent",
    "description": "Agent providing activity details."
  }技术支持

注意：.well-known/agent.json 在本项目中不被智能体直接使用，但它符合 A2A 规范，有助于智能体发现、内省，并与 LangGraph、CrewAI 或 Google 的智能体注册表等编排器保持未来兼容性。

flight_agent 与 stay_agent 也采用类似逻辑。

步骤 4：用 host_agent 进行协调

host_agent 在演示中充当中央规划者。host_agent 体现了多智能体系统中的控制器模式：它将决策与执行分离，使下游智能体各司其职，同时集中协调逻辑。这既简化了测试与扩展，也与分布式系统中的微服务架构相呼应。

它使用各智能体暴露的 /run API，将相同的 payload 发送给三个智能体并合并结果。让我们在 host_agent 文件夹中添加以下文件。

创建 agent.py 文件

先进行基础导入。

from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types技术支持

该导入块引入了使用 Google ADK 定义与运行基于 LLM 的智能体所需的核心构件：Agent 类、轻量级 LLM 封装器、用于处理执行的 Runner，以及内存会话管理。

host_agent = Agent(
    name="host_agent",
    model=LiteLlm("openai/gpt-4o"),
    description="Coordinates travel planning by calling flight, stay, and activity agents.",
    instruction="You are the host agent responsible for orchestrating trip planning tasks. "
                "You call external agents to gather flights, stays, and activities, then return a final result."
)
session_service = InMemorySessionService()
runner = Runner(
    agent=host_agent,
    app_name="host_app",
    session_service=session_service
)
USER_ID = "user_host"
SESSION_ID = "session_host"技术支持

上述代码定义了顶层 ADK 智能体，负责协调整个行程规划。尽管在本实现中没有从 LLM 内部调用子智能体，系统提示仍为后续扩展（例如由 LLM 处理工具调用与元推理）埋下伏笔。

async def execute(request):
    # Ensure session exists
    session_service.create_session(
        app_name="host_app",
        user_id=USER_ID,
        session_id=SESSION_ID
    )
    prompt = (
        f"Plan a trip to {request['destination']} from {request['start_date']} to {request['end_date']} "
        f"within a total budget of {request['budget']}. Call the flights, stays, and activities agents for results."
    )
    message = types.Content(role="user", parts=[types.Part(text=prompt)])
    async for event in runner.run_async(user_id=USER_ID, session_id=SESSION_ID, new_message=message):
        if event.is_final_response():
            return {"summary": event.content.parts[0].text}技术支持

该 execute() 函数作为 host 智能体 LLM 的主要入口，完成：

• 初始化会话（如需记忆支持）
• 动态构建用户提示
• 通过 ADK 的 runner.run_async() 方法发送给模型
• 最后等待并提取最终响应

创建 task_manager.py 文件

任务管理器通过调用远程智能体并处理完整的旅行规划工作流来执行编排逻辑。在具体实现中，我们为每个子智能体定义服务 URL。这些端点遵循 A2A 的 /run 协议，并期望接收共享的 TravelRequest` JSON 模式。

from common.a2a_client import call_agent
FLIGHT_URL = "http://localhost:8001/run"
STAY_URL = "http://localhost:8002/run"
ACTIVITIES_URL = "http://localhost:8003/run"技术支持

接下来，定义 payload。

async def run(payload):
    #Print what the host agent is sending
    print("Incoming payload:", payload)
    flights = await call_agent(FLIGHT_URL, payload)
    stay = await call_agent(STAY_URL, payload)
    activities = await call_agent(ACTIVITIES_URL, payload)
    # Log outputs
    print("flights:", flights)
    print("stay:", stay)
    print("activities:", activities)
    # Ensure all are dicts before access
    flights = flights if isinstance(flights, dict) else {}
    stay = stay if isinstance(stay, dict) else {}
    activities = activities if isinstance(activities, dict) else {}
    return {
        "flights": flights.get("flights", "No flights returned."),
        "stay": stay.get("stays", "No stay options returned."),
        "activities": activities.get("activities", "No activities found.")
    }技术支持

该函数使用 call_agent() 辅助函数将 payload 分发到各下游服务，并在开发时记录输入输出，便于观测调试。真正的编排逻辑基本都在此处完成。

创建 __main__.py 文件

__main__.py 是包装 host 智能体的 FastAPI 服务器入口文件。 

from common.a2a_server import create_app
from .task_manager import run
app = create_app(agent=type("Agent", (), {"execute": run}))
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, port=8000)技术支持

主文件完成以下工作：

• 使用 common/a2a_server.py 的 create_app() 生成带有标准 /run 端点的 FastAPI 应用。
• 传入一个带有 execute() 方法的简单类对象，该方法内部委托给 task_manager.run()。
• 最后使用 uvicorn 在指定端口（通常为 8000）启动 FastAPI 服务器。

这样就让 host 智能体的接口与其他下游智能体保持一致，从而在系统内维持统一性。

创建 .well-known/agent.json 文件

该文件体现了经典的多智能体模式：中心节点分派并组合任务。 

{
  "name": "host_agent",
  "description": "Coordinates travel planning among specialized agents."
}技术支持

尽管可选，但如前所述，建议在所有智能体目录中都包含该文件。

步骤 5：用 Streamlit 构建 UI

最后，我们来构建一个简单应用，用户可输入偏好并获得结构化行程。先在根目录创建 travel_ui.py 文件，并添加以下代码。

import streamlit as st
import requests技术支持

我们导入 Streamlit 与 requests 等基础库以支持 UI。

st.set_page_config(page_title="ADK-Powered Travel Planner", page_icon="✈️")
st.title("🌍 ADK-Powered Travel Planner")
origin = st.text_input("Where are you flying from?", placeholder="e.g., New York")
destination = st.text_input("Destination", placeholder="e.g., Paris")
start_date = st.date_input("Start Date")
end_date = st.date_input("End Date")
budget = st.number_input("Budget (in USD)", min_value=100, step=50)
if st.button("Plan My Trip ✨"):
    if not all([origin, destination, start_date, end_date, budget]):
        st.warning("Please fill in all the details.")
    else:
        payload = {
            "origin": origin,
            "destination": destination,
            "start_date": str(start_date),
            "end_date": str(end_date),
            "budget": budget
        }
        response = requests.post("http://localhost:8000/run", json=payload)
        if response.ok:
            data = response.json()
            st.subheader("✈️ Flights")
            st.markdown(data["flights"])
            st.subheader("🏨 Stays")
            st.markdown(data["stay"])
            st.subheader("🗺️ Activities")
            st.markdown(data["activities"])
        else:
            st.error("Failed to fetch travel plan. Please try again.")技术支持

该 Streamlit 应用提供了与基于 ADK 的多智能体旅行规划器交互的友好 UI。上述代码涉及：

• 使用 text_input、date_input 与 number_input 收集出发地、目的地、日期与预算。
• 点击 “Plan My Trip” 后，先进行校验，确保不留空。
• 若通过校验，则构造 JSON payload 并向 host_agent 的 http://localhost:8000/run 发送 POST 请求。
• host_agent 会调用所有子智能体（航班、住宿、活动），聚合其响应并返回统一的旅行计划。
• 使用 st.markdown() 分别在航班、住宿与活动的小节下渲染响应。
• 若后端失败，则用 st.error() 显示回退错误信息。

现在在本地终端运行以下命令：

uvicorn agents.host_agent.__main__:app --port 8000 &
uvicorn agents.flight_agent.__main__:app --port 8001 &
uvicorn agents.stay_agent.__main__:app --port 8002 &      
uvicorn agents.activities_agent.__main__:app --port 8003 &
streamlit run travel_ui.py技术支持

当用户点击 “Plan My Trip” 时，host 智能体开始工作，激活各智能体，并在 UI 中展示结果：

您的整体文件结构大致如下：

ADK_demo/
├── agents/
│   ├── host_agent/
│   │   ├── agent.py              
│   │   ├── task_manager.py      
│   │   ├── __main__.py         
│   │   └── .well-known/
│   │       └── agent.json   
│   │
│   ├── flight_agent/
│   │   ├── agent.py        
│   │   ├── task_manager.py    
│   │   ├── __main__.py         
│   │   └── .well-known/
│   │       └── agent.json        
│   │
│   ├── stay_agent/
│   │   ├── agent.py              
│   │   ├── task_manager.py       
│   │   ├── __main__.py           
│   │   └── .well-known/
│   │       └── agent.json     
│   │
│   └── activities_agent/
│       ├── agent.py              
│       ├── task_manager.py       
│       ├── __main__.py           
│       └── .well-known/
│           └── agent.json       
│
├── common/
│   ├── a2a_client.py             # Utility to send requests to other agents
│   └── a2a_server.py             # Shared FastAPI A2A-compatible server template
│
├── shared/
│   └── schemas.py                # Shared Pydantic schema
│
├── streamlit_app.py             # Frontend UI for user input and response rendering
├── requirements.txt           
└── README.md                    技术支持

就是这样！我已将我们构建的全部内容整理在这个 GitHub 项目中。

结语

借助少量 FastAPI 应用与 ADK 智能体，我们构建了一个协作式旅行规划器，能够：

• 通过 A2A 协议通信
• 为航班、住宿与活动分别使用 LLM 智能体
• 在简洁的 Streamlit UI 中聚合并展示结果

即使所有智能体底层都使用相同模型，该系统在行为上仍是多智能体：智能体角色各异、职责隔离、通信结构化。

以下资源可帮助您入门：

• A2A GitHub
• 使用 ADK 构建天气机器人
• A2A 技术文档