规则脚本引擎：Python/Groovy 编写高级规则
YAML DSL 能处理大部分业务规则，但遇到复杂的数据变换、外部 API 调用或自定义算法时就力不从心了。coomia-dip 的 FunctionRuntime 支持 Python、Groovy、TypeScript、Kotlin 和 WASM 五种语言编写高级规则脚本，通过 nsjail 沙箱隔离执行，确保安全性。本文深入解析脚本引擎架构、多语言运行时设计、沙箱安全模型、脚本生命周期管理以及与 Rete 规则引擎的集成方式。
Coomia发布于 2025年8月27日18 分钟阅读
分享本文Twitter / X
“系列：S5 智能决策 · 第 5 篇 | 难度：高级 | 阅读时间：20 分钟
规则脚本引擎：Python/Groovy 编写高级规则
#TL;DR
YAML DSL 能处理大部分业务规则，但遇到复杂的数据变换、外部 API 调用或自定义算法时就力不从心了。coomia-dip 的 FunctionRuntime 支持 Python、Groovy、TypeScript、Kotlin 和 WASM 五种语言编写高级规则脚本，通过 nsjail 沙箱隔离执行，确保安全性。本文深入解析脚本引擎架构、多语言运行时设计、沙箱安全模型、脚本生命周期管理以及与 Rete 规则引擎的集成方式。
#1. 脚本引擎定位
#1.1 YAML DSL vs. 脚本规则
Code
规则复杂度光谱：
简单 复杂
|------|---------|---------|---------|---------|------>
YAML YAML 脚本 脚本 自定义
单条件 多条件 Python Groovy ML 模型
比较 逻辑组合 数据变换 领域DSL
+------------+ +------------------+ +----------------+
| Low-Code | | Script Engine | | ML Layer |
| YAML Rules | | Python/Groovy/TS | | Custom Models |
+------------+ +------------------+ +----------------+
适合：80% 适合：15% 适合：5%
业务规则 高级规则 预测类规则
#1.2 适用场景
| 场景 | YAML 能力 | 脚本需求 | 示例 |
|---|---|---|---|
| 简单阈值 | 完全满足 | 不需要 | 库存 < 100 |
| 多条件组合 | 完全满足 | 不需要 | A AND B OR C |
| 日期计算 | 有限 | 需要 | 工作日+3天 |
| 外部 API | 不支持 | 需要 | 查询汇率 |
| 统计聚合 | 不支持 | 需要 | 移动平均 |
| 自定义评分 | 有限 | 需要 | 加权评分函数 |
| 递归逻辑 | 不支持 | 需要 | 层级审批 |
#2. 脚本引擎架构
#2.1 总体架构
Code
脚本引擎架构：
+----------------------------------------------------------+
| ScriptEngine (入口) |
+----------------------------------------------------------+
| |
| +------------------+ +-------------------+ |
| | Script Registry | | Script Compiler | |
| | (脚本注册表) | | (预编译/缓存) | |
| +------------------+ +-------------------+ |
| |
| +--------------------------------------------------+ |
| | Runtime Dispatcher | |
| | +----------+ +----------+ +----------+ | |
| | | Python | | Groovy | | TypeScript| | |
| | | Runtime | | Runtime | | Runtime | | |
| | +----------+ +----------+ +----------+ | |
| | +----------+ +----------+ | |
| | | Kotlin | | WASM | | |
| | | Runtime | | Runtime | | |
| | +----------+ +----------+ | |
| +--------------------------------------------------+ |
| |
| +--------------------------------------------------+ |
| | nsjail Sandbox Layer | |
| | - 文件系统隔离 - 网络隔离 - 资源限制 | |
| +--------------------------------------------------+ |
+----------------------------------------------------------+
#2.2 核心接口
Python
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any
from enum import Enum
import time
class ScriptLanguage(str, Enum):
PYTHON = "python"
GROOVY = "groovy"
TYPESCRIPT = "typescript"
KOTLIN = "kotlin"
WASM = "wasm"
@dataclass
class ScriptMetadata:
"""脚本元数据"""
script_id: str
name: str
language: ScriptLanguage
version: str
author: str
description: str
inputs: dict[str, str] # name -> type
outputs: dict[str, str] # name -> type
timeout_ms: int = 5000
max_memory_mb: int = 256
tags: list[str] = field(default_factory=list)
@dataclass
class ScriptResult:
"""脚本执行结果"""
success: bool
outputs: dict[str, Any]
logs: list[str]
elapsed_ms: float
memory_used_mb: float
error: str | None = None
class ScriptRuntime(ABC):
"""脚本运行时基类"""
@abstractmethod
def compile(self, source: str, metadata: ScriptMetadata) -> str:
"""预编译脚本，返回编译后的标识"""
...
@abstractmethod
def execute(self, compiled_id: str,
inputs: dict[str, Any]) -> ScriptResult:
"""执行已编译的脚本"""
...
@abstractmethod
def validate(self, source: str) -> list[str]:
"""验证脚本语法，返回错误列表"""
...
#3. Python 脚本运行时
#3.1 脚本编写规范
Python
# 规则脚本示例：动态信用评分
# 文件: scripts/credit_scoring.py
from rule_sdk import RuleContext, RuleResult
def evaluate(ctx: RuleContext) -> RuleResult:
"""
动态信用评分规则
根据多维度数据计算综合信用分数，
使用加权评分模型。
"""
# 获取输入
credit_score = ctx.get("credit_score", type=int)
income = ctx.get("annual_income", type=float)
debt_ratio = ctx.get("debt_ratio", type=float)
employment_years = ctx.get("employment_years", type=int)
industry = ctx.get("industry", type=str)
# 行业风险系数
industry_risk = {
"technology": 0.8,
"finance": 0.9,
"manufacturing": 1.0,
"real_estate": 1.2,
"crypto": 1.5,
}
risk_factor = industry_risk.get(industry, 1.0)
# 加权评分
score = (
credit_score * 0.35 +
min(income / 10000, 100) * 0.25 +
(1 - debt_ratio) * 100 * 0.20 +
min(employment_years * 5, 50) * 0.10 +
(1 / risk_factor) * 100 * 0.10
)
# 决策
if score >= 80:
decision = "approve"
confidence = min(score / 100, 0.99)
elif score >= 60:
decision = "conditional_approve"
confidence = score / 100
else:
decision = "reject"
confidence = (100 - score) / 100
return RuleResult(
decision=decision,
confidence=confidence,
outputs={
"composite_score": round(score, 2),
"risk_factor": risk_factor,
},
explanation=f"综合评分 {score:.1f}，行业风险系数 {risk_factor}",
)
#3.2 RuleContext SDK
Python
from dataclasses import dataclass, field
from typing import Any, TypeVar
T = TypeVar("T")
@dataclass
class RuleContext:
"""脚本规则的执行上下文"""
_inputs: dict[str, Any]
_variables: dict[str, Any] = field(default_factory=dict)
_logs: list[str] = field(default_factory=list)
def get(self, name: str, type: type[T] = str,
default: T | None = None) -> T:
"""获取输入变量（带类型检查）"""
value = self._inputs.get(name)
if value is None:
if default is not None:
return default
raise KeyError(f"必需的输入变量 '{name}' 未提供")
if not isinstance(value, type):
try:
value = type(value)
except (ValueError, TypeError) as e:
raise TypeError(
f"变量 '{name}' 类型转换失败: "
f"期望 {type.__name__}, 值为 {value!r}"
) from e
return value
def set_var(self, name: str, value: Any) -> None:
"""设置中间变量"""
self._variables[name] = value
def get_var(self, name: str, default: Any = None) -> Any:
return self._variables.get(name, default)
def log(self, message: str) -> None:
"""记录日志（会出现在执行结果中）"""
self._logs.append(message)
@property
def logs(self) -> list[str]:
return list(self._logs)
@dataclass
class RuleResult:
"""脚本规则的返回结果"""
decision: str
confidence: float = 1.0
outputs: dict[str, Any] = field(default_factory=dict)
explanation: str = ""
actions: list[dict] = field(default_factory=list)
#3.3 Python 运行时实现
Python
import importlib.util
import tempfile
import subprocess
import json
from pathlib import Path
class PythonScriptRuntime(ScriptRuntime):
"""Python 脚本运行时"""
def __init__(self, sandbox_enabled: bool = True):
self._sandbox = sandbox_enabled
self._compiled: dict[str, Path] = {}
self._cache_dir = Path(tempfile.mkdtemp(prefix="onto_scripts_"))
def compile(self, source: str, metadata: ScriptMetadata) -> str:
"""预编译 Python 脚本"""
# 语法检查
errors = self.validate(source)
if errors:
raise ValueError(f"语法错误: {errors}")
# 安全检查
security_issues = self._security_scan(source)
if security_issues:
raise SecurityError(f"安全违规: {security_issues}")
# 保存到缓存
script_path = self._cache_dir / f"{metadata.script_id}.py"
script_path.write_text(source)
compiled_id = f"{metadata.script_id}:{metadata.version}"
self._compiled[compiled_id] = script_path
return compiled_id
def execute(self, compiled_id: str,
inputs: dict[str, Any]) -> ScriptResult:
"""执行 Python 脚本"""
script_path = self._compiled.get(compiled_id)
if not script_path:
raise KeyError(f"脚本未编译: {compiled_id}")
start = time.monotonic()
if self._sandbox:
result = self._execute_sandboxed(script_path, inputs)
else:
result = self._execute_direct(script_path, inputs)
result.elapsed_ms = (time.monotonic() - start) * 1000
return result
def _execute_sandboxed(self, script_path: Path,
inputs: dict) -> ScriptResult:
"""通过 nsjail 沙箱执行"""
# 准备输入文件
input_file = self._cache_dir / "input.json"
input_file.write_text(json.dumps(inputs))
# nsjail 命令
cmd = [
"nsjail",
"--mode", "once",
"--chroot", "/",
"--rlimit_as", "256", # 内存限制 256MB
"--rlimit_cpu", "5", # CPU 限制 5 秒
"--rlimit_fsize", "0", # 禁止写文件
"--disable_clone_newnet", # 禁止网络
"--bindmount_ro", f"{script_path}:/script.py",
"--bindmount_ro", f"{input_file}:/input.json",
"--",
"python3", "/script.py", "/input.json",
]
try:
proc = subprocess.run(
cmd, capture_output=True, text=True, timeout=10,
)
if proc.returncode != 0:
return ScriptResult(
success=False, outputs={}, logs=[],
elapsed_ms=0, memory_used_mb=0,
error=proc.stderr,
)
output = json.loads(proc.stdout)
return ScriptResult(
success=True,
outputs=output.get("outputs", {}),
logs=output.get("logs", []),
elapsed_ms=0,
memory_used_mb=output.get("memory_mb", 0),
)
except subprocess.TimeoutExpired:
return ScriptResult(
success=False, outputs={}, logs=[],
elapsed_ms=10000, memory_used_mb=0,
error="执行超时 (10s)",
)
def _execute_direct(self, script_path: Path,
inputs: dict) -> ScriptResult:
"""直接执行（仅用于开发/测试环境）"""
spec = importlib.util.spec_from_file_location("rule_script", script_path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
ctx = RuleContext(_inputs=inputs)
result = module.evaluate(ctx)
return ScriptResult(
success=True,
outputs={
"decision": result.decision,
"confidence": result.confidence,
**result.outputs,
},
logs=ctx.logs,
elapsed_ms=0,
memory_used_mb=0,
)
def validate(self, source: str) -> list[str]:
"""验证 Python 语法"""
try:
compile(source, "<rule_script>", "exec")
return []
except SyntaxError as e:
return [f"第 {e.lineno} 行: {e.msg}"]
def _security_scan(self, source: str) -> list[str]:
"""安全扫描（禁止危险操作）"""
issues = []
forbidden = [
("import os", "禁止导入 os 模块"),
("import sys", "禁止导入 sys 模块"),
("import subprocess", "禁止导入 subprocess 模块"),
("__import__", "禁止动态导入"),
("eval(", "禁止使用 eval"),
("exec(", "禁止使用 exec"),
("open(", "禁止直接文件操作"),
]
for pattern, message in forbidden:
if pattern in source:
issues.append(message)
return issues
#4. Groovy 脚本运行时
#4.1 Groovy 规则示例
GROOVY
// 规则脚本：供应链风险评估
// 文件: scripts/supply_chain_risk.groovy
import onto.rules.RuleContext
import onto.rules.RuleResult
RuleResult evaluate(RuleContext ctx) {
def supplier = ctx.get("supplier_name", String)
def leadTime = ctx.get("lead_time_days", Integer)
def qualityScore = ctx.get("quality_score", Double)
def onTimeRate = ctx.get("on_time_delivery_rate", Double)
def alternativeCount = ctx.get("alternative_suppliers", Integer)
// 风险计算
def riskScore = 0.0
// 交付时间风险
if (leadTime > 30) riskScore += 25
else if (leadTime > 14) riskScore += 10
// 质量风险
if (qualityScore < 0.8) riskScore += 30
else if (qualityScore < 0.9) riskScore += 15
// 交付准时率风险
if (onTimeRate < 0.85) riskScore += 25
else if (onTimeRate < 0.95) riskScore += 10
// 供应商替代性风险
if (alternativeCount == 0) riskScore += 20
else if (alternativeCount == 1) riskScore += 10
// 决策
def decision
def confidence
if (riskScore >= 60) {
decision = "high_risk"
confidence = Math.min(riskScore / 100.0, 0.99)
} else if (riskScore >= 30) {
decision = "medium_risk"
confidence = 0.8
} else {
decision = "low_risk"
confidence = 0.9
}
ctx.log("供应商 ${supplier} 风险评分: ${riskScore}")
return new RuleResult(
decision: decision,
confidence: confidence,
outputs: [
risk_score: riskScore,
risk_breakdown: [
lead_time: leadTime > 14 ? "elevated" : "normal",
quality: qualityScore < 0.9 ? "concern" : "ok",
delivery: onTimeRate < 0.95 ? "concern" : "ok",
alternatives: alternativeCount < 2 ? "limited" : "ok",
]
],
explanation: "供应商 ${supplier} 综合风险评分 ${riskScore}"
)
}
#4.2 Groovy 运行时
Python
class GroovyScriptRuntime(ScriptRuntime):
"""Groovy 脚本运行时"""
def __init__(self, groovy_home: str = "/opt/groovy"):
self._groovy_home = groovy_home
self._compiled: dict[str, Path] = {}
def compile(self, source: str, metadata: ScriptMetadata) -> str:
"""预编译 Groovy 脚本"""
script_path = Path(tempfile.mkdtemp()) / f"{metadata.script_id}.groovy"
script_path.write_text(source)
# Groovy 编译检查
result = subprocess.run(
[f"{self._groovy_home}/bin/groovyc", str(script_path)],
capture_output=True, text=True,
)
if result.returncode != 0:
raise ValueError(f"Groovy 编译错误: {result.stderr}")
compiled_id = f"{metadata.script_id}:{metadata.version}"
self._compiled[compiled_id] = script_path
return compiled_id
def execute(self, compiled_id: str,
inputs: dict[str, Any]) -> ScriptResult:
"""通过 JVM 执行 Groovy 脚本"""
script_path = self._compiled.get(compiled_id)
if not script_path:
raise KeyError(f"脚本未编译: {compiled_id}")
start = time.monotonic()
# 准备输入
input_json = json.dumps(inputs)
# 通过 nsjail + JVM 执行
cmd = [
"nsjail", "--mode", "once",
"--rlimit_as", "512",
"--rlimit_cpu", "10",
"--",
f"{self._groovy_home}/bin/groovy",
"-cp", "/opt/onto/rule-sdk.jar",
str(script_path),
input_json,
]
try:
proc = subprocess.run(
cmd, capture_output=True, text=True, timeout=15,
)
elapsed = (time.monotonic() - start) * 1000
if proc.returncode != 0:
return ScriptResult(
success=False, outputs={}, logs=[],
elapsed_ms=elapsed, memory_used_mb=0,
error=proc.stderr,
)
output = json.loads(proc.stdout)
return ScriptResult(
success=True,
outputs=output.get("outputs", {}),
logs=output.get("logs", []),
elapsed_ms=elapsed,
memory_used_mb=output.get("memory_mb", 0),
)
except subprocess.TimeoutExpired:
return ScriptResult(
success=False, outputs={}, logs=[],
elapsed_ms=15000, memory_used_mb=0,
error="Groovy 执行超时 (15s)",
)
def validate(self, source: str) -> list[str]:
"""Groovy 语法验证"""
result = subprocess.run(
[f"{self._groovy_home}/bin/groovyc", "--dry-run", "-"],
input=source, capture_output=True, text=True,
)
if result.returncode != 0:
return [result.stderr]
return []
#5. 沙箱安全模型
#5.1 安全隔离层次
Code
安全隔离架构：
Layer 1: 静态分析（编译时）
+------------------------------------------+
| - 禁止危险 import (os, sys, subprocess) |
| - 禁止 eval/exec |
| - 禁止文件/网络操作 |
| - 禁止反射和动态类加载 |
+------------------------------------------+
|
v
Layer 2: nsjail 沙箱（运行时）
+------------------------------------------+
| - 独立 namespace（PID, NET, MNT, USER） |
| - 只读文件系统 |
| - 无网络访问 |
| - CPU 时间限制（5-10 秒） |
| - 内存限制（256-512 MB） |
| - 禁止文件创建 |
+------------------------------------------+
|
v
Layer 3: 结果验证（返回时）
+------------------------------------------+
| - 输出大小限制（< 1MB） |
| - 输出类型校验 |
| - 执行时间审计 |
+------------------------------------------+
#5.2 资源限制配置
Python
@dataclass
class SandboxConfig:
"""沙箱配置"""
max_cpu_seconds: int = 5
max_memory_mb: int = 256
max_output_bytes: int = 1_048_576 # 1MB
max_file_size: int = 0 # 禁止写文件
network_enabled: bool = False
allowed_imports: list[str] = field(default_factory=lambda: [
"math", "decimal", "datetime", "json",
"collections", "itertools", "functools",
"dataclasses", "typing", "enum",
"rule_sdk",
])
blocked_syscalls: list[str] = field(default_factory=lambda: [
"socket", "connect", "bind", "listen",
"fork", "clone", "execve",
"open", # 除白名单外禁止
])
class SandboxManager:
"""沙箱管理器"""
def __init__(self, config: SandboxConfig):
self._config = config
def build_nsjail_args(self, script_path: str,
input_path: str) -> list[str]:
return [
"nsjail",
"--mode", "once",
"--rlimit_as", str(self._config.max_memory_mb),
"--rlimit_cpu", str(self._config.max_cpu_seconds),
"--rlimit_fsize", str(self._config.max_file_size),
"--disable_clone_newnet",
"--bindmount_ro", f"{script_path}:/script",
"--bindmount_ro", f"{input_path}:/input.json",
"--cgroup_mem_max", str(self._config.max_memory_mb * 1024 * 1024),
]
#6. 脚本生命周期管理
#6.1 生命周期状态机
Code
脚本生命周期：
DRAFT --> VALIDATING --> VALIDATED --> COMPILING --> COMPILED
| | | | |
| v | v v
| INVALID | ERROR TESTING
| | |
| | +---------+--------+
| | | |
| | v v
| | TEST_PASSED TEST_FAILED
| | |
| | v
| +--------> ACTIVE <------- ROLLBACK
| |
| v
+------------------------------------ARCHIVED
#6.2 脚本注册表
Python
from datetime import datetime
class ScriptStatus(str, Enum):
DRAFT = "draft"
VALIDATED = "validated"
COMPILED = "compiled"
TESTING = "testing"
ACTIVE = "active"
ARCHIVED = "archived"
ERROR = "error"
@dataclass
class ScriptRecord:
"""脚本注册记录"""
metadata: ScriptMetadata
source: str
status: ScriptStatus
compiled_id: str | None = None
created_at: datetime = field(default_factory=datetime.utcnow)
updated_at: datetime = field(default_factory=datetime.utcnow)
test_results: dict | None = None
error_message: str | None = None
class ScriptRegistry:
"""脚本注册表"""
def __init__(self):
self._scripts: dict[str, ScriptRecord] = {}
self._active_by_domain: dict[str, list[str]] = {}
def register(self, metadata: ScriptMetadata, source: str) -> str:
key = f"{metadata.script_id}:{metadata.version}"
self._scripts[key] = ScriptRecord(
metadata=metadata,
source=source,
status=ScriptStatus.DRAFT,
)
return key
def update_status(self, key: str, status: ScriptStatus,
**kwargs) -> None:
record = self._scripts.get(key)
if not record:
raise KeyError(f"脚本未找到: {key}")
record.status = status
record.updated_at = datetime.utcnow()
if "compiled_id" in kwargs:
record.compiled_id = kwargs["compiled_id"]
if "error_message" in kwargs:
record.error_message = kwargs["error_message"]
if "test_results" in kwargs:
record.test_results = kwargs["test_results"]
def activate(self, key: str) -> None:
record = self._scripts[key]
record.status = ScriptStatus.ACTIVE
domain = record.metadata.tags[0] if record.metadata.tags else "default"
self._active_by_domain.setdefault(domain, []).append(key)
def get_active_scripts(self, domain: str) -> list[ScriptRecord]:
keys = self._active_by_domain.get(domain, [])
return [self._scripts[k] for k in keys if k in self._scripts]
def list_all(self) -> list[dict]:
return [
{
"script_id": r.metadata.script_id,
"name": r.metadata.name,
"language": r.metadata.language.value,
"version": r.metadata.version,
"status": r.status.value,
"updated_at": r.updated_at.isoformat(),
}
for r in self._scripts.values()
]
#7. 与 Rete 规则引擎集成
#7.1 脚本作为规则动作
Python
class ScriptActionExecutor:
"""将脚本作为 Rete 规则的 Action 执行"""
def __init__(self, engine: "ScriptEngine"):
self._engine = engine
def create_action(self, script_id: str,
version: str) -> callable:
"""创建 Rete 规则的 action 函数"""
compiled_id = f"{script_id}:{version}"
def action(token) -> dict:
# 从 Token 中提取输入
inputs = {}
for fact in token.facts:
for k, v in fact.attributes:
inputs[k] = v
result = self._engine.execute(compiled_id, inputs)
if result.success:
return result.outputs
else:
return {"error": result.error}
return action
#7.2 YAML 中引用脚本
YAML
apiVersion: rules/v1
kind: RuleSet
metadata:
name: advanced-credit-rules
domain: credit
version: "3.0.0"
spec:
inputs:
credit_score: { type: integer }
annual_income: { type: decimal }
industry: { type: string }
rules:
- id: ACR-001
name: "动态信用评分（脚本规则）"
priority: 100
when:
all:
- credit_score >= 500
- annual_income >= 50000
then:
script:
id: credit_scoring
version: "1.2.0"
language: python
input_mapping:
credit_score: "{{ credit_score }}"
annual_income: "{{ annual_income }}"
industry: "{{ industry }}"
#7.3 混合执行引擎
Code
混合规则执行流程：
输入事实
|
v
+---+---+
| Rete |----> YAML 规则匹配
| 网络 | |
+---+---+ v
| 条件满足?
| | |
| Yes No -> 下一条规则
| |
| v
| +-----+------+
| | Action 类型? |
| +-----+------+
| | |
| v v
| 简单动作 脚本动作
| (直接) (ScriptEngine)
| | |
| v v
| 结果 <------+
| |
v v
新事实注入工作记忆
#8. 多语言性能对比
#8.1 基准测试结果
Code
脚本执行延迟 (毫秒)：
语言 冷启动 热执行 吞吐(次/秒)
----------- -------- -------- ----------
Python | 120 | | 8 | | 125 |
Groovy | 850 | | 12 | | 83 |
TypeScript | 200 | | 5 | | 200 |
Kotlin | 900 | | 10 | | 100 |
WASM | 50 | | 2 | | 500 |
#8.2 语言选择指南
| 语言 | 最佳场景 | 优势 | 劣势 |
|---|---|---|---|
| Python | 数据计算、ML 集成 | 生态丰富、易学 | GIL 限制并发 |
| Groovy | 现有 Java 系统集成 | JVM 互操作 | 冷启动慢 |
| TypeScript | 前端规则一致性 | 类型安全 | V8 内存占用 |
| Kotlin | JVM 高性能规则 | 协程支持 | 冷启动慢 |
| WASM | 高频低延迟规则 | 极快执行 | 编写复杂 |
#9. gRPC 接口
PROTOBUF
// script_engine.proto
syntax = "proto3";
package onto.scripts.v1;
service ScriptEngineService {
rpc RegisterScript(RegisterRequest) returns (RegisterResponse);
rpc CompileScript(CompileRequest) returns (CompileResponse);
rpc ExecuteScript(ExecuteRequest) returns (ExecuteResponse);
rpc ValidateScript(ValidateRequest) returns (ValidateResponse);
rpc ListScripts(ListRequest) returns (ListResponse);
}
message RegisterRequest {
string script_id = 1;
string name = 2;
string language = 3;
string version = 4;
string source = 5;
map<string, string> inputs = 6;
map<string, string> outputs = 7;
int32 timeout_ms = 8;
}
message ExecuteRequest {
string compiled_id = 1;
map<string, string> inputs = 2;
}
message ExecuteResponse {
bool success = 1;
map<string, string> outputs = 2;
repeated string logs = 3;
double elapsed_ms = 4;
string error = 5;
}
#10. 最佳实践
#10.1 脚本编写规范
Code
+---------------------------------------------+
| 脚本规则编写 Checklist |
+---------------------------------------------+
| [ ] 函数签名：evaluate(ctx: RuleContext) |
| [ ] 所有输入通过 ctx.get() 获取 |
| [ ] 使用类型注解 |
| [ ] 返回 RuleResult 对象 |
| [ ] 无外部 I/O（文件、网络、数据库） |
| [ ] 无全局状态修改 |
| [ ] 执行时间 < 5 秒 |
| [ ] 内存使用 < 256 MB |
| [ ] 有单元测试 |
| [ ] 有文档字符串 |
+---------------------------------------------+
#10.2 选择 YAML 还是脚本
Code
决策树：
规则是否包含复杂计算？
| |
No Yes
| |
v v
是否需要 使用脚本
外部 API？
| |
No Yes
| |
v v
YAML 脚本
#Key Takeaways
- 脚本引擎补充 YAML DSL，处理剩余 15% 需要复杂逻辑的规则场景
- 五种语言（Python/Groovy/TS/Kotlin/WASM）覆盖不同技术栈和性能需求
- 三层安全模型（静态分析 + nsjail 沙箱 + 结果验证）确保脚本执行安全
- RuleContext SDK 提供类型安全的输入获取和结构化的结果返回
- 与 Rete 引擎集成，YAML 规则可以调用脚本作为高级动作
- WASM 运行时延迟仅 2ms，适合高频低延迟场景
- 脚本全生命周期管理（注册 -> 编译 -> 测试 -> 上线 -> 归档）
#Next Article
下一篇 S5-06 推理结果可解释性：为什么系统做了这个决策 将展示如何让每个推理和决策结果都可追溯、可解释、可审计。
tags: #script-engine #python #groovy #nsjail #sandbox #function-runtime #coomia-dip