人工智能领域正在经历一场代理革命（Agent Revolution）。随着大语言模型(LLM)能力的不断提升，基于LLM构建能够自主规划、决策和执行的智能代理（Agent）成为了行业热点。但开发功能完备、可靠稳定的AI代理系统并不容易，需要解决工具调用、多代理协同、评估测试等诸多挑战。Google近期推出的开源项目"Agent Development Kit"（简称ADK）正是为解决这些难题而生，它提供了一套完整的Python工具包，帮助开发者快速构建、评估和部署复杂的AI代理系统。本文将深入解析ADK框架的架构设计和核心特性，帮助读者快速入门这一强大框架。
ADK（Agent Development Kit）是Google开发的开源框架，它为AI代理开发提供了一套代码优先（code-first）的Python工具库。与其他Agent框架相比，ADK的设计理念是让Agent开发更像传统软件开发，提供灵活的模块化组件，支持从简单任务到复杂工作流的各类代理架构。
asyncdefquery_weather(url: str)-> dict[str, Any] |None: """Make a request to the NWS API with proper error handling.""" headers = { "X-QW-Api-Key": API_KEY, } asyncwithhttpx.AsyncClient()asclient: try: response =awaitclient.get(url, headers=headers, timeout=30.0) response.raise_for_status() returnresponse.json() exceptException: returnNone
asyncdefget_forecast(latitude: float, longitude: float)-> str: """Get weather forecast for a location.
Args: latitude: Latitude of the location longitude: Longitude of the location """ # First get the forecast grid endpoint weather_url =f"{API_BASE}/weather/7d?location={longitude},{latitude}" weather_data =awaitquery_weather(weather_url)
forecasts = [] forperiodinweather_data['daily']: # Only show next 5 periods forecast =f""" {period['fxDate']}{period['textDay']}: Temperature:{period['tempMin']}~{period['tempMax']}°C Wind:{period['windSpeedDay']}{period['windDirDay']} """ forecasts.append(forecast)
return"\n---\n".join(forecasts)
4. 构建汇率查询Agent
defget_exchange_rate( currency_from: str ="USD", currency_to: str ="EUR", currency_date: str ="latest", ): """Use this to get current exchange rate.
Args: currency_from: The currency to convert from (e.g., "USD"). currency_to: The currency to convert to (e.g., "EUR"). currency_date: The date for the exchange rate or "latest". Defaults to "latest".
Returns: A dictionary containing the exchange rate data, or an error message if the request fails. """ try: response = httpx.get( f"https://api.frankfurter.app/{currency_date}", params={"from": currency_from,"to": currency_to}, ) response.raise_for_status()
data = response.json() if"rates"notindata: return{"error":"Invalid API response format."} returndata excepthttpx.HTTPErrorase: return{"error":f"API request failed:{e}"} exceptValueError: return{"error":"Invalid JSON response from API."}
SYSTEM_INSTRUCTION = ( "You are a specialized assistant for currency conversions. " "Your sole purpose is to use the 'get_exchange_rate' tool to answer questions about currency exchange rates. " "If the user asks about anything other than currency conversion or exchange rates, " "politely state that you cannot help with that topic and can only assist with currency-related queries. " "Do not attempt to answer unrelated questions or use tools for other purposes." "Set response status to input_required if the user needs to provide more information." "Set response status to error if there is an error while processing the request." "Set response status to completed if the request is complete." )
rate_agent = LlmAgent( model=llm, name="rate_agent", instruction=SYSTEM_INSTRUCTION, description="Can respond to user when they ask about currency exchange rates.", tools=[get_exchange_rate] )
5. 构建代码助手Agent
code_agent = LlmAgent( model=llm, name="code_agent", description="Can respond to user when they ask about code.", instruction="You can write code and fix bugs.", )
6. 构建多代理系统
root_agent = LlmAgent( model=llm, name="coordinator", instruction="Route user requests:Use rate_agent to answer questions about currency exchange rates, use code_agent to write code and fix bugs, use weather_agent to get weather forecast for a location.", description="Main agent for routing user requests to the appropriate sub-agents.", sub_agents=[rate_agent, code_agent, weather_agent], )
APP_NAME ="multi_app" USER_ID ="user_1" SESSION_ID ="session_001"# Using a fixed ID for simplicity
asyncdefcall_agent_async(query: str, runner, user_id, session_id): """Sends a query to the agent and prints the final response.""" print(f"\n>>> User Query:{query}")
# Prepare the user's message in ADK format final_response_text ="Agent did not produce a final response."# Default
# Key Concept: run_async executes the agent logic and yields Events. # We iterate through events to find the final answer. asyncforeventinrunner.run_async( user_id=user_id, session_id=session_id, new_message=content ): # You can uncomment the line below to see *all* events during execution # print(f" [Event] Author: {event.author}, Type: {type(event).__name__}, Final: {event.is_final_response()}, Content: {event.content}")
# Key Concept: is_final_response() marks the concluding message for the turn. ifevent.is_final_response(): ifevent.contentandevent.content.parts: # Assuming text response in the first part final_response_text = event.content.parts[0].text elif( event.actionsandevent.actions.escalate ): # Handle potential errors/escalations final_response_text = ( f"Agent escalated:{event.error_messageor'No specific message.'}" ) # Add more checks here if needed (e.g., specific error codes) break# Stop processing events once the final response is found
