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