并行化模式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适用场景链式组合灵活轻量多智能体协作结构化流程
