从零开始使用ADK开发agent智能体(2)-多agent开发

上一篇文章我们构建了一个天气查询的agent，这个agent 功能比较单一，只能完成天气的查询，一个功能丰富的智能体系统，应该可以做很多事情，比如一个学术研究系统，会有信息搜索，文章总结，而信息搜索可能会使用到google 搜索，学术搜索，api 调用，RAG 检索等工具，文章总结可能包含大纲生成，图片生成，文章总结等等工具，且不同的agent 可能用到不同的模型，协调器的模型需要强大的规划推理模型，画图的agnet需要文生图的模型，文本总结可能一般的模型就可以胜任，这也是处于成本的考虑，不同的agent 只关注自身能力擅长的工作，对于复杂的系统，可能会由数量非常多的拥有不同工具能力的agent组成。

本文让我们来跟着官方文档尝试构建一个多agent 系统，我们在之前的天气查询agent 基础之上添加以下功能

不同的agent 使用不同的模型
添加子agent 功能
实现代理之间的智能委派

创建项目，在上一个demo 基础之上使用 uv 创建子项目

1	`uv init adk_multi_agent`

此时的目录结构为

.
├── README.md
├── adk_multi_agent
│   ├── README.md
│   ├── main.py
│   └── pyproject.toml
├── adk_starter 
│   ├── README.md
│   ├── __init__.py
│   ├── agent.py
│   ├── agent_tool.py
│   ├── main.py
│   └── pyproject.toml
├── main.py
├── pyproject.toml
└── uv.lock

先将上一节中的 adk_stater 目录下的所有文件拷贝到 adk_multi_agent 目录下，之后在这里进行修改。

定义agent¶

我们来定义三个agent

天气查询功能的 agent，这个就复用上一节的agent
用于查询当前时间的agent
用于提供问候和告别的agent

修改agent_tool.py

import httpx
import os
from pydantic import BaseModel
from typing import List, Dict, Any
from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm

class WeatherInfo(BaseModel):
    """天气信息"""
    status: str = "success"
    message: str = ""
    data: List[Dict[str, Any]]  = []

async def get_weather(city: str)-> dict:
    """获取天气
    Args:
        city (str): 要查询天气的城市名称, 例如：北京.

    Returns:
        dict: 该城市的天气信息 或者  错误信息.

    """
    weather_data = WeatherInfo()
    api_key = os.getenv("GAODE_KEY", "")
    # 和上一节代码一样
    return weather_data

MODEL_QWEN = "openai/qwen-plus"

weather_agent = Agent(
    name="weather_agent",
    model=LiteLlm(model=MODEL_QWEN),
    description=(
        "获取天气的agent."
    ),
    instruction=(
        "你是一个非常有用的助手，可以获取天气信息。"
    ),
    tools=[get_weather]
)

定义时间获取agent，新建一个agent.time.py

#!/usr/bin/env python
# -*- encoding: utf-8 -*-
'''
@File    :   agent_time.py
@Time    :   2025/05/18 18:26:28
@Author  :   YangYanxing 
@Desc    :   定义 获取时间的agent
'''
import datetime
from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm

async def get_current_time() -> str:
    """
    获取当前时间
    Returns:
        str: 当前的时间.
    """
    return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")


MODEL_QWEN = "openai/qwen-plus"

date_agent = Agent(
    name="time_agent",
    model=LiteLlm(model=MODEL_QWEN),
    description=(
        "获取当前时间的agent."
    ),
    instruction=(
        "你是一个非常有用的助手，可以获取当前的时间信息。"
    ),
    tools=[get_current_time]
)

定义一个问候语agent，新建agent_hello.py

from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm

def say_hello(name: str = "there") -> str:
    """提供以个友好的问候语，可选的 name 参数

    Args:
        name (str, optional): 需要问候的人名称. 默认值 "there".

    Returns:
        str: 一个友好的问候语.
    """
    return f"Hello, {name}!"

def say_goodbye() -> str:
    """提供一个简单的告别信息，以结束对话。"""
    return "Goodbye! Have a great day."

MODEL_QWEN = "openai/qwen-plus"

greetting_agent = Agent(
    name="greetting_agent",
    model=LiteLlm(model=MODEL_QWEN),
    description=(
        "问候和告别的agent."
    ),
    instruction=(
        "你是一个问候和告别的agent，你的工作只有两个任务，"
        "1. 你可以使用 say_hello 函数给用户一个友好的问候，并返回一个友好的问候语,如果用户提供了name，则需要传递到函数中"
        "2. 你可以给用say_goodbye 函数给用户一个友好的告别，并返回一个友好的告别信息。"
    ),
    tools=[say_hello, say_goodbye]
)

这三个agent 我都用的 qwen-plus 模型，可以根据实际情况进行调整。

定义好三个agent 以后，需要再定义一个 root_agent，当然这个也是为了使用adk 自带的测试页面使用。

修改agent.py 文件

from google.adk.agents import Agent
# 引入三个agent 
from adk_multi_agent.agent_tool import weather_agent
from adk_multi_agent.agent_time import date_agent
from adk_multi_agent.agent_hello import greetting_agent

from google.adk.models.lite_llm import LiteLlm


MODEL_QWEN = "openai/qwen-plus"

root_agent = Agent(
    name="weather_time_hello_agent",
    model=LiteLlm(model=MODEL_QWEN),
    description=(
        "获取天气和时间和大招呼的agent."
    ),
    instruction=(
        "你是一个非常有用的助手，可以获取天气和当前的时间信息。"
        "当用户向你问好是，"
    ),
    sub_agents=[weather_agent, date_agent, greetting_agent]
)

创建 .env 文件，配置好需要的环境变量，最终的目录结构为

.
├── README.md
├── adk_multi_agent
│   ├── README.md
│   ├── __init__.py
│   ├── agent.py
│   ├── agent_hello.py
│   ├── agent_time.py
│   ├── agent_tool.py
│   ├── main.py
│   └── pyproject.toml
├── adk_starter
│   ├── README.md
│   ├── __init__.py
│   ├── agent.py
│   ├── agent_tool.py
│   ├── main.py
│   └── pyproject.toml
├── main.py
├── pyproject.toml
└── uv.lock

测试agent¶

使用 adk web 启动webui 进行测试

你好，我叫 yyx, 现在几点了？帮我查一下北京明天的天气

可以看到，root_agent 先将任务分给了time_agent，获取到时间信息以后先将时间信息返回给用户，并提示天气查询交给别的agent 负责。

过了一会儿，负责天气的agent 将结果返回

上面的agent 工作的还算正常，但是并没有调用 greetting_agent 和我大招呼，于是我尝试先和他打招呼，看看能否调用 greetting_agent

从结果来看，并没有调用greetting_agent, 而是使用大模型自身的回复。

这并不是我想要的，尝试修改一下 root_agent 的description和instruction

root_agent = Agent(
    name="weather_time_hello_agent",
    model=LiteLlm(model=MODEL_QWEN),
    description=(
        "你是一个协调员代理，根据用户的对话提问，并调用其他代理进行任务。"
    ),
    instruction=(
        "你是一个非常有用的助手，可以获取天气和当前的时间信息。"
        "如果用户向你打招呼或者回复了再见，你可以调用greeting_agent 进行回复。"
    ),
    sub_agents=[weather_agent, date_agent, greetting_agent]
)

通过修改描述信息，和 instruction，指导agent 如何行动，这里给了它一个明显的指导，

如果用户向你打招呼或者回复了再见，你可以调用greeting_agent 进行回复

这样root_agent 就知道如何更好的调用子agent 了

提示¶

上面我定义的agent 方式和官方文档的有些不一样，官方文档是在 weather_agent 的定义处添加了 sub_agents, 也就是将weather_agent 作为总的协调者，由它来调用各种子agent。我觉得单独定义agent 然后再定义一个总的协调者会相对更加清晰一些，不过总体而言差异不大。总的思路是定义好agent，然后再定义协调者进行规划调用。

代码已经上传到 github

https://github.com/kevinkelin/a2a_demo

总结¶

本文演示了如何使用sdk 构建一个多agent 系统

先定义每个子agent，定义相应的工具，每个agent 使用不同的模型
创建root_agent，作为协调者，调用各种子agent