LangGraph 状态机实战:3步构建复杂多步骤 AI 工作流(附完整代码)

阿里云上云采购季 2核2G云服务器99元/年 续费同价

如果说 LangChain 是”函数调用”,那 LangGraph 就是”状态机”。当你的 AI 工作流需要处理分支、循环、回滚、历史记忆时,LangGraph 就是你需要的武器。


背景:为什么需要 LangGraph?

先看一个真实场景——智能客服工作流:

用户:我要退订套餐

AI:
1. 验证用户身份(身份验证)
   ├─ 验证成功 → 进入套餐查询
   └─ 验证失败 → 转人工
2. 查询当前套餐(查询API)
   ├─ 是合约套餐 → 提示违约金
   └─ 非合约套餐 → 继续
3. 发起退订(退订API)
   ├─ 退订成功 → 发送确认短信
   └─ 退订失败 → 记录工单
4. 更新CRM(CRM API)
5. 发送满意度调查

这个流程:
– 有条件分支(合约套餐 vs 非合约)
– 有失败处理(退订失败要转人工)
– 有状态持久化(每一步的结果要传给下一步)
– 有历史记录(方便客服接管时查看)

用普通代码写,你会发现状态管理越来越乱。LangGraph 就是来解决这个问题的——用状态机的思路来管理 AI 工作流。


LangGraph 核心概念

与 LangChain 的关键区别

特性 LangChain LangGraph
执行模型 线性 Chain(A→B→C) 图结构(可分支、可循环)
状态管理 无内置状态 内置状态机 + 检查点
错误处理 需手动 try/catch 内置 conditional 路由
可视化 无 内置 Mermaid 可视化
适用场景 简单问答、单一工具调用 复杂多步骤对话、Agent

LangGraph 四件套

概念 说明 类比
StateGraph 图结构容器,定义所有节点和边 流程图编辑器
State 在节点间流转的共享数据对象 流程表单
Node 图中的一个个处理函数 流程图中的方框
Reducer 状态更新规则 表单填写规则

第1步:搭建基础状态机(智能客服核心)

# smart客服_workflow.py
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
import operator

# ========== 1. 定义 State(流转的数据结构)==========

class CustomerServiceState(TypedDict):
    """客服工作流的状态结构"""
    # messages 是消息历史(append模式的 reducer)
    messages: Annotated[list, add_messages]

    # 用户信息
    user_id: str
    phone: str

    # 业务状态
    identity_verified: bool
    current_plan: str | None
    is_contract_plan: bool
    cancellation_fee: float

    # 处理结果
    cancellation_success: bool
    ticket_id: str | None
    error_message: str | None

    # 流程控制
    next_step: str  # 标记下一步去哪
    escalation_needed: bool  # 是否需要转人工


# ========== 2. 定义 Node(处理函数)==========

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

llm = ChatOpenAI(model="gpt-4o", temperature=0.1)

def verify_identity(state: CustomerServiceState) -> CustomerServiceState:
    """节点1:身份验证"""
    print(f"🔐 正在验证用户身份: {state.get('user_id')}")

    # 模拟调用身份验证 API
    # 实际项目中这里应该是: api.verify_user(state['phone'], state['user_id'])
    verified = len(state.get('user_id', '')) > 3

    return {
        "identity_verified": verified,
        "next_step": "end_session" if not verified else "query_plan",
        "messages": [AIMessage(content="身份验证" + ("通过" if verified else "失败"))],
    }


def query_plan(state: CustomerServiceState) -> CustomerServiceState:
    """节点2:查询当前套餐"""
    print(f"📋 查询用户套餐: {state.get('user_id')}")

    # 模拟查询套餐
    # 实际: plan = api.get_user_plan(state['user_id'])
    current_plan = "企业版-年套餐"
    is_contract = "年" in current_plan  # 年套餐 = 合约套餐
    fee = 299.0 if is_contract else 0.0

    return {
        "current_plan": current_plan,
        "is_contract_plan": is_contract,
        "cancellation_fee": fee,
        "next_step": "handle_cancellation",
        "messages": [AIMessage(
            content=f"当前套餐:{current_plan},退订手续费:¥{fee}"
        )],
    }


def handle_cancellation(state: CustomerServiceState) -> CustomerServiceState:
    """节点3:处理退订逻辑"""
    print(f"🚪 发起退订...")

    if state.get('is_contract_plan'):
        return {
            "next_step": "end_session",
            "messages": [AIMessage(
                content="您当前是合约套餐,退订需要支付违约金 ¥299。"
                "如需继续,请回复「确认退订」。"
            )],
        }

    # 模拟退订 API 调用
    success = True
    if success:
        return {
            "cancellation_success": True,
            "next_step": "update_crm",
            "messages": [AIMessage(content="退订成功!正在为您办理后续手续...")],
        }
    else:
        return {
            "cancellation_success": False,
            "escalation_needed": True,
            "next_step": "create_ticket",
            "error_message": "退订接口超时",
            "messages": [AIMessage(content="系统繁忙,退订失败,已为您创建工单。")],
        }


def update_crm(state: CustomerServiceState) -> CustomerServiceState:
    """节点4:更新 CRM 系统"""
    print(f"📝 更新 CRM...")

    # 模拟 CRM 更新
    # api.update_user_status(user_id, "cancelled")

    return {
        "next_step": "send_confirmation",
        "messages": [AIMessage(content="CRM 系统已更新。")],
    }


def send_confirmation(state: CustomerServiceState) -> CustomerServiceState:
    """节点5:发送确认短信"""
    print(f"📱 发送退订确认短信...")

    # 模拟发送短信
    # sms.send(state['phone'], "您的套餐已退订成功...")

    return {
        "next_step": "end_session",
        "messages": [AIMessage(
            content="退订完成!确认短信已发送至您手机,请注意查收。"
            "如有疑问,请回复「人工」转接客服。"
        )],
    }


def create_ticket(state: CustomerServiceState) -> CustomerServiceState:
    """失败处理:创建工单"""
    print(f"🎫 创建转接工单...")

    import uuid
    ticket_id = f"TKT-{uuid.uuid4().hex[:8].upper()}"

    return {
        "ticket_id": ticket_id,
        "next_step": "end_session",
        "messages": [AIMessage(
            content=f"工单已创建:{ticket_id},人工客服将在24小时内联系您。"
        )],
    }


def end_session(state: CustomerServiceState) -> CustomerServiceState:
    """结束会话"""
    print(f"✅ 会话结束")
    return {"next_step": END}


# ========== 3. 构建 StateGraph ==========

def build_customer_service_graph():
    """构建客服工作流图"""

    # 创建图
    graph = StateGraph(CustomerServiceState)

    # 注册节点
    graph.add_node("verify_identity", verify_identity)
    graph.add_node("query_plan", query_plan)
    graph.add_node("handle_cancellation", handle_cancellation)
    graph.add_node("update_crm", update_crm)
    graph.add_node("send_confirmation", send_confirmation)
    graph.add_node("create_ticket", create_ticket)
    graph.add_node("end_session", end_session)

    # 设置入口点
    graph.set_entry_point("verify_identity")

    # 定义边(条件路由)
    graph.add_conditional_edges(
        "verify_identity",
        lambda state: state["next_step"],
        {
            "query_plan": "query_plan",
            "end_session": "end_session",
        }
    )

    graph.add_conditional_edges(
        "query_plan",
        lambda state: state["next_step"],
        {"handle_cancellation": "handle_cancellation"}
    )

    # 退订结果分支
    def cancellation_router(state: CustomerServiceState) -> str:
        if state.get("escalation_needed"):
            return "create_ticket"
        elif state.get("cancellation_success"):
            return "update_crm"
        return "end_session"

    graph.add_conditional_edges(
        "handle_cancellation",
        cancellation_router,
        {
            "update_crm": "update_crm",
            "create_ticket": "create_ticket",
            "end_session": "end_session",
        }
    )

    graph.add_conditional_edges(
        "update_crm",
        lambda state: state["next_step"],
        {"send_confirmation": "send_confirmation"}
    )

    graph.add_conditional_edges(
        "send_confirmation",
        lambda state: state["next_step"],
        {"end_session": "end_session"}
    )

    graph.add_conditional_edges(
        "create_ticket",
        lambda state: state["next_step"],
        {"end_session": "end_session"}
    )

    # 结束节点
    graph.add_edge("end_session", END)

    return graph.compile()


# ========== 4. 运行工作流 ==========

if __name__ == "__main__":
    # 编译图
    workflow = build_customer_service_graph()

    # 可视化(Mermaid 格式)
    print("📊 工作流图结构:")
    print(workflow.get_graph().draw_mermaid())

    # 初始状态
    initial_state = CustomerServiceState(
        user_id="user123",
        phone="138****8888",
        messages=[
            HumanMessage(content="我要退订套餐")
        ],
        identity_verified=False,
        current_plan=None,
        is_contract_plan=False,
        cancellation_fee=0.0,
        cancellation_success=False,
        ticket_id=None,
        error_message=None,
        next_step="",
        escalation_needed=False,
    )

    # 运行
    print("\n🚀 开始执行工作流:\n")
    result = workflow.invoke(initial_state)

    print("\n📜 执行结果:")
    print(f"- 退订成功: {result['cancellation_success']}")
    print(f"- 当前套餐: {result['current_plan']}")
    print(f"- 工单号: {result['ticket_id']}")

第2步:添加检查点(状态持久化与恢复)

上面那个例子,如果用户说到一半退出了,下次进来要从头开始,体验很差。LangGraph 的 Checkpoint 功能可以持久化状态:

# checkpoint_workflow.py
from langgraph.checkpoint.sqlite import SqliteSaver

# 使用 SQLite 作为检查点存储(生产环境用 PostgreSQL)
checkpointer = SqliteSaver.from_conn_string(":memory:")  # 测试用内存
# 生产环境: checkpointer = SqliteSaver("/data/checkpoints/customerservice.db")

workflow = build_customer_service_graph().checkpointer(checkpointer)

# 首次运行(新建会话)
config = {"configurable": {"thread_id": "session-001"}}

# 第一轮对话
result1 = workflow.invoke(
    {
        **initial_state,
        "messages": [HumanMessage(content="我要退订套餐")]
    },
    config=config
)
print(result1["messages"][-1].content)
# 输出: 退订成功!正在为您办理后续手续...

# 模拟用户退出后重新进入
print("\n--- 用户重新进入,查看历史状态 ---\n")

# 只传入新消息,保留历史
result2 = workflow.invoke(
    {"messages": [HumanMessage(content="好的,谢谢")], "user_id": "user123"},
    config=config  # 同一个 thread_id!
)

# LangGraph 会自动恢复之前的完整状态
print(result2["messages"][-1].content)
# 输出: 退订完成!确认短信已发送至您手机...

# 查看完整消息历史(包括之前的)
for msg in result2["messages"]:
    role = "用户" if isinstance(msg, HumanMessage) else "AI"
    print(f"{role}: {msg.content}")

支持多种检查点存储

# PostgreSQL(生产环境推荐)
from langgraph.checkpoint.postgres import PostgresSaver

checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@host:5432/langgraph"
)

# Redis(高并发场景)
from langgraph.checkpoint.redis import RedisSaver

checkpointer = RedisSaver.from_url("redis://localhost:6379")

# Memory(单进程测试)
checkpointer = MemorySaver()

第3步:构建 ReAct Agent(带工具调用的状态机)

最强大的用法——把状态机变成 ReAct Agent:

# react_agent_workflow.py
from typing import Literal
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

# ========== 1. 定义工具 ==========

@tool
def get_weather(city: str) -> str:
    """查询城市天气"""
    weather_data = {
        "北京": "晴,26°C",
        "上海": "多云,28°C",
        "深圳": "雷阵雨,30°C",
    }
    return weather_data.get(city, f"{city}的天气数据暂不可用")

@tool
def search_news(keyword: str) -> list[dict]:
    """搜索新闻"""
    # 实际项目中调用新闻 API
    return [
        {"title": f"{keyword}的最新进展", "source": "科技日报", "time": "2小时前"},
        {"title": f"{keyword}引发行业讨论", "source": "财经周刊", "time": "5小时前"},
    ]

@tool
def send_email(to: str, subject: str, body: str) -> dict:
    """发送邮件"""
    print(f"📧 发送邮件: {to}")
    print(f"   主题: {subject}")
    print(f"   内容: {body[:50]}...")
    return {"status": "sent", "message_id": "msg-123"}


# 工具列表
tools = [get_weather, search_news, send_email]

# ========== 2. 创建 ReAct Agent ==========

llm = ChatOpenAI(model="gpt-4o")
llm_with_tools = llm.bind_tools(tools)

def create_agentic_workflow():
    """创建带 Agent 能力的工作流"""

    # 使用预建的 ReAct Agent
    agent = create_react_agent(llm_with_tools, tools)

    # 包装成 StateGraph
    graph = StateGraph(AgentState)

    graph.add_node("agent", agent)
    graph.add_node("human", human_feedback_node)  # 人工审核节点

    graph.set_entry_point("agent")

    # 条件边:Agent 判断是否需要人工
    def should_escalate(state: AgentState) -> str:
        if state.get("needs_human"):
            return "human"
        return END

    graph.add_conditional_edges(
        "agent",
        should_escalate,
        {"human": "human", END: END}
    )

    graph.add_edge("human", "agent")  # 人工审核后继续

    return graph.compile()

# ========== 3. 完整状态定义 ==========

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    needs_human: bool
    final_response: str


def human_feedback_node(state: AgentState) -> AgentState:
    """人工审核节点(实际项目中这里会暂停等待人工确认)"""
    print("\n" + "="*50)
    print("📋 人工审核请求")
    print("="*50)
    print(f"最后一条消息: {state['messages'][-1].content}")
    print("\n[自动通过 - 演示模式]")

    return {
        "needs_human": False,
        "messages": state["messages"] + [
            AIMessage(content="[人工审核已通过]")
        ]
    }


# ========== 4. 运行 ==========

if __name__ == "__main__":
    workflow = create_agentic_workflow()

    print("📊 Agent 工作流图:")
    print(workflow.get_graph().draw_mermaid())

    # 复杂任务:多工具协同
    result = workflow.invoke({
        "messages": [HumanMessage(content=(
            "帮我查一下深圳的天气,然后搜索一下深圳最近有什么科技新闻,"
            "最后把结果发到 [email protected]"
        ))],
        "needs_human": False,
    })

    print("\n📜 最终响应:")
    print(result["messages"][-1].content)

可视化:查看工作流结构

LangGraph 自带 Mermaid 可视化,运行后输出:

print(workflow.get_graph().draw_mermaid())

输出:

graph TD
    A[verify_identity] --> B{验证结果?}
    B -->|通过| C[query_plan]
    B -->|失败| Z[end_session]
    C --> D{合约套餐?}
    D -->|是| E[handle_cancellation]
    D -->|否| F[update_crm]
    E --> G[create_ticket]
    F --> H[send_confirmation]
    H --> Z
    G --> Z

你可以把这个 Mermaid 代码粘贴到 https://mermaid.live 生成可视化图。


性能对比

指标 原始方案(if-else) LangGraph
代码行数 300+ 行 150 行
新增步骤耗时 30 分钟 5 分钟
状态调试难度 高(要打印变量) 低(可视化追踪)
支持回滚 不支持 检查点恢复
并发处理 需自己实现 thread_id 自动隔离
测试难度 高(mock 复杂) 节点独立可测试

总结

LangGraph 的核心价值:

  1. 状态机思维:把复杂的 AI 工作流拆成节点和边,可维护性大增
  2. 内置检查点:用户会话中断没关系,从断点恢复,体验流畅
  3. 条件路由:复杂的分支逻辑不再是噩梦,路由函数一目了然
  4. 可视化调试:Mermaid 图直接看出整个流程哪里有问题

适用场景:智能客服、AI 审批流、多步骤数据处理、长期记忆的对话 Agent。


👤 作者简介

一枚在大中原腹地(河南)卖公有云的从业者,主营腾讯云/阿里云/火山云,曾踩坑无数,现专注AI大模型应用落地。关注公众号「AI热点日报」,围观AI前沿动态~

博客:y1cloud.cn

你的项目有没有遇到复杂 AI 工作流难以管理的问题?是用什么方案解决的?欢迎评论区交流!

发表评论