一、引言从“单兵作战”到“排兵布阵”早期的AI应用大多是“单兵作战”——一个模型调用、一次工具执行、一个答案返回。但当业务场景复杂起来后一个Agent单打独斗的模式就会捉襟见肘。跨境电商的“上架一条新Listing”这件事可能涉及产品信息提取、多语言翻译、类目匹配、库存同步、价格策略建议、广告计划创建——每个环节都需要不同的能力而且有严格的先后顺序。任务编排要解决的核心问题是如何把多个Agent、多个工具、多个步骤按照正确的顺序和条件组合起来形成一个可执行、可恢复、可观察的业务流程。这就像乐队指挥——不是每个乐手自己演奏得好就行而是要让它们在正确的时间、以正确的方式协同起来。本文将系统介绍AI Agent任务编排的几种核心模式并基于LangGraph给出完整的代码实现。二、任务编排的核心模式生产级Agent系统的编排有几种经过验证的模式。理解这些模式是设计复杂流程的第一步。2.1 顺序模式Sequential顺序模式是最基础的编排方式——Agent A的输出成为Agent B的输入像一条流水线。典型场景起草→翻译→润色→发布。微软Agent Framework的SequentialBuilder支持这种模式每个Agent专注于一个环节将结果传递给下一环节。2.2 路由模式Routing路由模式是先对输入进行分类再分配给不同的专家Agent。不同的问题类型走不同的处理路径不同复杂度的查询使用不同能力的模型。典型场景客服工单分流——退款类走退款Agent技术问题走技术支持Agent简单FAQ走知识库检索Agent。2.3 并行模式Parallelization并行模式是同时执行多个独立的任务然后在汇总节点合并结果。典型场景一个商品信息需要同时查价格、库存、销量、评论四个查询互不依赖并行执行可以大幅缩短总耗时。2.4 编排器-工作者模式Orchestrator-Workers编排器-工作者模式是最强大的模式之一。一个中心编排器Orchestrator理解整体任务目标动态拆解出子任务分发给多个工作者Worker执行然后汇总结果。核心特点是子任务集在运行时动态确定而非设计时固定。典型场景多Agent协作的代码修改任务——编排器分析需要改哪些文件然后为每个文件派发一个工作者Agent。2.5 评估器-优化器模式Evaluator-Optimizer评估器-优化器模式在循环中配对工作生成器Generator产生输出评估器Evaluator根据质量条件评分并提供反馈循环反复直到输出通过或达到最大迭代次数。典型场景自动代码审查、文学翻译、迭代内容优化。三、LangGraph实战构建跨境电商任务编排系统LangGraph是目前生产环境中编排Agent工作流的主流方案。它将Agent推理过程建模为有向图每个节点代表一个计算步骤每条边定义状态流转条件支持条件分支、并行执行和循环重试等复杂控制模式。3.1 环境准备pip install langgraph langchain-openai python-dotenv# .env配置OPENAI_API_KEYsk-xxxxx OPENAI_BASE_URLhttps://api.deepseek.com/v1# 或DeepSeek3.2 定义状态State状态是整个编排系统的核心数据结构负责在各个节点之间传递信息。fromtypingimportTypedDict,List,Dict,Any,OptionalfromtypingimportLiteralclassListingState(TypedDict,totalFalse):跨境电商商品上架任务的状态# 原始输入product_data:Dict[str,Any]# 商品原始数据图片、规格、价格等target_market:str# 目标市场US/UK/DE等# 各阶段产出extracted_info:Optional[Dict]# 提取后的结构化信息translated_content:Optional[Dict]# 多语言翻译结果category_match:Optional[str]# 匹配的类目IDprice_suggestion:Optional[Dict]# 价格建议listing_draft:Optional[str]# Listing草稿# 流程控制current_step:str# 当前步骤名称errors:List[str]# 错误记录retry_count:int# 重试计数# 最终输出final_listing_id:Optional[str]# 最终创建的Listing IDstatus:Literal[pending,running,completed,failed]3.3 定义各节点Agent/工具节点每个节点代表一个具体任务fromlanggraph.graphimportStateGraph,ENDfromlangchain_openaiimportChatOpenAIfromlangchain_core.messagesimportHumanMessage,SystemMessage llmChatOpenAI(modelgpt-4o-mini,temperature0.2)# 节点1产品信息提取 defextract_product_info(state:ListingState)-dict:从原始数据中提取结构化产品信息print(f [1/5] 提取产品信息...)promptf 请从以下商品原始数据中提取结构化信息{state[product_data]}提取字段品牌、型号、核心规格尺寸/重量/材质等、主要卖点3-5条 输出格式为JSON。 responsellm.invoke([HumanMessage(contentprompt)])# 模拟解析extracted{brand:TechGear,model:X-Pro 2026,specs:{screen:27 4K,weight:3.2kg},selling_points:[超高清画质,护眼技术,超薄设计]}return{extracted_info:extracted,current_step:extracted}# 节点2多语言翻译 deftranslate_content(state:ListingState)-dict:将产品信息翻译为目标市场语言print(f [2/5] 翻译为目标市场语言 ({state[target_market]})...)# 实际应调用翻译API或LLM翻译translations{US:27-inch 4K Ultra HD Monitor,DE:27-Zoll 4K Ultra-HD-Monitor,}translatedtranslations.get(state[target_market],4K Monitor)return{translated_content:{title:translated,lang:state[target_market]}}# 节点3类目匹配 defmatch_category(state:ListingState)-dict:匹配平台类目print(f [3/5] 匹配平台类目...)categories{monitor:Electronics Monitors,laptop:Electronics Laptops}# 简单关键词匹配matchedElectronics Monitorsifmonitorinstr(state[extracted_info]).lower()elseElectronicsreturn{category_match:matched}# 节点4定价建议 defsuggest_price(state:ListingState)-dict:生成定价建议考虑市场、竞品、汇率print(f [4/5] 生成定价建议...)# 模拟考虑目标市场的价格策略base_price299.99market_multiplier{US:1.0,DE:1.15,UK:1.20}multipliermarket_multiplier.get(state[target_market],1.0)suggested{suggested_price:round(base_price*multiplier,2),currency:USDifstate[target_market]USelseEUR,strategy:competitive}return{price_suggestion:suggested}# 节点5生成Listing草稿 defgenerate_listing(state:ListingState)-dict:生成最终的Listing描述print(f✍️ [5/5] 生成Listing草稿...)draftf Product:{state[translated_content][title]}Category:{state[category_match]}Price: ${state[price_suggestion][suggested_price]}Brand:{state[extracted_info][brand]}Selling Points:{, .join(state[extracted_info][selling_points])}return{listing_draft:draft,status:completed}3.4 定义条件路由不是所有步骤都需要执行。条件路由允许根据状态动态决定流程走向defshould_translate(state:ListingState)-Literal[translate,skip_translate]:判断是否需要翻译# 如果目标市场使用英语跳过翻译步骤ifstate[target_market]in[US,UK,AU]:returnskip_translatereturntranslatedefshould_continue(state:ListingState)-Literal[continue,end]:判断是否继续ifstate.get(status)failed:returnendreturncontinue3.5 构建工作流图有了节点和路由条件就可以组装成完整的执行图# 创建状态图workflowStateGraph(ListingState)# 添加节点workflow.add_node(extract,extract_product_info)workflow.add_node(translate,translate_content)workflow.add_node(match_category,match_category)workflow.add_node(suggest_price,suggest_price)workflow.add_node(generate_listing,generate_listing)# 设置入口workflow.set_entry_point(extract)# 顺序边workflow.add_edge(extract,match_category)workflow.add_edge(match_category,suggest_price)workflow.add_edge(suggest_price,generate_listing)# 条件边翻译步骤条件执行workflow.add_conditional_edges(extract,should_translate,{translate:translate,skip_translate:match_category})# 翻译后回到主流程workflow.add_edge(translate,match_category)# 结束边workflow.add_edge(generate_listing,END)# 编译为可执行应用appworkflow.compile()3.6 测试运行defrun_listing_workflow():执行商品上架工作流initial_state{product_data:{name:TechGear 27-inch 4K Monitor,specs:{resolution:3840x2160,refresh_rate:144Hz},price:299.99,description:Professional-grade monitor for designers and gamers},target_market:DE,# 德国市场触发翻译current_step:start,errors:[],retry_count:0,status:pending}print(*60)print( 启动商品上架工作流)print(f目标市场:{initial_state[target_market]})print(*60)resultapp.invoke(initial_state)print(\n*60)print(✅ 工作流执行完成)print(*60)print(f状态:{result.get(status)})print(f类目:{result.get(category_match)})print(f定价建议: ${result.get(price_suggestion, {}).get(suggested_price)})print(f\n Listing草稿:\n{result.get(listing_draft)})returnresultif__name____main__:run_listing_workflow()四、高级编排能力4.1 并行执行LangGraph支持在同一个图中实现并行扇出。例如“同时查询价格、库存、评分”这三个独立任务fromlanggraph.graphimportStateGraph# 三个独立的查询节点互不依赖workflow.add_node(query_price,query_price)workflow.add_node(query_stock,query_stock)workflow.add_node(query_rating,query_rating)# 从同一个节点扇出workflow.add_edge(start_parallel,query_price)workflow.add_edge(start_parallel,query_stock)workflow.add_edge(start_parallel,query_rating)# 汇总节点等待所有完成workflow.add_edge(query_price,aggregate)workflow.add_edge(query_stock,aggregate)workflow.add_edge(query_rating,aggregate)4.2 人工介入Human-in-the-Loop在电商场景中某些操作需要人工审批——例如退款超过阈值、价格调整、敏感操作。LangGraph支持在任意节点暂停等待人工输入fromlanggraph.checkpointimportMemorySaver# 在定价节点后添加审批暂停点defhuman_approval_node(state:ListingState)-dict:人工审批节点——实际生产环境中推送至Slack/钉钉print(⏸️ 需要人工审批定价建议...)print(f建议价格: ${state[price_suggestion][suggested_price]})print(等待审批...)# 模拟人工审批通过# 生产中会通过外部接口接收审批结果approvedTrueifapproved:return{status:approved}else:return{status:rejected,errors:[定价被人工拒绝]}# 使用MemorySaver支持暂停/恢复fromlanggraph.graphimportStateGraph memoryMemorySaver()appworkflow.compile(checkpointermemory)# 执行时遇到人工节点会自动暂停等待外部输入4.3 状态持久化与恢复生产环境中长流程可能执行数分钟甚至数小时。LangGraph支持检查点机制允许在任意节点中断后从断点恢复# 使用SQLite持久化状态取代MemorySaverfromlanggraph.checkpoint.sqliteimportSqliteSaverwithSqliteSaver.from_conn_string(checkpoints.db)assaver:appworkflow.compile(checkpointersaver)# 执行...# 中断后使用相同的thread_id恢复五、工程化要点5.1 错误处理与重试生产环境中节点执行可能失败。LangGraph支持在节点级别配置重试策略对于临时性失败如API超时配置指数退避重试对于业务逻辑错误如数据格式不符记录错误并转入人工处理路径。5.2 可观测性每个节点的输入输出、耗时、成功/失败状态都应被记录。可以与LangSmith集成获得完整的执行追踪可视化。5.3 分布式部署考虑如果Worker Agent需要独立扩展可以将编排器与工作者解耦为独立服务。AWS AgentCore Runtime等平台支持将Agent部署为独立容器通过消息队列SQS进行任务分发实现并行扩展。六、小结本文围绕AI Agent的任务编排系统介绍了五种核心编排模式顺序、路由、并行、编排器-工作者、评估器-优化器各适用不同场景LangGraph实战基于状态图构建跨境电商商品上架工作流覆盖提取→匹配→定价→生成全流程高级能力并行执行、人工介入审批、状态持久化与恢复工程化要点错误处理、可观测性、分布式部署任务编排的价值在于将多个AI能力从“各自为战”变为“协同作战”。当一个业务诉求需要10个Agent协作完成时工作流编排决定了这个系统是“优雅地跑起来”还是“崩溃地卡住”。关于多智能体协作的更深层次模式、子图嵌套与动态编排欢迎在评论区交流。
一、引言从“单兵作战”到“排兵布阵”早期的AI应用大多是“单兵作战”——一个模型调用、一次工具执行、一个答案返回。但当业务场景复杂起来后一个Agent单打独斗的模式就会捉襟见肘。跨境电商的“上架一条新Listing”这件事可能涉及产品信息提取、多语言翻译、类目匹配、库存同步、价格策略建议、广告计划创建——每个环节都需要不同的能力而且有严格的先后顺序。任务编排要解决的核心问题是如何把多个Agent、多个工具、多个步骤按照正确的顺序和条件组合起来形成一个可执行、可恢复、可观察的业务流程。这就像乐队指挥——不是每个乐手自己演奏得好就行而是要让它们在正确的时间、以正确的方式协同起来。本文将系统介绍AI Agent任务编排的几种核心模式并基于LangGraph给出完整的代码实现。二、任务编排的核心模式生产级Agent系统的编排有几种经过验证的模式。理解这些模式是设计复杂流程的第一步。2.1 顺序模式Sequential顺序模式是最基础的编排方式——Agent A的输出成为Agent B的输入像一条流水线。典型场景起草→翻译→润色→发布。微软Agent Framework的SequentialBuilder支持这种模式每个Agent专注于一个环节将结果传递给下一环节。2.2 路由模式Routing路由模式是先对输入进行分类再分配给不同的专家Agent。不同的问题类型走不同的处理路径不同复杂度的查询使用不同能力的模型。典型场景客服工单分流——退款类走退款Agent技术问题走技术支持Agent简单FAQ走知识库检索Agent。2.3 并行模式Parallelization并行模式是同时执行多个独立的任务然后在汇总节点合并结果。典型场景一个商品信息需要同时查价格、库存、销量、评论四个查询互不依赖并行执行可以大幅缩短总耗时。2.4 编排器-工作者模式Orchestrator-Workers编排器-工作者模式是最强大的模式之一。一个中心编排器Orchestrator理解整体任务目标动态拆解出子任务分发给多个工作者Worker执行然后汇总结果。核心特点是子任务集在运行时动态确定而非设计时固定。典型场景多Agent协作的代码修改任务——编排器分析需要改哪些文件然后为每个文件派发一个工作者Agent。2.5 评估器-优化器模式Evaluator-Optimizer评估器-优化器模式在循环中配对工作生成器Generator产生输出评估器Evaluator根据质量条件评分并提供反馈循环反复直到输出通过或达到最大迭代次数。典型场景自动代码审查、文学翻译、迭代内容优化。三、LangGraph实战构建跨境电商任务编排系统LangGraph是目前生产环境中编排Agent工作流的主流方案。它将Agent推理过程建模为有向图每个节点代表一个计算步骤每条边定义状态流转条件支持条件分支、并行执行和循环重试等复杂控制模式。3.1 环境准备pip install langgraph langchain-openai python-dotenv# .env配置OPENAI_API_KEYsk-xxxxx OPENAI_BASE_URLhttps://api.deepseek.com/v1# 或DeepSeek3.2 定义状态State状态是整个编排系统的核心数据结构负责在各个节点之间传递信息。fromtypingimportTypedDict,List,Dict,Any,OptionalfromtypingimportLiteralclassListingState(TypedDict,totalFalse):跨境电商商品上架任务的状态# 原始输入product_data:Dict[str,Any]# 商品原始数据图片、规格、价格等target_market:str# 目标市场US/UK/DE等# 各阶段产出extracted_info:Optional[Dict]# 提取后的结构化信息translated_content:Optional[Dict]# 多语言翻译结果category_match:Optional[str]# 匹配的类目IDprice_suggestion:Optional[Dict]# 价格建议listing_draft:Optional[str]# Listing草稿# 流程控制current_step:str# 当前步骤名称errors:List[str]# 错误记录retry_count:int# 重试计数# 最终输出final_listing_id:Optional[str]# 最终创建的Listing IDstatus:Literal[pending,running,completed,failed]3.3 定义各节点Agent/工具节点每个节点代表一个具体任务fromlanggraph.graphimportStateGraph,ENDfromlangchain_openaiimportChatOpenAIfromlangchain_core.messagesimportHumanMessage,SystemMessage llmChatOpenAI(modelgpt-4o-mini,temperature0.2)# 节点1产品信息提取 defextract_product_info(state:ListingState)-dict:从原始数据中提取结构化产品信息print(f [1/5] 提取产品信息...)promptf 请从以下商品原始数据中提取结构化信息{state[product_data]}提取字段品牌、型号、核心规格尺寸/重量/材质等、主要卖点3-5条 输出格式为JSON。 responsellm.invoke([HumanMessage(contentprompt)])# 模拟解析extracted{brand:TechGear,model:X-Pro 2026,specs:{screen:27 4K,weight:3.2kg},selling_points:[超高清画质,护眼技术,超薄设计]}return{extracted_info:extracted,current_step:extracted}# 节点2多语言翻译 deftranslate_content(state:ListingState)-dict:将产品信息翻译为目标市场语言print(f [2/5] 翻译为目标市场语言 ({state[target_market]})...)# 实际应调用翻译API或LLM翻译translations{US:27-inch 4K Ultra HD Monitor,DE:27-Zoll 4K Ultra-HD-Monitor,}translatedtranslations.get(state[target_market],4K Monitor)return{translated_content:{title:translated,lang:state[target_market]}}# 节点3类目匹配 defmatch_category(state:ListingState)-dict:匹配平台类目print(f [3/5] 匹配平台类目...)categories{monitor:Electronics Monitors,laptop:Electronics Laptops}# 简单关键词匹配matchedElectronics Monitorsifmonitorinstr(state[extracted_info]).lower()elseElectronicsreturn{category_match:matched}# 节点4定价建议 defsuggest_price(state:ListingState)-dict:生成定价建议考虑市场、竞品、汇率print(f [4/5] 生成定价建议...)# 模拟考虑目标市场的价格策略base_price299.99market_multiplier{US:1.0,DE:1.15,UK:1.20}multipliermarket_multiplier.get(state[target_market],1.0)suggested{suggested_price:round(base_price*multiplier,2),currency:USDifstate[target_market]USelseEUR,strategy:competitive}return{price_suggestion:suggested}# 节点5生成Listing草稿 defgenerate_listing(state:ListingState)-dict:生成最终的Listing描述print(f✍️ [5/5] 生成Listing草稿...)draftf Product:{state[translated_content][title]}Category:{state[category_match]}Price: ${state[price_suggestion][suggested_price]}Brand:{state[extracted_info][brand]}Selling Points:{, .join(state[extracted_info][selling_points])}return{listing_draft:draft,status:completed}3.4 定义条件路由不是所有步骤都需要执行。条件路由允许根据状态动态决定流程走向defshould_translate(state:ListingState)-Literal[translate,skip_translate]:判断是否需要翻译# 如果目标市场使用英语跳过翻译步骤ifstate[target_market]in[US,UK,AU]:returnskip_translatereturntranslatedefshould_continue(state:ListingState)-Literal[continue,end]:判断是否继续ifstate.get(status)failed:returnendreturncontinue3.5 构建工作流图有了节点和路由条件就可以组装成完整的执行图# 创建状态图workflowStateGraph(ListingState)# 添加节点workflow.add_node(extract,extract_product_info)workflow.add_node(translate,translate_content)workflow.add_node(match_category,match_category)workflow.add_node(suggest_price,suggest_price)workflow.add_node(generate_listing,generate_listing)# 设置入口workflow.set_entry_point(extract)# 顺序边workflow.add_edge(extract,match_category)workflow.add_edge(match_category,suggest_price)workflow.add_edge(suggest_price,generate_listing)# 条件边翻译步骤条件执行workflow.add_conditional_edges(extract,should_translate,{translate:translate,skip_translate:match_category})# 翻译后回到主流程workflow.add_edge(translate,match_category)# 结束边workflow.add_edge(generate_listing,END)# 编译为可执行应用appworkflow.compile()3.6 测试运行defrun_listing_workflow():执行商品上架工作流initial_state{product_data:{name:TechGear 27-inch 4K Monitor,specs:{resolution:3840x2160,refresh_rate:144Hz},price:299.99,description:Professional-grade monitor for designers and gamers},target_market:DE,# 德国市场触发翻译current_step:start,errors:[],retry_count:0,status:pending}print(*60)print( 启动商品上架工作流)print(f目标市场:{initial_state[target_market]})print(*60)resultapp.invoke(initial_state)print(\n*60)print(✅ 工作流执行完成)print(*60)print(f状态:{result.get(status)})print(f类目:{result.get(category_match)})print(f定价建议: ${result.get(price_suggestion, {}).get(suggested_price)})print(f\n Listing草稿:\n{result.get(listing_draft)})returnresultif__name____main__:run_listing_workflow()四、高级编排能力4.1 并行执行LangGraph支持在同一个图中实现并行扇出。例如“同时查询价格、库存、评分”这三个独立任务fromlanggraph.graphimportStateGraph# 三个独立的查询节点互不依赖workflow.add_node(query_price,query_price)workflow.add_node(query_stock,query_stock)workflow.add_node(query_rating,query_rating)# 从同一个节点扇出workflow.add_edge(start_parallel,query_price)workflow.add_edge(start_parallel,query_stock)workflow.add_edge(start_parallel,query_rating)# 汇总节点等待所有完成workflow.add_edge(query_price,aggregate)workflow.add_edge(query_stock,aggregate)workflow.add_edge(query_rating,aggregate)4.2 人工介入Human-in-the-Loop在电商场景中某些操作需要人工审批——例如退款超过阈值、价格调整、敏感操作。LangGraph支持在任意节点暂停等待人工输入fromlanggraph.checkpointimportMemorySaver# 在定价节点后添加审批暂停点defhuman_approval_node(state:ListingState)-dict:人工审批节点——实际生产环境中推送至Slack/钉钉print(⏸️ 需要人工审批定价建议...)print(f建议价格: ${state[price_suggestion][suggested_price]})print(等待审批...)# 模拟人工审批通过# 生产中会通过外部接口接收审批结果approvedTrueifapproved:return{status:approved}else:return{status:rejected,errors:[定价被人工拒绝]}# 使用MemorySaver支持暂停/恢复fromlanggraph.graphimportStateGraph memoryMemorySaver()appworkflow.compile(checkpointermemory)# 执行时遇到人工节点会自动暂停等待外部输入4.3 状态持久化与恢复生产环境中长流程可能执行数分钟甚至数小时。LangGraph支持检查点机制允许在任意节点中断后从断点恢复# 使用SQLite持久化状态取代MemorySaverfromlanggraph.checkpoint.sqliteimportSqliteSaverwithSqliteSaver.from_conn_string(checkpoints.db)assaver:appworkflow.compile(checkpointersaver)# 执行...# 中断后使用相同的thread_id恢复五、工程化要点5.1 错误处理与重试生产环境中节点执行可能失败。LangGraph支持在节点级别配置重试策略对于临时性失败如API超时配置指数退避重试对于业务逻辑错误如数据格式不符记录错误并转入人工处理路径。5.2 可观测性每个节点的输入输出、耗时、成功/失败状态都应被记录。可以与LangSmith集成获得完整的执行追踪可视化。5.3 分布式部署考虑如果Worker Agent需要独立扩展可以将编排器与工作者解耦为独立服务。AWS AgentCore Runtime等平台支持将Agent部署为独立容器通过消息队列SQS进行任务分发实现并行扩展。六、小结本文围绕AI Agent的任务编排系统介绍了五种核心编排模式顺序、路由、并行、编排器-工作者、评估器-优化器各适用不同场景LangGraph实战基于状态图构建跨境电商商品上架工作流覆盖提取→匹配→定价→生成全流程高级能力并行执行、人工介入审批、状态持久化与恢复工程化要点错误处理、可观测性、分布式部署任务编排的价值在于将多个AI能力从“各自为战”变为“协同作战”。当一个业务诉求需要10个Agent协作完成时工作流编排决定了这个系统是“优雅地跑起来”还是“崩溃地卡住”。关于多智能体协作的更深层次模式、子图嵌套与动态编排欢迎在评论区交流。
