状态机：对象生命周期管理与状态驱动的业务流程
业务系统中到处是"状态"——订单有状态、审批有状态、规则部署有状态、模型训练有状态。很多团队用 if-else 或 switch-case 管理状态转换，代码很快变得不可维护：
Coomia发布于 2025年12月19日15 分钟阅读
分享本文Twitter / X
状态机：对象生命周期管理与状态驱动的业务流程
“系列：S10 设计模式 · 第 5 篇 | 难度：高级 | 阅读时间：18 分钟
#TL;DR
- 有限状态机（FSM）是管理对象生命周期的经典模式。在 coomia-dip 中，每一个 Ontology 对象都可以拥有状态机定义，状态转换即业务流程推进。
- coomia-dip 将状态机嵌入 ObjectType 的 Schema 定义中，状态转换通过 Action 触发，转换条件由 Guard 函数校验，副作用（Side Effect）通过事件驱动执行。
- 结合 Saga 模式和事件溯源，状态机提供了可审计、可回放、可可视化的业务流程管理能力。
#引言：为什么需要状态机
业务系统中到处是"状态"——订单有状态、审批有状态、规则部署有状态、模型训练有状态。很多团队用 if-else 或 switch-case 管理状态转换，代码很快变得不可维护：
Python
# 反模式：散落各处的状态判断
def approve_order(order):
if order.status == "pending":
if order.amount > 10000:
if has_manager_approval(order):
order.status = "approved"
else:
order.status = "pending_manager"
else:
order.status = "approved"
elif order.status == "pending_manager":
if is_manager(current_user):
order.status = "approved"
else:
raise ValueError("Not authorized")
elif order.status == "approved":
raise ValueError("Already approved")
# ... 越来越多的分支
这段代码的问题：状态转换逻辑分散在业务代码中，没有全局视图，无法回答"这个对象可以从什么状态转换到什么状态"这个基本问题。
状态机模式将所有状态和转换集中定义，提供清晰的全局视图和严格的转换约束。
#一、状态机的核心模型
#1.1 coomia-dip 状态机定义
coomia-dip 在 Ontology Schema 层面定义状态机，使其成为 ObjectType 的一等公民：
Python
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, Awaitable
@dataclass(frozen=True)
class State:
"""A state in the state machine."""
name: str
display_name: str
description: str
is_initial: bool = False
is_terminal: bool = False
metadata: dict[str, Any] = field(default_factory=dict)
@dataclass(frozen=True)
class Transition:
"""A transition between states."""
name: str
from_state: str
to_state: str
action: str # 触发此转换的 Action
guard: str | None = None # Guard 条件表达式
side_effects: list[str] = field(default_factory=list)
required_permissions: list[str] = field(default_factory=list)
metadata: dict[str, Any] = field(default_factory=dict)
@dataclass
class StateMachineDefinition:
"""Complete state machine definition for an ObjectType."""
object_type: str
states: list[State]
transitions: list[Transition]
initial_state: str
history_enabled: bool = True # 是否记录状态变更历史
version: int = 1
def get_available_transitions(self, current_state: str) -> list[Transition]:
"""Get all transitions available from the current state."""
return [t for t in self.transitions if t.from_state == current_state]
def validate(self) -> list[str]:
"""Validate the state machine definition."""
errors: list[str] = []
state_names = {s.name for s in self.states}
# 检查初始状态存在
if self.initial_state not in state_names:
errors.append(f"Initial state '{self.initial_state}' not found")
# 检查转换引用的状态存在
for t in self.transitions:
if t.from_state not in state_names:
errors.append(f"Transition '{t.name}': from_state '{t.from_state}' not found")
if t.to_state not in state_names:
errors.append(f"Transition '{t.name}': to_state '{t.to_state}' not found")
# 检查是否有不可达状态
reachable = {self.initial_state}
changed = True
while changed:
changed = False
for t in self.transitions:
if t.from_state in reachable and t.to_state not in reachable:
reachable.add(t.to_state)
changed = True
unreachable = state_names - reachable
if unreachable:
errors.append(f"Unreachable states: {unreachable}")
# 检查至少有一个终态
terminals = [s for s in self.states if s.is_terminal]
if not terminals:
errors.append("No terminal states defined")
return errors
#1.2 Guard 条件
Guard 是状态转换前的校验函数。只有 Guard 返回 True，转换才允许执行：
Python
class GuardRegistry:
"""Registry for state machine guard functions."""
def __init__(self):
self._guards: dict[str, Callable[..., Awaitable[bool]]] = {}
def register(self, name: str, guard_fn: Callable[..., Awaitable[bool]]) -> None:
"""Register a guard function."""
self._guards[name] = guard_fn
async def evaluate(
self,
guard_name: str,
obj: dict[str, Any],
context: "TransitionContext",
) -> bool:
"""Evaluate a guard condition."""
guard_fn = self._guards.get(guard_name)
if guard_fn is None:
raise ValueError(f"Guard '{guard_name}' not registered")
return await guard_fn(obj, context)
# 注册具体的 Guard 实现
guard_registry = GuardRegistry()
async def amount_below_threshold(obj: dict, ctx: "TransitionContext") -> bool:
"""Guard: order amount below auto-approval threshold."""
return obj.get("amount", 0) <= ctx.config.get("auto_approve_threshold", 10000)
async def has_required_approvals(obj: dict, ctx: "TransitionContext") -> bool:
"""Guard: all required approvals have been received."""
required = obj.get("required_approvers", [])
received = obj.get("approvals", [])
return all(a in received for a in required)
async def data_quality_passed(obj: dict, ctx: "TransitionContext") -> bool:
"""Guard: data quality checks have passed."""
return obj.get("quality_score", 0) >= ctx.config.get("min_quality_score", 0.95)
guard_registry.register("amount_below_threshold", amount_below_threshold)
guard_registry.register("has_required_approvals", has_required_approvals)
guard_registry.register("data_quality_passed", data_quality_passed)
#1.3 状态机执行引擎
Python
@dataclass
class TransitionContext:
"""Context for a state transition."""
actor_id: str
actor_type: str
tenant_id: str
world_id: str
timestamp: datetime
config: dict[str, Any] = field(default_factory=dict)
metadata: dict[str, Any] = field(default_factory=dict)
@dataclass
class TransitionResult:
"""Result of a state transition attempt."""
success: bool
from_state: str
to_state: str | None
transition_name: str | None
error: str | None = None
side_effect_results: list[dict] = field(default_factory=list)
class StateMachineEngine:
"""Engine for executing state machine transitions."""
def __init__(
self,
definition: StateMachineDefinition,
guard_registry: GuardRegistry,
event_store: "EventStore",
side_effect_executor: "SideEffectExecutor",
):
self._definition = definition
self._guards = guard_registry
self._event_store = event_store
self._side_effects = side_effect_executor
async def transition(
self,
obj: dict[str, Any],
action: str,
context: TransitionContext,
) -> TransitionResult:
"""Attempt a state transition triggered by an action."""
current_state = obj.get("_state", self._definition.initial_state)
# 查找匹配的转换
matching = [
t for t in self._definition.transitions
if t.from_state == current_state and t.action == action
]
if not matching:
return TransitionResult(
success=False,
from_state=current_state,
to_state=None,
transition_name=None,
error=f"No transition for action '{action}' from state '{current_state}'",
)
# 评估 Guard 条件
for transition in matching:
if transition.guard:
guard_passed = await self._guards.evaluate(
transition.guard, obj, context
)
if not guard_passed:
continue
# 检查权限
if transition.required_permissions:
if not await self._check_permissions(
context.actor_id, transition.required_permissions
):
return TransitionResult(
success=False,
from_state=current_state,
to_state=transition.to_state,
transition_name=transition.name,
error="Insufficient permissions",
)
# 执行转换
obj["_state"] = transition.to_state
obj["_state_changed_at"] = context.timestamp.isoformat()
obj["_state_changed_by"] = context.actor_id
# 记录事件
await self._event_store.append([
DomainEvent(
event_id=generate_id(),
event_type="state_machine.transition",
aggregate_id=obj["_id"],
aggregate_type=self._definition.object_type,
sequence_number=await self._next_sequence(obj["_id"]),
timestamp=context.timestamp,
payload={
"from_state": current_state,
"to_state": transition.to_state,
"transition": transition.name,
"action": action,
},
metadata=EventMetadata(
actor_id=context.actor_id,
actor_type=context.actor_type,
tenant_id=context.tenant_id,
world_id=context.world_id,
source_plane="control",
trace_id=generate_trace_id(),
),
)
])
# 执行副作用
se_results = []
for se_name in transition.side_effects:
result = await self._side_effects.execute(se_name, obj, context)
se_results.append(result)
return TransitionResult(
success=True,
from_state=current_state,
to_state=transition.to_state,
transition_name=transition.name,
side_effect_results=se_results,
)
return TransitionResult(
success=False,
from_state=current_state,
to_state=None,
transition_name=None,
error="All guard conditions failed",
)
async def _check_permissions(
self, actor_id: str, required: list[str]
) -> bool:
"""Check if actor has required permissions."""
# 委托给权限服务
return True
async def _next_sequence(self, aggregate_id: str) -> int:
"""Get next sequence number for an aggregate."""
return 0
#二、实战：Ontology 对象的生命周期管理
#2.1 风控规则的生命周期
以风控规则为例，定义其完整的生命周期状态机：
Python
risk_rule_state_machine = StateMachineDefinition(
object_type="RiskRule",
initial_state="draft",
states=[
State("draft", "草稿", "规则正在编辑中", is_initial=True),
State("review", "审核中", "规则提交审核"),
State("testing", "测试中", "规则在沙箱中测试"),
State("approved", "已批准", "规则通过审核"),
State("deploying", "部署中", "规则正在部署到生产环境"),
State("active", "生效中", "规则在生产环境运行"),
State("suspended", "已暂停", "规则被临时暂停"),
State("deprecated", "已废弃", "规则已废弃"),
State("archived", "已归档", "规则已归档", is_terminal=True),
],
transitions=[
Transition("submit_review", "draft", "review", "submit",
guard="has_required_fields",
side_effects=["notify_reviewers"]),
Transition("approve", "review", "approved", "approve",
guard="has_required_approvals",
required_permissions=["rule.approve"],
side_effects=["notify_submitter"]),
Transition("reject", "review", "draft", "reject",
side_effects=["notify_submitter_rejection"]),
Transition("start_test", "approved", "testing", "test",
side_effects=["create_sandbox_instance"]),
Transition("test_passed", "testing", "approved", "pass_test",
guard="data_quality_passed",
side_effects=["record_test_results"]),
Transition("test_failed", "testing", "draft", "fail_test",
side_effects=["record_test_failures"]),
Transition("deploy", "approved", "deploying", "deploy",
required_permissions=["rule.deploy"],
side_effects=["trigger_deployment_saga"]),
Transition("deployment_complete", "deploying", "active", "activate",
side_effects=["notify_stakeholders"]),
Transition("deployment_failed", "deploying", "approved", "rollback",
side_effects=["notify_ops_team"]),
Transition("suspend", "active", "suspended", "suspend",
required_permissions=["rule.suspend"],
side_effects=["disable_rule_execution"]),
Transition("resume", "suspended", "active", "resume",
required_permissions=["rule.resume"],
side_effects=["enable_rule_execution"]),
Transition("deprecate", "active", "deprecated", "deprecate",
side_effects=["notify_dependents"]),
Transition("deprecate_suspended", "suspended", "deprecated", "deprecate",
side_effects=["notify_dependents"]),
Transition("archive", "deprecated", "archived", "archive",
side_effects=["cleanup_resources"]),
],
)
#2.2 状态机可视化
coomia-dip Platform Console 自动从 StateMachineDefinition 生成可视化的状态转换图：
Python
class StateMachineVisualizer:
"""Generate visual representations of state machines."""
def to_mermaid(self, definition: StateMachineDefinition) -> str:
"""Generate Mermaid state diagram."""
lines = ["stateDiagram-v2"]
# 初始状态
lines.append(f" [*] --> {definition.initial_state}")
# 转换
for t in definition.transitions:
label = t.name
if t.guard:
label += f" [{t.guard}]"
lines.append(f" {t.from_state} --> {t.to_state}: {label}")
# 终态
for s in definition.states:
if s.is_terminal:
lines.append(f" {s.name} --> [*]")
return "\n".join(lines)
def to_dot(self, definition: StateMachineDefinition) -> str:
"""Generate Graphviz DOT representation."""
lines = [
"digraph StateMachine {",
" rankdir=LR;",
' node [shape=box, style=rounded];',
]
for s in definition.states:
attrs = []
if s.is_initial:
attrs.append("peripheries=2")
if s.is_terminal:
attrs.append("shape=doublecircle")
attr_str = f" [{', '.join(attrs)}]" if attrs else ""
lines.append(f' {s.name} [label="{s.display_name}"{attr_str}];')
for t in definition.transitions:
label = t.action
if t.guard:
label += f"\\n[{t.guard}]"
lines.append(f' {t.from_state} -> {t.to_state} [label="{label}"];')
lines.append("}")
return "\n".join(lines)
#2.3 状态历史查询
启用 history_enabled 后，可以查询对象的完整状态变更历史：
Python
class StateHistoryService:
"""Service for querying state transition history."""
async def get_history(
self,
object_type: str,
object_id: str,
) -> list[dict[str, Any]]:
"""Get complete state transition history for an object."""
events = await self._event_store.get_events(
aggregate_id=object_id,
aggregate_type=object_type,
event_type="state_machine.transition",
)
return [
{
"from_state": e.payload["from_state"],
"to_state": e.payload["to_state"],
"transition": e.payload["transition"],
"action": e.payload["action"],
"actor": e.metadata.actor_id,
"timestamp": e.timestamp.isoformat(),
}
for e in events
]
async def get_state_at(
self,
object_type: str,
object_id: str,
timestamp: datetime,
) -> str:
"""Get the state of an object at a specific point in time."""
events = await self._event_store.get_events(
aggregate_id=object_id,
aggregate_type=object_type,
event_type="state_machine.transition",
before=timestamp,
)
if not events:
# 获取状态机定义的初始状态
definition = await self._get_definition(object_type)
return definition.initial_state
return events[-1].payload["to_state"]
async def get_dwell_time(
self,
object_type: str,
object_id: str,
state: str,
) -> timedelta:
"""Calculate how long an object has been in a specific state."""
events = await self._event_store.get_events(
aggregate_id=object_id,
aggregate_type=object_type,
event_type="state_machine.transition",
)
total = timedelta()
entered_at: datetime | None = None
for e in events:
if e.payload["to_state"] == state:
entered_at = e.timestamp
elif e.payload["from_state"] == state and entered_at:
total += e.timestamp - entered_at
entered_at = None
if entered_at:
total += datetime.utcnow() - entered_at
return total
#三、层次化状态机
#3.1 嵌套状态
对于复杂的业务流程，coomia-dip 支持层次化状态机——状态可以包含子状态机：
Python
@dataclass
class HierarchicalState(State):
"""A state that contains a sub-state machine."""
sub_machine: StateMachineDefinition | None = None
# 例如 "testing" 状态内部有子状态
testing_sub_machine = StateMachineDefinition(
object_type="RiskRule.testing",
initial_state="unit_test",
states=[
State("unit_test", "单元测试", "执行单元测试", is_initial=True),
State("integration_test", "集成测试", "执行集成测试"),
State("performance_test", "性能测试", "执行性能测试"),
State("sandbox_validation", "沙箱验证", "沙箱环境验证"),
State("test_complete", "测试完成", "所有测试通过", is_terminal=True),
],
transitions=[
Transition("unit_pass", "unit_test", "integration_test", "pass_unit"),
Transition("integration_pass", "integration_test", "performance_test", "pass_integration"),
Transition("perf_pass", "performance_test", "sandbox_validation", "pass_performance"),
Transition("sandbox_pass", "sandbox_validation", "test_complete", "pass_sandbox"),
],
)
#3.2 并行状态
某些场景需要并行的状态区域（Parallel Regions），例如一个审批流程同时需要法务审核和技术审核：
Python
@dataclass
class ParallelRegion:
"""A parallel region in a state machine."""
name: str
sub_machine: StateMachineDefinition
required: bool = True # 是否必须完成
@dataclass
class ParallelState(State):
"""A state with parallel regions."""
regions: list[ParallelRegion] = field(default_factory=list)
join_condition: str = "all" # all | any
# 审批状态同时需要法务和技术两个并行审核
legal_review = StateMachineDefinition(
object_type="RiskRule.review.legal",
initial_state="pending",
states=[
State("pending", "待审核", "", is_initial=True),
State("approved", "已通过", "", is_terminal=True),
State("rejected", "已拒绝", "", is_terminal=True),
],
transitions=[
Transition("approve", "pending", "approved", "legal_approve"),
Transition("reject", "pending", "rejected", "legal_reject"),
],
)
tech_review = StateMachineDefinition(
object_type="RiskRule.review.tech",
initial_state="pending",
states=[
State("pending", "待审核", "", is_initial=True),
State("approved", "已通过", "", is_terminal=True),
State("rejected", "已拒绝", "", is_terminal=True),
],
transitions=[
Transition("approve", "pending", "approved", "tech_approve"),
Transition("reject", "pending", "rejected", "tech_reject"),
],
)
#四、状态机与 Saga 的协作
#4.1 状态转换触发 Saga
当状态机转换涉及跨 Layer 操作时，转换的副作用（Side Effect）触发一个 Saga：
Python
class DeploymentSideEffect:
"""Side effect that triggers a deployment Saga."""
async def execute(
self,
obj: dict[str, Any],
context: TransitionContext,
) -> dict:
"""Trigger deployment Saga when rule transitions to 'deploying'."""
saga = CreateRiskRuleSaga().build({
"schema": obj["schema"],
"storage": obj["storage_config"],
"reasoning": obj["reasoning_config"],
"monitoring": obj["monitoring_config"],
})
result = await self._saga_orchestrator.execute(saga)
if result.status == "completed":
# Saga 成功，触发下一个状态转换
await self._state_engine.transition(
obj, "activate", context
)
else:
# Saga 失败，回滚状态
await self._state_engine.transition(
obj, "rollback", context
)
return {"saga_id": saga.saga_id, "status": result.status}
#4.2 Saga 步骤作为状态转换
反过来，Saga 的每个步骤的执行也可以建模为 Saga 实例自身的状态转换：
Code
Saga 状态机：
CREATED → STEP_1_RUNNING → STEP_1_COMPLETED → STEP_2_RUNNING → ...
↘ STEP_1_FAILED → COMPENSATING → COMPENSATED
#五、生产环境实践
#5.1 状态转换的并发安全
多个用户可能同时尝试对同一对象进行状态转换。coomia-dip 使用乐观锁确保并发安全：
Python
class ConcurrentStateMachine:
"""State machine with optimistic concurrency control."""
async def transition(
self, obj_id: str, action: str, context: TransitionContext
) -> TransitionResult:
"""Execute transition with optimistic locking."""
obj = await self._repository.get(obj_id)
current_version = obj["_version"]
result = await self._engine.transition(obj, action, context)
if result.success:
# 使用版本号进行乐观锁更新
updated = await self._repository.update(
obj_id, obj, expected_version=current_version
)
if not updated:
raise ConcurrencyError(
f"Object {obj_id} was modified concurrently"
)
return result
#5.2 状态转换超时
长时间停留在某个中间状态可能表示问题。coomia-dip 支持状态超时配置：
Python
@dataclass
class StateTimeout:
"""Timeout configuration for a state."""
state: str
timeout: timedelta
timeout_action: str # 超时后触发的 Action
notification_before: timedelta | None = None # 超时前通知
class StateTimeoutMonitor:
"""Monitor for state timeouts."""
async def check_timeouts(self) -> list[dict]:
"""Check all objects for state timeouts."""
timed_out = []
for timeout_config in self._timeout_configs:
objects = await self._repository.find_in_state(
timeout_config.state,
entered_before=datetime.utcnow() - timeout_config.timeout,
)
for obj in objects:
timed_out.append({
"object_id": obj["_id"],
"state": timeout_config.state,
"timeout_action": timeout_config.timeout_action,
})
# 自动触发超时 Action
await self._engine.transition(
obj, timeout_config.timeout_action,
TransitionContext(
actor_id="system",
actor_type="timeout_monitor",
tenant_id=obj["_tenant_id"],
world_id=obj["_world_id"],
timestamp=datetime.utcnow(),
),
)
return timed_out
#5.3 状态统计与分析
Python
class StateAnalyticsService:
"""Analytics for state machine usage."""
async def get_state_distribution(
self, object_type: str, tenant_id: str
) -> dict[str, int]:
"""Get current distribution of objects across states."""
return await self._repository.count_by_state(object_type, tenant_id)
async def get_avg_dwell_time(
self, object_type: str, tenant_id: str
) -> dict[str, float]:
"""Get average dwell time per state."""
pass
async def get_transition_frequency(
self, object_type: str, tenant_id: str, time_range: tuple
) -> dict[str, int]:
"""Get frequency of each transition in a time range."""
pass
async def get_bottleneck_states(
self, object_type: str, tenant_id: str
) -> list[dict]:
"""Identify states where objects tend to get stuck."""
distribution = await self.get_state_distribution(object_type, tenant_id)
dwell_times = await self.get_avg_dwell_time(object_type, tenant_id)
bottlenecks = []
for state, count in distribution.items():
avg_dwell = dwell_times.get(state, 0)
if count > 10 and avg_dwell > 86400: # 超过 1 天
bottlenecks.append({
"state": state,
"count": count,
"avg_dwell_seconds": avg_dwell,
})
return sorted(bottlenecks, key=lambda x: x["count"], reverse=True)
#六、反模式与最佳实践
#6.1 避免状态爆炸
状态数量过多会导致状态机难以理解和维护。经验法则：
- 单层状态机不超过 12 个状态
- 使用层次化状态机管理复杂流程
- 用"元数据"替代"微状态"——不需要为每个细微差异创建新状态
#6.2 状态转换应该是原子的
一个转换要么完全成功（状态变更 + 副作用执行），要么完全失败（状态不变）。不要出现"状态变了但副作用没执行"的情况。
#6.3 终态必须是稳定的
终态（Terminal State）意味着对象不再接受任何状态转换。不要在终态上定义出边（outgoing transition）。如果需要"重新激活"已归档的对象，应该创建新对象而不是修改终态。
#Key Takeaways
- 一等公民：状态机是 coomia-dip ObjectType Schema 的一等公民，状态转换即业务流程推进
- Guard + Side Effect：Guard 条件控制转换是否允许，Side Effect 执行转换后的副作用
- 事件驱动：每次状态转换都记录为事件，支持审计、回放和时间旅行查询
- 层次化：复杂流程通过嵌套状态机和并行区域管理
- 与 Saga 协作：跨 Layer 的状态转换通过 Saga 确保最终一致性
- 可视化：自动从定义生成状态转换图，提供全局视图
#Next Article
下一篇我们将探讨沙箱模式（Sandbox Pattern）——coomia-dip 如何为规则测试、策略验证和 Agent 调试提供安全隔离的执行环境。
S10-06: 沙箱模式：安全隔离的执行环境
#Tags
#设计模式 #状态机 #FSM #生命周期 #Guard #SideEffect #层次化状态机 #并行状态 #对象管理