用LangGraph构建多步骤AI工作流：从设计到部署的完整教程

为什么需要多步骤工作流？

单次LLM调用能做的事很有限。你问一个问题，它给你一个答案——就完了。但真实场景很少这么简单：用户提交一个文档，你需要先分类、再提取关键信息、然后判断是否需要进一步处理、最后生成回复。这种多步骤流程，单次调用搞不定。

LangGraph是LangChain团队推出的一个专门解决这个问题的框架。它不是又一个AI框架，而是让你把LLM调用编排成有状态的图（Graph）。每个节点是一个处理步骤，边是条件跳转逻辑。和传统的链式调用（Chain）相比，图的好处是你能处理分支、循环、并行——这些真实工作流需要的模式。

这篇文章带你从零搭建一个完整的多步骤文档处理工作流，包含可运行的代码。读完你就能直接用在自己项目里。

先搞清楚场景

假设你有一个AI客服系统，用户提交工单后需要：

判断工单类型（技术问题/账号问题/投诉建议）
根据类型提取关键信息
判断能否自动回复
能则生成回复，不能则升级给人工

这个场景覆盖了分类→提取→决策→生成四个典型步骤，适合用来演示LangGraph的核心能力。

安装和环境准备

先安装依赖。我用的是Python 3.11+和LangGraph 0.2+版本。

pip install langgraph langchain-openai

然后设置API密钥。为了安全，用环境变量而不是硬编码：

import os
os.environ["OPENAI_API_KEY"] = "你的API密钥"

注意：LangGraph本身不绑定任何模型提供商。你可以用OpenAI、Anthropic、本地模型，或者自己封装的API。这里用OpenAI做演示，但实际切换成本很低。

定义状态模型

LangGraph的核心概念之一是状态（State）。每个节点都可以读取和写入状态，图就是状态在这些节点之间的流转。

from typing import TypedDict, Optional
from langgraph.graph import StateGraph, END

class TicketState(TypedDict):
    raw_text: str  # 用户原始输入
    category: Optional[str]  # 分类结果
    extracted_info: Optional[dict]  # 提取的信息
    can_auto_reply: Optional[bool]  # 能否自动回复
    reply: Optional[str]  # 最终回复
    escalated: Optional[bool]  # 是否升级

状态模型定义了工作流中流动的所有数据。这比传统Chain的黑盒传递清晰得多——你能精确控制每个节点能读写什么。

构建处理节点

每个节点就是一个函数，接收当前状态，返回要更新的字段。

先写分类节点：

from langchain_openai import ChatOpenAI

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

def classify_ticket(state: TicketState) -> dict:
    prompt = f"""请将以下用户工单分类为：技术问题、账号问题、投诉建议。
只返回分类名称，不要多余内容。

工单内容：{state['raw_text']}"""
    category = llm.invoke(prompt).content.strip()
    return {"category": category}

然后提取信息节点：

def extract_info(state: TicketState) -> dict:
    prompt = f"""从以下{state['category']}类型的工单中提取关键信息。
以JSON格式返回。

工单内容：{state['raw_text']}"""
    result = llm.invoke(prompt).content.strip()
    return {"extracted_info": result}

决策节点：

def decide_action(state: TicketState) -> dict:
    # 简单的规则：投诉建议类型全部升级人工
    if state['category'] == '投诉建议':
        return {"can_auto_reply": False, "escalated": True}
    return {"can_auto_reply": True, "escalated": False}

回复生成节点：

def generate_reply(state: TicketState) -> dict:
    if state['escalated']:
        return {"reply": "您的工单已升级给人工客服，我们会在24小时内联系您。"}
    prompt = f"""根据以下信息生成回复：
工单类型：{state['category']}
提取信息：{state['extracted_info']}
请生成友好、专业的回复。"""
    reply = llm.invoke(prompt).content.strip()
    return {"reply": reply}

搭建图结构

有了节点和状态，把它们组合成图：

workflow = StateGraph(TicketState)

# 添加节点
workflow.add_node("classify", classify_ticket)
workflow.add_node("extract", extract_info)
workflow.add_node("decide", decide_action)
workflow.add_node("respond", generate_reply)

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

# 添加边
workflow.add_edge("classify", "extract")
workflow.add_edge("extract", "decide")

# 条件边：根据决策结果分流
workflow.add_conditional_edges(
    "decide",
    lambda state: "respond" if state["can_auto_reply"] else "respond",
    {"respond": "respond"}
)

workflow.add_edge("respond", END)

app = workflow.compile()

注意条件边的设计：这里虽然两个分支都走到respond，但respond节点内部会根据escalated标志产生不同回复。你完全可以让升级分支走到一个单独的escalate节点，结构更清晰。

跑一个完整的流程

result = app.invoke({
    "raw_text": "我登录后一直提示验证码错误，换了浏览器也不行，能帮我重置一下吗？",
    "category": None,
    "extracted_info": None,
    "can_auto_reply": None,
    "reply": None,
    "escalated": None
})

print(f"分类：{result['category']}")
print(f"回复：{result['reply']}")

输出：

分类：账号问题
回复：您好，感谢您反馈登录验证码问题。请尝试以下步骤：
1. 清除浏览器缓存和Cookie后重试
2. 使用无痕模式登录
3. 如果仍无法解决，请确认您的账号绑定邮箱是否有效
如问题持续，请提供更多信息，我们将进一步为您排查。

扩展到真实部署

上面的代码直接跑没问题，但生产环境需要考虑几件事：

持久化状态：LangGraph支持Checkpoint机制，可以把中间状态保存到数据库。这样如果处理到一半崩溃了，下次能从断点继续
超时控制：每个节点可以设置单独的timeout，防止LLM调用卡死
并行节点：如果你的工作流中有多个独立步骤，可以用add_node的parallel参数并行执行
人机协作：LangGraph支持interrupt机制，可以在关键节点停下来等人工确认再继续

用FastAPI包装一下就能做成API服务：

from fastapi import FastAPI
from pydantic import BaseModel

app_api = FastAPI()

class TicketRequest(BaseModel):
    text: str

@app_api.post("/process-ticket")
async def process_ticket(request: TicketRequest):
    result = app.invoke({"raw_text": request.text, ...})
    return {"category": result["category"], "reply": result["reply"]}