从零开始使用ADK开发agent(1)-单agent开发

目前市场上有很多的agent 框架，如 autogen,langgraph,crewai,agno 等等，不同的框架对于agent 的实现有些差异，有的框架侧重点在流程编排如 langGraph，有的框架提供很多内置的工具如crewai，但总体而言，这些agent 框架的基本能力包含工具调用，agent 协同。

前段时间 google 提出了A2A 协议，意在统一这些框架直接的交互方式。本系列我准备从零开始学习一下agent 的开发。

框架我选择了google 的 adk，一来这个框架使用起来比较简单，二来这个框架由google开发维护，对于A2A 协议有着很好的支持。

ADK 介绍

ADK (Agent Development Kit), 是谷歌在2025年4月正式发布的一款开源智能体开发框架，旨在简化复杂Agent应用程序的整个端到端开发生命周期。该框架与谷歌自身产品中用于支持Agent的框架相同，现在可供各地的开发人员使用。

主要功能

1. 多智能体架构：支持构建由多个专业智能体组成的层次化应用，实现复杂的协调和委派。开发者可定义不同层级的智能体，每个专注特定任务，提高系统整体效率和可扩展性。
2. 丰富的工具生态系统：提供预构建工具（如搜索、代码执行）、自定义函数和第三方库集成，开发者能轻松扩展智能体能力，满足多样化需求。
3. 灵活的编排：内置多种工作流智能体（如SequentialAgent、ParallelAgent、LoopAgent），支持LLM驱动的动态路由，可灵活定义复杂工作流程，满足不同场景任务需求。
4. 集成开发工具：提供命令行界面（CLI）和开发者UI，支持运行智能体、检查执行步骤、调试交互和可视化智能体定义，帮助开发者快速开发、调试和优化智能体。
5. 原生流式支持：支持双向流式交互（文本和音频），与底层能力（如Gemini Developer API）无缝集成，使智能体能实时响应用户输入，提供更流畅交互体验。
6. 广泛的LLM支持：虽与谷歌的Gemini模型深度集成，但通过BaseLlm接口，也支持与各种LLM（如OpenAI、Anthropic、Meta、Mistral AI等）集成，为开发者提供更多选择和灵活性。

废话不多说，接下来从头开始跟着官方文档来一起学习如何使用ADK 吧。

初始化项目

这里我使用uv 来管理项目，当然也可以使用pip，差异不大。

uv init adk_project --python 3.12
cd adk_project
uv sync 
uv add google-adk

# 创建子项目
uv init adk_starter

我这里创建子项目的原因是想以后每个章节使用独立的子项目，这样每个子项目可以公用一套虚拟环境，关于uv 中使用 workspace 的教程可以参考我之前的文章，如果只是想跑demo，完全可以不用workspce，单纯的再根目录下运行。

此时的目录结构为

.
├── adk_starter
│   ├── main.py
│   ├── pyproject.toml
│   └── README.md
├── main.py
├── mult_agents
├── pyproject.toml
├── README.md
└── uv.lock

创建智能体

作为ADK系列的第一篇，我们准备实现一个可以在线查询天气的agent，这个也是众多教程中都快包浆了的工具，虽然很多教程都在使用，包括官方的文档中也是用的天气查询，但是作为学习的工具，我们只需要了解它的运行机制，工作原理就行，之后我们可以在工作中实现真实的agent。

官方文档使用的是fake 数据，只是模拟了天气查询，我这里使用高德地图的接口，真实的进行天气查询。

创建工具函数

[!NOTE] 注意 以下创建的文件名和类名，名字不能变

在 adk_starter 目录下，先创建 agent_tool,py 写入一下代码

# -*- coding: utf-8 -*-
"""
 @Time    : 2025-05-16 22:29
 @Author  : YangYanxing
 @File    : agent_tool.py
 @Description : agent 用到的tool 定义
"""

import datetime
import httpx
import os
from pydantic import BaseModel
from typing import List, Dict, Any

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", "")
    if not api_key:
        weather_data.status = "error"
        weather_data.message = "未配置高德地图API Key"
        return weather_data.model_dump()
    api_domain = 'https://restapi.amap.com/v3'
    url = f"{api_domain}/config/district?keywords={city}"f"&subdistrict=0&extensions=base&key={api_key}"
    headers = {"Content-Type": "application/json; charset=utf-8"}
    async with httpx.AsyncClient(headers=headers, verify=False) as client:
        response = await client.get(url)
        if response.status_code != 200:
            weather_data.status = "error"
            weather_data.message = "查询失败"
            return weather_data.model_dump()
        
        city_info = response.json()
        if city_info["info"] != "OK":
            weather_data.status = "error"
            weather_data.message = "获取城市信息查询失败"
            return weather_data.model_dump()
        CityCode = city_info['districts'][0]['adcode']
        weather_url = f"{api_domain}/weather/weatherInfo?city={CityCode}&extensions=all&key={api_key}"
        weatherInfo_response = await client.get(weather_url)
        if weatherInfo_response.status_code != 200:
            weather_data.message = "查询天气信息失败"
            weather_data.status = "error"
            return weather_data.model_dump()
        weatherInfo = weatherInfo_response.json()
        if weatherInfo['info'] != "OK":
            weather_data.message = "查询天气信息失败"
            weather_data.status = "error"
            return weather_data.model_dump()
        weatherInfo_data = weatherInfo_response.json()
        contents = []
        if len(weatherInfo_data.get('forecasts')) <= 0:
            weather_data.message = "没有获取到该城市的天气信息"
            weather_data.status = "error"
            return weather_data.model_dump()
        for item in weatherInfo_data['forecasts'][0]['casts']:
            content = {
                'date': item.get('date'),
                'week': item.get('week'),
                'dayweather': item.get('dayweather'),
                'daytemp_float': item.get('daytemp_float'),
                'daywind': item.get('daywind'),
                'nightweather': item.get('nightweather'),
                'nighttemp_float': item.get('nighttemp_float')
            }
            contents.append(content)
        weather_data.data = contents
        weather_data.status = "success"
        weather_data.message = "获取天气成功"
        return weather_data.model_dump()
    

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

这里定义了两个工具函数：

1. get_weather， 调用高德天气查询接口获取天气信息
2. get_current_time，返回当前的时间

具体的函数实现很简单，就不用讲解了，只要记住，这就是两个普通的函数。

每个函数在注释部分写了函数的描述，以及参数和返回值说明，这个非常重要！大模型之后分析解析参数是否准确依赖于这里的说明是否清晰明确。

创建智能体

在 adk_starter 目录下新建 agent.py 文件

from google.adk.agents import Agent

from adk_starter.agent_tool import get_weather, get_current_time
from google.adk.models.lite_llm import LiteLlm


MODEL_QWEN = "openai/qwen-plus"

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

这个文件名必须 叫 agent.py ，里面的Agent 实例也必须叫 root_agent，这个先这样写，因为之后在使用官方的测试工具进行测试的时候，调用的对象都是固定写死的。

Agent 类有很多初始化参数，这里只使用了几个重要参数

name: 智能体的名字，这个理论上叫什么都学，甚至用个随机字符串都行，但是为了可读性，还是起个有意义的名字把。

model: 这个参数非常重要，是大模型agent 的核心，这个参数可以是 str 或者 BaseLlm，如果是str, 如 gemini-1.5-flash， 那么会用到 google 供应商，这个也是ADK 深度整合的，但是由于一些众所周知的原因是访问不了的，所以这里我使用了阿里的qwen 模型，需要在阿里云申请key， 然后使用 litellm 进行代理， 需要安装 litellm 和 openai

uv add litellm openai

litellm 初始化时model 参数为 供应商/模型名 的形式, 如这里的 openai/qwen-plus。

description: 这个参数是描述智能体能做什么，需要根据智能体的能力来定义。

instruction: 这个参数是用法说明，可以指导agent 在什么情况下调用哪个工具，具体的工作流程是什么样的。

tools: 工具列表，定义这个agent 可以调用哪些工具，这里填写上面的两个函数名称。

[!important] 注意 description 和 instruction 非常重要，决定这智能体是否可以正常工作，以及工作的效果的好坏

关于agent 类型

上面示例中使用的是 google.adk.agents.Agent, 这个agent 是 LlmAgent 的别名，是一个基于大模型的agent，除了这个agent，adk 中还定义了很多实用的 agent ：

1. SequentialAgent： 顺序执行子agent 的 agent
2. LangGraphAgent： 基于langgraph 实现的agent
3. LoopAgent： 循环执行的 agent
4. ParallelAgent: 并行执行的 agent

我们还可以自定义agent ,这个在之后用到的时候再详细记录下，目前我们只需要使用 LlmAgent 即可。

编辑 __init__.py 文件

编辑 adk_starter 下的 __init__.py, 如果没有这个文件，需要新创建一个，这一步也是为了之后使用官方测试工具而准备的，后期我们在开发自己的agent 平台时是不需要的

from . import agent

创建环境变量文件 .env

在 adk_starter 目录下创建 .env 文件， 将agent 运行时需要的环境变量写进去，上面使用的是openai 兼容格式的供应商，所以这里还需要配置以下 OPENAI_API_KEY， 高德天气查询函数里需要 GAODE_KEY

GAODE_KEY=xxxxx
OPENAI_API_KEY=阿里key
OPENAI_API_BASE=https://dashscope.aliyuncs.com/compatible-mode/v1

经过上面的步骤以后，一个最简单的agent 就完成了，我们来测试一下。

测试

有多种测试方法，官方sdk中提供三种，我们都来实验一下。

webui

返回到 adk_starter 上层目录，也就是 adk_project 根目录， 运行

adk web

webui 的方式我认为是最友好的，用户可以在浏览器上与agent 进行交互

打开 http://127.0.0.1:8000

这个页面主要分为两部分，左边为agent 运行时的状态信息，右边是一个对话页面。让我们来问几个问题。

这个对话框会将agent 运行过程中调用了哪个工具以及执行结果详细的展示出来，并且将最终的结果返回。

点击工具调用的按钮，左边就会展示出工具的名称和入参，这个很方便的进行调试

我问了一个问题

现在几点了？ 我明天要去上海，那边的天气怎么样？

可以看出，这个agent 分别调用了两个工具函数。

api_server

api_server 与 webui 的方式差不多，只是它不提供web 页面，而是通过http 接口调用

启动命令

adk api_server

启动成功以后，也是使用 8000 端口对外提供服务, 使用curl 命令或者postman 发送请求

curl -X POST http://0.0.0.0:8000/run \
-H "Content-Type: application/json" \
-d '{
"app_name": "adk_starter",
"user_id": "u_123",
"session_id": "aaaa",
"new_message": {
    "role": "user",
    "parts": [{
    "text": "北京的天气"
    }]
}
}'

session_id 是记录对话上下文的，同一个用户通过不同的 seseion_id 来管理对话，在使用webui 进行测试是，页面顶端会显示当前的session_id，这个看起来像是uuid生成的，如果刷新页面或者点击 New Session，这个id 就会变，将开启新的一轮对话。

cli

最后是在shell 中运行命令，在shell 中进行交互

adk run adk_starter

我觉得从交互体验上来讲 webui>api_server>cli， 你们也可以选择自己喜欢的方式。

手搓 runner

上面有提到，agent.py 和 agent.py 里的 root_agent 都是固定的，需要写死，这个只是为了配合官方的测试工具，我们也可以手动的运行agent，这样就不用受限于文件名称和类对象名称。

修改 adk_project 下的main.py，注意这个是adk_project 根目录下的 main.py， 不是 adk_starter 目录下的main.py。

from adk_starter.agent import root_agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
import anyio
from dotenv import load_dotenv

load_dotenv("adk_starter/.env")

async def call_agent_async(query: str):
    session_service = InMemorySessionService()

    APP_NAME = "weather_and_time"
    USER_ID = "user_1"
    SESSION_ID = "session_001" 

    # # Create the specific session where the conversation will happen
    session = session_service.create_session(
        app_name=APP_NAME,
        user_id=USER_ID,
        session_id=SESSION_ID
    )

    runner = Runner(
        agent=root_agent, # The agent we want to run
        app_name=APP_NAME,   # Associates runs with our app
        session_service=session_service # Uses our session manager
    )
    content = types.Content(role='user', parts=[types.Part(text=query)])
    final_response_text = "对不起，agent 不能回答您的问题"
    async for event in runner.run_async(user_id=USER_ID, session_id=SESSION_ID, new_message=content):
      if event.is_final_response():
          if event.content and event.content.parts:
             final_response_text = event.content.parts[0].text
          elif event.actions and event.actions.escalate: # Handle potential errors/escalations
             final_response_text = f"Agent escalated: {event.error_message or 'No specific message.'}"
          # Add more checks here if needed (e.g., specific error codes)
          break # Stop processing events once the final response is found

    print(f"<<< Agent Response: {final_response_text}")

if __name__ == "__main__":
    anyio.run(call_agent_async, "北京明天的天气")

agent 在执行过程中，需要使用Runner 对象，这个对象可以帮助保存对话内容，执行工具调用等，这个Runner 对象非常重要。接下来看下执行过程分析：

1. session_service = InMemorySessionService() 初始化一个内存类型的SessionService，这类型的SessionService 是将与用户交互的上下文保存到内存中，如果重启了就没了，后面我们尝试使用外部存储如mysql。
2. runner = Runner() 初始化一个Runner 对象，之后主要就是使用这个对象进行调用
3. runner.run_async() 异步进行agent 调用。

未来我们会大量使用手搓Runner 调用。

注意

上面的代码有几点注意

1. 工具函数的定义从一开始就需要考虑异步，尽量使用 async def 定义，无论是官方的测试webui 还是未来我们自己用fastAPI写后台服务，都要求写异步函数
2. Runner 类本身有 run 和 run_async 方法，run 方法用于同步调用，官方也不推荐使用，还是使用 run_async 方法吧

之前的工具使用async def 定义的异步函数

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

这个函数没有什么IO操作，所以也看不出来什么问题，但是如果改成下面的代码:

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

首先把 async def 改为的 def， 并且在函数体添加了阻塞的 time.sleep(30) 来模拟阻塞式 IO 操作，如果 agent 执行到了这段代码，则会将整个系统卡住，无论是web 还是api 访问，都将阻塞住。这个问题之前我也写个多次了。

与MCP的区别

上面的代码看上去和MCP 差不多，MCP 也是定义工具，绑定服务运行，有大模型来决定使用哪个工具，这里的agent 和MCP 有哪些区别呢？

首先，从这个简单的demo 来看，确实实现的功能和MCP 差不多的，但是这个是由于这个agent 实在是太简单了，没有体现出agent 的优势，上面有提到，定义agent 的description 和 instruction 非常重要，可以通过instruction 来指导agent 行动，之后我们会实现多agent示例，agent 完成的是一个task， agent 在完成task 时可能还会调用别的agent，完成的工作单元是 task，而mcp 完成的是一个个工具调用，完成的工作单元是tool。

这个在之后介绍复杂的多agent 时会有明显的体现。

总结

本文通过 adk 实现了最最基础的单 agent 多工具的调用，主要学习adk 创建agent 以及agent 如何运行，有了简单初步的认识

创建agent 的流程为

1. 定义工具执行函数，注意要使用异步的，在函数注释中写清楚函数的作用，以及参数的含义，这样大模型才能分析出是否要调用该工具
2. 定义agent，本文使用 LlmAgent 类
3. 运行测试，官方提供三种测试工具，主要以 webui 测试为主，交互比较好

后面会使用多agent 进行自主规划， 不同agent 执行模式，以及A2A协议的实践。

代码已经放到github, 有兴趣可以查看一下

https://github.com/kevinkelin/a2a_demo