1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
| # Copyright (c) 2025 Marco Fago
#
# This code is licensed under the MIT License.
# See the LICENSE file in the repository for the full license text.
import uuid
from typing import Dict, Any, Optional
from google.adk.agents import Agent
from google.adk.runners import InMemoryRunner
from google.adk.tools import FunctionTool
from google.genai import types
from google.adk.events import Event
# Colab 代码链接：https://colab.research.google.com/drive/10wxRlPyDJ70pPEtWWA3Aa3tjvfobrB3m
# 安装依赖
# pip install google-genai google-adk
# export GOOGLE_API_KEY=...
# --- Define Tool Functions ---
# These functions simulate the actions of the specialist agents.
# --- 定义工具函数 ---
# 这些函数模拟了专业智能体的具体行动。
def booking_handler(request: str) -> str:
"""
Handles booking requests for flights and hotels.
Args:
request: The user's request for a booking.
Returns:
A confirmation message that the booking was handled.
"""
print("-------------------------- Booking Handler Called ----------------------------")
return f"Booking action for '{request}' has been simulated."
def info_handler(request: str) -> str:
"""
Handles general information requests.
Args:
request: The user's question.
Returns:
A message indicating the information request was handled.
"""
print("-------------------------- Info Handler Called ----------------------------")
return f"Information request for '{request}'. Result: Simulated information retrieval."
def unclear_handler(request: str) -> str:
"""Handles requests that couldn't be delegated."""
return f"Coordinator could not delegate request: '{request}'. Please clarify."
# --- Create Tools from Functions ---
# --- 从函数创建工具 ---
booking_tool = FunctionTool(booking_handler)
info_tool = FunctionTool(info_handler)
# Define specialized sub-agents equipped with their respective tools
# --- 定义配备了各自工具的专业子智能体 ---
booking_agent = Agent(
name="Booker",
model="gemini-2.0-flash",
description="A specialized agent that handles all flight and hotel booking requests by calling the booking tool.",
tools=[booking_tool]
)
info_agent = Agent(
name="Info",
model="gemini-2.0-flash",
description="A specialized agent that provides general information and answers user questions by calling the info tool.",
tools=[info_tool]
)
# Define the parent agent with explicit delegation instructions
# --- 定义带有明确委派指令的父智能体 ---
coordinator = Agent(
name="Coordinator",
model="gemini-2.0-flash",
instruction=(
"You are the main coordinator. Your only task is to analyze incoming user requests "
"and delegate them to the appropriate specialist agent. Do not try to answer the user directly.\n"
"- For any requests related to booking flights or hotels, delegate to the 'Booker' agent.\n"
"- For all other general information questions, delegate to the 'Info' agent."
),
description="A coordinator that routes user requests to the correct specialist agent.",
# The presence of sub_agents enables LLM-driven delegation (Auto-Flow) by default.
# 定义了 sub_agents 默认就会启用由大语言模型驱动的委派。
sub_agents=[booking_agent, info_agent]
)
# --- Execution Logic ---
# --- 执行逻辑 ---
def run_coordinator(runner: InMemoryRunner, request: str):
"""Runs the coordinator agent with a given request and delegates."""
"""用给定的请求运行协调员智能体并进行委派。"""
print(f"\n--- Running Coordinator with request: '{request}' ---")
final_result = ""
try:
user_id = "user_123"
session_id = str(uuid.uuid4())
runner.session_service.create_session(
app_name=runner.app_name, user_id=user_id, session_id=session_id
)
for event in runner.run(
user_id=user_id,
session_id=session_id,
new_message=types.Content(
role='user',
parts=[types.Part(text=request)]
),
):
if event.is_final_response() and event.content:
# Try to get text directly from event.content to avoid iterating parts
if hasattr(event.content, 'text') and event.content.text:
final_result = event.content.text
elif event.content.parts:
# Fallback: Iterate through parts and extract text (might trigger warning)
text_parts = [part.text for part in event.content.parts if part.text]
final_result = "".join(text_parts)
# Assuming the loop should break after the final response
break
print(f"Coordinator Final Response: {final_result}")
return final_result
except Exception as e:
print(f"An error occurred while processing your request: {e}")
return f"An error occurred while processing your request: {e}"
def main():
"""Main function to run the ADK example."""
"""运行 ADK 示例的主函数。"""
print("--- Google ADK Routing Example (ADK Auto-Flow Style) ---")
print("Note: This requires Google ADK installed and authenticated.")
runner = InMemoryRunner(coordinator)
# Example Usage
# 使用示例
result_a = run_coordinator(runner, "Book me a hotel in Paris.")
print(f"Final Output A: {result_a}")
result_b = run_coordinator(runner, "What is the highest mountain in the world?")
print(f"Final Output B: {result_b}")
result_c = run_coordinator(runner, "Tell me a random fact.") # Should go to Info
print(f"Final Output C: {result_c}")
result_d = run_coordinator(runner, "Find flights to Tokyo next month.") # Should go to Booker
print(f"Final Output D: {result_d}")
if __name__ == "__main__":
main()
|