并行化模式Parallelization復雜智能體任務通常包含多個可以同時執行的子任務而不是一個接一個地串行處理。此時就需要并行化設計模式。并行化是指同時執行多個組件比如 LLM 調用、工具使用甚至整個子智能體。與等待上一步完成再開始下一步不同并行執行允許獨立任務同時運行大幅縮短可拆分任務的整體執行時間。核心思想識別無依賴的子任務同時執行最后匯總結果串行 vs 并行對比以研究某主題并總結結果為例串行流程搜索來源 A總結來源 A搜索來源 B總結來源 B綜合 A 和 B 的摘要生成最終答案并行流程同時搜索來源 A 和 B兩個搜索完成后同時總結來源 A 和 B綜合 A 和 B 的摘要生成最終答案仍然串行因為需等待前兩步完成串行A? → A? → B? → B? → 匯總 總耗時 所有步驟之和 并行A? ──→ A? ──┐ B? ──→ B? ──┴→ 匯總 總耗時 ≈ 最慢分支 匯總并行化的核心思想是識別流程中彼此無依賴的部分并將它們并行執行。尤其是涉及外部服務調用如搜索、數據庫查詢、API 請求有延遲時可以同時發起多個請求顯著提升效率。實踐應用與場景并行化是優化智能體性能的強大模式適用于多種場景1. 信息收集與調研同時從多個來源收集信息是典型用例。應用場景智能體調研某公司并行任務同時搜索新聞、拉取股票數據、檢查社交媒體、查詢公司數據庫優勢比串行查找更快獲得全面視角2. 數據處理與分析并行應用不同分析方法或處理不同數據片段。應用場景智能體分析客戶反饋并行任務同時進行情感分析、關鍵詞提取、分類、緊急問題識別優勢快速獲得多維度分析結果3. 多 API 或工具交互并行調用多個獨立 API 或工具獲取不同信息或執行不同操作。應用場景旅行規劃智能體并行任務同時查機票、酒店、當地活動、餐廳推薦優勢更快生成完整旅行方案4. 多組件內容生成并行生成復雜內容的不同部分。應用場景智能體創建營銷郵件并行任務同時生成主題、正文、圖片、CTA 按鈕文案優勢更高效地組裝最終郵件5. 驗證與校驗并行執行多個獨立校驗任務。應用場景智能體驗證用戶輸入并行任務同時檢查郵箱格式、手機號、地址數據庫校驗、敏感詞檢測優勢更快反饋輸入有效性6. 多模態處理并行處理同一輸入的不同模態文本、圖片、音頻。應用場景智能體分析帶圖片的社交媒體帖子并行任務同時分析文本情感與關鍵詞、圖片中的物體與場景優勢更快整合多模態洞察7. A/B 測試或多方案生成并行生成多個響應或輸出便于選擇最佳方案。應用場景智能體生成多種創意文案并行任務同時用不同 prompt 或模型生成三種標題優勢快速對比并選出最佳選項并行化是智能體設計中的基礎優化技術開發者可通過并發執行獨立任務構建更高性能、更具響應性的應用。關鍵要點主要結論如下并行化是一種通過并發執行獨立任務提升效率的設計模式尤其適用于涉及外部資源如 API 調用等待的場景并發/并行架構會增加設計、調試和日志等開發復雜度與成本LangChain、Google ADK 等框架均支持并行執行定義與管理LCEL 中 RunnableParallel 是并行運行多個 runnable 的關鍵構造Google ADK 可通過 LLM 驅動的委托實現協調智能體并行處理子任務并行化可顯著降低整體延遲讓智能體系統在復雜任務下更具響應性代碼示例示例 1LangChain 實現importasynciofromtypingimportOptionalfromdotenvimportload_dotenvfromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.runnablesimportRunnable,RunnableParallel,RunnablePassthrough load_dotenv()# --- 配置 ---# 確保環境變量已設置 API key如 OPENAI_API_KEYtry:llm:Optional[ChatOpenAI]ChatOpenAI(modelgpt-4o-mini,temperature0.7)exceptExceptionase:print(f初始化語言模型出錯{e})llmNone# --- 定義獨立鏈 ---# 三個鏈分別執行不同任務可并行運行summarize_chain:Runnable(ChatPromptTemplate.from_messages([(system,請簡明扼要地總結以下主題),(user,{topic})])|llm|StrOutputParser())questions_chain:Runnable(ChatPromptTemplate.from_messages([(system,請針對以下主題生成三個有趣的問題),(user,{topic})])|llm|StrOutputParser())terms_chain:Runnable(ChatPromptTemplate.from_messages([(system,請從以下主題中提取 5-10 個關鍵詞用逗號分隔),(user,{topic})])|llm|StrOutputParser())# --- 構建并行 匯總鏈 ---# 1. 定義并行任務塊結果與原始 topic 一起傳遞到下一步map_chainRunnableParallel({summary:summarize_chain,questions:questions_chain,key_terms:terms_chain,topic:RunnablePassthrough(),# 傳遞原始 topic})# 2. 定義最終匯總 prompt整合并行結果synthesis_promptChatPromptTemplate.from_messages([(system,根據以下信息 摘要{summary} 相關問題{questions} 關鍵詞{key_terms} 請綜合生成完整答案。),(user,原始主題{topic})])# 3. 構建完整鏈將并行結果直接傳遞給匯總 prompt再由 LLM 和輸出解析器處理full_parallel_chainmap_chain|synthesis_prompt|llm|StrOutputParser()# --- 運行鏈 ---asyncdefrun_parallel_example(topic:str)-None: 異步調用并行處理鏈輸出綜合結果。 Args: topic: 傳遞給 LangChain 的主題輸入 ifnotllm:print(LLM 未初始化無法運行示例。)returnprint(f\n--- 并行 LangChain 示例主題{topic} ---)try:# ainvoke 的輸入是單個 topic 字符串# 會傳遞給 map_chain 中的每個 runnableresponseawaitfull_parallel_chain.ainvoke(topic)print(\n--- 最終響應 ---)print(response)exceptExceptionase:print(f\n鏈執行出錯{e})if__name____main__:test_topic太空探索的歷史# Python 3.7 推薦用 asyncio.run 執行異步函數asyncio.run(run_parallel_example(test_topic))在 LangChain 框架中并行執行由 LangChain Expression LanguageLCEL實現。主要方法是將多個 Runnable 組件結構化為字典或列表當這些集合被傳遞給鏈中的下一個組件時LCEL 運行時會并發執行其中的 Runnable。在 LangGraph 中這一原理體現在圖的拓撲結構——通過設計圖結構使多個無直接依賴的節點可由同一節點并發啟動這些并行路徑獨立執行結果在后續匯聚節點整合。示例流程針對單一主題并發執行三個獨立任務最后匯總┌→ [摘要鏈] ────┐ 主題 → RunnableParallel ├→ [問題鏈] ────┼→ [匯總 Prompt] → LLM → 最終結果 └→ [關鍵詞鏈] ──┘三條獨立鏈分別負責主題摘要、問題生成、關鍵詞提取每條鏈由ChatPromptTemplate | LLM | StrOutputParser組成RunnableParallel將三條鏈打包并發執行同時用RunnablePassthrough保留原始輸入匯總鏈整合summary、questions、key_terms和topic由 LLM 生成綜合答案注意asyncio提供的是并發而非真正的并行——它通過事件循環在任務空閑如等待網絡請求時智能切換實現多個任務同時推進但實際仍在單線程下受 GIL 限制。示例 2Google ADK 實現fromgoogle.adk.agentsimportLlmAgent,ParallelAgent,SequentialAgentfromgoogle.adk.toolsimportgoogle_search GEMINI_MODELgemini-2.0-flash# --- 1. 定義并行運行的調研子智能體 ---# 調研員 1可再生能源researcher_agent_1LlmAgent(nameRenewableEnergyResearcher,modelGEMINI_MODEL,instruction你是一名專注于能源領域的 AI 調研助手。 調研可再生能源最新進展使用 Google Search 工具。 請簡明總結關鍵發現1-2 句只輸出摘要。 ,description調研可再生能源。,tools[google_search],output_keyrenewable_energy_result)# 調研員 2電動汽車researcher_agent_2LlmAgent(nameEVResearcher,modelGEMINI_MODEL,instruction你是一名專注于交通領域的 AI 調研助手。 調研電動汽車技術最新進展使用 Google Search 工具。 請簡明總結關鍵發現1-2 句只輸出摘要。 ,description調研電動汽車技術。,tools[google_search],output_keyev_technology_result)# 調研員 3碳捕集researcher_agent_3LlmAgent(nameCarbonCaptureResearcher,modelGEMINI_MODEL,instruction你是一名專注于氣候解決方案的 AI 調研助手。 調研碳捕集方法現狀使用 Google Search 工具。 請簡明總結關鍵發現1-2 句只輸出摘要。 ,description調研碳捕集方法。,tools[google_search],output_keycarbon_capture_result)# --- 2. 創建并行智能體并發運行調研員---parallel_research_agentParallelAgent(nameParallelWebResearchAgent,sub_agents[researcher_agent_1,researcher_agent_2,researcher_agent_3],description并行運行多個調研智能體收集信息。)# --- 3. 定義合并智能體并行智能體完成后運行---merger_agentLlmAgent(nameSynthesisAgent,modelGEMINI_MODEL,instruction你是一名負責整合調研結果的 AI 助手。 你的任務是將以下調研摘要合成為結構化報告并明確歸屬。每個主題用標題分段確保內容連貫僅整合輸入摘要。 **注意你的全部回答必須嚴格基于下方輸入摘要不得添加任何外部知識或細節。** **輸入摘要** * **可再生能源** {renewable_energy_result} * **電動汽車** {ev_technology_result} * **碳捕集** {carbon_capture_result} **輸出格式** ## 可持續技術最新進展摘要 ### 可再生能源發現 基于 RenewableEnergyResearcher 的摘要僅整合上述內容 ### 電動汽車發現 基于 EVResearcher 的摘要僅整合上述內容 ### 碳捕集發現 基于 CarbonCaptureResearcher 的摘要僅整合上述內容 ### 總結 僅基于上述內容簡要總結 1-2 句 只輸出結構化報告嚴格按上述格式不加其他說明。 ,description整合并行智能體的調研結果生成結構化報告僅基于輸入內容。,)# --- 4. 創建串行智能體總流程控制---sequential_pipeline_agentSequentialAgent(nameResearchAndSynthesisPipeline,sub_agents[parallel_research_agent,merger_agent],description協調并行調研與結果整合。)root_agentsequential_pipeline_agentGoogle ADK 框架提供了ParallelAgent、SequentialAgent等原語用于構建高效并發智能體流程。示例流程三個調研智能體并行調研完成后由合并智能體生成報告┌→ [可再生能源調研員] ──┐ SequentialAgent → ParallelAgent ├→ [電動汽車調研員] ────┼→ [合并智能體] → 結構化報告 └→ [碳捕集調研員] ────┘三個LlmAgent調研員分別聚焦可再生能源、電動汽車、碳捕集各自使用google_search工具摘要結果通過output_key存入 session stateParallelAgent并行運行三位調研員調研任務同步進行LlmAgent合并整合調研結果要求輸出僅基于輸入摘要結構化分段不添加外部知識SequentialAgent串行執行并行調研 → 結果整合作為主流程入口兩種實現的核心區別對比LangChainGoogle ADK并行原語RunnableParallel字典/列表結構ParallelAgent智能體容器串行控制LCEL 管道|SequentialAgent結果傳遞字典鍵值自動傳遞output_key寫入 session state適用場景鏈式組合靈活輕量多智能體協作結構化流程 【Agent 設計模式】并行化模式（Parallelization）
并行化模式Parallelization復雜智能體任務通常包含多個可以同時執行的子任務而不是一個接一個地串行處理。此時就需要并行化設計模式。并行化是指同時執行多個組件比如 LLM 調用、工具使用甚至整個子智能體。與等待上一步完成再開始下一步不同并行執行允許獨立任務同時運行大幅縮短可拆分任務的整體執行時間。核心思想識別無依賴的子任務同時執行最后匯總結果串行 vs 并行對比以研究某主題并總結結果為例串行流程搜索來源 A總結來源 A搜索來源 B總結來源 B綜合 A 和 B 的摘要生成最終答案并行流程同時搜索來源 A 和 B兩個搜索完成后同時總結來源 A 和 B綜合 A 和 B 的摘要生成最終答案仍然串行因為需等待前兩步完成串行A? → A? → B? → B? → 匯總 總耗時 所有步驟之和 并行A? ──→ A? ──┐ B? ──→ B? ──┴→ 匯總 總耗時 ≈ 最慢分支 匯總并行化的核心思想是識別流程中彼此無依賴的部分并將它們并行執行。尤其是涉及外部服務調用如搜索、數據庫查詢、API 請求有延遲時可以同時發起多個請求顯著提升效率。實踐應用與場景并行化是優化智能體性能的強大模式適用于多種場景1. 信息收集與調研同時從多個來源收集信息是典型用例。應用場景智能體調研某公司并行任務同時搜索新聞、拉取股票數據、檢查社交媒體、查詢公司數據庫優勢比串行查找更快獲得全面視角2. 數據處理與分析并行應用不同分析方法或處理不同數據片段。應用場景智能體分析客戶反饋并行任務同時進行情感分析、關鍵詞提取、分類、緊急問題識別優勢快速獲得多維度分析結果3. 多 API 或工具交互并行調用多個獨立 API 或工具獲取不同信息或執行不同操作。應用場景旅行規劃智能體并行任務同時查機票、酒店、當地活動、餐廳推薦優勢更快生成完整旅行方案4. 多組件內容生成并行生成復雜內容的不同部分。應用場景智能體創建營銷郵件并行任務同時生成主題、正文、圖片、CTA 按鈕文案優勢更高效地組裝最終郵件5. 驗證與校驗并行執行多個獨立校驗任務。應用場景智能體驗證用戶輸入并行任務同時檢查郵箱格式、手機號、地址數據庫校驗、敏感詞檢測優勢更快反饋輸入有效性6. 多模態處理并行處理同一輸入的不同模態文本、圖片、音頻。應用場景智能體分析帶圖片的社交媒體帖子并行任務同時分析文本情感與關鍵詞、圖片中的物體與場景優勢更快整合多模態洞察7. A/B 測試或多方案生成并行生成多個響應或輸出便于選擇最佳方案。應用場景智能體生成多種創意文案并行任務同時用不同 prompt 或模型生成三種標題優勢快速對比并選出最佳選項并行化是智能體設計中的基礎優化技術開發者可通過并發執行獨立任務構建更高性能、更具響應性的應用。關鍵要點主要結論如下并行化是一種通過并發執行獨立任務提升效率的設計模式尤其適用于涉及外部資源如 API 調用等待的場景并發/并行架構會增加設計、調試和日志等開發復雜度與成本LangChain、Google ADK 等框架均支持并行執行定義與管理LCEL 中 RunnableParallel 是并行運行多個 runnable 的關鍵構造Google ADK 可通過 LLM 驅動的委托實現協調智能體并行處理子任務并行化可顯著降低整體延遲讓智能體系統在復雜任務下更具響應性代碼示例示例 1LangChain 實現importasynciofromtypingimportOptionalfromdotenvimportload_dotenvfromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.runnablesimportRunnable,RunnableParallel,RunnablePassthrough load_dotenv()# --- 配置 ---# 確保環境變量已設置 API key如 OPENAI_API_KEYtry:llm:Optional[ChatOpenAI]ChatOpenAI(modelgpt-4o-mini,temperature0.7)exceptExceptionase:print(f初始化語言模型出錯{e})llmNone# --- 定義獨立鏈 ---# 三個鏈分別執行不同任務可并行運行summarize_chain:Runnable(ChatPromptTemplate.from_messages([(system,請簡明扼要地總結以下主題),(user,{topic})])|llm|StrOutputParser())questions_chain:Runnable(ChatPromptTemplate.from_messages([(system,請針對以下主題生成三個有趣的問題),(user,{topic})])|llm|StrOutputParser())terms_chain:Runnable(ChatPromptTemplate.from_messages([(system,請從以下主題中提取 5-10 個關鍵詞用逗號分隔),(user,{topic})])|llm|StrOutputParser())# --- 構建并行 匯總鏈 ---# 1. 定義并行任務塊結果與原始 topic 一起傳遞到下一步map_chainRunnableParallel({summary:summarize_chain,questions:questions_chain,key_terms:terms_chain,topic:RunnablePassthrough(),# 傳遞原始 topic})# 2. 定義最終匯總 prompt整合并行結果synthesis_promptChatPromptTemplate.from_messages([(system,根據以下信息 摘要{summary} 相關問題{questions} 關鍵詞{key_terms} 請綜合生成完整答案。),(user,原始主題{topic})])# 3. 構建完整鏈將并行結果直接傳遞給匯總 prompt再由 LLM 和輸出解析器處理full_parallel_chainmap_chain|synthesis_prompt|llm|StrOutputParser()# --- 運行鏈 ---asyncdefrun_parallel_example(topic:str)-None: 異步調用并行處理鏈輸出綜合結果。 Args: topic: 傳遞給 LangChain 的主題輸入 ifnotllm:print(LLM 未初始化無法運行示例。)returnprint(f\n--- 并行 LangChain 示例主題{topic} ---)try:# ainvoke 的輸入是單個 topic 字符串# 會傳遞給 map_chain 中的每個 runnableresponseawaitfull_parallel_chain.ainvoke(topic)print(\n--- 最終響應 ---)print(response)exceptExceptionase:print(f\n鏈執行出錯{e})if__name____main__:test_topic太空探索的歷史# Python 3.7 推薦用 asyncio.run 執行異步函數asyncio.run(run_parallel_example(test_topic))在 LangChain 框架中并行執行由 LangChain Expression LanguageLCEL實現。主要方法是將多個 Runnable 組件結構化為字典或列表當這些集合被傳遞給鏈中的下一個組件時LCEL 運行時會并發執行其中的 Runnable。在 LangGraph 中這一原理體現在圖的拓撲結構——通過設計圖結構使多個無直接依賴的節點可由同一節點并發啟動這些并行路徑獨立執行結果在后續匯聚節點整合。示例流程針對單一主題并發執行三個獨立任務最后匯總┌→ [摘要鏈] ────┐ 主題 → RunnableParallel ├→ [問題鏈] ────┼→ [匯總 Prompt] → LLM → 最終結果 └→ [關鍵詞鏈] ──┘三條獨立鏈分別負責主題摘要、問題生成、關鍵詞提取每條鏈由ChatPromptTemplate | LLM | StrOutputParser組成RunnableParallel將三條鏈打包并發執行同時用RunnablePassthrough保留原始輸入匯總鏈整合summary、questions、key_terms和topic由 LLM 生成綜合答案注意asyncio提供的是并發而非真正的并行——它通過事件循環在任務空閑如等待網絡請求時智能切換實現多個任務同時推進但實際仍在單線程下受 GIL 限制。示例 2Google ADK 實現fromgoogle.adk.agentsimportLlmAgent,ParallelAgent,SequentialAgentfromgoogle.adk.toolsimportgoogle_search GEMINI_MODELgemini-2.0-flash# --- 1. 定義并行運行的調研子智能體 ---# 調研員 1可再生能源researcher_agent_1LlmAgent(nameRenewableEnergyResearcher,modelGEMINI_MODEL,instruction你是一名專注于能源領域的 AI 調研助手。 調研可再生能源最新進展使用 Google Search 工具。 請簡明總結關鍵發現1-2 句只輸出摘要。 ,description調研可再生能源。,tools[google_search],output_keyrenewable_energy_result)# 調研員 2電動汽車researcher_agent_2LlmAgent(nameEVResearcher,modelGEMINI_MODEL,instruction你是一名專注于交通領域的 AI 調研助手。 調研電動汽車技術最新進展使用 Google Search 工具。 請簡明總結關鍵發現1-2 句只輸出摘要。 ,description調研電動汽車技術。,tools[google_search],output_keyev_technology_result)# 調研員 3碳捕集researcher_agent_3LlmAgent(nameCarbonCaptureResearcher,modelGEMINI_MODEL,instruction你是一名專注于氣候解決方案的 AI 調研助手。 調研碳捕集方法現狀使用 Google Search 工具。 請簡明總結關鍵發現1-2 句只輸出摘要。 ,description調研碳捕集方法。,tools[google_search],output_keycarbon_capture_result)# --- 2. 創建并行智能體并發運行調研員---parallel_research_agentParallelAgent(nameParallelWebResearchAgent,sub_agents[researcher_agent_1,researcher_agent_2,researcher_agent_3],description并行運行多個調研智能體收集信息。)# --- 3. 定義合并智能體并行智能體完成后運行---merger_agentLlmAgent(nameSynthesisAgent,modelGEMINI_MODEL,instruction你是一名負責整合調研結果的 AI 助手。 你的任務是將以下調研摘要合成為結構化報告并明確歸屬。每個主題用標題分段確保內容連貫僅整合輸入摘要。 **注意你的全部回答必須嚴格基于下方輸入摘要不得添加任何外部知識或細節。** **輸入摘要** * **可再生能源** {renewable_energy_result} * **電動汽車** {ev_technology_result} * **碳捕集** {carbon_capture_result} **輸出格式** ## 可持續技術最新進展摘要 ### 可再生能源發現 基于 RenewableEnergyResearcher 的摘要僅整合上述內容 ### 電動汽車發現 基于 EVResearcher 的摘要僅整合上述內容 ### 碳捕集發現 基于 CarbonCaptureResearcher 的摘要僅整合上述內容 ### 總結 僅基于上述內容簡要總結 1-2 句 只輸出結構化報告嚴格按上述格式不加其他說明。 ,description整合并行智能體的調研結果生成結構化報告僅基于輸入內容。,)# --- 4. 創建串行智能體總流程控制---sequential_pipeline_agentSequentialAgent(nameResearchAndSynthesisPipeline,sub_agents[parallel_research_agent,merger_agent],description協調并行調研與結果整合。)root_agentsequential_pipeline_agentGoogle ADK 框架提供了ParallelAgent、SequentialAgent等原語用于構建高效并發智能體流程。示例流程三個調研智能體并行調研完成后由合并智能體生成報告┌→ [可再生能源調研員] ──┐ SequentialAgent → ParallelAgent ├→ [電動汽車調研員] ────┼→ [合并智能體] → 結構化報告 └→ [碳捕集調研員] ────┘三個LlmAgent調研員分別聚焦可再生能源、電動汽車、碳捕集各自使用google_search工具摘要結果通過output_key存入 session stateParallelAgent并行運行三位調研員調研任務同步進行LlmAgent合并整合調研結果要求輸出僅基于輸入摘要結構化分段不添加外部知識SequentialAgent串行執行并行調研 → 結果整合作為主流程入口兩種實現的核心區別對比LangChainGoogle ADK并行原語RunnableParallel字典/列表結構ParallelAgent智能體容器串行控制LCEL 管道|SequentialAgent結果傳遞字典鍵值自動傳遞output_key寫入 session state適用場景鏈式組合靈活輕量多智能體協作結構化流程