Multi-Agents

Agent Team

这篇文章里，我们实现一个 Manager-Specialist 架构的 Agent Team.

简单来说，就是有一个负责分发任务的 Manager 和若干个负责处理事务的 Specialist

代码实现起来其实特别简单。我们先写两个 agent，一个专门用于打招呼，一个专门用于送别：

Greeting Agent, Farewell Agent

实现和 single agent 差不太多

import os, asyncio
import warnings, logging
from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner
from google.genai import types
from tools import greetings, farewell

greeting_agent = Agent(
    model=LiteLlm(
        model="openrouter/meta-llama/llama-4-scout:free",
        api_key=os.getenv("OpenRouter_API"),
        api_base=os.getenv("OpenRouter_URL"),
    ),
    name="greeting_agent",
    instruction="You are the Greeting Agent. Your ONLY task is to provide a friendly greeting to the user. "
    "Use the 'greetings' tool to generate the greeting. "
    "If the user provides their name, make sure to pass it to the tool. "
    "Do not engage in any other conversation or tasks.",
    description="Handles simple greetings and hellos using the 'greetings' tool.",  # Crucial for delegation
    tools=[greetings],
)

farewell_agent = Agent(
    model=LiteLlm(
        model="openrouter/meta-llama/llama-4-scout:free",
        api_key=os.getenv("OpenRouter_API"),
        api_base=os.getenv("OpenRouter_URL"),
    ),
    name="farewell_agent",
    instruction="You are the Farewell Agent. Your ONLY task is to provide a friendly farewell to the user. "
    "Use the 'farewell' tool to generate the farewell message. "
    "If the user provides their name, make sure to pass it to the tool. "
    "Do not engage in any other conversation or tasks.",
    description="Handles simple farewells and goodbyes using the 'farewell' tool.",  # Crucial for delegation
    tools=[farewell],
)

我们现在来定义 weather agent team. 唯一有区别的在于，我们在 Agent class 初始化的时候，传入 subagent=[] 参数来告诉 agent 下一步该调用哪个 agent. 至于如何调用、在什么条件下调用，完全由 LLM 决策、由 Google ADK 解析决策结果并执行.

weather_agent_team = Agent(
    name="weather_agent_team",
    model=LiteLlm(
        model="openrouter/meta-llama/llama-4-scout:free",
        api_key=os.getenv("OpenRouter_API"),
        api_base=os.getenv("OpenRouter_URL"),
    ),
    description="The main coordinator agent. Handles weather requests and delegates greetings/farewells to specialists.",
    instruction="You are the main Weather Agent coordinating a team. Your primary responsibility is to provide weather information. "
    "Use the 'get_weather' tool ONLY for specific weather requests (e.g., 'weather in London'). "
    "You have specialized sub-agents: "
    "1. 'greeting_agent': Handles simple greetings like 'Hi', 'Hello'. Delegate to it for these. "
    "2. 'farewell_agent': Handles simple farewells like 'Bye', 'See you'. Delegate to it for these. "
    "Analyze the user's query. If it's a greeting, delegate to 'greeting_agent'. If it's a farewell, delegate to 'farewell_agent'. "
    "If it's a weather request, handle it yourself using 'get_weather'. "
    "For anything else, respond appropriately or state you cannot handle it.",
    tools=[get_weather],  # Root agent still needs the weather tool for its core task
    sub_agents=[greeting_agent, farewell_agent],
)

然后我们随便写点测试函数试一下：

Testing Function

修改第一个 call_agent_async() 的 query= 参数，加上/去掉名字，LLM 也会对应输出名字. 这就是 InMemorySession 的作用~

async def run_team_async():
    USER_ID = "user_123"
    SESSION_ID = "session_abc"
    APP_NAME = "weather_agent_team_app"

    session_service = InMemorySessionService()
    session = await session_service.create_session(
        app_name=APP_NAME,
        user_id=USER_ID,
        session_id=SESSION_ID,
    )

    runner_agent_team = Runner(
        app_name=APP_NAME,
        agent=weather_agent_team,
        session_service=session_service,
    )

    await call_agent_async(
        # query="Hello there! My name is Faslir",
        query="Hello there!",
        runner=runner_agent_team,
        user_id=USER_ID,
        session_id=SESSION_ID,
    )
    await call_agent_async(
        query="What is the weather in New York?",
        runner=runner_agent_team,
        user_id=USER_ID,
        session_id=SESSION_ID,
    )
    await call_agent_async(
        query="Thanks, bye!",
        runner=runner_agent_team,
        user_id=USER_ID,
        session_id=SESSION_ID,
    )


if __name__ == "__main__":
    asyncio.run(run_team_async())