“

系列：S12 开发者教程 · 第 5 篇 | 难度：入门 | 阅读时间：15 分钟

第一条规则：用 YAML 创建自动化推理

#引言

在 coomia-dip 中，规则（Rule）是实现自动化推理和业务逻辑的核心机制。通过 YAML 声明式定义规则，你可以让平台自动响应数据变化、执行业务策略和触发工作流。本教程将带你创建第一条规则，并理解 coomia-dip 推理引擎的工作原理。

#核心概念

#什么是规则？

规则是一个"条件-动作"对：当某个条件满足时，自动执行对应的操作。coomia-dip 的规则引擎运行在 Intelligence Layer（Reasoning & Decision Layer）上，基于 Python + FastAPI 构建，通过 gRPC 与其他 Layer 通信。

规则的结构：

YAML

name: rule_name
trigger: event_or_schedule
condition: boolean_expression
actions:
  - action_to_execute

#规则类型

coomia-dip 支持三种规则类型：

事件驱动规则（Event-Driven）：数据变化时触发
定时规则（Scheduled）：按时间计划执行
推理规则（Inference）：基于知识图谱推理

#创建事件驱动规则

#场景：高优先级项目自动通知

当项目的优先级被设为 "critical" 时，自动发送通知给部门负责人。

YAML

# rules/critical_project_alert.yaml
name: critical_project_alert
display_name: 关键项目预警
description: 当项目优先级变为 critical 时自动通知

trigger:
  type: data_change
  object_type: Project
  events: [update]
  watch_properties: [priority]

condition:
  all:
    - property: priority
      operator: eq
      value: critical
    - property: status
      operator: in
      values: [planning, in_progress]

actions:
  - type: execute_action
    action_name: send_notification
    parameters:
      channel: "dingtalk"
      template: "critical_project_alert"
      recipients:
        query:
          relation: owns_project
          direction: source
          target: $trigger.object_rid
          select: [dept_id, name]

  - type: update_object
    object_type: Project
    target: $trigger.object_rid
    properties:
      tags:
        append: "needs-attention"

metadata:
  priority: high
  enabled: true
  created_by: admin

#加载和激活规则

Python

from ontology_sdk import OntoPlatform

platform = OntoPlatform(
    control_plane_url="localhost:50051",
    intelligence_plane_url="localhost:50053"
)

# 从 YAML 加载规则
rule = platform.rules.load_from_yaml("rules/critical_project_alert.yaml")
print(f"规则已加载: {rule.name}")

# 激活规则
platform.rules.activate(rule.name)
print(f"规则已激活")

# 测试触发
result = platform.rules.test(
    rule_name="critical_project_alert",
    test_data={
        "object_type": "Project",
        "event": "update",
        "object_rid": "ri.ontology.object.project.test-001",
        "changed_properties": {
            "priority": {"old": "high", "new": "critical"}
        }
    }
)
print(f"测试结果: {result.would_trigger}: {result.actions_to_execute}")

#创建定时规则

#场景：每日项目进度检查

YAML

# rules/daily_project_check.yaml
name: daily_project_check
display_name: 每日项目进度检查
description: 每天早上 9 点检查所有进行中的项目

trigger:
  type: schedule
  cron: "0 9 * * *"
  timezone: Asia/Shanghai

condition:
  always: true

actions:
  - type: query_and_process
    query:
      object_type: Project
      filter:
        status:
          eq: in_progress
        end_date:
          lt: $today_plus_7d
    for_each: project
    actions:
      - type: execute_action
        action_name: send_notification
        parameters:
          channel: email
          template: project_deadline_reminder
          recipients:
            query:
              relation: participates_in
              direction: source
              target: $project.rid
              filter:
                role: owner
          data:
            project_name: $project.properties.name
            end_date: $project.properties.end_date
            days_remaining: $project.properties.end_date.diff_days($today)

metadata:
  priority: medium
  enabled: true

#创建推理规则

#场景：自动计算部门风险等级

YAML

# rules/department_risk_inference.yaml
name: department_risk_inference
display_name: 部门风险推理
description: 根据部门下项目状态自动推理部门风险等级

trigger:
  type: data_change
  object_type: Project
  events: [create, update, delete]
  watch_properties: [status, priority]

condition:
  always: true

actions:
  - type: inference
    logic: |
      # 获取项目所属部门
      department = get_related(
          relation="owns_project",
          direction="source",
          target=$trigger.object_rid
      )

      # 统计部门下各状态项目数量
      projects = query(
          object_type="Project",
          relation="owns_project",
          source=department.rid
      )

      critical_count = count(projects, priority="critical")
      overdue_count = count(projects, status="in_progress", end_date < today())
      total_count = len(projects)

      # 推理风险等级
      if critical_count >= 3 or overdue_count >= 2:
          risk_level = "high"
      elif critical_count >= 1 or overdue_count >= 1:
          risk_level = "medium"
      else:
          risk_level = "low"

      # 更新部门风险属性
      update(department, derived_risk_level=risk_level)

metadata:
  priority: high
  enabled: true

#规则链（Rule Chaining）

多条规则可以形成链式反应：

YAML

# rules/cascade_risk_escalation.yaml
name: cascade_risk_escalation
display_name: 级联风险升级
description: 当部门风险变为 high 时，通知上级部门

trigger:
  type: data_change
  object_type: Department
  events: [update]
  watch_properties: [derived_risk_level]

condition:
  property: derived_risk_level
  operator: eq
  value: high

actions:
  - type: query_and_process
    query:
      relation: parent_department
      source: $trigger.object_rid
    for_each: parent_dept
    actions:
      - type: execute_action
        action_name: send_notification
        parameters:
          channel: dingtalk
          template: risk_escalation
          data:
            child_dept: $trigger.object.properties.name
            risk_level: high

#规则管理

#查看所有规则

Python

rules = platform.rules.list()
for r in rules:
    status = "Active" if r.enabled else "Disabled"
    print(f"  [{status}] {r.name}: {r.display_name}")
    print(f"    触发: {r.trigger.type}")
    print(f"    最后触发: {r.last_triggered_at}")

#查看规则执行日志

Python

logs = platform.rules.get_execution_logs(
    rule_name="critical_project_alert",
    limit=20
)
for log in logs:
    print(f"  {log.timestamp}: {log.status}")
    print(f"    触发对象: {log.trigger_object_rid}")
    print(f"    执行动作: {log.actions_executed}")
    if log.error:
        print(f"    错误: {log.error}")

#暂停和恢复规则

Python

# 暂停规则
platform.rules.deactivate("critical_project_alert")

# 恢复规则
platform.rules.activate("critical_project_alert")

# 删除规则
platform.rules.delete("critical_project_alert")

#条件表达式语法

coomia-dip 支持丰富的条件表达式：

YAML

# 简单条件
condition:
  property: status
  operator: eq
  value: active

# AND 条件
condition:
  all:
    - property: priority
      operator: eq
      value: critical
    - property: budget
      operator: gt
      value: 100000

# OR 条件
condition:
  any:
    - property: status
      operator: eq
      value: on_hold
    - property: end_date
      operator: lt
      value: $today

# NOT 条件
condition:
  not:
    property: is_active
    operator: eq
    value: false

# 嵌套条件
condition:
  all:
    - any:
        - property: priority
          operator: eq
          value: critical
        - property: priority
          operator: eq
          value: high
    - property: status
      operator: neq
      value: completed

#与 Palantir Foundry 对比

特性	Palantir Foundry	coomia-dip
规则定义	Ontology Rules	YAML + SDK
事件触发	Object Events	Data Change Events
定时任务	Scheduled Rules	Cron Expressions
推理能力	Limited	知识图谱推理
规则链	支持	支持
运行时	JVM	Python (FastAPI)

#最佳实践

单一职责：每条规则只做一件事
条件精确：避免过于宽泛的触发条件
幂等设计：规则动作应该是幂等的
错误处理：为关键规则配置告警
测试优先：使用 rules.test() 在激活前验证
文档化：为每条规则写清楚 description

#总结

本教程介绍了 coomia-dip 的规则引擎，包括三种规则类型（事件驱动、定时、推理）、YAML 声明式定义、条件表达式语法和规则管理。规则是实现业务自动化的关键能力，让你的数据平台从"被动存储"升级为"主动智能"。

本文是 coomia-dip 开发者教程系列的第 5 篇。 项目地址：https://github.com/coomia-dip/coomia-dip↗ | 许可证：Apache License 2.0