别再用 if-else 堆屎山了！用状态机优雅地管理复杂流程

发表于 更新于 分类于 ， 本文字数： 22k 阅读时长 ≈ 20 分钟

状态机是一种描述对象生命周期的数学模型。简单来说，它告诉我们一个东西在什么条件下会从一个状态变成另一个状态。

举个生活中的例子：红绿灯。

红绿灯有三种状态：红灯、黄灯、绿灯。它的状态转换规则是：

  • 红灯 → 绿灯（等待时间结束）
  • 绿灯 → 黄灯（等待时间结束）
  • 黄灯 → 红灯（等待时间结束）

用 Mermaid 状态图来表示就是：

stateDiagram-v2
    [*] --> 红灯
    红灯 --> 绿灯: 等待时间结束
    绿灯 --> 黄灯: 等待时间结束
    黄灯 --> 红灯: 等待时间结束

这就是最简单的状态机——一个有限状态机（Finite State Machine, FSM），它有有限个状态，状态之间的转换规则是确定的。

状态机的核心要素

一个完整的状态机需要包含以下几个要素：

  1. 状态（State）：对象所处的情况或阶段。比如红绿灯的”红灯”就是一个状态。
  2. 事件（Event）：触发状态变化的条件。比如”等待时间结束”就是一个事件。
  3. 转换（Transition）：从一个状态到另一个状态的变化过程。比如从红灯变成绿灯。
  4. 动作（Action）：状态转换时执行的操作。比如切换到绿灯时同时启动计时器。

它们之间的关系可以用下图表示：

flowchart LR
    A["事件 (Event)"] -->|"触发"| B["转换 (Transition)"]
    B -->|"执行"| C["动作 (Action)"]
    C -->|"到达"| D["新状态 (State)"]
    D -->|"等待事件"| A

状态机的使用场景

状态机在软件开发中非常常见，以下是一些典型场景：

1. 工作流审批

比如请假审批、报销审批等。一个申请可能经历”待审批 → 部门经理审批 → 总监审批 → 完成”这样的流程。

flowchart LR
    A[待提交] -->|提交申请| B[待审批]
    B -->|部门经理审批| C[总监审批]
    C -->|审批通过| D[已完成]
    B -->|审批拒绝| E[已终止]
    C -->|审批拒绝| E

2. 订单状态管理

电商订单的状态流转是状态机最经典的应用之一。用户下单后，订单会经历多个状态，每个状态只能按规则流转到下一个状态：

stateDiagram-v2
    [*] --> 待付款: 用户下单
    待付款 --> 已付款: 用户支付
    待付款 --> 已取消: 用户取消/超时
    已付款 --> 已发货: 商家发货
    已付款 --> 已退款: 申请退款
    已发货 --> 已收货: 用户确认收货
    已收货 --> 已完成: 评价完成
    已完成 --> [*]
    已取消 --> [*]
    已退款 --> [*]

3. 游戏角色状态

游戏角色的状态管理是状态机在娱乐领域的典型应用。角色在不同状态下有不同的行为表现：

stateDiagram-v2
    [*] --> 站立
    站立 --> 行走: 按下方向键
    行走 --> 站立: 松开方向键
    行走 --> 奔跑: 按住加速键
    奔跑 --> 行走: 松开加速键
    站立 --> 跳跃: 按下跳跃键
    行走 --> 跳跃: 按下跳跃键
    跳跃 --> 站立: 落地
    站立 --> 攻击: 按下攻击键
    攻击 --> 站立: 攻击结束
    站立 --> 死亡: 生命值归零
    行走 --> 死亡: 生命值归零
    跳跃 --> 死亡: 生命值归零
    死亡 --> [*]

4. 网络连接

网络连接状态：未连接 → 连接中 → 已连接 → 断开连接。

stateDiagram-v2
    [*] --> 未连接
    未连接 --> 连接中: 发起连接
    连接中 --> 已连接: 连接成功
    连接中 --> 未连接: 连接失败
    已连接 --> 断开连接: 连接中断
    断开连接 --> 未连接: 超时
    断开连接 --> 连接中: 自动重连

5. 表单验证

表单的状态：未填写 → 填写中 → 验证通过 → 提交成功。

stateDiagram-v2
    [*] --> 未填写
    未填写 --> 填写中: 开始输入
    填写中 --> 验证通过: 所有字段合法
    填写中 --> 验证失败: 字段不合法
    验证失败 --> 填写中: 修改字段
    验证通过 --> 提交成功: 点击提交
    提交成功 --> [*]

从简单到复杂的实例

实例1：最简单的状态机 - 红绿灯

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
class TrafficLight:
    """红绿灯状态机"""
    
    def __init__(self):
        self.state = "red"  # 初始状态
    
    def next(self):
        """切换到下一个状态"""
        if self.state == "red":
            self.state = "green"
        elif self.state == "green":
            self.state = "yellow"
        elif self.state == "yellow":
            self.state = "red"
    
    def get_state(self):
        return self.state

# 使用示例
light = TrafficLight()
print(light.get_state())  # red
light.next()
print(light.get_state())  # green
light.next()
print(light.get_state())  # yellow
light.next()
print(light.get_state())  # red

实例2：带条件判断的状态机 - 简单审批

stateDiagram-v2
    [*] --> pending: 创建申请
    pending --> approved: 审批通过
    pending --> rejected: 审批拒绝
    approved --> [*]
    rejected --> [*]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
class SimpleApproval:
    """简单审批状态机"""
    
    def __init__(self):
        self.status = "pending"  # pending: 待审批, approved: 已通过, rejected: 已拒绝
    
    def approve(self):
        """审批通过"""
        if self.status == "pending":
            self.status = "approved"
            return True, "审批通过"
        return False, "当前状态无法审批"
    
    def reject(self):
        """审批拒绝"""
        if self.status == "pending":
            self.status = "rejected"
            return True, "审批拒绝"
        return False, "当前状态无法拒绝"

# 使用示例
approval = SimpleApproval()
print(approval.status)  # pending
approval.approve()
print(approval.status)  # approved

实例3：多步骤状态机 - 请假审批流程

这个实例展示了多步骤状态机的实现：一个请假申请需要经过3个审批步骤，每个步骤都可以通过或拒绝，拒绝时可以打回到任意步骤重新审批。

flowchart LR
    S1[步骤1: 部门经理] -->|通过| S2[步骤2: HR审核]
    S2 -->|通过| S3[步骤3: 总监审批]
    S3 -->|通过| DONE[审批完成]
    S1 -->|拒绝| REJ[审批拒绝]
    S2 -->|拒绝| REJ
    S3 -->|打回| S1
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
class LeaveApproval:
    """请假审批状态机 - 多步骤"""
    
    def __init__(self):
        self.status = "pending"
        self.current_step = 1
        self.total_steps = 3  # 总共3个审批步骤
        self.approval_records = []
    
    def approve(self, approver_name):
        """审批通过"""
        if self.status != "pending":
            return False, "当前状态无法审批"
        
        # 记录审批
        self.approval_records.append({
            "step": self.current_step,
            "approver": approver_name,
            "action": "approved"
        })
        
        # 检查是否还有下一步
        if self.current_step < self.total_steps:
            self.current_step += 1
            return True, f"已通过，流转到步骤{self.current_step}"
        else:
            self.status = "approved"
            return True, "所有步骤通过，审批完成"
    
    def reject(self, approver_name, reject_to_step=None):
        """审批拒绝"""
        if self.status != "pending":
            return False, "当前状态无法拒绝"
        
        # 记录审批
        self.approval_records.append({
            "step": self.current_step,
            "approver": approver_name,
            "action": "rejected"
        })
        
        # 打回到指定步骤
        if reject_to_step and 1 <= reject_to_step <= self.total_steps:
            self.current_step = reject_to_step
            return True, f"已拒绝，打回到步骤{reject_to_step}"
        else:
            self.status = "rejected"
            return True, "审批拒绝"

# 使用示例
leave = LeaveApproval()
print(f"状态: {leave.status}, 步骤: {leave.current_step}")

leave.approve("王经理")
print(f"状态: {leave.status}, 步骤: {leave.current_step}")

leave.approve("钱总监")
print(f"状态: {leave.status}, 步骤: {leave.current_step}")

leave.reject("周总", reject_to_step=1)
print(f"状态: {leave.status}, 步骤: {leave.current_step}")

实例4：带状态图的复杂状态机

这个实例展示了一个更接近实际业务的订单状态机，它使用枚举定义状态，用字典定义转换规则，并支持状态转换时的回调动作：

stateDiagram-v2
    [*] --> pending
    pending --> paid: 支付
    pending --> cancelled: 取消
    paid --> shipped: 发货
    paid --> cancelled: 取消
    shipped --> delivered: 收货
    delivered --> completed: 完成
    completed --> [*]
    cancelled --> [*]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
from enum import Enum

class OrderStatus(Enum):
    """订单状态枚举"""
    PENDING = "pending"
    PAID = "paid"
    SHIPPED = "shipped"
    DELIVERED = "delivered"
    COMPLETED = "completed"
    CANCELLED = "cancelled"

class OrderStateMachine:
    """订单状态机 - 带状态图"""
    
    # 定义状态转换规则
    TRANSITIONS = {
        OrderStatus.PENDING: [OrderStatus.PAID, OrderStatus.CANCELLED],
        OrderStatus.PAID: [OrderStatus.SHIPPED, OrderStatus.CANCELLED],
        OrderStatus.SHIPPED: [OrderStatus.DELIVERED],
        OrderStatus.DELIVERED: [OrderStatus.COMPLETED],
        OrderStatus.COMPLETED: [],
        OrderStatus.CANCELLED: []
    }
    
    def __init__(self):
        self.status = OrderStatus.PENDING
    
    def can_transition(self, new_status):
        """检查是否可以转换到新状态"""
        return new_status in self.TRANSITIONS.get(self.status, [])
    
    def transition(self, new_status, action=None):
        """执行状态转换"""
        if not self.can_transition(new_status):
            return False, f"无法从{self.status.value}转换到{new_status.value}"
        
        old_status = self.status
        self.status = new_status
        
        # 执行附加动作
        if action:
            action(old_status, new_status)
        
        return True, f"状态从{old_status.value}转换到{new_status.value}"

# 使用示例
def on_status_change(old_status, new_status):
    print(f"状态变化: {old_status.value} -> {new_status.value}")

order = OrderStateMachine()
print(f"当前状态: {order.status.value}")

order.transition(OrderStatus.PAID, on_status_change)
order.transition(OrderStatus.SHIPPED, on_status_change)
order.transition(OrderStatus.DELIVERED, on_status_change)
order.transition(OrderStatus.COMPLETED, on_status_change)

状态机的分类

根据输出方式的不同，有限状态机主要分为两类：

Mealy 型状态机

输出取决于当前状态和输入。状态转换时立即产生输出，适合需要根据输入实时响应的场景。

flowchart LR
    S1[状态1] -->|"输入A / 输出X"| S2[状态2]
    S2 -->|"输入B / 输出Y"| S1

Moore 型状态机

输出仅取决于当前状态，与输入无关。每个状态对应一个固定的输出，适合状态本身就有明确含义的场景。

flowchart LR
    S1["状态1 (输出X)"] -->|"输入A"| S2["状态2 (输出Y)"]
    S2 -->|"输入B"| S1

在实际开发中，大多数状态机实现更接近 Mealy 型，因为我们需要根据不同的触发条件执行不同的动作。

常用 Python 状态机库

库名 特点 适用场景
transitions 功能全面，支持回调、条件、时间触发 复杂业务逻辑
pytransitions transitions 的异步版本 异步应用
statecraft 轻量级，API 简洁 简单场景
automat 基于状态模式 面向对象设计

下面是一个使用 transitions 库的示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from transitions import Machine

class TrafficLight:
    states = ['red', 'green', 'yellow']
    
    transitions = [
        {'trigger': 'next', 'source': 'red', 'dest': 'green'},
        {'trigger': 'next', 'source': 'green', 'dest': 'yellow'},
        {'trigger': 'next', 'source': 'yellow', 'dest': 'red'},
    ]
    
    def __init__(self):
        self.state = 'red'
        self.machine = Machine(model=self, states=self.states, transitions=self.transitions, initial='red')

light = TrafficLight()
print(light.state)  # red
light.next()
print(light.state)  # green

transitions 库自动为我们生成了状态转换方法、条件守卫、进入/离开回调等功能，大大减少了样板代码。

pytransitions — 异步场景的 transitions

pytransitions 是 transitions 的异步版本，支持 async/await，适合在 FastAPI、aiohttp 等异步框架中使用：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
import asyncio
from pytransitions.extensions import AioMachine as Machine

class TrafficLight:
    states = ['red', 'green', 'yellow']

    transitions = [
        {'trigger': 'next', 'source': 'red', 'dest': 'green'},
        {'trigger': 'next', 'source': 'green', 'dest': 'yellow'},
        {'trigger': 'next', 'source': 'yellow', 'dest': 'red'},
    ]

    def __init__(self):
        self.state = 'red'
        self.machine = Machine(model=self, states=self.states, transitions=self.transitions, initial='red')

async def main():
    light = TrafficLight()
    print(light.state)   # red
    await light.next()   # 异步切换
    print(light.state)   # green
    await light.next()
    print(light.state)   # yellow

asyncio.run(main())

核心区别在于 AioMachine 让所有转换方法变成协程，可以在 await 中调用，不会阻塞事件循环。

statecraft — 极简主义之选

statecraft 是一个极轻量的状态机库，API 非常简洁，适合不想引入太多依赖的场景：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from statecraft import StateMachine, transition

class TrafficLight(StateMachine):
    initial = 'red'

    @transition('red', 'green')
    def next(self):
        pass

    @transition('green', 'yellow')
    def next(self):
        pass

    @transition('yellow', 'red')
    def next(self):
        pass

light = TrafficLight()
print(light.state)  # red
light.next()
print(light.state)  # green

statecraft 用装饰器定义转换，代码量最少，但功能也相对有限——没有条件守卫、回调等高级特性。适合简单场景快速原型开发。

automat — 面向对象的状态模式

automat 是 Hynek Schlawack 开发的库，采用**状态模式（State Pattern）**的设计理念，把每个状态封装成一个类，适合追求严格面向对象设计的团队：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
from automat import MethodicalMachine

class TrafficLight:
    _machine = MethodicalMachine()

    @_machine.state(initial=True)
    def red(self):
        """红灯状态"""

    @_machine.state()
    def green(self):
        """绿灯状态"""

    @_machine.state()
    def yellow(self):
        """黄灯状态"""

    @_machine.input()
    def next(self):
        """切换到下一个状态"""

    @red.upon(next)
    def go_green(self):
        """红灯 -> 绿灯"""
        return (green,)

    @green.upon(next)
    def go_yellow(self):
        """绿灯 -> 黄灯"""
        return (yellow,)

    @yellow.upon(next)
    def go_red(self):
        """黄灯 -> 红灯"""
        return (red,)

light = TrafficLight()
print(light.red.entered)  # True
light.next()
print(light.green.entered)  # True

automat 的特点是状态转换逻辑分散在各个状态的方法中，每个状态决定收到输入后转向哪个新状态。这种设计让状态逻辑高度内聚，但学习曲线稍高。

四个库的对比总结

特性 transitions pytransitions statecraft automat
异步支持 否 是 否 否
条件守卫 是 是 否 否
进入/离开回调 是 是 否 是
代码量 中 中 极少 多
学习成本 低 低 极低 中
适合场景 通用 异步应用 快速原型 OOP 设计

AI应用中的状态机：以LangGraph为例

在AI应用中，状态机也有广泛应用。LangGraph是一个用于构建AI Agent工作流的框架，它本质上也是一个状态机。LangGraph 将传统的状态机概念与图结构相结合，使得 AI Agent 的工作流可以用**节点（Node）和边（Edge）**来描述。

LangGraph的核心概念

LangGraph的核心概念包括：

  1. 状态（State）：Agent工作流中的数据，通常是一个类型化的字典（TypedDict），包含消息列表、当前步骤等信息。
  2. 节点（Node）：处理数据的函数，每个节点接收当前状态并返回更新后的状态。
  3. 边（Edge）：节点之间的连接，定义了状态转换路径。可以是无条件的（直接走到下一个节点），也可以是有条件的（根据状态决定走哪个节点）。
flowchart TD
    A["开始 (START)"] --> B["process: 处理输入"]
    B --> C["analyze: 分析数据"]
    C --> D["respond: 生成响应"]
    D --> E["结束 (END)"]

LangGraph的简单示例

下面是一个最简单的 LangGraph 示例，展示了如何用图结构定义一个三步工作流：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END

# 定义状态
class AgentState(TypedDict):
    messages: list
    current_step: str

# 定义节点
def process_input(state: AgentState):
    """处理用户输入"""
    return {"messages": state["messages"] + ["处理输入"], "current_step": "process"}

def analyze(state: AgentState):
    """分析数据"""
    return {"messages": state["messages"] + ["分析数据"], "current_step": "analyze"}

def respond(state: AgentState):
    """生成响应"""
    return {"messages": state["messages"] + ["生成响应"], "current_step": "respond"}

# 构建状态机
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("process", process_input)
workflow.add_node("analyze", analyze)
workflow.add_node("respond", respond)

# 定义边（状态转换）
workflow.set_entry_point("process")
workflow.add_edge("process", "analyze")
workflow.add_edge("analyze", "respond")
workflow.add_edge("respond", END)

# 编译状态机
app = workflow.compile()

# 运行
result = app.invoke({"messages": [], "current_step": "start"})
print(result["messages"])

LangGraph与传统状态机的对比

特性 传统状态机 LangGraph
状态定义 简单的枚举或字符串 类型化的字典（TypedDict）
状态转换 显式的条件判断 图结构的边（可带条件）
执行逻辑 在状态转换中 在节点函数中
适用场景 业务流程、工作流 AI Agent、对话流程
复杂度 低到中 中到高
可视化 需要额外工具 内置图可视化
持久化 需要自行实现 原生支持

下面是一个带条件分支的 LangGraph 示例，展示了如何根据用户意图选择不同的处理路径：

flowchart TD
    A["开始"] --> B["classifier: 意图识别"]
    B -->|"查询类"| C["search: 搜索知识库"]
    B -->|"对话类"| D["chat: 闲聊回复"]
    B -->|"任务类"| E["executor: 执行任务"]
    C --> F["respond: 生成回复"]
    D --> F
    E --> F
    F --> G["结束"]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END

class AgentState(TypedDict):
    messages: list
    intent: str

def classifier(state: AgentState):
    """意图识别节点 - 根据输入判断用户意图"""
    last_msg = state["messages"][-1] if state["messages"] else ""
    if "搜索" in last_msg or "查询" in last_msg:
        intent = "search"
    elif "执行" in last_msg or "帮我" in last_msg:
        intent = "task"
    else:
        intent = "chat"
    return {"messages": state["messages"], "intent": intent}

def search(state: AgentState):
    """搜索节点"""
    return {"messages": state["messages"] + ["搜索结果: ..."], "intent": state["intent"]}

def chat(state: AgentState):
    """闲聊节点"""
    return {"messages": state["messages"] + ["你好！有什么可以帮你的？"], "intent": state["intent"]}

def executor(state: AgentState):
    """任务执行节点"""
    return {"messages": state["messages"] + ["任务执行完成"], "intent": state["intent"]}

def respond(state: AgentState):
    """统一回复节点"""
    return {"messages": state["messages"], "intent": state["intent"]}

def route_intent(state: AgentState):
    """条件路由函数 - 根据意图选择下一个节点"""
    return state.get("intent", "chat")

workflow = StateGraph(AgentState)
workflow.add_node("classifier", classifier)
workflow.add_node("search", search)
workflow.add_node("chat", chat)
workflow.add_node("executor", executor)
workflow.add_node("respond", respond)

workflow.set_entry_point("classifier")
workflow.add_conditional_edges("classifier", route_intent, {
    "search": "search",
    "chat": "chat",
    "task": "executor"
})
workflow.add_edge("search", "respond")
workflow.add_edge("chat", "respond")
workflow.add_edge("executor", "respond")
workflow.add_edge("respond", END)

app = workflow.compile()

LangGraph的优势

  1. 可视化：状态图可以可视化，便于理解。LangGraph 内置了 get_graph().draw_mermaid_png() 方法，可以直接生成状态图的图片。
  2. 可扩展：可以轻松添加新的节点和边，不需要修改已有代码。
  3. 持久化：支持状态持久化（Checkpointer），可以中断和恢复执行，适合长时间运行的任务。
  4. 人机交互：支持人在回路中（Human-in-the-loop），可以在关键节点暂停等待人工审核。
  5. 流式输出：支持流式处理，可以实时返回中间结果。

下面是一个带有人工审核环节的 LangGraph 示例：

flowchart TD
    A["开始"] --> B["generate: 生成内容"]
    B --> C{"需要审核?"}
    C -->|"是"| D["暂停等待人工审核"]
    D --> E{"审核通过?"}
    E -->|"是"| F["publish: 发布内容"]
    E -->|"否"| B
    C -->|"否"| F
    F --> G["结束"]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
from typing import TypedDict
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

class ContentState(TypedDict):
    content: str
    needs_review: bool
    approved: bool

def generate(state: ContentState):
    """生成内容"""
    return {"content": "AI 生成的文章内容...", "needs_review": True, "approved": False}

def review_check(state: ContentState):
    """判断是否需要人工审核"""
    return "human_review" if state.get("needs_review") else "publish"

def human_review(state: ContentState):
    """等待人工审核 - 会暂停执行"""
    pass

def after_review(state: ContentState):
    """审核后的处理"""
    return state

def publish(state: ContentState):
    """发布内容"""
    return {"content": state["content"], "needs_review": False, "approved": True}

workflow = StateGraph(ContentState)
workflow.add_node("generate", generate)
workflow.add_node("human_review", human_review)
workflow.add_node("after_review", after_review)
workflow.add_node("publish", publish)

workflow.set_entry_point("generate")
workflow.add_conditional_edges("generate", review_check, {
    "human_review": "human_review",
    "publish": "publish"
})
workflow.add_edge("human_review", "after_review")
workflow.add_edge("after_review", "publish")
workflow.add_edge("publish", END)

# 使用 Checkpointer 支持中断和恢复
memory = MemorySaver()
app = workflow.compile(checkpointer=memory, interrupt_before=["human_review"])

# 运行时会在 human_review 节点前暂停，等待人工审核

如何选择状态机实现方式

面对不同的业务场景，选择合适的状态机实现方式至关重要：

场景复杂度 推荐方案 示例
简单（2-3个状态） if-else / 字典映射 红绿灯、开关
中等（4-8个状态） 状态机类 + 枚举 订单状态、审批流程
复杂（8+状态，多分支） transitions 库 工作流引擎
AI Agent LangGraph 对话系统、任务编排

选择时需要考虑以下因素：

  • 状态数量：状态越少，实现越简单
  • 转换规则复杂度：是否有条件分支、守卫条件
  • 是否需要持久化：长时间运行的任务需要状态持久化
  • 是否需要可视化：团队协作时可视化很重要
  • 是否涉及 AI：LangGraph 是 AI Agent 的首选

最佳实践

（请假审批流系统）是一个典型的状态机应用案例，展示了如何在实际项目中使用状态机模式。

项目中的状态机实现

1. 状态定义

在database.py中定义了请假申请的状态：

1
2
3
4
5
class LeaveRequest(Base):
    # 状态：pending（待审批）、approved（已通过）、rejected（已拒绝）
    status = Column(String, default="pending")
    # 当前审批步骤（从1开始）
    current_step = Column(Integer, default=1)

2. 状态转换规则

在approval_engine.py中实现了状态转换逻辑：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
class ApprovalEngine:
    @staticmethod
    def approve(db, request_id, approver_id, comment=None):
        """同意审批"""
        # 验证权限
        is_valid, error_msg, current_step = ApprovalEngine.validate_approver(
            db, request_id, approver_id
        )
        if not is_valid:
            return False, error_msg
        
        # 创建审批记录
        create_approval_record(...)
        
        # 检查是否还有下一步
        steps = get_flow_steps(db, request.flow_id)
        next_step_order = current_step.step_order + 1
        
        if next_step_order > len(steps):
            # 所有步骤都通过了
            update_leave_request_status(db, request_id, "approved", current_step.step_order)
            return True, "审批通过，流程已完成"
        else:
            # 还有下一步
            update_leave_request_status(db, request_id, "pending", next_step_order)
            return True, f"审批通过，已流转到下一步：步骤{next_step_order}"

3. 状态转换图

flowchart TD
    A[待提交] -->|提交| B[待审批]
    B -->|审批通过| C{是否最后一步?}
    C -->|否| D[下一步审批]
    C -->|是| E[审批通过/已完成]
    B -->|审批拒绝| F[已拒绝/已终止]
    B -->|打回重审| G[返回指定步骤]
    G --> B
    D --> B
    E --> H([结束])
    F --> H

项目的最佳实践

1. 分离状态和行为

  • 状态定义：在database.py中定义数据模型
  • 状态行为：在approval_engine.py中实现业务逻辑
  • API接口：在main.py中提供对外服务

这种分离使得代码更易于维护和测试。

2. 使用枚举或常量管理状态

1
2
3
4
5
# 在schemas.py中定义状态常量
class LeaveRequestStatus:
    PENDING = "pending"
    APPROVED = "approved"
    REJECTED = "rejected"

3. 记录状态转换历史

1
2
3
4
5
6
7
class ApprovalRecord(Base):
    """审批记录表 - 记录每一步的审批情况"""
    request_id = Column(Integer, ForeignKey("leave_requests.id"))
    step_order = Column(Integer, nullable=False)
    action = Column(String, nullable=False)  # approved/rejected
    comment = Column(Text)
    created_at = Column(DateTime, default=datetime.now)

4. 支持复杂的转换规则

1
2
3
4
class ApprovalFlowStep(Base):
    """审批流程步骤表"""
    can_reject_to_previous = Column(Boolean, default=True)  # 是否可以拒绝回上一步
    can_reject_to_any = Column(Boolean, default=False)  # 是否可以拒绝到任意步骤

5. 提供清晰的API接口

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
@app.post("/api/leave-requests/{request_id}/approve")
async def approve_leave_request(
    request_id: int,
    approval_action: ApprovalAction,
    db: Session = Depends(get_db)
):
    """审批操作接口"""
    if approval_action.action == "approved":
        success, message = ApprovalEngine.approve(
            db, request_id, approval_action.approver_id, approval_action.comment
        )
    elif approval_action.action == "rejected":
        success, message = ApprovalEngine.reject(
            db, request_id, approval_action.approver_id, 
            approval_action.comment, approval_action.reject_to_step
        )
    
    if success:
        return {"message": message}
    else:
        raise HTTPException(status_code=400, detail=message)

项目的优势

  1. 清晰的架构：状态定义、业务逻辑、API接口分离
  2. 完整的记录：每次状态转换都有详细记录
  3. 灵活的配置：审批流程可配置，支持不同的审批规则
  4. 易于扩展：可以轻松添加新的状态和转换规则
  5. 良好的文档：有详细的技术文档和API文档

如何在自己的项目中应用

  1. 明确状态：首先明确你的对象有哪些状态
  2. 定义转换规则：明确什么条件下可以从一个状态转换到另一个状态
  3. 选择实现方式：
    • 简单场景：使用if-else或switch-case
    • 复杂场景：使用状态机库或自己实现状态机类
  4. 记录状态变化：记录每次状态变化，便于追踪和调试
  5. 提供清晰的接口：让使用者能够方便地触发状态转换

总结

状态机是一种简单而强大的设计模式，它可以帮助我们更好地管理对象的生命周期。无论是简单的工作流还是复杂的AI应用，状态机都能提供清晰的结构和可预测的行为。

本指南从以下几个层面介绍了状态机：

  1. 基础概念：状态、事件、转换、动作四大核心要素
  2. 生活实例：红绿灯、订单、游戏角色等贴近生活的案例
  3. 代码实现：从简单的 if-else 到完整的状态机类
  4. AI应用：LangGraph 如何将状态机理念应用于 AI Agent
  5. 最佳实践：分离关注点、记录历史、灵活配置

关键要点：

  • 简单优先：不要过度设计，简单场景用简单方案
  • 状态明确：每个状态要有明确的含义和行为
  • 转换可控：所有状态转换都应该是可预测的
  • 记录追踪：记录状态变化历史，便于调试和审计
  • 可视化：用图表展示状态机，让团队成员更容易理解

状态机不仅仅是一种技术实现，更是一种思维方式——它帮助我们用结构化的方式思考系统的行为。当你下次遇到”这个东西有几种状态？什么时候会变？”这样的问题时，不妨画一张状态图，答案往往就清晰了。

参考资料

  • 状态机 - 维基百科
  • 有限状态机 - 百度百科
  • LangGraph 官方文档
  • 状态机模式 - Refactoring.Guru