ActionType 详解：把业务操作变成平台原生能力

“

系列：S4 本体建模 · 第 5 篇 | 难度：中级 | 阅读时间：18 分钟

#TL;DR

ActionType 将"业务操作"从代码中抽离为 Schema 一等公民——审批、分配、计算、通知等操作都成为可发现、可审计、可复用的平台原生能力，终结了"业务逻辑散落在一百个微服务里"的混乱。
**10 种执行器类型（Function / Webhook / gRPC / Workflow / SQL / Script / Notification / Approval / Schedule / Composite）**覆盖从简单字段更新到多步骤审批流的全部场景。
参数验证 + 幂等保障 + 审计日志三重防线确保每次操作可信、可追溯、可重试。

#1. 为什么需要 ActionType

在传统架构中，业务操作分散在各个服务中：

Code

传统方式（操作散落）：

用户点击"审批订单" →
  前端调 POST /api/orders/{id}/approve →
    后端写了一坨逻辑 →
      改状态 + 发通知 + 记日志 + 扣库存 + ...

问题：
├── 每个操作都是定制 API，无法统一管理
├── 参数校验逻辑重复写（前端一遍、后端一遍）
├── 没有统一的审计日志（有的记了有的没记）
├── 重试安全性不确定（多扣了库存？多发了通知？）
├── 新人入职看不到系统有哪些业务操作
└── 权限控制粒度到 API 而非业务操作

coomia-dip 方式（ActionType）：

用户点击"审批订单" →
  SDK 调 executeAction("ApproveOrder", params) →
    平台统一处理：
    ├── 1. 参数验证（Schema 定义的规则）
    ├── 2. 权限检查（RBAC + ABAC）
    ├── 3. 执行器执行（10 种类型之一）
    ├── 4. 审计日志（自动记录）
    ├── 5. 幂等检查（防止重复执行）
    └── 6. 事件发布（通知下游系统）

#2. ActionType 数据模型

#2.1 核心定义

Python

from ontology_sdk import OntologyClient

client = OntologyClient(base_url="http://localhost:8080")

# 创建一个完整的 ActionType
action = client.schema.create_action_type({
    "apiName": "ApproveOrder",
    "displayName": "审批订单",
    "description": "主管审批待处理订单，通过后自动分配仓库和物流",

    # 绑定的 ObjectType
    "objectType": "Order",

    # 参数定义
    "parameters": {
        "orderId": {
            "type": "STRING",
            "required": True,
            "description": "订单 ID",
        },
        "approved": {
            "type": "BOOLEAN",
            "required": True,
            "description": "是否通过",
        },
        "comment": {
            "type": "STRING",
            "required": False,
            "maxLength": 500,
            "description": "审批意见",
        },
        "priority": {
            "type": "ENUM",
            "enumValues": ["LOW", "NORMAL", "HIGH", "URGENT"],
            "defaultValue": "NORMAL",
        },
    },

    # 执行器配置
    "executor": {
        "type": "FUNCTION",
        "functionId": "approve-order-fn",
        "timeout": 30000,  # 30 秒超时
        "retryPolicy": {
            "maxRetries": 3,
            "backoffMs": 1000,
        },
    },

    # 幂等配置
    "idempotency": {
        "enabled": True,
        "keyExpression": "orderId + '-' + executorUserId",
        "ttlSeconds": 3600,  # 1 小时内不重复执行
    },

    # 前置条件
    "preconditions": [
        {
            "type": "OBJECT_STATE",
            "field": "status",
            "operator": "eq",
            "value": "PENDING_APPROVAL",
            "errorMessage": "只有待审批状态的订单才能审批",
        },
        {
            "type": "PERMISSION",
            "requiredRole": "ORDER_APPROVER",
        },
    ],

    # 后置动作
    "sideEffects": [
        {
            "type": "UPDATE_OBJECT",
            "updates": {
                "status": "${approved ? 'APPROVED' : 'REJECTED'}",
                "approvedBy": "${currentUser.id}",
                "approvedAt": "${now()}",
            },
        },
        {
            "type": "SEND_NOTIFICATION",
            "template": "order-approval-result",
            "recipients": ["${object.createdBy}"],
        },
    ],
})

#2.2 Protobuf Schema

PROTOBUF

message ActionType {
    string api_name = 1;
    string display_name = 2;
    string description = 3;
    string object_type = 4;

    map<string, ParameterDef> parameters = 5;
    ExecutorConfig executor = 6;
    IdempotencyConfig idempotency = 7;

    repeated Precondition preconditions = 8;
    repeated SideEffect side_effects = 9;

    LifecycleState lifecycle = 10;
    AuditInfo audit = 11;
}

message ParameterDef {
    string type = 1;
    bool required = 2;
    string description = 3;
    string default_value = 4;
    repeated ValidationRule validations = 5;
}

message ExecutorConfig {
    ExecutorType type = 1;
    string function_id = 2;
    string webhook_url = 3;
    string grpc_service = 4;
    int32 timeout_ms = 5;
    RetryPolicy retry_policy = 6;
}

enum ExecutorType {
    FUNCTION = 0;
    WEBHOOK = 1;
    GRPC = 2;
    WORKFLOW = 3;
    SQL = 4;
    SCRIPT = 5;
    NOTIFICATION = 6;
    APPROVAL = 7;
    SCHEDULE = 8;
    COMPOSITE = 9;
}

#3. 10 种执行器类型详解

#3.1 FUNCTION — 函数执行器

Code

最常用的执行器类型，调用平台注册的 Function

┌──────────┐      ┌──────────────────┐      ┌──────────┐
│  Action  │─────►│ Function Runtime  │─────►│  Result  │
│  Request │      │  (Python/Java)    │      │          │
└──────────┘      └──────────────────┘      └──────────┘

Python

# 注册一个 Function
client.functions.register({
    "functionId": "calculate-risk-score",
    "runtime": "python3.11",
    "handler": "risk_module.calculate_score",
    "timeout": 60000,
})

# 创建使用 Function 执行器的 ActionType
client.schema.create_action_type({
    "apiName": "CalculateRiskScore",
    "objectType": "Customer",
    "parameters": {
        "customerId": {"type": "STRING", "required": True},
        "includeHistory": {"type": "BOOLEAN", "defaultValue": "true"},
    },
    "executor": {
        "type": "FUNCTION",
        "functionId": "calculate-risk-score",
    },
})

#3.2 WEBHOOK — HTTP 回调执行器

Python

# 调用外部系统的 Webhook
client.schema.create_action_type({
    "apiName": "SyncToERP",
    "objectType": "Order",
    "executor": {
        "type": "WEBHOOK",
        "webhookUrl": "https://erp.company.com/api/sync-order",
        "method": "POST",
        "headers": {
            "Authorization": "Bearer ${secrets.ERP_TOKEN}",
            "Content-Type": "application/json",
        },
        "bodyTemplate": {
            "orderId": "${params.orderId}",
            "items": "${object.lineItems}",
        },
        "successCodes": [200, 201, 202],
        "timeout": 15000,
    },
})

#3.3 GRPC — gRPC 服务调用

Python

# 调用内部 gRPC 服务
client.schema.create_action_type({
    "apiName": "AllocateWarehouse",
    "objectType": "Order",
    "executor": {
        "type": "GRPC",
        "grpcService": "warehouse-service",
        "grpcMethod": "AllocateWarehouse",
        "protoMessage": "AllocateWarehouseRequest",
        "fieldMapping": {
            "order_id": "${params.orderId}",
            "region": "${object.shippingRegion}",
            "weight_kg": "${object.totalWeight}",
        },
    },
})

#3.4 WORKFLOW — 工作流执行器

Python

# 触发多步骤工作流（基于 Temporal）
client.schema.create_action_type({
    "apiName": "OnboardNewEmployee",
    "objectType": "Employee",
    "executor": {
        "type": "WORKFLOW",
        "workflowId": "employee-onboarding",
        "taskQueue": "hr-workflows",
        "steps": [
            {"name": "create_accounts", "timeout": 60000},
            {"name": "assign_equipment", "timeout": 120000},
            {"name": "schedule_training", "timeout": 30000},
            {"name": "notify_team", "timeout": 10000},
        ],
    },
})

#3.5 SQL — 数据库操作执行器

Python

# 直接执行 SQL（适用于批量更新）
client.schema.create_action_type({
    "apiName": "BatchUpdatePrices",
    "objectType": "Product",
    "executor": {
        "type": "SQL",
        "dataSource": "doris-main",
        "sqlTemplate": """
            UPDATE products
            SET price = price * (1 + :adjustmentPercent / 100.0),
                updated_at = NOW()
            WHERE category = :category
              AND price BETWEEN :minPrice AND :maxPrice
        """,
        "parameterMapping": {
            "adjustmentPercent": "${params.adjustmentPercent}",
            "category": "${params.category}",
            "minPrice": "${params.minPrice}",
            "maxPrice": "${params.maxPrice}",
        },
    },
})

#3.6 SCRIPT — 脚本执行器

Python

# 运行 Python 脚本片段
client.schema.create_action_type({
    "apiName": "GenerateReport",
    "objectType": "Project",
    "executor": {
        "type": "SCRIPT",
        "language": "python",
        "script": """
import pandas as pd
from datetime import datetime

project = context.get_object("Project", params["projectId"])
tasks = context.traverse(project, "Project_hasTasks_Task")

df = pd.DataFrame([t.to_dict() for t in tasks])
summary = {
    "total": len(df),
    "completed": len(df[df.status == "DONE"]),
    "overdue": len(df[(df.status != "DONE") & (df.due_date < datetime.now())]),
    "completion_rate": f"{len(df[df.status == 'DONE']) / len(df) * 100:.1f}%",
}

context.update_object(project, {"lastReport": summary})
return {"success": True, "summary": summary}
        """,
        "timeout": 120000,
    },
})

#3.7 NOTIFICATION — 通知执行器

Python

# 发送多渠道通知
client.schema.create_action_type({
    "apiName": "SendAlertNotification",
    "objectType": "Equipment",
    "executor": {
        "type": "NOTIFICATION",
        "channels": ["email", "sms", "webhook"],
        "template": "equipment-alert",
        "recipientExpression": """
            object.assignedTechnicians
            + object.factoryManager
        """,
        "variables": {
            "equipmentName": "${object.name}",
            "alertType": "${params.alertType}",
            "severity": "${params.severity}",
        },
    },
})

#3.8 APPROVAL — 审批执行器

Python

# 多级审批流
client.schema.create_action_type({
    "apiName": "ApprovePurchaseRequest",
    "objectType": "PurchaseRequest",
    "executor": {
        "type": "APPROVAL",
        "approvalChain": [
            {
                "level": 1,
                "approverExpression": "${object.departmentManager}",
                "condition": "params.amount < 10000",
                "autoApproveAfterHours": 48,
            },
            {
                "level": 2,
                "approverExpression": "${object.financeDirector}",
                "condition": "params.amount >= 10000 && params.amount < 100000",
                "autoApproveAfterHours": 72,
            },
            {
                "level": 3,
                "approverExpression": "${getCEO()}",
                "condition": "params.amount >= 100000",
            },
        ],
        "onApproved": {
            "type": "FUNCTION",
            "functionId": "create-purchase-order",
        },
        "onRejected": {
            "type": "NOTIFICATION",
            "template": "purchase-rejected",
        },
    },
})

#3.9 SCHEDULE — 定时执行器

Python

# 定时或延迟执行
client.schema.create_action_type({
    "apiName": "ScheduleMaintenanceReminder",
    "objectType": "Equipment",
    "executor": {
        "type": "SCHEDULE",
        "scheduleExpression": "0 8 * * MON",  # 每周一早上 8 点
        "action": {
            "type": "FUNCTION",
            "functionId": "check-maintenance-due",
        },
    },
})

#3.10 COMPOSITE — 组合执行器

Python

# 按顺序或并行执行多个子操作
client.schema.create_action_type({
    "apiName": "ProcessNewOrder",
    "objectType": "Order",
    "executor": {
        "type": "COMPOSITE",
        "strategy": "SEQUENTIAL",  # SEQUENTIAL 或 PARALLEL
        "steps": [
            {
                "name": "validate_inventory",
                "executor": {"type": "FUNCTION", "functionId": "check-inventory"},
                "onFailure": "ABORT",  # 失败则终止
            },
            {
                "name": "charge_payment",
                "executor": {"type": "GRPC", "grpcService": "payment-service", "grpcMethod": "Charge"},
                "onFailure": "COMPENSATE",  # 失败则补偿
                "compensateAction": "RefundPayment",
            },
            {
                "name": "allocate_warehouse",
                "executor": {"type": "FUNCTION", "functionId": "allocate-warehouse"},
                "onFailure": "COMPENSATE",
                "compensateAction": "ReleaseWarehouse",
            },
            {
                "name": "notify_customer",
                "executor": {"type": "NOTIFICATION", "template": "order-confirmed"},
                "onFailure": "IGNORE",  # 通知失败不影响主流程
            },
        ],
    },
})

#4. 参数验证体系

#4.1 内置验证规则

Python

# 丰富的参数验证
client.schema.create_action_type({
    "apiName": "TransferFunds",
    "objectType": "Account",
    "parameters": {
        "sourceAccountId": {
            "type": "STRING",
            "required": True,
            "validations": [
                {"rule": "pattern", "value": "^ACC-[0-9]{10}$", "message": "无效的账户 ID 格式"},
            ],
        },
        "targetAccountId": {
            "type": "STRING",
            "required": True,
            "validations": [
                {"rule": "pattern", "value": "^ACC-[0-9]{10}$"},
                {"rule": "notEqual", "referenceParam": "sourceAccountId", "message": "不能转账给自己"},
            ],
        },
        "amount": {
            "type": "DOUBLE",
            "required": True,
            "validations": [
                {"rule": "min", "value": 0.01, "message": "金额必须大于 0"},
                {"rule": "max", "value": 1000000, "message": "单次转账不能超过 100 万"},
            ],
        },
        "currency": {
            "type": "ENUM",
            "enumValues": ["CNY", "USD", "EUR", "GBP", "JPY"],
            "defaultValue": "CNY",
        },
        "memo": {
            "type": "STRING",
            "required": False,
            "validations": [
                {"rule": "maxLength", "value": 200},
                {"rule": "noScript", "message": "备注中不能包含脚本代码"},
            ],
        },
    },
})

#4.2 自定义验证函数

Python

# 注册自定义验证器
client.schema.create_action_type({
    "apiName": "ChangeEquipmentStatus",
    "objectType": "Equipment",
    "parameters": {
        "equipmentId": {"type": "STRING", "required": True},
        "newStatus": {"type": "ENUM", "enumValues": ["RUNNING", "MAINTENANCE", "STOPPED", "DECOMMISSIONED"]},
    },
    "validationFunction": "validate-status-transition",
})

# 验证函数代码
# def validate_status_transition(params, object, context):
#     valid_transitions = {
#         "RUNNING": ["MAINTENANCE", "STOPPED"],
#         "MAINTENANCE": ["RUNNING", "STOPPED"],
#         "STOPPED": ["RUNNING", "MAINTENANCE", "DECOMMISSIONED"],
#         "DECOMMISSIONED": [],  # 终态，不可转换
#     }
#     current = object.status
#     target = params["newStatus"]
#     if target not in valid_transitions.get(current, []):
#         raise ValidationError(f"不能从 {current} 转换到 {target}")

#5. 幂等性保障

#5.1 幂等机制原理

Code

幂等性保障流程：

请求到达 → 计算幂等 Key → 查询幂等存储
                              │
                    ┌─────────┴─────────┐
                    ▼                   ▼
               Key 存在              Key 不存在
               (已执行过)            (首次执行)
                    │                   │
                    ▼                   ▼
              返回上次结果         执行 Action
                                       │
                                       ▼
                                  存储幂等 Key + 结果
                                       │
                                       ▼
                                   返回结果

Python

# 幂等执行示例
result1 = client.actions.execute("ApproveOrder", {
    "orderId": "order-001",
    "approved": True,
    "comment": "同意",
}, idempotency_key="approval-order-001-user-admin")

# 相同的幂等 Key，不会重复执行
result2 = client.actions.execute("ApproveOrder", {
    "orderId": "order-001",
    "approved": True,
    "comment": "同意",
}, idempotency_key="approval-order-001-user-admin")

assert result1.execution_id == result2.execution_id  # 同一次执行
assert result2.was_deduplicated == True  # 标记为去重

#5.2 幂等 Key 策略

Code

┌─────────────────────────────────────────────────────┐
│              幂等 Key 生成策略                        │
├─────────────────────────────────────────────────────┤
│                                                      │
│  1. 客户端指定（最灵活）：                             │
│     idempotency_key = "user-123-approve-order-456"  │
│                                                      │
│  2. 参数派生（自动计算）：                             │
│     keyExpression = "orderId + '-' + userId"         │
│     → "order-456-user-123"                           │
│                                                      │
│  3. 业务规则（语义级）：                              │
│     keyExpression = "orderId + '-approve'"           │
│     → 同一订单只能审批一次（无论谁审批）               │
│                                                      │
│  TTL 策略：                                          │
│  ├── 短期幂等（1 小时）：防止按钮双击                  │
│  ├── 中期幂等（24 小时）：防止当日重复操作             │
│  └── 永久幂等（无过期）：业务上只能执行一次的操作       │
└─────────────────────────────────────────────────────┘

#6. 审计日志

#6.1 自动审计

Python

# 每次 Action 执行都会自动生成审计日志
result = client.actions.execute("ApproveOrder", {
    "orderId": "order-001",
    "approved": True,
})

# 查询审计日志
audit_logs = client.audit.query({
    "actionType": "ApproveOrder",
    "objectType": "Order",
    "objectId": "order-001",
    "timeRange": {"from": "2026-03-01", "to": "2026-03-24"},
})

for log in audit_logs:
    print(f"[{log.timestamp}] {log.user} executed {log.action_type}")
    print(f"  Parameters: {log.parameters}")
    print(f"  Result: {log.result_status}")
    print(f"  Duration: {log.duration_ms} ms")
    print(f"  Changes: {log.object_changes}")

#6.2 审计日志结构

Code

┌───────────────────────────────────────────────┐
│              审计日志条目                       │
├───────────────────────────────────────────────┤
│  execution_id    : "exec-uuid-001"            │
│  action_type     : "ApproveOrder"             │
│  object_type     : "Order"                    │
│  object_id       : "order-001"                │
│  user_id         : "user-admin"               │
│  timestamp       : "2026-03-24T10:30:00Z"     │
│  parameters      : {orderId, approved, ...}   │
│  result_status   : "SUCCESS"                  │
│  duration_ms     : 245                        │
│  idempotency_key : "approval-order-001-..."   │
│  was_deduplicated: false                      │
│  object_before   : {status: "PENDING"}        │
│  object_after    : {status: "APPROVED"}       │
│  side_effects    : ["notification-sent"]      │
│  ip_address      : "192.168.1.100"            │
│  user_agent      : "onto-sdk/1.0"             │
└───────────────────────────────────────────────┘

#7. Action 权限控制

Python

# ActionType 权限模型
client.schema.create_action_type({
    "apiName": "DeleteCustomer",
    "objectType": "Customer",
    "permissions": {
        # RBAC：角色级权限
        "requiredRoles": ["CUSTOMER_ADMIN"],

        # ABAC：属性级权限
        "attributeRules": [
            {
                "description": "只能删除自己区域的客户",
                "expression": "user.region == object.region",
            },
            {
                "description": "VIP 客户需要总监审批",
                "expression": "object.tier != 'VIP' || user.role == 'DIRECTOR'",
            },
        ],

        # 数据级权限
        "fieldAccess": {
            "params.reason": "ALL",      # 所有人都可以填写原因
            "params.force": "ADMIN_ONLY", # 只有管理员可以强制删除
        },
    },
})

#8. Action 执行生命周期

Code

Action 执行状态机：

┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐
│ SUBMITTED│───►│VALIDATING│───►│ EXECUTING│───►│COMPLETED │
└──────────┘    └─────┬────┘    └─────┬────┘    └──────────┘
                      │               │
                      ▼               ▼
               ┌──────────┐    ┌──────────┐
               │VALIDATION│    │  FAILED  │
               │_FAILED   │    │          │
               └──────────┘    └─────┬────┘
                                     │
                                     ▼
                              ┌──────────┐
                              │ RETRYING │──► EXECUTING (重试)
                              └──────────┘

每个阶段的职责：
├── SUBMITTED    → 接收请求，分配 execution_id
├── VALIDATING   → 参数验证 + 前置条件检查 + 权限检查
├── EXECUTING    → 调用执行器
├── COMPLETED    → 执行成功，触发后置动作
├── FAILED       → 执行失败，根据重试策略决定是否重试
└── RETRYING     → 退避等待后重新执行

Python

# 异步执行长时间 Action
execution = client.actions.execute_async("OnboardNewEmployee", {
    "employeeId": "emp-new-001",
    "department": "engineering",
    "startDate": "2026-04-01",
})

print(f"Execution ID: {execution.id}")
print(f"Status: {execution.status}")  # SUBMITTED

# 轮询执行状态
import time
while execution.status not in ["COMPLETED", "FAILED"]:
    time.sleep(2)
    execution = client.actions.get_execution(execution.id)
    print(f"Status: {execution.status}, Progress: {execution.progress}%")

# 查看执行结果
if execution.status == "COMPLETED":
    print(f"Result: {execution.result}")
else:
    print(f"Error: {execution.error}")

#9. 批量操作

Python

# 批量执行 Action
batch_result = client.actions.batch_execute("UpdateEquipmentStatus", [
    {"equipmentId": "equip-001", "newStatus": "MAINTENANCE"},
    {"equipmentId": "equip-002", "newStatus": "MAINTENANCE"},
    {"equipmentId": "equip-003", "newStatus": "MAINTENANCE"},
    {"equipmentId": "equip-004", "newStatus": "STOPPED"},
], batch_options={
    "concurrency": 5,           # 并发度
    "stopOnFirstError": False,  # 不因单个失败停止
    "transactional": False,     # 非事务（各自独立）
})

print(f"总数: {batch_result.total}")
print(f"成功: {batch_result.success_count}")
print(f"失败: {batch_result.failure_count}")

for failure in batch_result.failures:
    print(f"  {failure.params['equipmentId']}: {failure.error}")

#10. ActionType 与事件驱动

Python

# Action 执行完成后自动发布事件
client.schema.create_action_type({
    "apiName": "UpdateInventory",
    "objectType": "Product",
    "executor": {
        "type": "FUNCTION",
        "functionId": "update-inventory",
    },
    "events": {
        "onSuccess": {
            "topic": "inventory.updated",
            "payload": {
                "productId": "${params.productId}",
                "oldQuantity": "${before.quantity}",
                "newQuantity": "${after.quantity}",
                "delta": "${params.quantityDelta}",
            },
        },
        "onFailure": {
            "topic": "inventory.update-failed",
            "payload": {
                "productId": "${params.productId}",
                "error": "${error.message}",
            },
        },
    },
})

# 其他系统可以订阅这些事件
client.events.subscribe("inventory.updated", handler=lambda event:
    print(f"库存变更: {event.productId} 从 {event.oldQuantity} 变为 {event.newQuantity}")
)

#11. 实战案例：制造业设备巡检

Python

# 完整的设备巡检 ActionType

# 1. 开始巡检
client.schema.create_action_type({
    "apiName": "StartInspection",
    "objectType": "Equipment",
    "displayName": "开始巡检",
    "parameters": {
        "equipmentId": {"type": "STRING", "required": True},
        "inspectorId": {"type": "STRING", "required": True},
        "inspectionType": {
            "type": "ENUM",
            "enumValues": ["ROUTINE", "SPECIAL", "EMERGENCY"],
            "required": True,
        },
    },
    "preconditions": [
        {"type": "OBJECT_STATE", "field": "status", "operator": "in", "value": ["RUNNING", "STOPPED"]},
    ],
    "executor": {
        "type": "COMPOSITE",
        "strategy": "SEQUENTIAL",
        "steps": [
            {
                "name": "create_inspection_record",
                "executor": {"type": "FUNCTION", "functionId": "create-inspection"},
            },
            {
                "name": "update_equipment_status",
                "executor": {
                    "type": "SQL",
                    "sqlTemplate": "UPDATE equipment SET inspection_status = 'IN_PROGRESS' WHERE id = :equipmentId",
                },
            },
            {
                "name": "notify_maintenance_team",
                "executor": {"type": "NOTIFICATION", "template": "inspection-started"},
                "onFailure": "IGNORE",
            },
        ],
    },
    "idempotency": {
        "enabled": True,
        "keyExpression": "equipmentId + '-' + inspectorId + '-' + today()",
        "ttlSeconds": 86400,
    },
})

# 2. 记录巡检结果
client.schema.create_action_type({
    "apiName": "RecordInspectionResult",
    "objectType": "Equipment",
    "displayName": "记录巡检结果",
    "parameters": {
        "inspectionId": {"type": "STRING", "required": True},
        "overallStatus": {"type": "ENUM", "enumValues": ["PASS", "WARNING", "FAIL"], "required": True},
        "checkItems": {
            "type": "ARRAY",
            "itemType": "STRUCT",
            "structFields": {
                "itemName": {"type": "STRING"},
                "result": {"type": "ENUM", "enumValues": ["OK", "ABNORMAL", "CRITICAL"]},
                "note": {"type": "STRING"},
                "photoUrls": {"type": "ARRAY", "itemType": "STRING"},
            },
        },
        "recommendations": {"type": "STRING", "maxLength": 2000},
    },
    "executor": {
        "type": "FUNCTION",
        "functionId": "process-inspection-result",
    },
})

# 使用示例
result = client.actions.execute("StartInspection", {
    "equipmentId": "equip-CNC-001",
    "inspectorId": "tech-zhang",
    "inspectionType": "ROUTINE",
})
print(f"巡检已开始: {result.data['inspectionId']}")

#Key Takeaways

ActionType 是业务操作的 Schema 化表达——每个操作都有明确的参数、执行器、权限和审计，让业务能力可发现、可管理、可复用。
10 种执行器类型从简单的函数调用到复杂的多步骤审批流全覆盖，COMPOSITE 执行器支持 Saga 模式的分布式事务补偿。
参数验证前移到 Schema 层——类型检查、范围校验、正则匹配、自定义验证函数，一次定义，前后端统一执行。
幂等性是生产系统的刚需——通过幂等 Key + TTL 机制，确保网络抖动、用户重试不会导致重复执行。
全量审计日志自动生成——谁在什么时间对什么对象做了什么操作，入参出参全记录，是合规和问题排查的基石。

#Next Article

下一篇 S4-06 InterfaceType 与 StructType 将探讨 Ontology 类型系统的高级特性——如何通过接口继承实现多态查询，如何用结构体嵌套构建复杂值对象。

#ontology #action-type #executor #idempotency #audit-log #parameter-validation #workflow