决策追踪链：从输入到执行的端到端可追溯
企业级决策系统必须回答一个核心问题："这个决策是怎么做出的？" coomia-dip 构建了从原始数据输入到最终执行结果的 端到端决策追踪链（Decision Trace Chain），基于 OpenTelemetry 分布式追踪标准，将 Sense、Think、Decide、Act 四个阶段的所有中间状态串联为一条完整的因果链路。本文深入解析追踪链的数据模型、Span 设计、存储方案和查询 API。
Coomia发布于 2025年9月2日13 分钟阅读
分享本文Twitter / X
“系列：S5 智能决策 · 第 11 篇 | 难度：高级 | 阅读时间：20 分钟
决策追踪链：从输入到执行的端到端可追溯
#TL;DR
企业级决策系统必须回答一个核心问题："这个决策是怎么做出的？" coomia-dip 构建了从原始数据输入到最终执行结果的 端到端决策追踪链（Decision Trace Chain），基于 OpenTelemetry 分布式追踪标准，将 Sense、Think、Decide、Act 四个阶段的所有中间状态串联为一条完整的因果链路。本文深入解析追踪链的数据模型、Span 设计、存储方案和查询 API。
#1. 为什么需要端到端追踪
#1.1 决策审计的法规要求
Code
法规合规要求：
GDPR (欧盟) 金融监管 医疗合规
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ 数据主体有权 │ │ 模型可解释性 │ │ 临床决策必须 │
│ 了解自动化 │ │ 要求 │ │ 完整记录 │
│ 决策的逻辑 │ │ │ │ │
└─────────────┘ └─────────────┘ └─────────────┘
│ │ │
└─────────┬───────┘─────────────────┘
│
▼
┌─────────────────┐
│ 需要完整的 │
│ 决策追踪链 │
└─────────────────┘
#1.2 追踪链覆盖范围
Code
Decision Trace Chain 全链路：
数据输入 ──→ 特征计算 ──→ 规则匹配 ──→ ML推理 ──→ 融合决策
│ │ │ │ │
▼ ▼ ▼ ▼ ▼
[Span] [Span] [Span] [Span] [Span]
│ │ │ │ │
└────────────┴────────────┴───────────┴──────────┘
│
▼
──→ 审批流程 ──→ 执行动作 ──→ 结果反馈
│ │ │
▼ ▼ ▼
[Span] [Span] [Span]
│ │ │
└───────────┴───────────┘
│
▼
TraceID: 统一串联
#2. 追踪数据模型
#2.1 核心模型
Python
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any
import uuid
class TracePhase(Enum):
"""追踪阶段"""
SENSE = "sense"
THINK = "think"
DECIDE = "decide"
ACT = "act"
APPROVE = "approve"
class SpanKind(Enum):
"""Span 类型"""
DATA_INPUT = "data_input"
FEATURE_COMPUTE = "feature_compute"
RULE_EVALUATION = "rule_evaluation"
ML_INFERENCE = "ml_inference"
FUSION = "fusion"
DECISION = "decision"
APPROVAL = "approval"
ACTION_EXECUTION = "action_execution"
NOTIFICATION = "notification"
@dataclass
class DecisionTrace:
"""决策追踪"""
trace_id: str
decision_id: str
domain: str
initiator: str
start_time: datetime
end_time: datetime | None = None
status: str = "in_progress" # in_progress, completed, failed
spans: list[TraceSpan] = field(default_factory=list)
metadata: dict[str, Any] = field(default_factory=dict)
@staticmethod
def new(domain: str, initiator: str) -> DecisionTrace:
trace_id = str(uuid.uuid4()).replace("-", "")[:32]
return DecisionTrace(
trace_id=trace_id,
decision_id=f"dec-{trace_id[:12]}",
domain=domain,
initiator=initiator,
start_time=datetime.utcnow(),
)
@dataclass
class TraceSpan:
"""追踪 Span"""
span_id: str
parent_span_id: str | None
trace_id: str
phase: TracePhase
kind: SpanKind
name: str
start_time: datetime
end_time: datetime | None = None
status: str = "ok" # ok, error
attributes: dict[str, Any] = field(default_factory=dict)
events: list[SpanEvent] = field(default_factory=list)
links: list[SpanLink] = field(default_factory=list)
@staticmethod
def new(trace_id: str, phase: TracePhase, kind: SpanKind,
name: str, parent_id: str | None = None) -> TraceSpan:
return TraceSpan(
span_id=str(uuid.uuid4()).replace("-", "")[:16],
parent_span_id=parent_id,
trace_id=trace_id,
phase=phase,
kind=kind,
name=name,
start_time=datetime.utcnow(),
)
@dataclass
class SpanEvent:
"""Span 事件"""
name: str
timestamp: datetime
attributes: dict[str, Any] = field(default_factory=dict)
@dataclass
class SpanLink:
"""Span 关联"""
linked_trace_id: str
linked_span_id: str
relationship: str = "caused_by"
#2.2 追踪上下文传播
Python
from contextvars import ContextVar
_current_trace: ContextVar[DecisionTrace | None] = ContextVar(
"current_trace", default=None
)
_current_span: ContextVar[TraceSpan | None] = ContextVar(
"current_span", default=None
)
class TraceContext:
"""追踪上下文管理"""
@staticmethod
def start_trace(domain: str, initiator: str) -> DecisionTrace:
trace = DecisionTrace.new(domain, initiator)
_current_trace.set(trace)
return trace
@staticmethod
def current_trace() -> DecisionTrace | None:
return _current_trace.get()
@staticmethod
def start_span(phase: TracePhase, kind: SpanKind,
name: str) -> TraceSpan:
trace = _current_trace.get()
if trace is None:
raise RuntimeError("No active trace")
parent = _current_span.get()
span = TraceSpan.new(
trace_id=trace.trace_id,
phase=phase,
kind=kind,
name=name,
parent_id=parent.span_id if parent else None,
)
trace.spans.append(span)
_current_span.set(span)
return span
@staticmethod
def end_span(span: TraceSpan, status: str = "ok") -> None:
span.end_time = datetime.utcnow()
span.status = status
# 恢复父 Span
trace = _current_trace.get()
if trace and span.parent_span_id:
parent = next(
(s for s in trace.spans if s.span_id == span.parent_span_id),
None
)
_current_span.set(parent)
else:
_current_span.set(None)
@staticmethod
def add_event(name: str, **attributes) -> None:
span = _current_span.get()
if span:
span.events.append(SpanEvent(
name=name,
timestamp=datetime.utcnow(),
attributes=attributes,
))
@staticmethod
def set_attribute(key: str, value: Any) -> None:
span = _current_span.get()
if span:
span.attributes[key] = value
#3. 各阶段 Span 设计
#3.1 Sense 阶段
Python
class SenseTracer:
"""感知阶段追踪"""
@staticmethod
def trace_data_input(source: str, record_count: int,
schema: dict) -> TraceSpan:
span = TraceContext.start_span(
TracePhase.SENSE, SpanKind.DATA_INPUT,
f"data_input.{source}"
)
TraceContext.set_attribute("source.type", source)
TraceContext.set_attribute("source.record_count", record_count)
TraceContext.set_attribute("source.schema", str(schema))
return span
@staticmethod
def trace_feature_compute(features: dict[str, Any]) -> TraceSpan:
span = TraceContext.start_span(
TracePhase.SENSE, SpanKind.FEATURE_COMPUTE,
"feature_computation"
)
TraceContext.set_attribute("features.count", len(features))
for name, value in features.items():
TraceContext.set_attribute(f"feature.{name}", str(value))
return span
#3.2 Think 阶段
Python
class ThinkTracer:
"""推理阶段追踪"""
@staticmethod
def trace_rule_evaluation(rule_set: str,
rules_count: int) -> TraceSpan:
span = TraceContext.start_span(
TracePhase.THINK, SpanKind.RULE_EVALUATION,
f"rule_eval.{rule_set}"
)
TraceContext.set_attribute("rules.set", rule_set)
TraceContext.set_attribute("rules.total_count", rules_count)
return span
@staticmethod
def trace_rule_match(rule_id: str, matched: bool,
conditions: list[dict]) -> None:
TraceContext.add_event(
"rule_match",
rule_id=rule_id,
matched=matched,
conditions=str(conditions),
)
@staticmethod
def trace_ml_inference(model_name: str, model_version: str,
prediction: str,
confidence: float) -> TraceSpan:
span = TraceContext.start_span(
TracePhase.THINK, SpanKind.ML_INFERENCE,
f"ml_inference.{model_name}"
)
TraceContext.set_attribute("model.name", model_name)
TraceContext.set_attribute("model.version", model_version)
TraceContext.set_attribute("model.prediction", prediction)
TraceContext.set_attribute("model.confidence", confidence)
return span
#3.3 Decide 阶段
Python
class DecideTracer:
"""决策阶段追踪"""
@staticmethod
def trace_fusion(method: str, rule_result: str,
ml_result: str, final: str,
confidence: float) -> TraceSpan:
span = TraceContext.start_span(
TracePhase.DECIDE, SpanKind.FUSION,
f"fusion.{method}"
)
TraceContext.set_attribute("fusion.method", method)
TraceContext.set_attribute("fusion.rule_result", rule_result)
TraceContext.set_attribute("fusion.ml_result", ml_result)
TraceContext.set_attribute("fusion.final_decision", final)
TraceContext.set_attribute("fusion.confidence", confidence)
return span
@staticmethod
def trace_decision(decision: str, confidence: float,
needs_approval: bool) -> TraceSpan:
span = TraceContext.start_span(
TracePhase.DECIDE, SpanKind.DECISION,
"decision_output"
)
TraceContext.set_attribute("decision.result", decision)
TraceContext.set_attribute("decision.confidence", confidence)
TraceContext.set_attribute("decision.needs_approval", needs_approval)
return span
#3.4 Act 阶段
Python
class ActTracer:
"""执行阶段追踪"""
@staticmethod
def trace_action_execution(action_type: str,
target: str) -> TraceSpan:
span = TraceContext.start_span(
TracePhase.ACT, SpanKind.ACTION_EXECUTION,
f"action.{action_type}"
)
TraceContext.set_attribute("action.type", action_type)
TraceContext.set_attribute("action.target", target)
return span
@staticmethod
def trace_notification(channel: str, recipient: str,
template: str) -> TraceSpan:
span = TraceContext.start_span(
TracePhase.ACT, SpanKind.NOTIFICATION,
f"notify.{channel}"
)
TraceContext.set_attribute("notification.channel", channel)
TraceContext.set_attribute("notification.recipient", recipient)
TraceContext.set_attribute("notification.template", template)
return span
#4. 追踪存储
#4.1 存储架构
Code
追踪数据存储分层：
实时层 (< 24h) 近线层 (1-30天) 离线层 (> 30天)
┌───────────┐ ┌───────────┐ ┌───────────┐
│ PostgreSQL│ │ Iceberg │ │ Iceberg │
│ (JSONB) │ │ (Parquet) │ │ (Archive) │
└─────┬─────┘ └─────┬─────┘ └─────┬─────┘
│ │ │
▼ ▼ ▼
索引: trace_id 分区: day/domain 分区: month
索引: decision_id 压缩: Snappy 保留: 7年
索引: timestamp TTL: 30天 冷存储
#4.2 存储实现
Python
import json
from datetime import datetime
class TraceStore:
"""追踪数据存储"""
def __init__(self, db_pool, iceberg_catalog):
self._db = db_pool
self._iceberg = iceberg_catalog
async def save_trace(self, trace: DecisionTrace) -> None:
"""保存完整追踪"""
async with self._db.acquire() as conn:
await conn.execute(
"""
INSERT INTO decision_traces
(trace_id, decision_id, domain, initiator,
start_time, end_time, status, spans, metadata)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
ON CONFLICT (trace_id) DO UPDATE SET
end_time = $6, status = $7, spans = $8
""",
trace.trace_id,
trace.decision_id,
trace.domain,
trace.initiator,
trace.start_time,
trace.end_time,
trace.status,
json.dumps([self._span_to_dict(s) for s in trace.spans]),
json.dumps(trace.metadata),
)
async def get_trace(self, trace_id: str) -> DecisionTrace | None:
"""按 trace_id 查询"""
async with self._db.acquire() as conn:
row = await conn.fetchrow(
"SELECT * FROM decision_traces WHERE trace_id = $1",
trace_id,
)
if row:
return self._row_to_trace(row)
return None
async def query_traces(self, domain: str | None = None,
start: datetime | None = None,
end: datetime | None = None,
status: str | None = None,
limit: int = 100) -> list[DecisionTrace]:
"""条件查询追踪"""
conditions = []
params = []
idx = 1
if domain:
conditions.append(f"domain = ${idx}")
params.append(domain)
idx += 1
if start:
conditions.append(f"start_time >= ${idx}")
params.append(start)
idx += 1
if end:
conditions.append(f"start_time <= ${idx}")
params.append(end)
idx += 1
if status:
conditions.append(f"status = ${idx}")
params.append(status)
idx += 1
where = " AND ".join(conditions) if conditions else "TRUE"
query = f"""
SELECT * FROM decision_traces
WHERE {where}
ORDER BY start_time DESC
LIMIT {limit}
"""
async with self._db.acquire() as conn:
rows = await conn.fetch(query, *params)
return [self._row_to_trace(r) for r in rows]
def _span_to_dict(self, span: TraceSpan) -> dict:
return {
"span_id": span.span_id,
"parent_span_id": span.parent_span_id,
"phase": span.phase.value,
"kind": span.kind.value,
"name": span.name,
"start_time": span.start_time.isoformat(),
"end_time": span.end_time.isoformat() if span.end_time else None,
"status": span.status,
"attributes": span.attributes,
"events": [
{
"name": e.name,
"timestamp": e.timestamp.isoformat(),
"attributes": e.attributes,
}
for e in span.events
],
}
def _row_to_trace(self, row) -> DecisionTrace:
spans_data = json.loads(row["spans"])
return DecisionTrace(
trace_id=row["trace_id"],
decision_id=row["decision_id"],
domain=row["domain"],
initiator=row["initiator"],
start_time=row["start_time"],
end_time=row["end_time"],
status=row["status"],
spans=[self._dict_to_span(s, row["trace_id"]) for s in spans_data],
metadata=json.loads(row["metadata"]),
)
def _dict_to_span(self, data: dict, trace_id: str) -> TraceSpan:
return TraceSpan(
span_id=data["span_id"],
parent_span_id=data.get("parent_span_id"),
trace_id=trace_id,
phase=TracePhase(data["phase"]),
kind=SpanKind(data["kind"]),
name=data["name"],
start_time=datetime.fromisoformat(data["start_time"]),
end_time=(
datetime.fromisoformat(data["end_time"])
if data.get("end_time") else None
),
status=data.get("status", "ok"),
attributes=data.get("attributes", {}),
)
#5. gRPC 查询服务
#5.1 Protobuf 定义
PROTOBUF
syntax = "proto3";
package onto.trace.v1;
service TraceService {
rpc GetTrace(GetTraceRequest) returns (TraceResponse);
rpc QueryTraces(QueryRequest) returns (QueryResponse);
rpc GetTraceTimeline(TimelineRequest) returns (TimelineResponse);
rpc GetTraceGraph(GraphRequest) returns (GraphResponse);
}
message GetTraceRequest {
string trace_id = 1;
}
message TraceResponse {
string trace_id = 1;
string decision_id = 2;
string domain = 3;
string status = 4;
repeated SpanData spans = 5;
string timeline_ascii = 6;
}
message SpanData {
string span_id = 1;
string parent_span_id = 2;
string phase = 3;
string kind = 4;
string name = 5;
int64 start_time_ms = 6;
int64 duration_ms = 7;
string status = 8;
map<string, string> attributes = 9;
}
#6. 追踪可视化
#6.1 时间线视图
Code
决策追踪时间线 (trace_id: a1b2c3d4e5f6)
═══════════════════════════════════════════════
时间(ms) 0 10 20 30 40 50 60 70
SENSE
├ data_input.api [████] 12ms
├ feature_compute [ ████] 15ms
│
THINK
├ rule_eval.credit [ ████] 8ms
├ ml_inference.v3 [ ██████████] 22ms
│
DECIDE
├ fusion.weighted [ ███] 6ms
├ decision_output [ ██] 4ms
│
APPROVE
├ approval.submit [ ██] 3ms
├ approval.wait [ ........] pending
│
ACT
└ (等待审批完成)
总耗时: 70ms (不含审批等待)
决策: conditional_approve (confidence: 0.72)
#6.2 因果图视图
Code
决策因果图：
[data_input.api] ──→ [feature_compute]
│
┌─────────┼─────────┐
▼ ▼ ▼
[rule_eval] [ml_inference]
│ │
└────┬────┘
▼
[fusion.weighted]
│
▼
[decision_output]
│
┌────┴────┐
▼ ▼
[approval] [notification]
│
▼
[action_exec]
#7. OpenTelemetry 集成
#7.1 导出器
Python
from opentelemetry import trace as otel_trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import (
OTLPSpanExporter,
)
class OTelTraceExporter:
"""将决策追踪导出到 OpenTelemetry"""
def __init__(self, endpoint: str = "localhost:4317"):
provider = TracerProvider()
exporter = OTLPSpanExporter(endpoint=endpoint)
provider.add_span_processor(BatchSpanProcessor(exporter))
otel_trace.set_tracer_provider(provider)
self._tracer = otel_trace.get_tracer("coomia-dip-decision")
def export_trace(self, trace: DecisionTrace) -> None:
"""将 DecisionTrace 转换并导出为 OTel Spans"""
span_map: dict[str, Any] = {}
# 按依赖顺序处理
root_spans = [s for s in trace.spans if s.parent_span_id is None]
child_spans = [s for s in trace.spans if s.parent_span_id is not None]
for span in root_spans:
otel_span = self._create_otel_span(span, None)
span_map[span.span_id] = otel_span
for span in child_spans:
parent_ctx = span_map.get(span.parent_span_id)
otel_span = self._create_otel_span(span, parent_ctx)
span_map[span.span_id] = otel_span
def _create_otel_span(self, span: TraceSpan,
parent_context) -> Any:
ctx = otel_trace.set_span_in_context(parent_context) if parent_context else None
with self._tracer.start_as_current_span(
span.name,
context=ctx,
attributes={
"decision.phase": span.phase.value,
"decision.kind": span.kind.value,
**{k: str(v) for k, v in span.attributes.items()},
},
) as otel_span:
for event in span.events:
otel_span.add_event(
event.name,
attributes={k: str(v) for k, v in event.attributes.items()},
)
return otel_span
#8. 追踪分析
#8.1 追踪统计
Python
class TraceAnalyzer:
"""追踪分析器"""
def analyze(self, trace: DecisionTrace) -> dict:
"""分析单条追踪"""
spans = trace.spans
total_duration = (
(trace.end_time - trace.start_time).total_seconds() * 1000
if trace.end_time else 0
)
phase_durations: dict[str, float] = {}
for span in spans:
if span.end_time:
duration = (span.end_time - span.start_time).total_seconds() * 1000
phase = span.phase.value
phase_durations[phase] = phase_durations.get(phase, 0) + duration
bottleneck = max(phase_durations.items(), key=lambda x: x[1]) if phase_durations else ("none", 0)
return {
"trace_id": trace.trace_id,
"total_duration_ms": total_duration,
"span_count": len(spans),
"phase_durations_ms": phase_durations,
"bottleneck_phase": bottleneck[0],
"bottleneck_ms": bottleneck[1],
"error_spans": [
s.name for s in spans if s.status == "error"
],
"decision": next(
(s.attributes.get("decision.result")
for s in spans if s.kind == SpanKind.DECISION),
"unknown"
),
}
def analyze_batch(self, traces: list[DecisionTrace]) -> dict:
"""批量追踪分析"""
analyses = [self.analyze(t) for t in traces]
durations = [a["total_duration_ms"] for a in analyses if a["total_duration_ms"] > 0]
if not durations:
return {"count": 0}
durations.sort()
return {
"count": len(analyses),
"avg_duration_ms": sum(durations) / len(durations),
"p50_duration_ms": durations[len(durations) // 2],
"p99_duration_ms": durations[int(len(durations) * 0.99)],
"error_rate": sum(1 for a in analyses if a["error_spans"]) / len(analyses),
"decision_distribution": self._count_decisions(analyses),
}
def _count_decisions(self, analyses: list[dict]) -> dict[str, int]:
dist: dict[str, int] = {}
for a in analyses:
d = a.get("decision", "unknown")
dist[d] = dist.get(d, 0) + 1
return dist
#9. 安全与隐私
#9.1 数据脱敏
Python
class TraceSanitizer:
"""追踪数据脱敏"""
SENSITIVE_KEYS = {
"ssn", "id_number", "phone", "email",
"credit_card", "password", "secret",
}
@classmethod
def sanitize(cls, trace: DecisionTrace) -> DecisionTrace:
"""脱敏追踪数据"""
for span in trace.spans:
sanitized_attrs = {}
for key, value in span.attributes.items():
if any(sk in key.lower() for sk in cls.SENSITIVE_KEYS):
sanitized_attrs[key] = cls._mask(str(value))
else:
sanitized_attrs[key] = value
span.attributes = sanitized_attrs
for event in span.events:
for key in list(event.attributes.keys()):
if any(sk in key.lower() for sk in cls.SENSITIVE_KEYS):
event.attributes[key] = cls._mask(
str(event.attributes[key])
)
return trace
@staticmethod
def _mask(value: str) -> str:
if len(value) <= 4:
return "****"
return value[:2] + "*" * (len(value) - 4) + value[-2:]
#10. 实战案例
Python
# 完整的决策追踪流程
# 1. 开始追踪
trace = TraceContext.start_trace(domain="credit", initiator="api-gateway")
# 2. Sense 阶段
span = SenseTracer.trace_data_input("api", record_count=1, schema={"credit_score": "int"})
TraceContext.end_span(span)
span = SenseTracer.trace_feature_compute({"credit_score": 620, "debt_ratio": 0.45})
TraceContext.end_span(span)
# 3. Think 阶段
span = ThinkTracer.trace_rule_evaluation("credit_rules", rules_count=5)
ThinkTracer.trace_rule_match("CR-001", matched=False, conditions=[{"credit_score >= 700": False}])
ThinkTracer.trace_rule_match("CR-003", matched=True, conditions=[{"credit_score >= 550": True}])
TraceContext.end_span(span)
span = ThinkTracer.trace_ml_inference("credit_v3", "3.2.1", "conditional_approve", 0.68)
TraceContext.end_span(span)
# 4. Decide 阶段
span = DecideTracer.trace_fusion("weighted", "conditional_approve", "conditional_approve",
"conditional_approve", 0.72)
TraceContext.end_span(span)
span = DecideTracer.trace_decision("conditional_approve", 0.72, needs_approval=True)
TraceContext.end_span(span)
# 5. 保存
trace_obj = TraceContext.current_trace()
trace_obj.end_time = datetime.utcnow()
trace_obj.status = "completed"
await trace_store.save_trace(trace_obj)
# 6. 分析
analyzer = TraceAnalyzer()
analysis = analyzer.analyze(trace_obj)
# {
# "total_duration_ms": 52.3,
# "span_count": 6,
# "bottleneck_phase": "think",
# "bottleneck_ms": 30.0,
# "decision": "conditional_approve",
# }
#Key Takeaways
- 端到端追踪链 串联 Sense-Think-Decide-Act 四阶段的所有中间状态
- Span 分层设计 按阶段和类型组织追踪数据，支持细粒度分析
- ContextVar 传播 实现零侵入的追踪上下文传递
- 三层存储 实时/近线/离线分层，平衡查询性能与存储成本
- OpenTelemetry 兼容 可导出到标准 OTel 生态（Jaeger/Zipkin）
- 数据脱敏 自动识别和遮蔽敏感字段，满足隐私合规
- 追踪分析 支持单条和批量分析，识别性能瓶颈和决策分布
#Next Article
下一篇 S5-12 Action Engine：10 种执行器的统一编排 将深入解析 coomia-dip ActionEngine 如何统一编排 gRPC、HTTP、消息队列等 10 种不同的执行器。
tags: #decision-trace #opentelemetry #distributed-tracing #audit #compliance #span #coomia-dip