源码精读：DerivedPropertyService — 依赖 DAG 与级联重算

“

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

#TL;DR

DerivedPropertyService 是 coomia-dip 推理决策层（Reasoning & Decision Layer）中的派生属性引擎，实现了本体实例上的虚拟计算属性。它支持四种计算模式——FunctionRuntime 函数、SQL 查询、算术表达式、Reducer 聚合——以及三种存储策略（VIRTUAL/CACHED/MATERIALIZED）。通过 DerivedPropertyEngine 核心引擎，它管理属性定义、依赖 DAG、缓存失效、级联重算和三轴版本模型（数据版本 x Schema 版本 x 计算逻辑版本）。本文逐行剖析 Proto 合约中的四种计算模式优先级规则、Servicer 层的 Pydantic-Proto 双向转换、缓存统计的可观测性设计、以及 FEAT-012 Reducer 聚合的多跳链路遍历机制。

#1. 整体架构：Engine + Servicer 分层

DerivedPropertyService 的代码分布在三个层次：

Code

intelligence-Layer/src/reasoning_decision_plane/
├── derived_property/
│   ├── engine.py                 # DerivedPropertyEngine — 核心引擎
│   └── models.py                 # Domain 模型（Pydantic v2）
├── api/grpc/
│   └── derived_property_servicer.py  # gRPC Servicer 适配层
proto/plane_d/
└── derived_property.proto        # 463 行核心合约

Servicer 采用与 FunctionRuntime 相同的"可注入但有默认值"模式：

Python

class DerivedPropertyServicer:
    def __init__(
        self,
        engine: DerivedPropertyEngine | None = None,
    ) -> None:
        self._engine = engine or get_derived_property_engine()

#2. 四种计算模式与优先级规则

DerivedPropertyDefinition 支持四种互斥的计算模式，通过不同字段承载：

PROTOBUF

message DerivedPropertyDefinition {
  string property_id = 1;
  string property_name = 2;
  string ontology_type = 3;
  string return_type = 4;
  StorageMode storage_mode = 5;
  string function_id = 6;              // 计算模式 1: FunctionRuntime
  repeated string dependencies = 7;

  SqlComputation sql_computation = 15;       // 计算模式 2: SQL 查询
  ExpressionComputation expression_computation = 16;  // 计算模式 3: 算术表达式
  ReducerDefinition reducer = 17;            // 计算模式 4: Reducer 聚合
}

优先级规则在 Proto 注释中明确定义：

Code

sql_computation > expression_computation > reducer > function_id

这意味着如果同时设置了 sql_computation 和 function_id，引擎会使用 SQL 计算。这个设计允许渐进式迁移——先用 function_id 快速实现，后续替换为更高效的 SQL 或表达式计算时只需添加新字段，无需删除旧字段。

#3. 三种存储策略：VIRTUAL / CACHED / MATERIALIZED

PROTOBUF

enum StorageMode {
  STORAGE_MODE_VIRTUAL = 1;           // 每次查询都计算
  STORAGE_MODE_CACHED = 2;            // 缓存在 Redis
  STORAGE_MODE_MATERIALIZED = 3;      // 存储在 Iceberg
}

每种存储策略对应不同的配置：

CACHED 模式需要 CacheConfig：

PROTOBUF

message CacheConfig {
  int32 ttl_seconds = 1;
  repeated string invalidation_events = 2;
}

invalidation_events 列出了会触发缓存失效的事件模式，例如 "entity.Employee.updated" 会在任何 Employee 实例更新时清除该属性的缓存。

MATERIALIZED 模式需要 MaterializationConfig：

PROTOBUF

message MaterializationConfig {
  TriggerType trigger_type = 1;
  string schedule = 2;                    // Cron 表达式
  repeated string events = 3;
  string debounce = 4;                    // 如 "5m"
  int32 batch_size = 6;
  int32 parallelism = 7;
  int32 retry_count = 8;
}

debounce 是一个关键优化——当短时间内有大量变更事件时，延迟物化执行避免重复计算。

#4. 三轴版本模型：数据 x Schema x 计算逻辑

DerivedProperty 引入了"三轴版本模型"确保计算的确定性和可重现性：

PROTOBUF

message DerivedPropertyDefinition {
  // ... 其他字段 ...
  optional string compute_version = 13;
  BranchBindingPolicy branch_binding_policy = 14;
}

三个轴分别是：

数据版本：通过 WorldContext.commit_hash（Nessie commit）确定
Schema 版本：本体类型定义的版本
计算逻辑版本：compute_version 字段，当设为具体版本号（如 "1.2.0"）时，function_id 会解析到该特定版本的函数

当 compute_version 为空或 "latest" 时，使用最新版本的函数——适合开发环境。在生产环境中锁定具体版本确保同样的输入始终产出同样的结果。

#5. DerivedPropertyServicer：请求模型设计

Servicer 为每种计算模式都定义了专用的请求模型：

Python

class SqlComputationRequest(BaseModel):
    connection_id: str = ""
    sql_template: str = ""
    params: list[SqlTemplateParamRequest] = Field(default_factory=list)
    result_column: str = ""
    timeout_seconds: int = 5
    max_rows: int = 1

class ExpressionComputationRequest(BaseModel):
    expression: str = ""
    return_type: str = "float"

class ReducerDefinitionRequest(BaseModel):
    reducer_id: str = ""
    name: str = ""
    source_object_type: str = ""
    link_path: list[LinkPathStepRequest] = Field(default_factory=list)
    target_property: str = ""
    aggregation: str = "SUM"
    strategy: str = "VIRTUAL"

DefineDerivedPropertyRequest 将所有模式合并在一个请求中，通过各字段是否为 None 来区分使用哪种计算模式：

Python

class DefineDerivedPropertyRequest(BaseModel):
    property_id: str = ""
    property_name: str = ""
    ontology_type: str = ""
    return_type: str = "string"
    storage_mode: str = "VIRTUAL"
    function_id: str = ""
    dependencies: list[str] = Field(default_factory=list)
    cache_config: CacheConfigRequest | None = None
    materialization_config: MaterializationConfigRequest | None = None
    sql_computation: SqlComputationRequest | None = None
    expression_computation: ExpressionComputationRequest | None = None
    reducer: ReducerDefinitionRequest | None = None
    description: str = ""

#6. 属性定义：四种计算模式的构建流程

define_derived_property 方法展示了四种计算模式的构建流程：

Python

async def define_derived_property(
    self, request: DefineDerivedPropertyRequest,
) -> DerivedPropertyResponse:
    # 解析存储模式（容错降级为 VIRTUAL）
    try:
        storage_mode = StorageMode(request.storage_mode.lower())
    except ValueError:
        storage_mode = StorageMode.VIRTUAL

    # 构建 SQL 计算（检查 connection_id 是否非空）
    sql_comp: SqlComputation | None = None
    if request.sql_computation and request.sql_computation.connection_id:
        sql_comp = SqlComputation(
            connection_id=request.sql_computation.connection_id,
            sql_template=request.sql_computation.sql_template,
            params=[
                SqlTemplateParam(
                    placeholder=p.placeholder,
                    source=p.source,
                    default_value=p.default_value,
                )
                for p in request.sql_computation.params
            ],
            result_column=request.sql_computation.result_column,
            timeout_seconds=request.sql_computation.timeout_seconds,
            max_rows=request.sql_computation.max_rows,
        )

    # 构建表达式计算（检查 expression 是否非空）
    expr_comp: ExpressionComputation | None = None
    if request.expression_computation and request.expression_computation.expression:
        expr_comp = ExpressionComputation(
            expression=request.expression_computation.expression,
            return_type=request.expression_computation.return_type,
        )

注意每种计算模式的构建都有"存在性检查"——不仅检查请求对象是否存在，还检查关键字段是否非空。空的 connection_id 或 expression 会被视为未配置该计算模式。

Reducer 构建更复杂，涉及枚举解析和链路步骤转换：

Python

    # 构建 Reducer（检查 name 是否非空）
    reducer_def: ReducerDefinition | None = None
    if request.reducer and request.reducer.name:
        try:
            agg_func = AggregationFunction(request.reducer.aggregation.lower())
        except ValueError:
            agg_func = AggregationFunction.SUM

        reducer_def = ReducerDefinition(
            reducer_id=request.reducer.reducer_id or f"red-{request.property_name}",
            name=request.reducer.name,
            source_object_type=request.reducer.source_object_type,
            link_path=[
                LinkPathStep(
                    link_type_id=s.link_type_id,
                    target_object_type=s.target_object_type,
                )
                for s in request.reducer.link_path
            ],
            target_property=request.reducer.target_property,
            aggregation=agg_func,
            # ...
        )

#7. 属性计算：缓存策略与 WorldContext

compute_derived_property 方法展示了计算请求的处理流程：

Python

async def compute_derived_property(
    self, request: ComputeDerivedPropertyRequest,
) -> ComputeDerivedPropertyResponse:
    world_context: WorldContext | None = None
    if request.world_id:
        world_context = WorldContext(world_id=request.world_id)

    values, metrics = await self._engine.compute(
        property_id=request.property_id,
        object_ids=request.object_ids,
        world_context=world_context,
        force_recompute=request.force_recompute,
        skip_cache=request.skip_cache,
    )

ComputeOptions 提供了三个缓存控制开关：

force_recompute：忽略缓存，强制重新计算
skip_cache：计算结果不写入缓存
include_metadata：返回计算元数据

响应中的 PropertyValue 包含了值的来源信息：

PROTOBUF

enum PropertyOrigin {
  PROPERTY_ORIGIN_COMPUTED = 1;       // 新鲜计算
  PROPERTY_ORIGIN_CACHED = 2;         // 来自缓存
  PROPERTY_ORIGIN_MATERIALIZED = 3;   // 来自存储
}

结合 ttl_remaining_seconds 字段，客户端可以判断缓存值的新鲜度。

#8. Reducer 聚合：多跳链路遍历

FEAT-012 引入的 Reducer 是四种计算模式中最复杂的一种：

PROTOBUF

message ReducerDefinition {
  string source_object_type = 3;           // 源对象类型
  repeated LinkPathStep link_path = 4;     // 链路遍历路径（最多 3 跳）
  string target_property = 5;              // 目标属性
  AggregationFunction aggregation = 6;     // 聚合函数
  string filter_expression = 7;            // 可选过滤条件
  ReducerExecutionStrategy strategy = 8;   // 执行策略
}

多跳链路遍历通过 LinkPathStep 实现：

PROTOBUF

message LinkPathStep {
  string link_type_id = 1;          // 关系类型 ID
  string target_object_type = 2;    // 目标对象类型
}

例如，计算"部门的所有员工的平均薪资"，链路路径为：Department --[has_employee]--> Employee，目标属性为 salary，聚合函数为 AVG。最多支持 3 跳的限制防止了图遍历的爆炸性增长。

九种聚合函数覆盖了常见的统计需求：

PROTOBUF

enum AggregationFunction {
  AGGREGATION_FUNCTION_SUM = 1;
  AGGREGATION_FUNCTION_COUNT = 2;
  AGGREGATION_FUNCTION_AVG = 3;
  AGGREGATION_FUNCTION_MIN = 4;
  AGGREGATION_FUNCTION_MAX = 5;
  AGGREGATION_FUNCTION_COLLECT = 6;           // 收集为列表
  AGGREGATION_FUNCTION_FIRST = 7;
  AGGREGATION_FUNCTION_LAST = 8;
  AGGREGATION_FUNCTION_COUNT_DISTINCT = 9;
}

COLLECT 和 FIRST/LAST 是非标准聚合——COLLECT 将所有值收集为 JSON 数组，FIRST/LAST 需要配合 order_by_property 字段指定排序依据。

#9. 缓存管理：失效与统计

缓存管理提供了两个核心操作：

Python

async def invalidate_cache(
    self, request: InvalidateCacheRequest,
) -> InvalidateCacheResponse:
    count = await self._engine.invalidate_cache(
        property_id=request.property_id or None,
        object_ids=request.object_ids if request.object_ids else None,
    )

当 property_id 和 object_ids 都为空时，清除所有缓存。当只提供 property_id 时，清除该属性的所有对象缓存。两者都提供时，精确清除特定对象的特定属性缓存。

缓存统计通过 GetCacheStats 返回：

Python

class GetCacheStatsResponse(BaseModel):
    total_entries: int = 0
    hits: int = 0
    misses: int = 0
    hit_rate: float = 0.0

Proto 层面提供了更细粒度的按属性统计：

PROTOBUF

message CacheStatsResponse {
  int64 total_entries = 1;
  int64 memory_used_bytes = 2;
  double hit_rate = 3;
  int64 evictions = 4;
  map<string, PropertyCacheStats> by_property = 5;
}

evictions 字段报告被驱逐的缓存条目数——如果此值持续增长，说明缓存容量不足需要扩容。

#10. BranchBindingPolicy：分支上的版本策略

当 Nessie 分支创建时，派生属性的计算逻辑版本如何处理？

PROTOBUF

enum BranchBindingPolicy {
  BRANCH_BINDING_POLICY_PIN_AT_CREATION = 1;  // 默认：冻结计算版本
  BRANCH_BINDING_POLICY_FOLLOW_LATEST = 2;    // 开发模式：始终使用最新
}

PIN_AT_CREATION（默认）：分支创建时继承并冻结当时的 compute_version。这对 A/B 测试至关重要——两个分支使用不同的数据但相同的计算逻辑，确保比较的公平性。

FOLLOW_LATEST：分支始终使用最新的计算逻辑版本。适合开发/沙箱环境，开发者希望看到最新函数变更的效果。

#11. Key Takeaways

四种计算模式覆盖了从简单表达式到复杂函数的完整计算需求谱系，优先级规则允许渐进式迁移。
三种存储策略平衡了实时性和性能，VIRTUAL 用于低频高精度、CACHED 用于高频中等延迟、MATERIALIZED 用于大规模批量。
三轴版本模型（数据 x Schema x 计算逻辑）确保了计算的确定性和可重现性。
Reducer 的多跳链路最多 3 跳的限制在功能性和安全性之间取得平衡。
BranchBindingPolicy 在 A/B 测试和开发体验之间提供了灵活选择。
缓存可观测性通过 hit_rate 和 evictions 指标帮助运维团队优化缓存配置。

#下一篇

S9-16：PipelineService — DSL 到 DAG 的编译，我们将深入 Pipeline & Orchestration Layer 的管道引擎，了解声明式 Pipeline DSL 如何编译为可执行的 DAG。

Tags: #coomia-dip #source-code-reading #derived-property #dependency-dag #cascade #grpc #Layer-d