2026/8/11 7:42:12
FEATURED · 精选文章 
一、引言为什么单轮对话已不够用我在上篇文章中搭建了一个基于LangChain的跨境AI智能体原型用RAG解决了知识检索问题。但很快就会发现一个现实困境真实的业务场景很少是“一问一答”就能搞定的。想象一下这个场景——一个跨境卖家收到客户邮件“我买的显示器到手有坏点想退货但我已经把包装扔了。”如果要让AI处理这件事它至少需要判断意图是“退货”还是“投诉”查询订单状态和客户历史信用根据退货政策和金额决定是否可自动处理起草回复邮件如果金额超标或政策例外转人工审批。传统的ReAct Agent做这件事很容易翻车——在工具调用之间反复横跳、上下文越滚越大、逻辑混乱甚至陷入死循环。这正是Agentic Workflow智能体工作流要解决的问题。吴恩达曾在多个场合强调2024年以后AI应用的关键趋势不是模型本身更强而是让模型在可控的流程中工作。本文将深入剖析AI Agent工作流的三大核心机制——任务拆解、工具调用、人工审核闭环并用LangGraph从零构建一个带完整兜底机制的跨境电商客服工单处理系统。二、核心机制一任务拆解2.1 为什么需要显式拆解如果让一个Agent直接面对“处理退款工单”这样的复杂目标它会怎么做通常是一次性把所有工具都塞进上下文→让模型自己决定先查什么后查什么→结果经常跑偏。问题出在模型擅长推理但不擅长流程管理。更好的做法是把一个复杂任务拆成多个有明确边界的小任务节点让确定性逻辑来控制流程走向让LLM只在需要判断和生成的节点发挥作用。这能带来几个好处Token消耗大幅降低每个节点只看自己需要的上下文失败可以局部重试而不是全部重来流程可观测、可审计、可修改。2.2 任务拆解的设计原则对于“处理工单”这个场景我们可以拆解为5个步骤意图分类判断是退款、技术咨询还是垃圾邮件数据查询根据分类结果查询对应业务系统决策判断根据数据与规则决定是否可自动处理内容生成起草回复邮件人工审核条件触发高风险或边界情况转人工这种“分类→查询→判断→生成→审核”的流水线模式本质上是将LLM的推理能力与确定性业务逻辑组合起来。三、核心机制二工具调用3.1 工具的定义与权限控制在Agent工作流中每个节点可以调用自己所需的一组工具。关键在于权限最小化原则——不要把全部工具暴露给整个流程而是按节点精细化授权。跨境电商场景下常见的工具包括查询订单状态对接电商平台API或数据库查询客户历史信用等级、退款次数查询退货政策RAG检索知识库生成回复邮件草稿3.2 工具调用的三种模式根据不同场景工具调用可以设计为三种模式模式适用场景特点自动执行低风险、确定性操作无需人工介入直接放行需审批执行涉及资金、敏感数据触发Human-in-the-Loop暂停点拒绝执行高风险、违规操作直接拒绝并记录原因核心原则只读操作自动放行写入操作需审批。这种“只读自动放行高危写入拦截”的组合在企业级部署中几乎是标配。四、核心机制三人工审核闭环4.1 为什么必须有人工介入无论LLM多强大在某些场景下它都不能替代人类的判断金额超过某个阈值的退款申请涉及法律合规的内容首次出现的、模型置信度低于阈值的caseHuman-in-the-LoopHITL机制的核心功能是在Agent执行到关键节点时暂停将当前上下文和决策建议推送给人工等待人工审批后再继续执行。4.2 闭环学习机制更完善的系统还应该具备学习闭环人工修改Agent生成的回复后修改内容和理由会被记录下来用作后续Few-shot示例持续优化Agent的判断能力。这样每一次人工介入都在为系统积累经验而不是一次性的干预。五、实战用LangGraph构建完整的Agent工作流现在来看完整的代码实现。我们将构建一个“跨境电商客诉工单自动化处理系统”包含以上全部三个核心机制。5.1 环境准备pipinstalllanggraph langchain-openai langchain-core python-dotenv环境变量配置.env文件OPENAI_API_KEYsk-xxxxx OPENAI_BASE_URLhttps://api.deepseek.com/v1 # 或使用DeepSeek等国内服务5.2 完整代码实现importosimportjsonfromtypingimportTypedDict,Literalfromdotenvimportload_dotenvfromlangchain_openaiimportChatOpenAIfromlangchain_core.messagesimportHumanMessage,SystemMessagefromlanggraph.graphimportStateGraph,END load_dotenv()# # 1. 定义工作流状态 (State)# classTicketState(TypedDict):在节点之间传递的全局状态ticket_content:str# 原始工单内容category:str# 分类结果order_id:str# 订单IDorder_amount:float# 订单金额customer_tier:str# 客户等级: gold/silver/bronzerefund_count:int# 历史退款次数ai_draft_reply:str# AI生成的回复草稿needs_human_review:bool# 是否需要人工审核human_review_decision:str# 人工决策: approve/reject/modifyfinal_response:str# 最终回复内容confidence_score:float# 模型置信度# # 2. 初始化LLM# llmChatOpenAI(modelgpt-4o-mini,temperature0)# # 3. 定义业务节点 (Nodes)# defclassify_intent(state:TicketState)-dict:节点1意图分类print(⚙️ [节点1] 正在分析工单意图...)promptf 请对以下客户工单进行分类只能从以下选项中选择一个 - refund: 退款/退货请求 - tech_support: 技术咨询/产品问题 - spam: 垃圾邮件/无关内容 只输出分类英文单词不要输出其他内容。 工单内容{state[ticket_content]}responsellm.invoke([HumanMessage(contentprompt)])categoryresponse.content.strip().lower()print(f✅ 分类结果:{category})return{category:category}deffetch_order_data(state:TicketState)-dict:节点2查询订单数据模拟对接电商平台APIprint(⚙️ [节点2] 正在查询订单信息...)# 模拟从Amazon/Walmart API或ERP数据库查询# 实际生产环境替换为真实API调用mock_orders{ORD-001:{amount:299.00,customer_tier:gold,refund_count:0},ORD-002:{amount:599.00,customer_tier:silver,refund_count:2},ORD-003:{amount:49.99,customer_tier:bronze,refund_count:1},}# 从工单中提取订单ID模拟order_idORD-001# 实际应通过NER或正则提取order_datamock_orders.get(order_id,{amount:0,customer_tier:unknown,refund_count:0})print(f✅ 订单查询完成: 金额${order_data[amount]}, 等级{order_data[customer_tier]})return{order_id:order_id,order_amount:order_data[amount],customer_tier:order_data[customer_tier],refund_count:order_data[refund_count],}defevaluate_and_route(state:TicketState)-Literal[draft_reply,human_review]:节点3决策判断 条件路由print(⚙️ [节点3] 正在评估处理策略...)amountstate[order_amount]tierstate[customer_tier]refund_countstate[refund_count]# 确定性规则哪些情况需要人工介入needs_humanFalsereason# 规则1金额超过$500需要人工审批ifamount500:needs_humanTruereasonf订单金额${amount}超过$500审批阈值# 规则2历史退款超过3次需要人工审核elifrefund_count3:needs_humanTruereasonf该客户历史退款{refund_count}次超过3次上限# 规则3非黄金会员且退款金额超过$200eliftier!goldandamount200:needs_humanTruereasonf{tier}等级会员退款金额${amount}超过$200# 规则4置信度检查用LLM判断是否有歧义else:promptf 以下工单是否需要人工介入请判断是否存在模糊信息、潜在风险或政策例外。 只回答 yes 或 no。 工单{state[ticket_content]}订单金额${amount}客户等级{tier}responsellm.invoke([HumanMessage(contentprompt)])ifyesinresponse.content.strip().lower():needs_humanTruereasonLLM判断工单存在潜在风险或歧义ifneeds_human:print(f⚠️ 需要人工介入:{reason})returnhuman_reviewelse:print(✅ 自动处理进入草稿生成阶段)returndraft_replydefgenerate_draft(state:TicketState)-dict:节点4a生成回复草稿自动处理路径print(⚙️ [节点4] 正在生成AI回复草稿...)promptf 你是一位专业的跨境电商客服。请根据以下信息生成一封友好的回复邮件草稿。 工单内容{state[ticket_content]}订单金额${state[order_amount]}客户等级{state[customer_tier]}要求 1. 语言专业、礼貌、富有同理心 2. 明确说明处理方案 3. 格式为标准的英文商务邮件 responsellm.invoke([HumanMessage(contentprompt)])draftresponse.contentprint(✅ 回复草稿生成完成)return{ai_draft_reply:draft,needs_human_review:False,confidence_score:0.9}defhuman_review(state:TicketState)-dict:节点4b人工审核需要人工介入的路径print(⛔ [节点4] 进入人工审核流程...)print(f 工单内容:{state[ticket_content][:100]}...)print(f 订单金额: ${state[order_amount]})print(f 客户等级:{state[customer_tier]})# 在实际系统中这里会# 1. 将请求推送到人工审核队列如Slack、钉钉、内部审批系统# 2. 等待人工决策approve/reject/modify# 3. 将人工决策结果写回状态# 模拟人工审批通过human_decisionapprove# 实际从外部输入获得human_response Dear Customer, We have reviewed your refund request for order ORD-001. We are pleased to inform you that your refund has been approved. The amount of $299.00 will be credited back to your original payment method within 5-7 business days. We apologize for the inconvenience and appreciate your understanding. Best regards, Customer Service Team print(✅ 人工审核完成决策: approve)return{needs_human_review:True,human_review_decision:human_decision,final_response:human_response,confidence_score:1.0,}deffinalize_response(state:TicketState)-dict:节点5生成最终回复print(⚙️ [节点5] 生成最终回复...)ifstate.get(human_review_decision)approve:# 使用人工审核的回复finalstate[final_response]else:# 使用AI生成的草稿finalstate[ai_draft_reply]# 如果内容为空生成兜底回复ifnotfinal:finalWe have received your request and will get back to you within 24 hours.print(✅ 最终回复已就绪)return{final_response:final}# # 4. 构建工作流图 (Graph)# defbuild_workflow():使用LangGraph构建完整工作流# 创建状态图workflowStateGraph(TicketState)# 添加节点workflow.add_node(classify,classify_intent)workflow.add_node(fetch_order,fetch_order_data)workflow.add_node(evaluate,evaluate_and_route)# 注意这是条件路由函数也是节点workflow.add_node(draft_reply,generate_draft)workflow.add_node(human_review,human_review)workflow.add_node(finalize,finalize_response)# 定义边流程走向workflow.set_entry_point(classify)workflow.add_edge(classify,fetch_order)workflow.add_edge(fetch_order,evaluate)# 条件边根据评估结果走不同路径workflow.add_conditional_edges(evaluate,lambdastate:state.get(route,draft_reply),# 返回下一个节点名称{draft_reply:draft_reply,human_review:human_review,})# 两条路径汇合到最终节点workflow.add_edge(draft_reply,finalize)workflow.add_edge(human_review,finalize)# 结束workflow.add_edge(finalize,END)# 编译为可执行图returnworkflow.compile()# # 5. 运行与测试# defmain():# 构建工作流appbuild_workflow()# 模拟用户输入test_ticket Dear Support Team, I ordered a 4K monitor (Order #ORD-001) from your store last week. Unfortunately, there are several dead pixels on the screen. I would like to request a refund or replacement. Please let me know how to proceed. Thanks, John # 初始化状态initial_state{ticket_content:test_ticket,category:,order_id:,order_amount:0.0,customer_tier:,refund_count:0,ai_draft_reply:,needs_human_review:False,human_review_decision:,final_response:,confidence_score:0.0,route:# 用于条件路由的临时字段}print(*60)print( 开始执行智能工单处理工作流)print(*60)# 执行工作流final_stateapp.invoke(initial_state)print(\n*60)print(✅ 工作流执行完成)print(*60)print(f\n 分类结果:{final_state[category]})print(f 订单金额: ${final_state[order_amount]})print(f 客户等级:{final_state[customer_tier]})print(f 是否需要人工审核:{final_state[needs_human_review]})print(f\n 最终回复:\n{final_state[final_response]})if__name____main__:main()5.3 代码解读这段代码展示了工作流的三个核心机制任务拆解通过TicketState定义节点间共享的状态将“处理工单”这个大任务拆成了classify→fetch_order→evaluate→(draft_reply或human_review)→finalize五个明确的小任务。条件路由evaluate_and_route函数既是一个节点也是一个路由器。它根据金额、客户等级、历史退款次数等确定性规则判断走哪条路——这是企业级AI与纯LLM应用的关键区别用确定性逻辑控制流程让LLM只在需要的地方发挥作用。人工审核闭环human_review节点模拟了人工介入。在生产环境中它会把工单推送到Slack、钉钉或内部审批系统等待人工决策后恢复执行。人工修改的内容还可被记录为Few-shot示例持续优化Agent的后续表现。六、工程化建议6.1 可观测性在生产环境中务必为每个节点添加日志追踪记录每个节点的输入输出、耗时、Token消耗、是否命中规则。LangGraph天然支持与LangSmith集成可以可视化整个流程图。6.2 失败恢复考虑某次执行在第4步失败的情况。基于LangGraph的状态管理你可以从失败节点恢复执行已完成的节点不用重跑。这对长流程来说尤为重要。6.3 权限管理根据节点收窄工具权限classify节点只需要文本分析能力fetch_order节点需要数据库查询权限但不需要写权限human_review节点需要写权限但只针对审批结果字段。权限最小化能有效降低误操作风险。七、小结本文从“单轮对话不够用”这个痛点出发深入剖析了AI Agent工作流的三大核心机制任务拆解用有向图定义明确节点让流程可预测、可恢复工具调用按节点收窄权限只读操作自动放行写入操作严格管控人工审核闭环在关键节点引入Human-in-the-Loop让判断与规则结合通过LangGraph的完整代码示例我们构建了一个可运行的跨境电商工单处理系统。这套架构的核心思想——“让确定性逻辑控场让LLM在需要的地方发光”——是任何企业级AI应用落地的关键。下一步你可以在此基础上接入真实的电商API、完善审批仪表板、建立基于人工修正的学习闭环。关于多智能体协作、流程版本管理等进阶话题欢迎在评论区交流。
AI Agent工作流深度剖析：任务拆解、工具调用与人工审核闭环机制
一、引言为什么单轮对话已不够用我在上篇文章中搭建了一个基于LangChain的跨境AI智能体原型用RAG解决了知识检索问题。但很快就会发现一个现实困境真实的业务场景很少是“一问一答”就能搞定的。想象一下这个场景——一个跨境卖家收到客户邮件“我买的显示器到手有坏点想退货但我已经把包装扔了。”如果要让AI处理这件事它至少需要判断意图是“退货”还是“投诉”查询订单状态和客户历史信用根据退货政策和金额决定是否可自动处理起草回复邮件如果金额超标或政策例外转人工审批。传统的ReAct Agent做这件事很容易翻车——在工具调用之间反复横跳、上下文越滚越大、逻辑混乱甚至陷入死循环。这正是Agentic Workflow智能体工作流要解决的问题。吴恩达曾在多个场合强调2024年以后AI应用的关键趋势不是模型本身更强而是让模型在可控的流程中工作。本文将深入剖析AI Agent工作流的三大核心机制——任务拆解、工具调用、人工审核闭环并用LangGraph从零构建一个带完整兜底机制的跨境电商客服工单处理系统。二、核心机制一任务拆解2.1 为什么需要显式拆解如果让一个Agent直接面对“处理退款工单”这样的复杂目标它会怎么做通常是一次性把所有工具都塞进上下文→让模型自己决定先查什么后查什么→结果经常跑偏。问题出在模型擅长推理但不擅长流程管理。更好的做法是把一个复杂任务拆成多个有明确边界的小任务节点让确定性逻辑来控制流程走向让LLM只在需要判断和生成的节点发挥作用。这能带来几个好处Token消耗大幅降低每个节点只看自己需要的上下文失败可以局部重试而不是全部重来流程可观测、可审计、可修改。2.2 任务拆解的设计原则对于“处理工单”这个场景我们可以拆解为5个步骤意图分类判断是退款、技术咨询还是垃圾邮件数据查询根据分类结果查询对应业务系统决策判断根据数据与规则决定是否可自动处理内容生成起草回复邮件人工审核条件触发高风险或边界情况转人工这种“分类→查询→判断→生成→审核”的流水线模式本质上是将LLM的推理能力与确定性业务逻辑组合起来。三、核心机制二工具调用3.1 工具的定义与权限控制在Agent工作流中每个节点可以调用自己所需的一组工具。关键在于权限最小化原则——不要把全部工具暴露给整个流程而是按节点精细化授权。跨境电商场景下常见的工具包括查询订单状态对接电商平台API或数据库查询客户历史信用等级、退款次数查询退货政策RAG检索知识库生成回复邮件草稿3.2 工具调用的三种模式根据不同场景工具调用可以设计为三种模式模式适用场景特点自动执行低风险、确定性操作无需人工介入直接放行需审批执行涉及资金、敏感数据触发Human-in-the-Loop暂停点拒绝执行高风险、违规操作直接拒绝并记录原因核心原则只读操作自动放行写入操作需审批。这种“只读自动放行高危写入拦截”的组合在企业级部署中几乎是标配。四、核心机制三人工审核闭环4.1 为什么必须有人工介入无论LLM多强大在某些场景下它都不能替代人类的判断金额超过某个阈值的退款申请涉及法律合规的内容首次出现的、模型置信度低于阈值的caseHuman-in-the-LoopHITL机制的核心功能是在Agent执行到关键节点时暂停将当前上下文和决策建议推送给人工等待人工审批后再继续执行。4.2 闭环学习机制更完善的系统还应该具备学习闭环人工修改Agent生成的回复后修改内容和理由会被记录下来用作后续Few-shot示例持续优化Agent的判断能力。这样每一次人工介入都在为系统积累经验而不是一次性的干预。五、实战用LangGraph构建完整的Agent工作流现在来看完整的代码实现。我们将构建一个“跨境电商客诉工单自动化处理系统”包含以上全部三个核心机制。5.1 环境准备pipinstalllanggraph langchain-openai langchain-core python-dotenv环境变量配置.env文件OPENAI_API_KEYsk-xxxxx OPENAI_BASE_URLhttps://api.deepseek.com/v1 # 或使用DeepSeek等国内服务5.2 完整代码实现importosimportjsonfromtypingimportTypedDict,Literalfromdotenvimportload_dotenvfromlangchain_openaiimportChatOpenAIfromlangchain_core.messagesimportHumanMessage,SystemMessagefromlanggraph.graphimportStateGraph,END load_dotenv()# # 1. 定义工作流状态 (State)# classTicketState(TypedDict):在节点之间传递的全局状态ticket_content:str# 原始工单内容category:str# 分类结果order_id:str# 订单IDorder_amount:float# 订单金额customer_tier:str# 客户等级: gold/silver/bronzerefund_count:int# 历史退款次数ai_draft_reply:str# AI生成的回复草稿needs_human_review:bool# 是否需要人工审核human_review_decision:str# 人工决策: approve/reject/modifyfinal_response:str# 最终回复内容confidence_score:float# 模型置信度# # 2. 初始化LLM# llmChatOpenAI(modelgpt-4o-mini,temperature0)# # 3. 定义业务节点 (Nodes)# defclassify_intent(state:TicketState)-dict:节点1意图分类print(⚙️ [节点1] 正在分析工单意图...)promptf 请对以下客户工单进行分类只能从以下选项中选择一个 - refund: 退款/退货请求 - tech_support: 技术咨询/产品问题 - spam: 垃圾邮件/无关内容 只输出分类英文单词不要输出其他内容。 工单内容{state[ticket_content]}responsellm.invoke([HumanMessage(contentprompt)])categoryresponse.content.strip().lower()print(f✅ 分类结果:{category})return{category:category}deffetch_order_data(state:TicketState)-dict:节点2查询订单数据模拟对接电商平台APIprint(⚙️ [节点2] 正在查询订单信息...)# 模拟从Amazon/Walmart API或ERP数据库查询# 实际生产环境替换为真实API调用mock_orders{ORD-001:{amount:299.00,customer_tier:gold,refund_count:0},ORD-002:{amount:599.00,customer_tier:silver,refund_count:2},ORD-003:{amount:49.99,customer_tier:bronze,refund_count:1},}# 从工单中提取订单ID模拟order_idORD-001# 实际应通过NER或正则提取order_datamock_orders.get(order_id,{amount:0,customer_tier:unknown,refund_count:0})print(f✅ 订单查询完成: 金额${order_data[amount]}, 等级{order_data[customer_tier]})return{order_id:order_id,order_amount:order_data[amount],customer_tier:order_data[customer_tier],refund_count:order_data[refund_count],}defevaluate_and_route(state:TicketState)-Literal[draft_reply,human_review]:节点3决策判断 条件路由print(⚙️ [节点3] 正在评估处理策略...)amountstate[order_amount]tierstate[customer_tier]refund_countstate[refund_count]# 确定性规则哪些情况需要人工介入needs_humanFalsereason# 规则1金额超过$500需要人工审批ifamount500:needs_humanTruereasonf订单金额${amount}超过$500审批阈值# 规则2历史退款超过3次需要人工审核elifrefund_count3:needs_humanTruereasonf该客户历史退款{refund_count}次超过3次上限# 规则3非黄金会员且退款金额超过$200eliftier!goldandamount200:needs_humanTruereasonf{tier}等级会员退款金额${amount}超过$200# 规则4置信度检查用LLM判断是否有歧义else:promptf 以下工单是否需要人工介入请判断是否存在模糊信息、潜在风险或政策例外。 只回答 yes 或 no。 工单{state[ticket_content]}订单金额${amount}客户等级{tier}responsellm.invoke([HumanMessage(contentprompt)])ifyesinresponse.content.strip().lower():needs_humanTruereasonLLM判断工单存在潜在风险或歧义ifneeds_human:print(f⚠️ 需要人工介入:{reason})returnhuman_reviewelse:print(✅ 自动处理进入草稿生成阶段)returndraft_replydefgenerate_draft(state:TicketState)-dict:节点4a生成回复草稿自动处理路径print(⚙️ [节点4] 正在生成AI回复草稿...)promptf 你是一位专业的跨境电商客服。请根据以下信息生成一封友好的回复邮件草稿。 工单内容{state[ticket_content]}订单金额${state[order_amount]}客户等级{state[customer_tier]}要求 1. 语言专业、礼貌、富有同理心 2. 明确说明处理方案 3. 格式为标准的英文商务邮件 responsellm.invoke([HumanMessage(contentprompt)])draftresponse.contentprint(✅ 回复草稿生成完成)return{ai_draft_reply:draft,needs_human_review:False,confidence_score:0.9}defhuman_review(state:TicketState)-dict:节点4b人工审核需要人工介入的路径print(⛔ [节点4] 进入人工审核流程...)print(f 工单内容:{state[ticket_content][:100]}...)print(f 订单金额: ${state[order_amount]})print(f 客户等级:{state[customer_tier]})# 在实际系统中这里会# 1. 将请求推送到人工审核队列如Slack、钉钉、内部审批系统# 2. 等待人工决策approve/reject/modify# 3. 将人工决策结果写回状态# 模拟人工审批通过human_decisionapprove# 实际从外部输入获得human_response Dear Customer, We have reviewed your refund request for order ORD-001. We are pleased to inform you that your refund has been approved. The amount of $299.00 will be credited back to your original payment method within 5-7 business days. We apologize for the inconvenience and appreciate your understanding. Best regards, Customer Service Team print(✅ 人工审核完成决策: approve)return{needs_human_review:True,human_review_decision:human_decision,final_response:human_response,confidence_score:1.0,}deffinalize_response(state:TicketState)-dict:节点5生成最终回复print(⚙️ [节点5] 生成最终回复...)ifstate.get(human_review_decision)approve:# 使用人工审核的回复finalstate[final_response]else:# 使用AI生成的草稿finalstate[ai_draft_reply]# 如果内容为空生成兜底回复ifnotfinal:finalWe have received your request and will get back to you within 24 hours.print(✅ 最终回复已就绪)return{final_response:final}# # 4. 构建工作流图 (Graph)# defbuild_workflow():使用LangGraph构建完整工作流# 创建状态图workflowStateGraph(TicketState)# 添加节点workflow.add_node(classify,classify_intent)workflow.add_node(fetch_order,fetch_order_data)workflow.add_node(evaluate,evaluate_and_route)# 注意这是条件路由函数也是节点workflow.add_node(draft_reply,generate_draft)workflow.add_node(human_review,human_review)workflow.add_node(finalize,finalize_response)# 定义边流程走向workflow.set_entry_point(classify)workflow.add_edge(classify,fetch_order)workflow.add_edge(fetch_order,evaluate)# 条件边根据评估结果走不同路径workflow.add_conditional_edges(evaluate,lambdastate:state.get(route,draft_reply),# 返回下一个节点名称{draft_reply:draft_reply,human_review:human_review,})# 两条路径汇合到最终节点workflow.add_edge(draft_reply,finalize)workflow.add_edge(human_review,finalize)# 结束workflow.add_edge(finalize,END)# 编译为可执行图returnworkflow.compile()# # 5. 运行与测试# defmain():# 构建工作流appbuild_workflow()# 模拟用户输入test_ticket Dear Support Team, I ordered a 4K monitor (Order #ORD-001) from your store last week. Unfortunately, there are several dead pixels on the screen. I would like to request a refund or replacement. Please let me know how to proceed. Thanks, John # 初始化状态initial_state{ticket_content:test_ticket,category:,order_id:,order_amount:0.0,customer_tier:,refund_count:0,ai_draft_reply:,needs_human_review:False,human_review_decision:,final_response:,confidence_score:0.0,route:# 用于条件路由的临时字段}print(*60)print( 开始执行智能工单处理工作流)print(*60)# 执行工作流final_stateapp.invoke(initial_state)print(\n*60)print(✅ 工作流执行完成)print(*60)print(f\n 分类结果:{final_state[category]})print(f 订单金额: ${final_state[order_amount]})print(f 客户等级:{final_state[customer_tier]})print(f 是否需要人工审核:{final_state[needs_human_review]})print(f\n 最终回复:\n{final_state[final_response]})if__name____main__:main()5.3 代码解读这段代码展示了工作流的三个核心机制任务拆解通过TicketState定义节点间共享的状态将“处理工单”这个大任务拆成了classify→fetch_order→evaluate→(draft_reply或human_review)→finalize五个明确的小任务。条件路由evaluate_and_route函数既是一个节点也是一个路由器。它根据金额、客户等级、历史退款次数等确定性规则判断走哪条路——这是企业级AI与纯LLM应用的关键区别用确定性逻辑控制流程让LLM只在需要的地方发挥作用。人工审核闭环human_review节点模拟了人工介入。在生产环境中它会把工单推送到Slack、钉钉或内部审批系统等待人工决策后恢复执行。人工修改的内容还可被记录为Few-shot示例持续优化Agent的后续表现。六、工程化建议6.1 可观测性在生产环境中务必为每个节点添加日志追踪记录每个节点的输入输出、耗时、Token消耗、是否命中规则。LangGraph天然支持与LangSmith集成可以可视化整个流程图。6.2 失败恢复考虑某次执行在第4步失败的情况。基于LangGraph的状态管理你可以从失败节点恢复执行已完成的节点不用重跑。这对长流程来说尤为重要。6.3 权限管理根据节点收窄工具权限classify节点只需要文本分析能力fetch_order节点需要数据库查询权限但不需要写权限human_review节点需要写权限但只针对审批结果字段。权限最小化能有效降低误操作风险。七、小结本文从“单轮对话不够用”这个痛点出发深入剖析了AI Agent工作流的三大核心机制任务拆解用有向图定义明确节点让流程可预测、可恢复工具调用按节点收窄权限只读操作自动放行写入操作严格管控人工审核闭环在关键节点引入Human-in-the-Loop让判断与规则结合通过LangGraph的完整代码示例我们构建了一个可运行的跨境电商工单处理系统。这套架构的核心思想——“让确定性逻辑控场让LLM在需要的地方发光”——是任何企业级AI应用落地的关键。下一步你可以在此基础上接入真实的电商API、完善审批仪表板、建立基于人工修正的学习闭环。关于多智能体协作、流程版本管理等进阶话题欢迎在评论区交流。 
