🔄 有状态Agent系统 | 条件分支 + 循环 + 并行执行 | 完整客服工单系统实现

---

📖 为什么需要工作流编排？

传统Chain的局限性

# LangChain Chain - 线性执行
chain = prompt | llm | output_parser

# 问题：
❌ 只能顺序执行
❌ 无法根据结果做决策
❌ 无法循环重试
❌ 状态管理困难

LangGraph的优势

# LangGraph - 图结构执行
graph = StateGraph()
graph.add_node("research", research_node)
graph.add_node("write", write_node)
graph.add_edge("research", "write")
graph.add_conditional_edges("write", should_revise)

# 优势：
✅ 支持条件分支
✅ 支持循环和重试
✅ 支持并行执行
✅ 内置状态管理
✅ 可视化工作流

---

🏗️ LangGraph核心概念

1. State（状态）

from typing import TypedDict, Annotated
from langgraph.graph import add_messages

class AgentState(TypedDict):
    """Agent状态定义"""
    messages: Annotated[list, add_messages]  # 对话历史
    research_results: list                    # 研究结果
    draft: str                                # 草稿
    revision_count: int                       # 修订次数
    final_output: str                         # 最终输出

关键点：

• TypedDict 定义状态结构
• Annotated 指定如何合并更新
• add_messages 自动追加消息

---

2. Node（节点）

def research_node(state: AgentState) -> dict:
    """研究节点"""
    query = state["messages"][-1].content
    
    # 执行搜索
    results = search_web(query)
    
    return {
        "research_results": results
    }

def write_node(state: AgentState) -> dict:
    """写作节点"""
    context = "\n".join(state["research_results"])
    
    prompt = f"""
基于以下资料撰写文章：

{context}

要求：
- 结构清晰
- 语言流畅
- 800-1000字
"""
    
    draft = llm.invoke(prompt).content
    
    return {
        "draft": draft,
        "revision_count": state.get("revision_count", 0) + 1
    }

关键点：

• 函数接收state，返回dict
• 返回的dict会merge到state
• 可以访问和修改任何状态字段

---

3. Edge（边）

from langgraph.graph import END

# 普通边（无条件）
graph.add_edge("research", "write")

# 条件边（根据状态决策）
def should_revise(state: AgentState) -> str:
    """判断是否需要修订"""
    if state["revision_count"] >= 3:
        return "end"  # 最多修订3次
    
    quality_score = evaluate_quality(state["draft"])
    
    if quality_score < 0.8:
        return "revise"  # 质量不够，需要修订
    else:
        return "finalize"  # 质量合格，完成

graph.add_conditional_edges(
    "write",
    should_revise,
    {
        "revise": "write",      # 回到写作节点
        "finalize": "end",      # 结束
        "end": END
    }
)

---

💻 实战案例1：智能客服工单系统

业务场景

用户提交问题 → 分类 → 检索知识库 → 生成回复 → 
人工审核（可选）→ 发送回复 → 记录工单

完整实现

from langgraph.graph import StateGraph, END
from typing import TypedDict, Literal
from enum import Enum

# ==================== 状态定义 ====================

class TicketStatus(Enum):
    NEW = "new"
    IN_PROGRESS = "in_progress"
    NEEDS_REVIEW = "needs_review"
    RESOLVED = "resolved"
    ESCALATED = "escalated"

class TicketState(TypedDict):
    ticket_id: str
    customer_message: str
    category: str                    # 问题分类
    urgency: str                     # 紧急程度
    knowledge_results: list          # 知识库检索结果
    draft_response: str              # 回复草稿
    needs_human_review: bool         # 是否需要人工审核
    final_response: str              # 最终回复
    status: TicketStatus
    history: list                    # 操作历史

# ==================== 节点函数 ====================

def classify_ticket(state: TicketState) -> dict:
    """分类工单"""
    message = state["customer_message"]
    
    # 使用LLM分类
    prompt = f"""
分类以下客户问题：

问题：{message}

类别选项：
- technical（技术问题）
- billing（账单问题）
- account（账户问题）
- general（一般咨询）

紧急程度：
- low（低）
- medium（中）
- high（高）
- critical（紧急）

返回JSON格式：{{"category": "...", "urgency": "..."}}
"""
    
    result = llm.invoke(prompt)
    classification = json.loads(result.content)
    
    return {
        "category": classification["category"],
        "urgency": classification["urgency"],
        "status": TicketStatus.IN_PROGRESS,
        "history": [f"分类为: {classification['category']}, 紧急程度: {classification['urgency']}"]
    }

def retrieve_knowledge(state: TicketState) -> dict:
    """检索知识库"""
    query = state["customer_message"]
    category = state["category"]
    
    # 根据分类检索相关知识
    results = vector_db.similarity_search(
        query=query,
        filter={"category": category},
        k=5
    )
    
    return {
        "knowledge_results": [doc.page_content for doc in results],
        "history": state["history"] + [f"检索到 {len(results)} 条相关知识"]
    }

def generate_response(state: TicketState) -> dict:
    """生成回复"""
    context = "\n".join(state["knowledge_results"])
    message = state["customer_message"]
    
    prompt = f"""
你是专业的客服代表。请基于以下知识回答客户问题。

客户问题：
{message}

相关知识：
{context}

要求：
1. 语气友好专业
2. 回答准确完整
3. 如果有不确定的地方，诚实说明
4. 提供具体的解决步骤

回复：
"""
    
    response = llm.invoke(prompt).content
    
    # 判断是否需要人工审核
    needs_review = (
        state["urgency"] in ["high", "critical"] or
        "refund" in message.lower() or
        "cancel" in message.lower()
    )
    
    return {
        "draft_response": response,
        "needs_human_review": needs_review,
        "status": TicketStatus.NEEDS_REVIEW if needs_review else TicketStatus.RESOLVED,
        "history": state["history"] + ["生成回复草稿"]
    }

def human_review_node(state: TicketState) -> dict:
    """人工审核节点（模拟）"""
    # 实际应该等待人工审核
    # 这里模拟审核通过
    approved = True  # 模拟审核结果
    
    if approved:
        return {
            "final_response": state["draft_response"],
            "status": TicketStatus.RESOLVED,
            "history": state["history"] + ["人工审核通过"]
        }
    else:
        return {
            "status": TicketStatus.ESCALATED,
            "history": state["history"] + ["人工审核拒绝，已升级"]
        }

def send_response(state: TicketState) -> dict:
    """发送回复"""
    # 实际应该调用邮件/API发送
    print(f"发送回复给工单 {state['ticket_id']}")
    print(f"内容: {state['final_response'][:100]}...")
    
    return {
        "history": state["history"] + ["回复已发送"]
    }

def log_ticket(state: TicketState) -> dict:
    """记录工单"""
    # 保存到数据库
    ticket_record = {
        "ticket_id": state["ticket_id"],
        "status": state["status"].value,
        "category": state["category"],
        "resolution_time": datetime.now(),
        "history": state["history"]
    }
    
    database.save_ticket(ticket_record)
    
    return {"history": state["history"] + ["工单已记录"]}

# ==================== 条件函数 ====================

def route_after_generation(state: TicketState) -> Literal["human_review", "send_response"]:
    """路由：生成回复后"""
    if state["needs_human_review"]:
        return "human_review"
    else:
        return "send_response"

def route_after_review(state: TicketState) -> Literal["send_response", "escalate"]:
    """路由：审核后"""
    if state["status"] == TicketStatus.RESOLVED:
        return "send_response"
    else:
        return "escalate"

# ==================== 构建图 ====================

def create_ticket_workflow():
    """创建工单处理工作流"""
    
    workflow = StateGraph(TicketState)
    
    # 添加节点
    workflow.add_node("classify", classify_ticket)
    workflow.add_node("retrieve", retrieve_knowledge)
    workflow.add_node("generate", generate_response)
    workflow.add_node("human_review", human_review_node)
    workflow.add_node("send", send_response)
    workflow.add_node("log", log_ticket)
    
    # 添加入口
    workflow.set_entry_point("classify")
    
    # 添加边
    workflow.add_edge("classify", "retrieve")
    workflow.add_edge("retrieve", "generate")
    
    # 条件边：生成后
    workflow.add_conditional_edges(
        "generate",
        route_after_generation,
        {
            "human_review": "human_review",
            "send_response": "send"
        }
    )
    
    # 条件边：审核后
    workflow.add_conditional_edges(
        "human_review",
        route_after_review,
        {
            "send_response": "send",
            "escalate": END
        }
    )
    
    # 发送后记录
    workflow.add_edge("send", "log")
    workflow.add_edge("log", END)
    
    return workflow.compile()

# ==================== 使用示例 ====================

if __name__ == "__main__":
    # 创建工作流
    app = create_ticket_workflow()
    
    # 模拟工单
    initial_state = {
        "ticket_id": "TKT-2026-001",
        "customer_message": "我的账户被扣了两次费用，请帮我退款",
        "category": "",
        "urgency": "",
        "knowledge_results": [],
        "draft_response": "",
        "needs_human_review": False,
        "final_response": "",
        "status": TicketStatus.NEW,
        "history": []
    }
    
    # 执行工作流
    result = app.invoke(initial_state)
    
    print("=" * 80)
    print(f"工单ID: {result['ticket_id']}")
    print(f"状态: {result['status'].value}")
    print(f"分类: {result['category']}")
    print(f"紧急程度: {result['urgency']}")
    print(f"\n处理历史:")
    for i, step in enumerate(result['history'], 1):
        print(f"  {i}. {step}")
    print(f"\n最终回复:")
    print(result['final_response'])
    print("=" * 80)

---

🚀 实战案例2：多轮对话Agent

带记忆的对话系统

from langgraph.graph import MessagesState
from langchain_core.messages import HumanMessage, AIMessage

class ChatState(MessagesState):
    """对话状态（继承MessagesState）"""
    context: dict                      # 上下文信息
    intent: str                        # 用户意图
    entities: dict                     # 提取的实体
    should_escalate: bool              # 是否需要转人工

def intent_recognition(state: ChatState) -> dict:
    """意图识别"""
    last_message = state["messages"][-1].content
    
    prompt = f"""
识别用户意图：

用户消息：{last_message}

可能的意图：
- inquiry（咨询）
- complaint（投诉）
- request（请求）
- greeting（问候）
- other（其他）

返回JSON：{{"intent": "...", "confidence": 0.95}}
"""
    
    result = llm.invoke(prompt)
    intent_data = json.loads(result.content)
    
    return {
        "intent": intent_data["intent"],
        "context": {"intent_confidence": intent_data["confidence"]}
    }

def entity_extraction(state: ChatState) -> dict:
    """实体提取"""
    last_message = state["messages"][-1].content
    
    # 提取关键信息
    entities = {
        "product": extract_product(last_message),
        "date": extract_date(last_message),
        "amount": extract_amount(last_message)
    }
    
    return {"entities": entities}

def response_generation(state: ChatState) -> dict:
    """生成回复"""
    intent = state["intent"]
    entities = state["entities"]
    history = state["messages"]
    
    # 根据意图选择策略
    if intent == "greeting":
        response = "您好！有什么可以帮助您的吗？"
    
    elif intent == "inquiry":
        # 查询知识库
        kb_result = search_knowledge_base(entities)
        response = f"关于您的问题：\n\n{kb_result}"
    
    elif intent == "complaint":
        # 投诉需要安抚
        response = "非常抱歉给您带来不便。我会立即为您处理这个问题。"
        state["should_escalate"] = True
    
    else:
        # 通用回复
        response = llm.invoke(history).content
    
    return {
        "messages": [AIMessage(content=response)]
    }

def check_escalation(state: ChatState) -> Literal["respond", "escalate"]:
    """检查是否需要转人工"""
    if state.get("should_escalate"):
        return "escalate"
    
    # 检查情绪
    last_message = state["messages"][-1].content
    sentiment = analyze_sentiment(last_message)
    
    if sentiment == "negative":
        return "escalate"
    
    return "respond"

def escalate_to_human(state: ChatState) -> dict:
    """转人工"""
    print("⚠️ 转接人工客服...")
    
    return {
        "messages": [AIMessage(content="正在为您转接人工客服，请稍候...")]
    }

# 构建对话图
chat_graph = StateGraph(ChatState)

chat_graph.add_node("recognize_intent", intent_recognition)
chat_graph.add_node("extract_entities", entity_extraction)
chat_graph.add_node("generate_response", response_generation)
chat_graph.add_node("escalate", escalate_to_human)

chat_graph.set_entry_point("recognize_intent")
chat_graph.add_edge("recognize_intent", "extract_entities")
chat_graph.add_edge("extract_entities", "generate_response")

chat_graph.add_conditional_edges(
    "generate_response",
    check_escalation,
    {
        "respond": END,
        "escalate": "escalate"
    }
)

chat_graph.add_edge("escalate", END)

chat_app = chat_graph.compile()

# 使用
messages = [HumanMessage(content="我要投诉你们的产品质量")]
result = chat_app.invoke({"messages": messages})

---

📊 性能优化技巧

1. 并行执行

from langgraph.graph import START

def parallel_research(state: AgentState) -> dict:
    """并行研究多个来源"""
    
    # 同时执行多个任务
    web_results = search_web(state["query"])
    db_results = search_database(state["query"])
    internal_docs = search_internal(state["query"])
    
    return {
        "research_results": {
            "web": web_results,
            "database": db_results,
            "internal": internal_docs
        }
    }

# 或者使用Send API进行动态并行
def fan_out(state: AgentState):
    """扇出：为每个子任务创建分支"""
    topics = extract_topics(state["query"])
    
    return [
        Send("research_topic", {"topic": topic})
        for topic in topics
    ]

graph.add_conditional_edges(START, fan_out)

2. 断点续传

from langgraph.checkpoint.memory import MemorySaver

# 使用持久化检查点
checkpointer = MemorySaver()

app = workflow.compile(checkpointer=checkpointer)

# 执行并保存状态
thread_id = "thread_001"
config = {"configurable": {"thread_id": thread_id}}

# 第一次执行
result1 = app.invoke(initial_state, config=config)

# 中断后恢复
result2 = app.invoke(None, config=config)  # 从上次中断处继续

3. 流式输出

# 流式获取中间状态
for event in app.stream(initial_state, stream_mode="values"):
    print("当前状态:", event)
    print("---")

# 或者只关注特定节点
for event in app.stream(initial_state, stream_mode="updates"):
    for node_name, update in event.items():
        print(f"节点 {node_name} 的更新:", update)

---

🎯 最佳实践

1. 状态设计原则

# ✅ 好的状态设计
class GoodState(TypedDict):
    messages: Annotated[list, add_messages]  # 使用annotated
    step_count: int                           # 简单的计数器
    results: dict                             # 结构化数据

# ❌ 不好的状态设计
class BadState(TypedDict):
    data: Any  # 太宽泛
    temp_var: str  # 临时变量不应该在state中

2. 节点设计原则

# ✅ 好的节点：单一职责
def good_node(state):
    """只做一件事"""
    result = do_one_thing(state["input"])
    return {"output": result}

# ❌ 不好的节点：职责过多
def bad_node(state):
    """做了太多事"""
    result1 = do_thing_1(...)
    result2 = do_thing_2(...)
    result3 = do_thing_3(...)
    return {...}  # 返回太多东西

3. 错误处理

def robust_node(state: AgentState) -> dict:
    """健壮的节点"""
    try:
        result = risky_operation(state["input"])
        return {"result": result, "error": None}
    
    except Exception as e:
        return {
            "result": None,
            "error": str(e),
            "retry_count": state.get("retry_count", 0) + 1
        }

def should_retry(state: AgentState) -> str:
    if state.get("error") and state["retry_count"] < 3:
        return "retry"
    else:
        return "continue"

---

📈 实际应用案例

案例1：自动化报告生成

工作流：

数据收集 → 数据分析 → 图表生成 → 文字撰写 → 质量检查 → 发布

效果：

• 原来需要4小时手动完成
• 现在15分钟自动生成
• 质量更稳定

案例2：智能代码审查

工作流：

代码解析 → 静态分析 → 安全扫描 → 性能评估 → 生成建议 → 人工确认

效果：

• 审查时间从30分钟降到5分钟
• 发现问题更全面
• 减少人为疏漏

---

🎯 总结

LangGraph的核心价值：

1. ✅ 有状态 - 内置状态管理
2. ✅ 灵活 - 支持任意图结构
3. ✅ 可靠 - 断点续传、错误处理
4. ✅ 可观测 - 流式输出、可视化
5. ✅ 生产就绪 - 持久化、并发控制

适用场景：

• 复杂的多步骤工作流
• 需要条件分支和循环
• 需要状态持久化
• 需要人工介入

下一步：

• 尝试LangGraph Studio（可视化工具）
• 学习高级模式（Map-Reduce、Hierarchical）
• 集成到生产环境

---

完整代码和详细教程： 👉 GitHub仓库