引言：从“单点对话”走向“复杂工作流（Agentic Workflow）”

在前两篇文章中，我们探讨了用 RAG 解决知识库问答，用 Agent 解决数据查询。但在真实的商业环境中，很多业务并不是“一问一答”就能解决的，而是

一条包含多个步骤、条件分支和异常处理的复杂流水线

例如：处理一个客户退款工单。AI 需要：

传统的 LangChain Agent（如 ReAct 模式）在处理这种“线性+条件分支”的复杂流程时，往往会陷入死循环或逻辑混乱。

吴恩达（Andrew Ng）曾指出：2024年以后的AI趋势是 Agentic Workflow（智能体工作流）。

今天，我们将使用目前企业级最火的编排框架 

“智能客诉工单自动化处理系统”

LangGraph 是 LangChain 团队推出的扩展库，它用

。企业不敢用 AI 的核心原因是“黑盒”，而 LangGraph 允许你像画流程图一样，把 AI 的思考过程变成白盒，并在关键节点强制让人类接管。

二、 实战演练：智能客诉工单处理系统

我们将构建一个系统，自动读取客户邮件，进行分类、查询模拟订单库，并根据“退款金额”决定是否拦截给人类主管审批。

pip install langgraph langchain-openai langchain-core python-dotenv


OPENAI_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxx
OPENAI_BASE_URL=https://api.openai.com/v1
# 同样兼容 DeepSeek、通义千问等国内大模型接口


import os
import json
from typing import TypedDict, Literal
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, END

# 加载环境变量
load_dotenv()

# ==========================================
# 1. 定义工作流的状态 (State)
# ==========================================
class TicketState(TypedDict):
    ticket_content: str       # 原始工单内容
    category: str             # 分类结果 (refund, tech_support, spam)
    order_amount: float       # 涉及的订单金额
    draft_reply: str          # AI 生成的回复草稿
    needs_human_review: bool  # 是否需要人工介入

# ==========================================
# 2. 定义各个业务节点 (Nodes)
# ==========================================
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def classify_ticket(state: TicketState):
    """节点1：意图分类"""
    print("⚙️ [节点1] 正在分析工单意图...")
    prompt = f"""
    请对以下客户工单进行分类。只能从 ['refund', 'tech_support', 'spam'] 中选择一个。
    只输出分类英文单词，不要输出其他内容。
    工单内容：{state['ticket_content']}
    """
    response = llm.invoke([HumanMessage(content=prompt)])
    category = response.content.strip().lower()
    return {"category": category}

def query_order_system(state: TicketState):
    """节点2：模拟查询 CRM/订单系统"""
    print("⚙️ [节点2] 正在查询订单系统...")
    # 模拟数据库查询逻辑
    if "退款" in state["ticket_content"] or "refund" in state["ticket_content"]:
        # 假设从数据库查出该订单金额为 899.0
        return {"order_amount": 899.0} 
    return {"order_amount": 0.0}

def draft_response(state: TicketState):
    """节点3：生成回复草稿"""
    print("⚙️ [节点3] 正在生成回复草稿...")
    prompt = f"""
    你是一名专业的客服。请根据工单内容和订单金额，写一封简短、礼貌的回复邮件草稿。
    工单内容：{state['ticket_content']}
    订单金额：{state['order_amount']}元
    """
    response = llm.invoke([HumanMessage(content=prompt)])
    return {"draft_reply": response.content}

def human_review_node(state: TicketState):
    """节点4：人工介入 (模拟断点)"""
    print("⚠️ [节点4] 触发高风险拦截，等待人工主管审批...")
    # 在真实的 LangGraph 部署中，这里会使用 interrupt() 暂停图执行，等待前端人类点击"同意"
    # 这里我们用 print 模拟人类审核通过
    print("👨‍💼 [人工主管] 已审核退款申请，批准通过。")
    return {"needs_human_review": False}

# ==========================================
# 3. 定义条件路由 (Conditional Edges)
# ==========================================
def route_ticket(state: TicketState) -> Literal["query_order_system", "draft_response", "end_spam"]:
    """根据分类结果决定下一步走向"""
    if state["category"] == "refund":
        return "query_order_system"
    elif state["category"] == "tech_support":
        return "draft_response" # 技术支持直接生成通用回复
    else:
        return "end_spam"       # 垃圾邮件直接结束

def check_refund_policy(state: TicketState) -> Literal["human_review_node", "draft_response"]:
    """企业核心风控：退款金额大于500元，必须人工介入"""
    if state["order_amount"] > 500.0:
        return "human_review_node"
    return "draft_response"

# ==========================================
# 4. 组装 LangGraph 工作流
# ==========================================
def build_workflow():
    workflow = StateGraph(TicketState)

    # 添加节点
    workflow.add_node("classify_ticket", classify_ticket)
    workflow.add_node("query_order_system", query_order_system)
    workflow.add_node("human_review_node", human_review_node)
    workflow.add_node("draft_response", draft_response)

    # 设置入口
    workflow.set_entry_point("classify_ticket")

    # 添加边 (路由)
    workflow.add_conditional_edges("classify_ticket", route_ticket, {
        "query_order_system": "query_order_system",
        "draft_response": "draft_response",
        "end_spam": END
    })
    
    workflow.add_conditional_edges("query_order_system", check_refund_policy, {
        "human_review_node": "human_review_node",
        "draft_response": "draft_response"
    })

    workflow.add_edge("human_review_node", "draft_response")
    workflow.add_edge("draft_response", END)

    return workflow.compile()

# ==========================================
# 5. 运行测试
# ==========================================
if __name__ == "__main__":
    app = build_workflow()
    
    # 模拟一封要求退款的客户邮件
    mock_ticket = "你好，我上周买的蓝牙耳机有严重杂音，要求立刻退款！订单号是 #9527。"
    
    print(f"📩 收到新工单: {mock_ticket}\n" + "="*50)
    
    # 执行工作流
    final_state = app.invoke({
        "ticket_content": mock_ticket,
        "category": "",
        "order_amount": 0.0,
        "draft_reply": "",
        "needs_human_review": True
    })otterly.cn  31225.t.kuaisou.com
    
    print("\n" + "="*50)
    print("✅ 工作流执行完毕！最终状态：")
    print(f"分类: {final_state['category']}")
    print(f"订单金额: {final_state['order_amount']}")
    print(f"是否经过人工审核: {'是' if 'human_review_node' in str(app.get_graph().nodes) else '否'}")
    print(f"\n📧 最终生成的回复草稿:\n{final_state['draft_reply']}")


当你运行这段代码时，控制台会清晰展示 AI 的“流水线”作业过程：

📩 收到新工单: 你好，我上周买的蓝牙耳机有严重杂音，要求立刻退款！订单号是 #9527。
==================================================
⚙️ [节点1] 正在分析工单意图...
⚙️ [节点2] 正在查询订单系统...
⚠️ [节点4] 触发高风险拦截，等待人工主管审批...
👨‍💼 [人工主管] 已审核退款申请，批准通过。
⚙️ [节点3] 正在生成回复草稿...

==================================================
✅ 工作流执行完毕！最终状态：
分类: refund
订单金额: 899.0

📧 最终生成的回复草稿:
尊敬的客户，您好！
非常抱歉您收到的蓝牙耳机存在杂音问题。我们已核实您的订单（#9527），金额899.0元的退款申请已获批准。退款将在3-5个工作日内原路退回...


：如果客户咨询的是“技术支持”或者“退款金额小于500元”，工作流会自动跳过 

，实现真正的自动化降本增效。

三、 企业级落地的 4 个“高阶玩法”

上面的代码展示了 LangGraph 的基础骨架，在真实的企业生产环境中，你需要掌握以下高阶特性：

1. 真正的 Human-in-the-loop（断点续传）

 模拟了人工审核。在真实的 Web 系统中，LangGraph 提供了 

app.compile(checkpointer=MemorySaver(), interrupt_before=["human_review_node"])

，将状态存入数据库。前端页面弹出审批框，主管点击“同意”后，系统通过 API 恢复（Resume）图的执行。

2. 引入 Checkpointer（状态持久化）

企业的长流程可能跨越好几天（如：等待客户提供补充材料）。 

。这样即使服务器重启，AI 也能记住“这个客户上周二提到了什么”，实现跨天的多轮复杂跟进。

3. 子图嵌套（Sub-graphs）处理复杂业务

真实企业的业务极其庞大，一个图无法塞下所有逻辑。 

：将“订单查询”、“物流追踪”、“财务退款”分别封装成独立的子图（Sub-graph）。主图就像一个“部门经理”，根据意图将任务分发给不同的“子部门（子图）”处理，实现代码的模块化和解耦。

4. 可观测性（Observability）与追踪

当 AI 出错时，如何排查是哪个节点出了问题？ 

。它们能记录 LangGraph 每一个节点的输入、输出、Token 消耗和延迟时间。你可以像看监控录像一样，回放 AI 处理某一个工单的完整思考链路。

从 RAG 到 Text-to-SQL，再到今天的 

，企业 AI 的落地正在从“玩具”走向“生产力工具”。

大模型本身只是一个“大脑”，而 

LangGraph 赋予了它“四肢”和“企业规章制度”

。通过状态图和条件路由，我们终于可以把不可控的 AI 生成，装进企业严谨的业务 SOP（标准作业程序）中。

打开你的代码编辑器，梳理一下你们公司每天耗时最长的人工流水线，用 LangGraph 把它画出来，这就是你为企业创造巨大商业价值的开始！

在前两篇文章中，我们探讨了用 RAG 解决知识库问答，用 Agent 解决数据查询。但在真实的商业环境中，很多业务并不是“一问一答”就能解决的，而是一条包含多个步骤、条件分支和异常处理的复杂流水线。

企业AI落地实战：用 LangGraph 构建“带人工介入”的自动化业务工作流

人工智能

编程语言

架构设计

本文介绍如何利用LangGraph框架构建智能客诉工单处理系统，实现AI自动化处理复杂业务工作流。通过节点、边和状态控制流程，结合Human-in-the-loop机制，确保企业级风控与效率。文章包含完整代码示例，解析退款工单处理流程，并分享企业落地的4个高阶玩法，如状态持久化、子图嵌套等，助力企业将AI融入严谨的业务SOP。

数据库

服务器

智能体

DeepSeek

路由

Agent

2026上云采购 | AI焕新·智启新局

lexiang

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

本文介绍如何利用LangGraph框架构建智能客诉工单处理系统，实现AI自动化处理复杂业务工作流。通过节点、边和状态控制流程，结合Human-in-the-loop机制，确保企业级风控与效率。文章包含完整代码示例，解析退款工单处理流程，并分享企业落地的4个高阶玩法，如状态持久化、子图嵌套等，助力企业将AI融入严谨的业务...

企业AI落地实战：用 LangGraph 构建“带人工介入”的自动化业务工作流

企业AI落地实战：用 LangGraph 构建“带人工介入”的自动化业务工作流

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐