源码精读：FunctionRuntime — 多语言沙箱的统一接口

“

系列：S9 源码精读 · 第 14 篇 | 难度：高级 | 阅读时间：25 分钟

#TL;DR

FunctionRuntimeService 是 coomia-dip 推理决策层（Reasoning & Decision Layer）中的函数执行引擎，为 Python、TypeScript、Groovy 三种语言提供统一的注册、调用和生命周期管理能力。它通过 FunctionRegistry 管理函数定义，通过 FunctionExecutor 实现沙箱隔离执行，支持五种函数类型（QUERY/DERIVED/ACTION/AGGREGATION/TIMESERIES）和三种执行模式（SERVERLESS/DEPLOYED/PIPELINE）。本文将深入剖析 Pydantic 请求模型与 Proto 合约的映射策略、Registry/Executor 双层架构、WorldContext 的函数注入机制、以及异步调用的状态追踪设计。

#1. 整体架构：Registry + Executor 双层设计

FunctionRuntime 采用经典的注册中心+执行器架构，将函数的"定义管理"与"运行时执行"完全分离：

Code

intelligence-Layer/src/reasoning_decision_plane/
├── function_runtime/
│   ├── registry.py               # FunctionRegistry — 函数注册中心
│   ├── executor.py               # FunctionExecutor — 沙箱执行器
│   └── models.py                 # Domain 模型（Pydantic v2）
├── api/grpc/
│   └── function_runtime_servicer.py  # gRPC Servicer 适配层
├── common/
│   ├── context.py                # WorldContext
│   └── exceptions.py             # 统一异常体系

关键设计决策：FunctionRuntimeServicer 通过构造函数注入 FunctionRegistry 和 FunctionExecutor，两者都提供了工厂函数 get_function_registry() 和 get_function_executor() 作为默认实例获取方式：

Python

class FunctionRuntimeServicer:
    def __init__(
        self,
        registry: FunctionRegistry | None = None,
        executor: FunctionExecutor | None = None,
    ) -> None:
        self._registry = registry or get_function_registry()
        self._executor = executor or get_function_executor()

这种"可注入但有默认值"的模式在测试中特别有用——单元测试可以注入 Mock 对象，而集成测试使用真实实例。

#2. Proto 合约：FunctionDefinition 的五类型三模式

Proto 定义中的 FunctionDefinition 承载了函数的完整元数据：

PROTOBUF

enum FunctionType {
  FUNCTION_TYPE_QUERY = 1;           // 只读查询
  FUNCTION_TYPE_DERIVED = 2;         // 派生属性计算
  FUNCTION_TYPE_ACTION = 3;          // 可变操作
  FUNCTION_TYPE_AGGREGATION = 4;     // ObjectSet 聚合
  FUNCTION_TYPE_TIMESERIES = 5;      // 时序处理
}

enum FunctionLanguage {
  FUNCTION_LANGUAGE_PYTHON = 1;
  FUNCTION_LANGUAGE_TYPESCRIPT = 2;
  FUNCTION_LANGUAGE_GROOVY = 3;
}

enum ExecutionMode {
  EXECUTION_MODE_SERVERLESS = 1;     // 按需执行
  EXECUTION_MODE_DEPLOYED = 2;       // 长驻进程
  EXECUTION_MODE_PIPELINE = 3;       // Pipeline 节点
}

五种函数类型的设计映射了 Palantir Foundry 的函数分类：

QUERY：只读函数，可安全缓存结果
DERIVED：派生属性计算，与 DerivedPropertyService 联动
ACTION：可变操作，由 ActionEngine 调用
AGGREGATION：对 ObjectSet 进行聚合计算
TIMESERIES：时序数据处理

三种执行模式平衡了资源效率和启动延迟：SERVERLESS 适合低频调用、DEPLOYED 适合高频热路径、PIPELINE 适合批处理场景。

#3. FunctionRuntimeServicer：gRPC 适配层

Servicer 的核心职责是 Proto 请求/响应模型与内部 Domain 模型之间的转换。它使用 Pydantic BaseModel 作为中间层：

Python

class RegisterFunctionRequest(BaseModel):
    function_id: str = ""
    function_name: str = ""
    function_type: str = "QUERY"
    language: str = "PYTHON"
    code: str = ""
    entrypoint: str = "main"
    description: str = ""
    timeout_seconds: int = 30
    max_memory_mb: int = 256

默认值策略：所有字段都有合理的默认值。function_type 默认为 "QUERY"（最安全的只读类型），language 默认为 "PYTHON"（平台主力语言），entrypoint 默认为 "main"（Python 惯例）。这使得最小化的注册请求只需提供 function_name 和 code。

FunctionResponse 同时承载成功和失败两种情况：

Python

class FunctionResponse(BaseModel):
    function_id: str = ""
    function_name: str = ""
    # ... 正常字段 ...
    error_code: str = ""
    error_message: str = ""

当 error_code 非空时表示操作失败。这种"统一响应"模式避免了 gRPC 层面抛出异常，让客户端可以统一处理成功和失败。

#4. 注册流程：从 Proto 到 Domain 的转换链

register_function 方法展示了完整的 Proto→Domain 转换链路：

Python

async def register_function(
    self,
    request: RegisterFunctionRequest,
) -> FunctionResponse:
    try:
        # 1. 解析语言枚举（容错降级为 PYTHON）
        try:
            language = FunctionLanguage(request.language.lower())
        except ValueError:
            language = FunctionLanguage.PYTHON

        # 2. 解析函数类型（容错降级为 QUERY）
        try:
            func_type = FunctionType(request.function_type.lower())
        except ValueError:
            func_type = FunctionType.QUERY

        # 3. 构建 FunctionConfig
        config = FunctionConfig(
            timeout_seconds=request.timeout_seconds or 30,
            max_memory_mb=request.max_memory_mb or 256,
        )

        # 4. 调用 Registry 注册
        func_def = await self._registry.register(
            function_id=request.function_id or None,
            function_name=request.function_name,
            type=func_type,
            language=language,
            code=request.code,
            entrypoint=request.entrypoint or "main",
            config=config,
            description=request.description,
        )

容错降级模式是关键设计点：枚举解析失败时不抛异常，而是降级到最安全的默认值（PYTHON/QUERY）。这使得系统在面对不规范的客户端输入时保持健壮。

function_id 为空时自动生成：当客户端未提供 function_id 时，传 None 给 Registry，由 Registry 内部生成 UUID。

#5. 函数调用：WorldContext 注入与沙箱执行

invoke_function 是最核心的方法，展示了函数调用的完整流程：

Python

async def invoke_function(
    self,
    request: InvokeFunctionRequest,
) -> InvokeFunctionResponse:
    try:
        # 构建 WorldContext
        world_context: WorldContext | None = None
        if request.world_id:
            world_context = WorldContext(world_id=request.world_id)

        # 执行函数
        result = await self._executor.invoke(
            function_id=request.function_id,
            arguments=request.arguments,
            world_context=world_context,
            timeout_override=request.timeout_override or None,
        )

WorldContext 注入：当调用者提供 world_id 时，构建 WorldContext 对象注入到执行器中。这使得函数内部可以感知当前的世界上下文——例如在 Nessie 的哪个分支上执行查询。

Proto 层面的 FunctionArgument 支持七种值类型的 oneof：

PROTOBUF

message FunctionArgument {
  string name = 1;
  oneof value {
    string string_value = 2;
    int64 int_value = 3;
    double double_value = 4;
    bool bool_value = 5;
    bytes json_value = 6;
    ObjectSetRef object_set = 7;
    ObjectRef object = 8;
  }
}

ObjectSetRef 和 ObjectRef 是函数系统与本体系统的桥梁——函数可以直接接收本体对象或对象集合作为参数。

#6. 异常分层：四级错误处理

FunctionRuntime 定义了四种专用异常，在 Servicer 中形成了清晰的异常处理层次：

Python

except FunctionNotFoundError as e:
    return InvokeFunctionResponse(
        invocation_id=f"inv-{uuid.uuid4().hex[:12]}",
        success=False,
        error_code="NOT_FOUND",
        error_message=str(e),
    )
except FunctionTimeoutError as e:
    return InvokeFunctionResponse(
        error_code="TIMEOUT",
        error_message=str(e),
    )
except FunctionExecutionError as e:
    return InvokeFunctionResponse(
        error_code="EXECUTION_ERROR",
        error_message=e.error_message,
        stack_trace=e.stack_trace or "",
    )
except Exception as e:
    return InvokeFunctionResponse(
        error_code="INTERNAL_ERROR",
        error_message=str(e),
    )

NOT_FOUND → TIMEOUT → EXECUTION_ERROR → INTERNAL_ERROR，从最具体到最通用。FunctionExecutionError 额外携带 stack_trace 字段，帮助开发者定位函数内部的错误。

注意即使在异常路径中，响应也会生成有效的 invocation_id（使用 UUID hex 前 12 位），确保客户端可以关联日志进行问题排查。

#7. 异步调用与状态追踪

Proto 定义了独立的异步调用接口和状态查询接口：

PROTOBUF

service FunctionRuntimeService {
  rpc InvokeFunction(InvokeFunctionRequest) returns (InvokeFunctionResponse);
  rpc InvokeFunctionAsync(InvokeFunctionRequest) returns (AsyncInvocationResponse);
  rpc GetInvocationStatus(GetInvocationStatusRequest) returns (InvocationStatus);
}

异步调用返回的 AsyncInvocationResponse 只包含 invocation_id 和初始状态：

PROTOBUF

message AsyncInvocationResponse {
  string invocation_id = 1;
  InvocationState state = 2;       // 通常为 PENDING
  com.onto.common.v1.ErrorInfo error = 3;
}

客户端通过轮询 GetInvocationStatus 获取执行进度和最终结果。InvocationStatus 包含完整的执行信息：

PROTOBUF

message InvocationStatus {
  string invocation_id = 1;
  InvocationState state = 2;
  FunctionResult result = 3;
  InvocationMetrics metrics = 4;
  google.protobuf.Timestamp started_at = 5;
  google.protobuf.Timestamp completed_at = 6;
  com.onto.common.v1.ErrorInfo error = 7;
}

七种调用状态覆盖了异步函数的完整生命周期：

Code

PENDING → RUNNING → COMPLETED
                  → FAILED
                  → TIMEOUT
PENDING → CANCELLED

#8. 代码绑定生成：GenerateCodeBindings

GenerateCodeBindings 是一个巧妙的开发体验特性：

PROTOBUF

message GenerateBindingsRequest {
  com.onto.common.v1.RequestContext context = 1;
  repeated string ontology_types = 2;     // 要生成绑定的类型
  FunctionLanguage target_language = 3;
}

message CodeBindingsResponse {
  map<string, string> bindings = 1;       // type_name -> generated code
  com.onto.common.v1.ErrorInfo error = 2;
}

它根据指定的本体类型和目标语言，自动生成类型安全的代码绑定。例如，对于 Employee 类型和 Python 语言，它会生成包含所有属性和关系的 Pydantic 模型类。返回的 bindings 是一个 map<string, string>，key 是类型名，value 是生成的代码文本。

#9. 函数更新：Config 合并策略

update_function 方法展示了一个精心设计的部分更新策略：

Python

async def update_function(
    self,
    request: UpdateFunctionRequest,
) -> FunctionResponse:
    config: FunctionConfig | None = None
    if request.timeout_seconds is not None or request.max_memory_mb is not None:
        # 获取现有函数以保留其他配置值
        existing = await self._registry.get_or_none(request.function_id)
        if existing:
            config = FunctionConfig(
                timeout_seconds=request.timeout_seconds
                if request.timeout_seconds is not None
                else existing.config.timeout_seconds,
                max_memory_mb=request.max_memory_mb
                if request.max_memory_mb is not None
                else existing.config.max_memory_mb,
                mode=existing.config.mode,
                allowed_apis=existing.config.allowed_apis,
                env_vars=existing.config.env_vars,
            )

合并而非覆盖：当更新请求只修改 timeout_seconds 时，其他配置字段（max_memory_mb、mode、allowed_apis、env_vars）全部从现有函数定义中保留。这避免了"更新一个字段导致其他字段被清空"的常见 API 陷阱。

请求模型中使用 int | None 来区分"未提供"和"设为 0"两种语义：

Python

class UpdateFunctionRequest(BaseModel):
    function_id: str = ""
    code: str | None = None
    entrypoint: str | None = None
    timeout_seconds: int | None = None
    max_memory_mb: int | None = None

#10. InvocationMetrics：性能可观测性

每次函数调用都返回详细的执行指标：

PROTOBUF

message InvocationMetrics {
  double execution_time_ms = 1;
  double memory_used_mb = 2;
  double cpu_time_ms = 3;
  int32 api_calls = 4;
}

四个维度的指标服务于不同的运维需求：

execution_time_ms：端到端延迟，用于 SLA 监控
memory_used_mb：内存使用，用于资源配额管理
cpu_time_ms：CPU 时间，用于计费和容量规划
api_calls：API 调用次数，用于检测函数是否过度调用外部服务

Servicer 在成功执行后将这些指标传递给调用者：

Python

if result.success:
    return InvokeFunctionResponse(
        invocation_id=result.invocation_id,
        result=result.result,
        success=True,
        execution_time_ms=result.execution_time_ms,
        memory_used_mb=result.memory_used_mb,
    )

#11. Key Takeaways

Registry/Executor 分离：函数定义管理和运行时执行的解耦使得两者可以独立扩展和测试。
容错降级模式：枚举解析失败时降级到最安全的默认值，而非抛出异常，提升了系统健壮性。
统一响应模式：成功和失败共用同一个 Response 类型，通过 error_code 字段区分，简化了客户端处理逻辑。
Config 合并更新：部分更新时保留未修改的配置字段，避免了"更新导致数据丢失"的 API 设计反模式。
七种参数类型：FunctionArgument 的 oneof value 覆盖了从基础类型到本体对象引用的完整类型谱系。
四维执行指标：时间/内存/CPU/API 调用的组合提供了全面的性能可观测性。

#下一篇

S9-15：DerivedPropertyService — 依赖 DAG 与级联重算，我们将深入 Reasoning & Decision Layer 的派生属性引擎，了解四种计算模式和三轴版本模型的实现。

Tags: #coomia-dip #source-code-reading #function-runtime #multi-language #sandbox #grpc #Layer-d