标准引擎 vs 自定义函数：

  标准引擎能覆盖的：              需要自定义函数的：
  ┌─────────────────┐          ┌─────────────────┐
  │ 通用规则匹配      │          │ 行业特定算法      │
  │ 标准 ML 模型      │          │ 自定义风控模型     │
  │ 常见约束求解      │          │ 复杂特征工程      │
  │ 基本数据转换      │          │ 外部API集成       │
  └─────────────────┘          └─────────────────┘
         │                           │
         ▼                           ▼
  覆盖 80% 场景               覆盖剩余 20%

FunctionRuntime 架构：

  函数定义 (FunctionSpec)
       │
       ▼
  ┌──────────────────────────────────────┐
  │         FunctionRegistry              │
  │  注册 / 发现 / 版本管理 / 权限控制      │
  └───────────────┬──────────────────────┘
                  │
       ┌──────────┼──────────┐
       ▼          ▼          ▼
  ┌────────┐ ┌────────┐ ┌────────┐
  │ Python │ │  TS    │ │ WASM   │ ...
  │Runtime │ │Runtime │ │Runtime │
  └───┬────┘ └───┬────┘ └───┬────┘
      │          │          │
      ▼          ▼          ▼
  ┌────────────────────────────────┐
  │        Sandbox Layer            │
  │  资源配额 | 超时 | 网络隔离       │
  └────────────────────────────────┘

from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any


class FunctionLanguage(Enum):
    PYTHON = "python"
    TYPESCRIPT = "typescript"
    GROOVY = "groovy"
    WASM = "wasm"
    KOTLIN = "kotlin"


class FunctionStatus(Enum):
    DRAFT = "draft"
    ACTIVE = "active"
    DEPRECATED = "deprecated"
    DISABLED = "disabled"


@dataclass
class FunctionSpec:
    """函数规格定义"""
    function_id: str
    name: str
    language: FunctionLanguage
    version: str
    source_code: str
    entry_point: str                          # 入口函数名
    description: str = ""
    input_schema: dict[str, Any] = field(default_factory=dict)
    output_schema: dict[str, Any] = field(default_factory=dict)
    dependencies: list[str] = field(default_factory=list)
    resource_limits: ResourceLimits = field(default_factory=lambda: ResourceLimits())
    tags: list[str] = field(default_factory=list)
    status: FunctionStatus = FunctionStatus.DRAFT
    created_at: datetime = field(default_factory=datetime.utcnow)
    updated_at: datetime = field(default_factory=datetime.utcnow)


@dataclass
class ResourceLimits:
    """资源限制"""
    max_memory_mb: int = 256
    max_cpu_ms: int = 5000              # CPU 时间限制 5s
    max_execution_time_ms: int = 30000  # 墙钟时间 30s
    max_output_size_kb: int = 1024
    network_enabled: bool = False
    filesystem_read_only: bool = True
    max_file_descriptors: int = 64


@dataclass
class FunctionInvocation:
    """函数调用"""
    invocation_id: str
    function_id: str
    inputs: dict[str, Any]
    outputs: dict[str, Any] | None = None
    status: str = "pending"             # pending, running, completed, failed, timeout
    error: str = ""
    duration_ms: float = 0.0
    memory_used_mb: float = 0.0
    started_at: datetime | None = None
    completed_at: datetime | None = None

from abc import ABC, abstractmethod


class LanguageRuntime(ABC):
    """语言运行时抽象基类"""

    @abstractmethod
    async def initialize(self, spec: FunctionSpec) -> None:
        """初始化运行环境"""
        ...

    @abstractmethod
    async def execute(self, spec: FunctionSpec,
                       inputs: dict[str, Any]) -> dict[str, Any]:
        """执行函数"""
        ...

    @abstractmethod
    async def cleanup(self, spec: FunctionSpec) -> None:
        """清理运行环境"""
        ...

    @abstractmethod
    def supported_language(self) -> FunctionLanguage:
        ...

import subprocess
import json
import tempfile
import os
from pathlib import Path


class PythonRuntime(LanguageRuntime):
    """Python 函数运行时"""

    def __init__(self, sandbox_type: str = "nsjail"):
        self._sandbox_type = sandbox_type
        self._venv_cache: dict[str, str] = {}

    def supported_language(self) -> FunctionLanguage:
        return FunctionLanguage.PYTHON

    async def initialize(self, spec: FunctionSpec) -> None:
        """创建虚拟环境并安装依赖"""
        venv_dir = f"/tmp/fn-venv/{spec.function_id}/{spec.version}"
        os.makedirs(venv_dir, exist_ok=True)

        subprocess.run(
            ["python", "-m", "venv", venv_dir],
            check=True, timeout=30,
        )

        if spec.dependencies:
            pip = os.path.join(venv_dir, "bin", "pip")
            subprocess.run(
                [pip, "install", "--no-cache-dir"] + spec.dependencies,
                check=True, timeout=120,
            )

        self._venv_cache[spec.function_id] = venv_dir

    async def execute(self, spec: FunctionSpec,
                       inputs: dict[str, Any]) -> dict[str, Any]:
        """在沙箱中执行 Python 函数"""
        # 构建执行脚本
        wrapper = self._build_wrapper(spec, inputs)

        with tempfile.NamedTemporaryFile(
            mode="w", suffix=".py", delete=False
        ) as f:
            f.write(wrapper)
            script_path = f.name

        try:
            venv_dir = self._venv_cache.get(spec.function_id, "")
            python_bin = os.path.join(venv_dir, "bin", "python") if venv_dir else "python"

            result = subprocess.run(
                [python_bin, script_path],
                capture_output=True,
                text=True,
                timeout=spec.resource_limits.max_execution_time_ms / 1000,
            )

            if result.returncode != 0:
                raise RuntimeError(f"Function failed: {result.stderr}")

            return json.loads(result.stdout)

        finally:
            os.unlink(script_path)

    def _build_wrapper(self, spec: FunctionSpec,
                        inputs: dict[str, Any]) -> str:
        return f"""
import json
import sys

# 用户函数代码
{spec.source_code}

# 执行入口
inputs = json.loads('''{json.dumps(inputs)}''')
result = {spec.entry_point}(**inputs)

# 输出结果
print(json.dumps(result))
"""

    async def cleanup(self, spec: FunctionSpec) -> None:
        import shutil
        venv_dir = self._venv_cache.pop(spec.function_id, None)
        if venv_dir and os.path.exists(venv_dir):
            shutil.rmtree(venv_dir)

class TypeScriptRuntime(LanguageRuntime):
    """TypeScript 函数运行时"""

    def supported_language(self) -> FunctionLanguage:
        return FunctionLanguage.TYPESCRIPT

    async def initialize(self, spec: FunctionSpec) -> None:
        work_dir = f"/tmp/fn-ts/{spec.function_id}/{spec.version}"
        os.makedirs(work_dir, exist_ok=True)

        # 写入 package.json
        package = {
            "name": spec.function_id,
            "version": spec.version,
            "type": "module",
            "dependencies": {
                dep.split("@")[0]: dep.split("@")[1] if "@" in dep else "latest"
                for dep in spec.dependencies
            },
        }
        with open(os.path.join(work_dir, "package.json"), "w") as f:
            json.dump(package, f)

        subprocess.run(
            ["npm", "install", "--production"],
            cwd=work_dir, check=True, timeout=120,
        )

    async def execute(self, spec: FunctionSpec,
                       inputs: dict[str, Any]) -> dict[str, Any]:
        wrapper = f"""
const inputs = {json.dumps(inputs)};

{spec.source_code}

const result = {spec.entry_point}(inputs);
Promise.resolve(result).then(r => {{
    process.stdout.write(JSON.stringify(r));
}});
"""

        with tempfile.NamedTemporaryFile(
            mode="w", suffix=".mjs", delete=False
        ) as f:
            f.write(wrapper)
            script_path = f.name

        try:
            result = subprocess.run(
                ["node", "--experimental-vm-modules", script_path],
                capture_output=True, text=True,
                timeout=spec.resource_limits.max_execution_time_ms / 1000,
            )

            if result.returncode != 0:
                raise RuntimeError(f"Function failed: {result.stderr}")

            return json.loads(result.stdout)
        finally:
            os.unlink(script_path)

    async def cleanup(self, spec: FunctionSpec) -> None:
        import shutil
        work_dir = f"/tmp/fn-ts/{spec.function_id}/{spec.version}"
        if os.path.exists(work_dir):
            shutil.rmtree(work_dir)

class FunctionRegistry:
    """函数注册表"""

    def __init__(self, store):
        self._store = store
        self._runtimes: dict[FunctionLanguage, LanguageRuntime] = {}

    def register_runtime(self, runtime: LanguageRuntime) -> None:
        self._runtimes[runtime.supported_language()] = runtime

    async def register_function(self, spec: FunctionSpec) -> str:
        """注册函数"""
        # 验证
        self._validate_spec(spec)

        # 初始化运行时
        runtime = self._runtimes.get(spec.language)
        if runtime is None:
            raise ValueError(f"Unsupported language: {spec.language}")

        await runtime.initialize(spec)

        # 保存
        spec.status = FunctionStatus.ACTIVE
        await self._store.save(spec)

        return spec.function_id

    async def invoke(self, function_id: str,
                      inputs: dict[str, Any]) -> FunctionInvocation:
        """调用函数"""
        spec = await self._store.get(function_id)
        if spec is None:
            raise ValueError(f"Function not found: {function_id}")
        if spec.status != FunctionStatus.ACTIVE:
            raise ValueError(f"Function not active: {function_id}")

        runtime = self._runtimes.get(spec.language)
        invocation = FunctionInvocation(
            invocation_id=f"inv-{function_id}-{datetime.utcnow().timestamp()}",
            function_id=function_id,
            inputs=inputs,
            started_at=datetime.utcnow(),
        )

        try:
            invocation.status = "running"
            import time
            start = time.monotonic()

            outputs = await runtime.execute(spec, inputs)

            invocation.outputs = outputs
            invocation.status = "completed"
            invocation.duration_ms = (time.monotonic() - start) * 1000

        except TimeoutError:
            invocation.status = "timeout"
            invocation.error = "Execution timeout"
        except Exception as e:
            invocation.status = "failed"
            invocation.error = str(e)

        invocation.completed_at = datetime.utcnow()
        return invocation

    def _validate_spec(self, spec: FunctionSpec) -> None:
        if not spec.function_id:
            raise ValueError("function_id is required")
        if not spec.source_code:
            raise ValueError("source_code is required")
        if not spec.entry_point:
            raise ValueError("entry_point is required")
        if spec.resource_limits.max_memory_mb > 1024:
            raise ValueError("max_memory_mb cannot exceed 1024")

    async def list_functions(self, language: FunctionLanguage | None = None,
                              tags: list[str] | None = None) -> list[FunctionSpec]:
        return await self._store.query(language=language, tags=tags)

    async def deprecate(self, function_id: str) -> None:
        spec = await self._store.get(function_id)
        if spec:
            spec.status = FunctionStatus.DEPRECATED
            await self._store.save(spec)

class TypeMapper:
    """Ontology 类型到各语言类型的映射"""

    _mappings = {
        "string": {
            FunctionLanguage.PYTHON: "str",
            FunctionLanguage.TYPESCRIPT: "string",
            FunctionLanguage.KOTLIN: "String",
            FunctionLanguage.GROOVY: "String",
        },
        "integer": {
            FunctionLanguage.PYTHON: "int",
            FunctionLanguage.TYPESCRIPT: "number",
            FunctionLanguage.KOTLIN: "Int",
            FunctionLanguage.GROOVY: "Integer",
        },
        "float": {
            FunctionLanguage.PYTHON: "float",
            FunctionLanguage.TYPESCRIPT: "number",
            FunctionLanguage.KOTLIN: "Double",
            FunctionLanguage.GROOVY: "Double",
        },
        "boolean": {
            FunctionLanguage.PYTHON: "bool",
            FunctionLanguage.TYPESCRIPT: "boolean",
            FunctionLanguage.KOTLIN: "Boolean",
            FunctionLanguage.GROOVY: "Boolean",
        },
        "datetime": {
            FunctionLanguage.PYTHON: "datetime",
            FunctionLanguage.TYPESCRIPT: "Date",
            FunctionLanguage.KOTLIN: "Instant",
            FunctionLanguage.GROOVY: "Instant",
        },
        "object": {
            FunctionLanguage.PYTHON: "dict",
            FunctionLanguage.TYPESCRIPT: "Record<string, unknown>",
            FunctionLanguage.KOTLIN: "Map<String, Any>",
            FunctionLanguage.GROOVY: "Map<String, Object>",
        },
        "array": {
            FunctionLanguage.PYTHON: "list",
            FunctionLanguage.TYPESCRIPT: "Array<unknown>",
            FunctionLanguage.KOTLIN: "List<Any>",
            FunctionLanguage.GROOVY: "List<Object>",
        },
    }

    @classmethod
    def map_type(cls, ontology_type: str,
                  language: FunctionLanguage) -> str:
        type_map = cls._mappings.get(ontology_type, {})
        return type_map.get(language, "Any")

class FunctionDecisionPlugin:
    """函数作为决策引擎的插件"""

    def __init__(self, registry: FunctionRegistry):
        self._registry = registry

    async def evaluate_with_function(self, function_id: str,
                                       context) -> dict:
        """使用自定义函数评估决策"""
        invocation = await self._registry.invoke(
            function_id, context.inputs
        )

        if invocation.status == "completed":
            return {
                "decision": invocation.outputs.get("decision", "unknown"),
                "confidence": invocation.outputs.get("confidence", 0.0),
                "source": "custom_function",
                "function_id": function_id,
                "duration_ms": invocation.duration_ms,
            }
        else:
            return {
                "decision": "error",
                "confidence": 0.0,
                "error": invocation.error,
            }

syntax = "proto3";
package onto.function.v1;

service FunctionService {
    rpc RegisterFunction(RegisterRequest) returns (RegisterResponse);
    rpc InvokeFunction(InvokeRequest) returns (InvokeResponse);
    rpc ListFunctions(ListRequest) returns (ListResponse);
    rpc GetFunction(GetRequest) returns (FunctionDetail);
    rpc UpdateFunction(UpdateRequest) returns (UpdateResponse);
    rpc DeprecateFunction(DeprecateRequest) returns (DeprecateResponse);
}

message RegisterRequest {
    string name = 1;
    string language = 2;
    string version = 3;
    string source_code = 4;
    string entry_point = 5;
    string input_schema_json = 6;
    string output_schema_json = 7;
    repeated string dependencies = 8;
    ResourceLimits limits = 9;
}

message InvokeRequest {
    string function_id = 1;
    string inputs_json = 2;
}

message InvokeResponse {
    string invocation_id = 1;
    string status = 2;
    string outputs_json = 3;
    double duration_ms = 4;
    string error = 5;
}

函数运行时安全层次：

  Layer 1: 代码审查
  ├── 静态分析：禁止 import os.system, subprocess
  ├── 依赖白名单：只允许经过审批的包
  │
  Layer 2: 沙箱隔离
  ├── nsjail / gVisor 容器沙箱
  ├── 只读文件系统
  ├── 禁止网络访问（默认）
  │
  Layer 3: 资源限制
  ├── CPU 时间上限
  ├── 内存上限
  ├── 文件描述符上限
  │
  Layer 4: 输出验证
  ├── 输出大小限制
  └── 输出 Schema 校验

# 注册一个自定义信贷评分函数

spec = FunctionSpec(
    function_id="fn-credit-custom-001",
    name="custom_credit_scorer",
    language=FunctionLanguage.PYTHON,
    version="1.0.0",
    entry_point="score",
    source_code="""
def score(credit_score: int, annual_income: float,
          debt_ratio: float, employment_years: int) -> dict:
    base_score = credit_score / 850 * 40
    income_score = min(annual_income / 500000, 1.0) * 25
    debt_penalty = max(0, (debt_ratio - 0.3) * 50)
    tenure_bonus = min(employment_years / 10, 1.0) * 10

    final_score = base_score + income_score - debt_penalty + tenure_bonus

    if final_score >= 60:
        decision = "approve"
    elif final_score >= 40:
        decision = "conditional_approve"
    else:
        decision = "reject"

    return {
        "decision": decision,
        "confidence": min(final_score / 100, 1.0),
        "score_breakdown": {
            "base": base_score,
            "income": income_score,
            "debt_penalty": debt_penalty,
            "tenure_bonus": tenure_bonus,
            "final": final_score,
        },
    }
""",
    dependencies=["numpy"],
    resource_limits=ResourceLimits(
        max_memory_mb=128,
        max_execution_time_ms=5000,
    ),
)

registry = FunctionRegistry(store)
await registry.register_function(spec)

# 调用
result = await registry.invoke("fn-credit-custom-001", {
    "credit_score": 680,
    "annual_income": 250000,
    "debt_ratio": 0.35,
    "employment_years": 8,
})
# result.outputs = {
#   "decision": "conditional_approve",
#   "confidence": 0.57,
#   "score_breakdown": {...},
# }