智能體執行器 Runner，負責完成一次用戶需求的響應，是ADK中真正讓Agent運行起來的引擎，其核心功能和Agents SDK中的Runner類似，具體作用如下：
-
會話管理：自動讀取/寫入 SessionService，維護歷史信息。
-
Agent調用：調用指定的Agent完成推理和工具調用
-
輸入輸出流轉：把用戶輸入交給Agent，把Agent輸出返回。
-
流程控制：支持多輪對話、子Agent委托、工具調用等。
-
生命周期管理：處理每次對話流程的完整生命周期
Runner 是Agent的管理者，負責組織每次對話交互的完整生命周期。
Runner 類(Class)
class Runner:"""Runner 類用于運行智能體。Runner 類用于在會話中管理智能體的執行，負責處理消息、生成事件，并與各類服務進行交互。"""app_name: str"""The app name of the runner."""agent: BaseAgent"""The root agent to run."""artifact_service: Optional[BaseArtifactService] = None"""The artifact service for the runner."""session_service: BaseSessionService"""The session service for the runner."""memory_service: Optional[BaseMemoryService] = None"""The memory service for the runner."""def __init__(self,*,app_name: str,agent: BaseAgent,artifact_service: Optional[BaseArtifactService] = None,session_service: BaseSessionService,memory_service: Optional[BaseMemoryService] = None,):"""Initializes the Runner.Args:app_name: The application name of the runner.agent: The root agent to run.artifact_service: The artifact service for the runner.session_service: The session service for the runner.memory_service: The memory service for the runner."""self.app_name = app_nameself.agent = agentself.artifact_service = artifact_serviceself.session_service = session_serviceself.memory_service = memory_service參數解析如下：
| 參數名 | 參數類型 | 含義 | 說明 |
|---|---|---|---|
| app_name | str | 應用名稱 | 應用的名稱，用于標識不同的應用實例。 |
| agent | BaseAgent | 要運行的主Agent | 負責執行實際的任務和推理 |
| artifact_service | Optional[BaseArtifactService] | 文件存儲服務(可選) | 可選的文件存儲服務，用于處理生成的工件(如:文件) |
| session_service | BaseSessionService | 會話服務 | 用于管理和維護對話歷史。 |
| memory_service | Optional[BaseMemoryService] | 內存服務(可選) | 用于長期和短期記憶管理 |
創建Runner
# 創建智能體執行器
runner = Runner(agent=agent,app_name="app01",session_service=session_service
)run_async(...)方法
async def run_async(self,*,user_id: str,session_id: str,new_message: types.Content,run_config: RunConfig = RunConfig(),) -> AsyncGenerator[Event, None]:# ...runner.run_async(...)方法是Runner類的核心方法之一，用于以異步的方式執行Agent的任務。這個方法接收用戶輸入的消息，并將其添加到會話中，然后啟動Agent來處理該消息，并生成一系列的事件(Event).
具體參數解析如下：
1、user_id (str)
-
該參數表示用戶的唯一標識，用于標記當前會話是屬于哪個用戶的，每個用戶可以有多個會話實例。
2、session_id (str)
-
該參數表示會話的唯一標識，用于區分不同用戶或同一個用戶的不同的會話，每次創建會話時都會為其分配一個唯一的session_id
3、new_message (types.Content)
-
這是ADK中的一個核心對象，表示傳遞給Agent的新消息。types.Content 是一個包含具體消息內容的對象，通常它包含 role(角色)、text(消息文本) 等字段。
-
例如：content: parts=[Part(video_metadata=None, thought=None, code_execution_result=None, executable_code=None, file_data=None, function_call=None, function_response=None, inline_data=None, text='寫一句心靈雞湯')] role='user'
4、run_config (RunConfig) 可選
-
這是一個配置對象，用于定制執行時的參數設置，可以根據需求設置一些如語音配置、響應方式、模型最大調用次數等參數。其默認值為 RunConfig()，這是一個內置的默認配置類。
RunConfig 核心字段解析：
-
speech_config: 用于配置語音識別和語音輸出的相關設置(例如：語音轉文本、語音輸出等)，通常與語音助手相關。
-
response_modalities：配置響應的方式，例如返回文本、語音等。
-
save_input_blobs_as_artifacts: 是否將輸入的文件保存為工件(例如：將用戶上傳的圖片保存為文件以供后續處理)，默認值為false
-
support_cfc: 是否支持CFC(Custom Function Calling),即是否允許Agent調用自定義函數。
-
streaming_mode：設置流模型(例如：是否開啟流式輸出),用于處理生成時響應時間較長的情況。
-
output_audio_transcription: 配置輸出的音頻轉錄設置(比如：支持語音輸出)。
-
max_llm_calls: 配置在一次會話中最大可以調用LLM次數的上限。默認值: 500
運行Agent
# 執行對話
async def call_agent_async(query: str, runner, user_id, session_id):"""向agent發送查詢并打印最終響應"""print(f"\n>>> 用戶: {query} \n")# 將用戶消息轉換為ADK格式content = types.Content(role='user', parts=[types.Part(text=query)])print("content:", content)final_response_text = "未生成最終響應。" # Default# Key Concept: run_async 會執行Agent邏輯并持續生成事件流（Events）。# 我們通過遍歷事件流來獲取最終答案。async for event in runner.run_async(user_id=user_id, session_id=session_id, new_message=content):# 您可以取消下面這行的注釋，以查看執行過程中的所有事件print(f" [Event] Author: {event.author}, Type: {type(event).__name__}, Final: {event.is_final_response()}, Content: {event.content}")# Key Concept: is_final_response() 用于標記當前對話輪次（turn）的終結消息if event.is_final_response():if event.content and event.content.parts:# 假設響應文本位于首部分final_response_text = event.content.parts[0].textelif event.actions and event.actions.escalate: # Handle potential errors/escalationsfinal_response_text = f"Agent escalated: {event.error_message or 'No specific message.'}"# Add more checks here if needed (e.g., specific error codes)break # 找到最終響應后停止處理事件print(f"<<< Agent Response: {final_response_text}")if __name__ == '__main__':query = "寫一句心靈雞湯"asyncio.run(call_agent_async(query=query,runner=runner,user_id=USER_ID,session_id=SESSION_ID))call_agent_async(...) 函數解析
🍂Part 1 導入依賴的模塊
from google.genai import types這個types模塊是用于在ADK中創建消息內容的，它允許你定義消息的"角色"(例如：user 或 assistant)，以及消息的內容部分。
🍂Part 2 定義 call_agent_async 函數
async def call_agent_async(query: str, runner, user_id, session_id):-
query: 用戶輸入的查詢問題。
-
runner: 創建的Runner對象，負責啟動Agent并管理會話。
-
user_id 和 session_id: 用于標記特定用戶的會話，確保模型能夠維持會話上下文。
🍂Part 3 將用戶消息轉換為ADK格式
content = types.Content(role='user', parts=[types.Part(text=query)])content中內容：
這里將用戶的查詢(query) 格式化為ADK的消息格式，并通過 types.Content 和 types.Part 封裝成可以傳遞給Agent的內容。 role="user" 表明消息是由用戶發送的。
🍂Part 4 準備獲取Agent最終回復
final_response_text = "未生成最終響應。" # Default默認情況下 final_response_text 被初始化為一個錯誤信息字符串，表明Agent沒有最終響應，后續會被更新為最終真正的回答.
🍂Part 5 調用Agent并監聽事件
async for event in runner.run_async(user_id=user_id, session_id=session_id, new_message=content):# 您可以取消下面這行的注釋，以查看執行過程中的所有事件print(f" [Event] Author: {event.author}, Type: {type(event).__name__}, Final: {event.is_final_response()}, Content: {event.content}")runner.run_async(...) 是一個異步生成器，負責執行Agent推理并生成一系列事件 (Events)，每個事件代表了交互中的一步操作(如用戶消息、模型回復、工具調用等)
你可以通過 async for...循環迭代所有的事件，并逐一檢查它們的內容。
Event 事件中的內容：
🍂Part 6 檢查是否是最終響應
if event.is_final_response():event.is_final_response() 用來判斷當前事件是否是最終響應，這個判斷標志著Agent已經完成了對用戶問題的處理，可以停止進一步處理。
if event.content and event.content.parts:final_response_text = event.content.parts[0].text如果事件 content 存在，并且包含了 parts(模型回復的文本內容) ,就將 最終回復 保存到 final_response_text 變量中。
elif event.actions and event.actions.escalate: # Handle potential errors/escalationsfinal_response_text = f"Agent escalated: {event.error_message or 'No specific message.'}"如果 event.actions.escalate為True , 表示有錯誤發生或需要升級，比如模型無法處理問題，或者需要人工干預，這時，會將相應的錯誤信息記錄在 final_response_text
-節點排錯)
——繼承)
