Mutation Rules：声明式状态变更编排
Mutation Rules 是 coomia-dip 中连接业务规则与 Action 执行的桥梁。业务人员通过声明式 YAML/JSON 定义"当条件满足时，执行什么操作"，系统自动将这些规则编译为 ActionRequest 序列并通过 ActionEngine 执行。本文深入解析 Mutation Rules 的语法设计、条件表达式引擎、规则冲突检测、执行优先级、与 Ontology 事件的绑定机制以及规则版本管理，展示如何用声明式方式编排复杂的状态变更逻辑。
Coomia发布于 2025年9月5日16 分钟阅读
分享本文Twitter / X
“系列：S5 智能决策 · 第 14 篇 | 难度：高级 | 阅读时间：20 分钟
Mutation Rules：声明式状态变更编排
#TL;DR
Mutation Rules 是 coomia-dip 中连接业务规则与 Action 执行的桥梁。业务人员通过声明式 YAML/JSON 定义"当条件满足时，执行什么操作"，系统自动将这些规则编译为 ActionRequest 序列并通过 ActionEngine 执行。本文深入解析 Mutation Rules 的语法设计、条件表达式引擎、规则冲突检测、执行优先级、与 Ontology 事件的绑定机制以及规则版本管理，展示如何用声明式方式编排复杂的状态变更逻辑。
#1. 为什么需要声明式变更规则
#1.1 命令式编排的痛点
传统方式中，状态变更逻辑分散在代码的各个角落：
Code
命令式方法（传统）：
if order.status == "approved" and order.amount > 10000:
create_payment_request(order)
update_inventory(order.items)
send_notification(order.owner, "approved")
if order.is_international:
trigger_compliance_check(order)
create_audit_log(order, "approved")
问题：
| 问题 | 说明 |
|---|---|
| 规则散落 | 变更逻辑分布在多个代码文件中 |
| 不可审计 | 无法一目了然看到所有变更规则 |
| 修改成本高 | 每次变更都需要改代码、部署 |
| 业务不可见 | 业务人员无法理解代码逻辑 |
| 测试困难 | 无法独立测试单个规则 |
#1.2 声明式的优势
Code
声明式方法（coomia-dip）：
┌────────────────────────────────────────────┐
│ Mutation Rule: "order_approval_actions" │
│ │
│ WHEN: │
│ object.type == "Order" │
│ AND object.status CHANGED TO "approved" │
│ AND object.amount > 10000 │
│ │
│ THEN: │
│ 1. CreateObject("PaymentRequest", ...) │
│ 2. InvokeFunction("update_inventory") │
│ 3. Notification("owner", "approved") │
│ │
│ IF object.is_international: │
│ 4. InvokeFunction("compliance_check") │
└────────────────────────────────────────────┘
#2. Mutation Rule 语法设计
#2.1 完整语法结构
YAML
# Mutation Rule 定义
apiVersion: coomia-dip/v1
kind: MutationRule
metadata:
name: order-approval-actions
namespace: supply-chain
version: "1.2.0"
labels:
domain: procurement
priority: high
annotations:
description: "订单审批通过后的自动化操作序列"
author: "business-team"
spec:
# 触发条件
trigger:
type: ontology_event # ontology_event | schedule | manual
object_type: Order
event: property_changed
filter:
property: status
from: ["pending", "reviewing"]
to: "approved"
# 前置条件（额外的守卫条件）
conditions:
- expr: "object.amount > 10000"
description: "仅大额订单"
- expr: "object.department != 'test'"
description: "排除测试部门"
# 执行动作序列
actions:
- name: create-payment-request
executor: CreateObject
params:
object_type: PaymentRequest
properties:
order_id: "{{ object.id }}"
amount: "{{ object.amount }}"
currency: "{{ object.currency }}"
requested_by: "{{ event.triggered_by }}"
status: pending
- name: update-inventory
executor: InvokeFunction
params:
function_rid: ri.function.inventory.reserve
arguments:
items: "{{ object.line_items }}"
warehouse: "{{ object.warehouse_id }}"
- name: notify-owner
executor: Notification
params:
channel: "{{ object.owner.preferred_channel }}"
template_id: tpl_order_approved
recipients:
- "{{ object.owner.email }}"
variables:
order_id: "{{ object.id }}"
amount: "{{ object.amount | format_currency }}"
- name: compliance-check
executor: InvokeFunction
when: "object.is_international == true"
params:
function_rid: ri.function.compliance.check
arguments:
order: "{{ object }}"
# 执行配置
execution:
mode: sequential # sequential | parallel | saga
stop_on_failure: true
timeout_seconds: 120
idempotency: true
dry_run_enabled: true
# 补偿策略（mode=saga 时生效）
compensation:
enabled: true
max_retries: 3
#2.2 模板表达式
Mutation Rules 使用 Jinja2 风格的模板表达式：
Python
class TemplateEngine:
"""规则模板表达式引擎"""
def __init__(self):
self.env = jinja2.Environment(
undefined=jinja2.StrictUndefined,
)
self._register_filters()
def _register_filters(self) -> None:
"""注册自定义过滤器"""
self.env.filters.update({
"format_currency": self._format_currency,
"to_json": json.dumps,
"first": lambda lst: lst[0] if lst else None,
"last": lambda lst: lst[-1] if lst else None,
"sum_field": lambda lst, f: sum(
item.get(f, 0) for item in lst
),
"now": lambda _: datetime.utcnow().isoformat(),
})
def render(self, template_str: str,
context: dict) -> Any:
"""渲染模板表达式"""
if not isinstance(template_str, str):
return template_str
if "{{" not in template_str:
return template_str
template = self.env.from_string(template_str)
result = template.render(**context)
# 自动类型推断
try:
return json.loads(result)
except (json.JSONDecodeError, TypeError):
return result
#2.3 条件表达式引擎
Python
class ConditionEvaluator:
"""条件表达式求值器"""
OPERATORS = {
"==": operator.eq,
"!=": operator.ne,
">": operator.gt,
">=": operator.ge,
"<": operator.lt,
"<=": operator.le,
"in": lambda a, b: a in b,
"not_in": lambda a, b: a not in b,
"contains": lambda a, b: b in a,
"starts_with": lambda a, b: a.startswith(b),
"ends_with": lambda a, b: a.endswith(b),
"matches": lambda a, b: bool(re.match(b, a)),
"is_null": lambda a, _: a is None,
"is_not_null": lambda a, _: a is not None,
}
def evaluate(self, expr: str, context: dict) -> bool:
"""
求值条件表达式。
支持的语法：
- object.field == "value"
- object.amount > 10000
- object.status in ["a", "b"]
- object.tags contains "urgent"
"""
ast = self._parse(expr)
return self._eval_node(ast, context)
def evaluate_change_condition(
self, filter_spec: dict, event: dict
) -> bool:
"""评估属性变更条件"""
prop = filter_spec["property"]
old_value = event.get("old_values", {}).get(prop)
new_value = event.get("new_values", {}).get(prop)
if "from" in filter_spec:
from_values = filter_spec["from"]
if isinstance(from_values, list):
if old_value not in from_values:
return False
elif old_value != from_values:
return False
if "to" in filter_spec:
if new_value != filter_spec["to"]:
return False
return True
#3. 规则编译与执行
#3.1 规则编译流程
Code
┌──────────┐ ┌──────────────┐ ┌──────────────┐ ┌──────────┐
│ YAML/JSON│────→│ Schema Valid.│────→│ Compile to │────→│ Register │
│ Rule Def │ │ (JSON Schema)│ │ ActionPlan │ │ in Engine│
└──────────┘ └──────┬───────┘ └──────┬───────┘ └──────────┘
│ │
Validation Optimization
Errors - Template pre-compile
- Condition index
- Dependency graph
Python
class MutationRuleCompiler:
"""规则编译器：将声明式规则编译为可执行的 ActionPlan"""
def compile(self, rule_yaml: str) -> CompiledRule:
"""编译规则定义"""
# 1. 解析 YAML
rule_def = yaml.safe_load(rule_yaml)
# 2. Schema 校验
self._validate_schema(rule_def)
# 3. 预编译模板
compiled_actions = []
for action_def in rule_def["spec"]["actions"]:
compiled = CompiledAction(
name=action_def["name"],
executor_type=ExecutorType(action_def["executor"]),
param_templates=self._precompile_templates(
action_def["params"]
),
condition=action_def.get("when"),
)
compiled_actions.append(compiled)
# 4. 构建条件索引
trigger = rule_def["spec"]["trigger"]
condition_index = ConditionIndex(
object_type=trigger["object_type"],
event_type=trigger["event"],
property_filter=trigger.get("filter"),
guard_conditions=rule_def["spec"].get("conditions", []),
)
return CompiledRule(
metadata=rule_def["metadata"],
condition_index=condition_index,
actions=compiled_actions,
execution_config=rule_def["spec"]["execution"],
compensation_config=rule_def["spec"].get("compensation"),
)
#3.2 规则执行引擎
Python
class MutationRuleEngine:
"""规则执行引擎"""
def __init__(self, action_engine: ActionScheduler):
self.action_engine = action_engine
self.rules: dict[str, CompiledRule] = {}
self.template_engine = TemplateEngine()
self.condition_eval = ConditionEvaluator()
async def on_ontology_event(self, event: dict) -> list[ActionResult]:
"""响应 Ontology 事件，匹配并执行规则"""
# 1. 匹配规则
matched_rules = self._match_rules(event)
# 2. 按优先级排序
matched_rules.sort(
key=lambda r: r.metadata.get("labels", {}).get(
"priority_weight", 0
),
reverse=True,
)
# 3. 冲突检测
self._check_conflicts(matched_rules, event)
# 4. 执行规则
all_results: list[ActionResult] = []
for rule in matched_rules:
results = await self._execute_rule(rule, event)
all_results.extend(results)
return all_results
def _match_rules(self, event: dict) -> list[CompiledRule]:
"""匹配适用的规则"""
matched = []
for rule in self.rules.values():
idx = rule.condition_index
# 对象类型匹配
if idx.object_type != event.get("object_type"):
continue
# 事件类型匹配
if idx.event_type != event.get("event_type"):
continue
# 属性过滤匹配
if idx.property_filter:
if not self.condition_eval.evaluate_change_condition(
idx.property_filter, event
):
continue
# 守卫条件
context = {"object": event.get("object"), "event": event}
guards_pass = all(
self.condition_eval.evaluate(g["expr"], context)
for g in idx.guard_conditions
)
if not guards_pass:
continue
matched.append(rule)
return matched
async def _execute_rule(
self, rule: CompiledRule, event: dict
) -> list[ActionResult]:
"""执行单个规则的所有动作"""
context = {
"object": event.get("object"),
"event": event,
"now": datetime.utcnow().isoformat(),
}
action_requests: list[ActionRequest] = []
for action in rule.actions:
# 检查条件动作
if action.condition:
if not self.condition_eval.evaluate(
action.condition, context
):
continue
# 渲染参数模板
params = self.template_engine.render_deep(
action.param_templates, context
)
request = ActionRequest(
executor_type=action.executor_type,
target_object_type=params.get("object_type"),
parameters=params,
triggered_by=f"rule:{rule.metadata['name']}",
context=context,
)
action_requests.append(request)
# 根据执行模式分发
mode = rule.execution_config.get("mode", "sequential")
if mode == "saga":
return await self._execute_as_saga(
action_requests, rule
)
elif mode == "parallel":
tasks = [
self.action_engine.dispatch(req)
for req in action_requests
]
return await asyncio.gather(*tasks)
else: # sequential
results = []
for req in action_requests:
result = await self.action_engine.dispatch(req)
results.append(result)
if (result.status == ActionStatus.FAILED
and rule.execution_config.get(
"stop_on_failure", True)):
break
return results
#4. 规则冲突检测
#4.1 冲突类型
Code
┌─────────────────────────────────────────────────────────────┐
│ 规则冲突检测 │
│ │
│ 类型 1: 写-写冲突 │
│ ┌────────────┐ ┌────────────┐ │
│ │ Rule A: │ │ Rule B: │ 同一对象的同一属性 │
│ │ set status │ │ set status │ 被两条规则修改 │
│ │ = "active" │ │ = "frozen" │ │
│ └────────────┘ └────────────┘ │
│ │
│ 类型 2: 顺序依赖 │
│ ┌────────────┐ ┌────────────┐ │
│ │ Rule A: │ │ Rule B: │ Rule B 依赖 │
│ │ create Obj │ │ update Obj │ Rule A 创建的对象 │
│ └────────────┘ └────────────┘ │
│ │
│ 类型 3: 循环触发 │
│ ┌────────────┐ ┌────────────┐ │
│ │ Rule A: │ │ Rule B: │ Rule A 触发 Rule B │
│ │ on X → Y │ │ on Y → X │ Rule B 又触发 Rule A │
│ └────────────┘ └────────────┘ │
└─────────────────────────────────────────────────────────────┘
#4.2 冲突检测器
Python
class ConflictDetector:
"""规则冲突检测器"""
def detect_conflicts(
self, rules: list[CompiledRule]
) -> list[Conflict]:
conflicts: list[Conflict] = []
# 检测写-写冲突
conflicts.extend(self._detect_write_write(rules))
# 检测循环触发
conflicts.extend(self._detect_cycles(rules))
return conflicts
def _detect_write_write(
self, rules: list[CompiledRule]
) -> list[Conflict]:
"""检测多条规则修改同一对象同一属性"""
write_map: dict[str, list[str]] = {} # "Type.prop" → [rule_names]
for rule in rules:
for action in rule.actions:
if action.executor_type in (
ExecutorType.UPDATE_OBJECT,
ExecutorType.CREATE_OBJECT,
):
obj_type = action.param_templates.get("object_type", "")
for prop in action.param_templates.get(
"properties", {}
).keys():
key = f"{obj_type}.{prop}"
write_map.setdefault(key, []).append(
rule.metadata["name"]
)
conflicts = []
for key, rule_names in write_map.items():
if len(rule_names) > 1:
conflicts.append(Conflict(
type="write_write",
target=key,
rules=rule_names,
severity="warning",
message=(
f"Property {key} is modified by multiple "
f"rules: {rule_names}"
),
))
return conflicts
def _detect_cycles(
self, rules: list[CompiledRule]
) -> list[Conflict]:
"""检测规则间的循环触发"""
# 构建触发图
graph: dict[str, set[str]] = {}
for rule in rules:
trigger_type = rule.condition_index.object_type
produced_types = set()
for action in rule.actions:
if action.executor_type == ExecutorType.CREATE_OBJECT:
produced_types.add(
action.param_templates.get("object_type", "")
)
elif action.executor_type == ExecutorType.UPDATE_OBJECT:
produced_types.add(
action.param_templates.get("object_type", "")
)
graph[trigger_type] = produced_types
# DFS 检测环
cycles = self._find_cycles(graph)
return [
Conflict(
type="cycle",
target=" → ".join(cycle),
rules=[],
severity="error",
message=f"Circular trigger detected: {' → '.join(cycle)}",
)
for cycle in cycles
]
#5. 规则优先级与排序
#5.1 优先级模型
Code
优先级决策树：
┌─────────────┐
│ 多条规则匹配 │
└──────┬──────┘
│
┌──────┴──────┐
│ 显式优先级？ │
└──────┬──────┘
Yes │ No
┌──────┴──────┐
│ 按 priority │
│ 排序执行 │
└─────────────┘
│
┌──────┴──────┐
│ 同优先级？ │
└──────┬──────┘
Yes │
┌──────┴──────┐
│ 按规则名称 │
│ 字母序 │
└─────────────┘
#5.2 优先级配置
YAML
# 高优先级：安全相关
metadata:
labels:
priority: critical # critical > high > normal > low
priority_weight: 1000
# 当多条同优先级规则冲突时的策略
spec:
execution:
conflict_resolution: first_match # first_match | all | merge
| 策略 | 说明 | 适用场景 |
|---|---|---|
| first_match | 只执行第一条匹配的规则 | 互斥规则 |
| all | 执行所有匹配的规则 | 独立规则 |
| merge | 合并多条规则的动作列表 | 补充规则 |
#6. 规则版本管理
#6.1 版本控制模型
Python
class RuleVersionManager:
"""规则版本管理器"""
async def publish(self, rule_yaml: str,
author: str) -> RuleVersion:
"""发布新版本"""
rule_def = yaml.safe_load(rule_yaml)
name = rule_def["metadata"]["name"]
version = rule_def["metadata"]["version"]
# 编译验证
compiled = self.compiler.compile(rule_yaml)
# 冲突检测
existing_rules = await self._get_active_rules()
conflicts = self.conflict_detector.detect_conflicts(
existing_rules + [compiled]
)
errors = [c for c in conflicts if c.severity == "error"]
if errors:
raise RuleConflictError(errors)
# 存储版本
rule_version = RuleVersion(
name=name,
version=version,
definition=rule_yaml,
compiled=compiled,
author=author,
status="draft",
created_at=datetime.utcnow(),
)
await self.store.save(rule_version)
return rule_version
async def activate(self, name: str, version: str) -> None:
"""激活指定版本（原子切换）"""
# 停用当前激活版本
current = await self.store.get_active(name)
if current:
current.status = "inactive"
await self.store.save(current)
# 激活新版本
target = await self.store.get(name, version)
target.status = "active"
await self.store.save(target)
# 注册到规则引擎
self.rule_engine.register(target.compiled)
async def rollback(self, name: str) -> RuleVersion:
"""回滚到上一个版本"""
history = await self.store.get_history(name)
if len(history) < 2:
raise ValueError("No previous version to rollback to")
previous = history[-2]
await self.activate(name, previous.version)
return previous
#6.2 版本历史
Code
┌──────────────────────────────────────────────────────┐
│ Rule: order-approval-actions │
│ │
│ Version │ Status │ Author │ Date │
│ ─────────┼──────────┼─────────┼────────────── │
│ v1.0.0 │ inactive │ alice │ 2026-01-15 │
│ v1.1.0 │ inactive │ alice │ 2026-02-01 │
│ v1.2.0 │ active │ bob │ 2026-03-10 ←当前 │
│ v1.3.0 │ draft │ carol │ 2026-03-24 │
└──────────────────────────────────────────────────────┘
#7. Dry-Run 与测试
#7.1 Dry-Run 模式
Python
class RuleDryRunner:
"""规则 dry-run 测试器"""
async def dry_run(
self, rule_name: str, mock_event: dict
) -> DryRunResult:
"""模拟执行规则，不实际修改数据"""
rule = self.rule_engine.rules.get(rule_name)
if not rule:
raise KeyError(f"Rule not found: {rule_name}")
# 评估条件
context = {
"object": mock_event.get("object"),
"event": mock_event,
}
conditions_met = all(
self.condition_eval.evaluate(g["expr"], context)
for g in rule.condition_index.guard_conditions
)
# 渲染动作（不执行）
planned_actions = []
for action in rule.actions:
if action.condition:
if not self.condition_eval.evaluate(
action.condition, context
):
planned_actions.append({
"name": action.name,
"skipped": True,
"reason": f"Condition not met: {action.condition}",
})
continue
params = self.template_engine.render_deep(
action.param_templates, context
)
planned_actions.append({
"name": action.name,
"executor": action.executor_type.value,
"resolved_params": params,
"skipped": False,
})
return DryRunResult(
rule_name=rule_name,
conditions_met=conditions_met,
planned_actions=planned_actions,
warnings=[],
)
#7.2 规则测试框架
Python
class RuleTestCase(BaseModel):
"""规则测试用例"""
name: str
description: str
mock_event: dict
expected_conditions_met: bool
expected_actions: list[str] # 期望执行的 action 名称列表
expected_skipped: list[str] = []
# 测试用例定义
test_cases = [
RuleTestCase(
name="approved_large_order",
description="大额订单审批通过应触发全部动作",
mock_event={
"object_type": "Order",
"event_type": "property_changed",
"old_values": {"status": "pending"},
"new_values": {"status": "approved"},
"object": {
"id": "ORD-001",
"amount": 50000,
"department": "procurement",
"is_international": False,
"owner": {"email": "owner@co.com",
"preferred_channel": "email"},
},
},
expected_conditions_met=True,
expected_actions=[
"create-payment-request",
"update-inventory",
"notify-owner",
],
expected_skipped=["compliance-check"],
),
RuleTestCase(
name="small_order_skipped",
description="小额订单不应触发规则",
mock_event={
"object_type": "Order",
"event_type": "property_changed",
"old_values": {"status": "pending"},
"new_values": {"status": "approved"},
"object": {
"id": "ORD-002",
"amount": 500,
"department": "procurement",
},
},
expected_conditions_met=False,
expected_actions=[],
),
]
#8. 与 Ontology 事件的绑定
#8.1 事件订阅架构
Code
┌─────────────────────────────────────────────────────┐
│ Ontology Event Bus │
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Create │ │ Update │ │ Delete │ │
│ │ Events │ │ Events │ │ Events │ │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘ │
│ └──────────────┼──────────────┘ │
│ │ │
│ ┌───────┴────────┐ │
│ │ Event Router │ │
│ └───────┬────────┘ │
│ │ │
│ ┌────────────┼────────────┐ │
│ ▼ ▼ ▼ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │Rule Eng. │ │CDC Stream│ │ Audit │ │
│ │Mutation │ │ (Kafka) │ │ Logger │ │
│ │Rules │ │ │ │ │ │
│ └──────────┘ └──────────┘ └──────────┘ │
└─────────────────────────────────────────────────────┘
#8.2 防循环触发
Python
class CircuitBreaker:
"""防止规则循环触发的断路器"""
def __init__(self, max_depth: int = 5,
max_executions_per_event: int = 50):
self.max_depth = max_depth
self.max_executions = max_executions_per_event
self._execution_counts: dict[str, int] = {}
def should_execute(self, event: dict,
rule_name: str) -> bool:
"""判断是否应该执行规则"""
# 检查触发深度
depth = event.get("_trigger_depth", 0)
if depth >= self.max_depth:
self.logger.warning(
f"Max trigger depth ({self.max_depth}) reached "
f"for rule {rule_name}"
)
return False
# 检查执行次数
root_event_id = event.get("_root_event_id", event.get("event_id"))
key = f"{root_event_id}:{rule_name}"
count = self._execution_counts.get(key, 0)
if count >= self.max_executions:
self.logger.warning(
f"Max executions ({self.max_executions}) reached "
f"for rule {rule_name}"
)
return False
self._execution_counts[key] = count + 1
return True
#9. 性能优化
#9.1 规则索引
Python
class RuleIndex:
"""基于对象类型和事件类型的规则索引"""
def __init__(self):
# 两级索引: object_type → event_type → [rules]
self._index: dict[str, dict[str, list[CompiledRule]]] = {}
def add(self, rule: CompiledRule) -> None:
obj_type = rule.condition_index.object_type
evt_type = rule.condition_index.event_type
self._index.setdefault(obj_type, {}).setdefault(
evt_type, []
).append(rule)
def match(self, object_type: str,
event_type: str) -> list[CompiledRule]:
"""O(1) 查找候选规则"""
return (
self._index
.get(object_type, {})
.get(event_type, [])
)
| 优化策略 | 效果 |
|---|---|
| 两级索引 | 事件匹配从 O(n) 降为 O(1) |
| 模板预编译 | 避免重复解析 Jinja2 模板 |
| 条件短路 | 第一个不满足的条件立即跳过 |
| 批量事件 | 合并同类事件减少规则匹配次数 |
#Key Takeaways
- 声明式优于命令式：Mutation Rules 将状态变更逻辑从代码中抽取为可审计、可测试的 YAML 声明
- 模板表达式：Jinja2 风格的模板支持动态参数渲染，引用事件上下文中的任何字段
- 冲突检测：编译时自动检测写-写冲突和循环触发，防止规则交互产生意外行为
- 多种执行模式：sequential、parallel、saga 三种模式适应不同的一致性和性能需求
- 版本管理：规则支持版本控制、灰度激活和一键回滚，变更风险可控
- Dry-Run：发布前可通过模拟事件测试规则行为，避免生产事故
#Next Article
下一篇 S5-15 Webhook 回写与外部系统集成 将详解如何通过 Webhook 执行器实现与外部系统的双向数据同步，包括签名验证、重试策略和幂等保证。
tags: mutation-rules, declarative, state-machine, ontology-event, rule-engine, coomia-dip