2024 年 11 月 Mike Tyson 对阵 Jake Paul 时，Netflix 向约 6,500 万个并发流交付了这场比赛，同时每秒记录超过 3,800 万个事件。那么，当首页渲染速度变慢 1.5 秒时会发生什么？告警会触发，但在 Netflix 自己的示例中，修复大约需要四个小时，主要是在重新构建工具已经知道的事实，因为一个工具称其为 application，而另一个工具称其为 service。Netflix 在伦敦 QCon 上展示的方案是本体驱动的可观测性：建立在一套共享约定之上的 knowledge graph。我们使用公开数据重建了它，并对其进行测试：
我们对 Netflix 架构的重建包含七个部分：
telemetry lake：15.24 GB Parquet 中的 1,922,290,480 行真实数据。
identity 和 ontology：每台机器只有一个名称，并通过 shapes 拒绝错误事实。
provenance store：每次运行使用一个 named graph，并支持针对某个时刻提问。
enrichers 和 ladder：记录覆盖范围，并让模型遵守 closed vocabulary。
correlation：沿着 call graph 跟踪链路。
root-cause agents：带有 typed tools 的 specialist、evidence verifier 和 reconciler。
exam 和 replay：双读者标签、精确测试，以及在一台笔记本电脑上完成的 fleet sizing。
最终，call-graph chains 有 26.8% 的概率将故障组件排在前三名，而一台笔记本电脑可以持续处理每秒 2,610 万个事件。
这项重建独立完成，与 Netflix 没有任何关联。全部代码都在 GitHub repository 中：
https://github.com/FareedKhan-dev/ontology-driven-observability
我们编写了以下文件：
ontology-observability/
├── ontology/
│ ├── ops.ttl # classes, properties and ranges
│ └── shapes.ttl # the rules every fact must pass
└── kg/
├── config.py # settings, models and budgets
├── data.py # checked Parquet
├── identity.py # one name per machine
├── ontology.py # vocabulary and shape checks
├── store.py # named graphs and provenance
├── enrich.py # the enricher contract
├── llm.py # the model client
├── ladder.py # the closed-vocabulary rung
├── signals.py # components and topology
├── correlate.py # chains and rankings
├── agents.py # specialists and the verifier
├── exam.py # agreement and exact tests
└── replay.py # the stream replay
Incident Context Gap
2024 年 4 月 17 日 09:18 UTC，Wikimedia eqiad data centre 中 Kubernetes 上的 MediaWiki 开始发生故障。一项变更将 mcrouter proxy 移入了独立的 service，DNS 查询量从每秒约 40,000 次上升到 110,000 次。1.3 天后发布的事后分析将原因归咎于 hostname 缺少最后一个句点。
每个工具都掌握了这段故事的一部分，但使用的是各自的名称：
在告警中，一台 database machine 是 db1127.eqiad.wmnet:3306，在聊天中是“the s7 primary”，而在 deploy log 中是 db1127。在 89 条书面时间线中，Wikimedia incident 的中位发现时间为 2 分钟，修复时间为 40 分钟，但直到 62 分钟后才被理解：
Netflix 自己的 telemetry 不公开。我们使用了形态相同的公开记录：Wikimedia 的 on-call chat、change log、alerts 和 incident reports，也就是 Netflix 进行关联的四类来源。OpenRCA 和 RCAEval 提供带有 answer keys 的 metrics、logs 和 traces，Wikipedia pageviews 则提供了数据量，总计 1,922,290,480 行。
在查看任何数据之前，我先列出了 Netflix 设计中的十二项可测试主张，最后几节将逐一对其进行测试。
Environment Setup
你需要 Python 3.12 和约 25 GB 磁盘空间。完整运行使用一台拥有 16 个 CPU 核心和 32 GB 内存的笔记本电脑。我们使用 uv 创建项目，并将所有下载内容保存在项目内部：
uv init ontology-observability --python 3.12 && cd ontology-observability && mkdir -p kg ontology
uv add duckdb pandas pyarrow rdflib pyshacl pyoxigraph httpx huggingface-hub scipy psutil
export KG_CONTACT="you@example.org" # Wikimedia asks every script for a contact
export LLM_BASE_URL="https://your-endpoint/v1" # any OpenAI-compatible server
export LLM_API_KEY="your key"
每个模块都会读取 config.py：路径、DuckDB 的限制、每个角色使用的模型，以及每个阶段的 credit budget。
"""config.py: the settings every other file reads."""
import os
from pathlib import Path
PROFILE = os.environ.get("KG_PROFILE", "FULL") # SMOKE takes a thin slice of every source
DATA = Path(os.environ.get("KG_DATA", "data")) # about 25 GB once everything is in
CONTACT = os.environ.get("KG_CONTACT", "") # Wikimedia asks every script for a contact
THREADS, MEMORY, SEED = 16, "18GB", 42 # a 16-core laptop with 31 GiB of memory
BASE_URL = os.environ.get("LLM_BASE_URL", "http://127.0.0.1:8000/v1") # any OpenAI-compatible server
API_KEY = os.environ.get("LLM_API_KEY", "") # read once, never printed
MODELS = {"flash": "Qwen/Qwen3.8-Flash-Next", "max": "Qwen/Qwen3.8-Max", "reader": "Qwen/Qwen3.7-Max"}
PRICE = {"Qwen/Qwen3.8-Flash-Next": 0.2, "Qwen/Qwen3.8-Max": 0.5, "Qwen/Qwen3.7-Max": 0.5} # credits a call
BUDGET = {"label extraction": 200, "chat classification": 60, "AutoRCA development": 140, "Wikimedia test": 330,
"OpenRCA test": 250, "experiments": 185, "reserve": 35} # 1,200 in all
整个构建建立在以下工具之上：
以下是 import，以及每次计时所使用的机器信息：
import platform
import duckdb # SQL over Parquet, without loading it into memory
import pyoxigraph # the triple store, written in Rust
import pyshacl # SHACL shapes, checked in Python
import rdflib # RDF graphs and Turtle files
import psutil # cores and memory of this machine
print(f" python {platform.python_version()}, duckdb {duckdb.__version__}, rdflib {rdflib.__version__}")
print(f" pyshacl {pyshacl.__version__}, pyoxigraph {pyoxigraph.__version__}, {psutil.cpu_count()} cores")
#### OUTPUT ####
python 3.12.14, duckdb 2.0.0, rdflib 7.6.0
pyshacl 0.40.2, pyoxigraph 0.5.13, 16 cores
Telemetry Lake
一次合规的数据采集为每个人都使用了 pseudonym，带回了 1,730,378 条聊天记录、538,717 条变更记录和 414 份 incident reports：
OpenRCA 的镜像包含 162 个 CSV 文件，共 68.65 GB；我们逐个将其转换：
data.py 打开一个使用全部核心并将临时数据溢写到磁盘的 DuckDB connection：
"""data.py: one DuckDB connection, and text turned into checked Parquet."""
from pathlib import Path
import duckdb
from kg.config import DATA, MEMORY, THREADS
def duck() -> duckdb.DuckDBPyConnection:
# every core, a memory cap, and a spill folder on disk
return duckdb.connect(config={"threads": THREADS, "memory_limit": MEMORY, "temp_directory": str(DATA / "spill"),
"preserve_insertion_order": False, "TimeZone": "UTC"})
Span duration 可能以 microseconds 或 milliseconds 表示，因此我们在 select 中统一单位。只有在两侧都完成行数统计后，才会删除 CSV：
# kg/data.py
SPANS = { # one trace layout, three systems, three clocks
"bank": 'to_timestamp(TRY_CAST("timestamp" AS BIGINT) / 1000) AS ts, cmdb_id AS entity, trace_id, parent_id, '
"TRY_CAST(duration AS DOUBLE) AS duration_ms",
"telecom": "to_timestamp(TRY_CAST(startTime AS BIGINT) / 1000) AS ts, cmdb_id AS entity, traceId AS trace_id, "
"pid AS parent_id, TRY_CAST(elapsedTime AS DOUBLE) AS duration_ms",
"market": 'to_timestamp(TRY_CAST("timestamp" AS BIGINT) / 1000) AS ts, cmdb_id AS entity, trace_id, '
"parent_span AS parent_id, TRY_CAST(duration AS DOUBLE) / 1000 AS duration_ms", # microseconds
}
def to_parquet(csv: Path, select: str, out: Path, con) -> int:
"""CSV in, zstd Parquet out, rows counted, then the CSV deleted."""
tmp = out.with_suffix(".tmp")
out.parent.mkdir(parents=True, exist_ok=True)
con.execute(f"CREATE OR REPLACE TEMP VIEW src AS SELECT * FROM read_csv('{csv.as_posix()}', all_varchar = true)")
rows = con.execute(f"COPY (SELECT {select} FROM src) TO '{tmp.as_posix()}' "
"(FORMAT parquet, COMPRESSION zstd, ROW_GROUP_SIZE 1000000)").fetchone()[0]
assert rows == con.execute("SELECT count(*) FROM src").fetchone()[0], f"rows lost in {csv.name}"
tmp.replace(out) # only a checked file gets its real name
csv.unlink() # so 68 GB of CSV never sits on disk at once
return rows
我们从运行示例所在的日期开始：
from pathlib import Path
from huggingface_hub import hf_hub_download
from kg.data import SPANS, duck, to_parquet
path = "Market/cloudbed-1/telemetry/2022_03_21/trace/trace_span.csv" # one of the 162 files
csv = Path(hf_hub_download("tracer-cloud/opensre", path, repo_type="dataset", local_dir="data/raw/openrca"))
out = Path("data/parquet/openrca/market_cb1/spans/2022_03_21__trace_span.parquet")
print(f" {to_parquet(csv, SPANS['market'], out, duck()):,} spans, CSV deleted: {not csv.exists()}")
#### OUTPUT ####
9,027,465 spans, CSV deleted: True
在全部 162 个文件中，68.65 GB 数据缩减为 7.11 GB，且没有丢失任何行。Pageviews 和 RCAEval 补全了 telemetry lake：
#### OUTPUT ####
Wikimedia operations record 4,609,863 0.16 GB
Wikimedia pageviews, page x hour 1,142,762,478 4.64 GB
OpenRCA metrics, logs, spans 524,547,719 7.11 GB
RCAEval metrics, logs, spans 250,370,420 3.32 GB
total 1,922,290,480 15.24 GB
从这里开始，我们一直跟踪的故障是 07:45 UTC 的 cartservice，它在三类 signal 中都有体现：
DuckDB 处理全部 397 million spans 所需时间不到六秒：
Identity Resolver
在两个事实能够进行 join 之前，同一台机器必须拥有同一个名称。在 Wikimedia 中，host 编号的第一位表示其 data centre：
"""identity.py: one name for every machine, however a tool happens to write it."""
import re
DC_BY_DIGIT = {"1": "eqiad", "2": "codfw", "3": "esams", "4": "ulsfo", "5": "eqsin", "6": "drmrs", "7": "magru"}
HOSTNAME = re.compile(r"(?<![\w.-])([a-z][a-z-]*\d{4}(?:-dev)?)(?:\.(?:mgmt\.)?(?:eqiad|codfw|esams|ulsfo|eqsin|drmrs"
r"|magru)\.wmnet)?(?::\d+)?(?![\w-])")
def canonical_host(token: str) -> tuple[str, str] | None:
"""(name, datacentre) for any spelling of a production host, or None for anything else."""
m = HOSTNAME.search((token or "").strip().lower())
if not m:
return None # 'T316186' is a task, 'php8.3' a version
dc = DC_BY_DIGIT.get(re.search(r"(\d)\d{3}(?:-dev)?$", m.group(1)).group(1)) # first digit = datacentre
return (m.group(1), dc) if dc else None
DB1127.eqiad.wmnet:3306 和 elastic1124.mgmt.eqiad.wmnet 会解析为 eqiad 中的 hosts，而 task id T316186 无法解析为任何内容。随后，catalogue 为每台 host 提供 role、team 和 cluster，而时间是薄弱环节：
在五年的 logs 中共有 5,593 台 hosts，但目前只有 44% 能与 catalogue 匹配。catalogue 知道 2026 年 hosts 的 81%，却只知道 2021 年 hosts 的 18%，因为机器会退役。
Ontology as Code
我们存储的每个事实都是一个 triple，例如“这次 deploy 发生在 db1127 上”，而由一个文件决定哪些 triples 可以存在。Netflix 将该词汇拆分为十二个 namespaces，我们又增加了两个：
任何事物都不会只是一个 ops:Component，每种 signal type 都位于 corr:Signal 之下。词汇表是 git 中的 Turtle 文件 ontology/ops.ttl，例如规定 ops:Host 是 ops:Component 的一种。ontology.py 与 shapes file 一起加载它：
"""ontology.py: the vocabulary and its rules, kept in files and tested like code."""
import hashlib
import re
from pathlib import Path
import pandas as pd
import rdflib
from pyshacl import validate
from rdflib import RDF, Namespace, URIRef
from rdflib.collection import Collection
from rdflib.namespace import SH
FILES = Path(__file__).resolve().parent.parent / "ontology" # ops.ttl and shapes.ttl
BASE = "https://kg.example.org/"
NS = {"ops": "ops#", "slack": "slack#", "alert": "alert#", "metric": "metric#", "log": "log#", "tl": "timeline#",
"incident": "incident#", "rca": "rca#", "exp": "experiment#", "e2e": "e2e#", "corr": "correlation#",
"harvest": "harvest#", "code": "code#", "users": "users#"} # twelve from Netflix's post, two ours
OPS, CORR, TL, ALERT, HARVEST, SLACK = (Namespace(BASE + NS[p]) for p in ("ops", "corr", "tl", "alert", "harvest", "slack"))
PREFIXES = "".join(f"PREFIX {p}: <{BASE}{s}>\n" for p, s in NS.items()) + \
"PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>\nPREFIX xsd: <http://www.w3.org/2001/XMLSchema#>\n"
TBOX = rdflib.Graph().parse(FILES / "ops.ttl") # classes, properties, ranges
SHAPES = rdflib.Graph().parse(FILES / "shapes.ttl") # what data must pass to be written
一个函数负责生成每个 URI，另一个函数将某个时刻映射到对应的 hour node：
# kg/ontology.py
def slug(text: str) -> str:
s = re.sub(r"[^a-z0-9._~-]+", "-", str(text).strip().lower()).strip("-")
return s or hashlib.sha256(str(text).encode()).hexdigest()[:12] # a name made only of symbols
def mint(tenant: str, kind: str, key: str) -> URIRef:
# one function names every thing, so the same thing always gets the same URI
return URIRef(f"{BASE}id/{slug(tenant)}/{slug(kind)}/{slug(key)}")
def hour_node(ts) -> URIRef:
# the hour as a node of its own turns a time window into a lookup, not a scan
return URIRef(f"{BASE}id/hour/{pd.Timestamp(ts).tz_convert('UTC'):%Y-%m-%dT%H}")
完整词汇表包含 272 个 triples、43 个 classes 和 77 个 properties；带有多余空格的 DB1127 与 db1127 会生成相同的 URI。
Shape Validation Layer
ontology 规定哪些内容可以存在，但无法阻止垃圾数据。SHACL shapes 可以做到：它们是事实写入前必须通过的 closed-world rules。一个数字就能说明原因：
以文本形式存储的 confidence "0.92" 会使 filter 两次失效。rdflib 返回全部三行，包括单词 "high"；Oxigraph 则一行都不返回，而且两者都不会报错。修复必须在写入时完成，写入 ontology/shapes.ttl，其中包含十六条规则。它的 chain shape 要求每个 corr:confidence 都是 0 到 1 之间的 xsd:decimal，并附带消息“confidence 是 0 到 1 之间的数字，不能是文本”。
我的第一个 validator 开启了 RDFS inference，结果它认定拼写错误的 host 必然是 component。当前 checker 会保留 ontology 的可见性，但不进行推断：
# kg/ontology.py
def check(delta: rdflib.Graph, context: rdflib.Graph | None = None) -> tuple[bool, list[tuple]]:
"""Judge the nodes a delta writes, with their stored context. Subclasses count, nothing is inferred."""
focus = sorted({s for s in delta.subjects() if isinstance(s, URIRef)})
ok, report, _ = validate(delta + (context or rdflib.Graph()), shacl_graph=SHAPES, ont_graph=TBOX,
inference="none", focus_nodes=focus or None)
rows = []
for r in report.subjects(RDF.type, SH.ValidationResult):
rule = str(report.value(r, SH.sourceConstraintComponent)).rsplit("#", 1)[-1].removesuffix("ConstraintComponent")
path = report.value(r, SH.resultPath)
rows.append((str(report.value(r, SH.focusNode)).rsplit("/", 1)[-1], path.n3(TBOX.namespace_manager)
if path is not None else "", rule, str(report.value(r, SH.resultMessage) or "")))
return ok, sorted(rows)
系统只检查新 delta 的 subjects，并使用 store 中已有的相关事实。我在一个小文件中植入了六个错误节点：
import rdflib
from kg.ontology import check
ok, rows = check(rdflib.Graph().parse("planted.ttl")) # six broken nodes, two good ones
for node, path, rule, why in rows[:4]:
print(f" {node:<7}{path:<17}{rule:<13}{why}")
#### OUTPUT ####
chain1 corr:confidence Datatype a confidence is a number between 0 and 1, never text
chain1 corr:confidence MaxInclusive a confidence is a number between 0 and 1, never text
chain1 corr:confidence MinInclusive a confidence is a number between 0 and 1, never text
cov1 harvest:status In a harvest is ok, partial or missing, nothing else
八项违规全部被捕获。
Provenance Store
一个事实来自哪里？何时被知道？每个事实都存在于写入它的 run 所对应的 named graph 中，并由 W3C PROV-O record 描述每个 run：
store 使用 Oxigraph，class tree 则存放在其自己的 graph 中：
"""store.py: Oxigraph quads, one named graph per run, every run described in PROV-O."""
import pandas as pd
import pyoxigraph as ox
import rdflib
from rdflib import RDF, XSD, Literal, Namespace, URIRef
from kg.ontology import BASE, HARVEST, OPS, PREFIXES, TBOX, mint, slug
PROV = Namespace("http://www.w3.org/ns/prov#")
META, TBOX_GRAPH = URIRef(f"{BASE}graph/meta"), URIRef(f"{BASE}graph/tbox") # runs described, classes
def dt_literal(ts) -> Literal:
return Literal(pd.Timestamp(ts).tz_convert("UTC").strftime("%Y-%m-%dT%H:%M:%SZ"), datatype=XSD.dateTime)
class GraphStore:
"""Facts live in the named graph of the run that wrote them."""
def __init__(self, path: str | None = None):
self.store = ox.Store(path) if path else ox.Store() # on disk for real, in memory for tests
self._load(TBOX, TBOX_GRAPH) # the class tree, so subClassOf* can be followed
def _load(self, g: rdflib.Graph, into: URIRef) -> None:
self.store.load(g.serialize(format="nt").encode(), format=ox.RdfFormat.N_TRIPLES, to_graph=ox.NamedNode(into))
再次写入同一个 run 会替换原内容，因此 rebuilds 不会产生重复数据：
# kg/store.py
def write(self, facts: rdflib.Graph, tenant: str, enricher: str, run: str, used: list[str], started) -> URIRef:
g = URIRef(f"{BASE}graph/{slug(tenant)}/{slug(enricher)}/{slug(run)}")
if self.store.contains_named_graph(ox.NamedNode(g)):
self.drop(g) # a rerun replaces, never duplicates
self._load(facts, g)
rid, meta = URIRef(g.replace("/graph/", "/id/run/")), rdflib.Graph()
for p, o in ((RDF.type, HARVEST.Run), (HARVEST.enricher, Literal(enricher)), (PROV.startedAtTime,
dt_literal(started)), (PROV.endedAtTime, dt_literal(pd.Timestamp.now(tz="UTC"))),
(HARVEST.triplesAdded, Literal(len(facts), datatype=XSD.integer)),
*((PROV.used, URIRef(u)) for u in used)): # the sources this run read
meta.add((rid, p, o))
meta.add((g, PROV.wasGeneratedBy, rid))
meta.add((g, OPS.tenant, mint(tenant, "organisation", tenant)))
self._load(meta, META)
return g
删除错误的 run 会同时删除其 facts 和 record。context 会收集 delta 周围已经存储的事实：
# kg/store.py
def drop(self, g: URIRef) -> None:
# its facts and its PROV-O record, nothing else
self.store.remove_graph(ox.NamedNode(g))
self.store.update(f"DELETE WHERE {{ GRAPH <{META}> {{ <{g}> <{PROV.wasGeneratedBy}> ?run . ?run ?p ?o }} }}")
self.store.update(f"DELETE WHERE {{ GRAPH <{META}> {{ <{g}> ?p ?o }} }}")
def ask(self, sparql: str, graphs: list | None = None) -> list:
# a caller may name the only graphs it is allowed to read
kw = {"default_graph": [ox.NamedNode(g) for g in graphs]} if graphs else {"use_default_graph_as_union": True}
return list(self.store.query(PREFIXES + f"PREFIX prov: <{PROV}>\n" + sparql, **kw))
def context(self, delta: rdflib.Graph) -> rdflib.Graph:
# what the store already says about a delta's nodes
nodes = " ".join(sorted({f"<{n}>" for s, _, o in delta for n in (s, o) if str(n).startswith(BASE + "id/")}))
q = f"CONSTRUCT {{ ?s ?p ?o . ?o a ?c }} WHERE {{ VALUES ?s {{ {nodes} }} ?s ?p ?o OPTIONAL {{ ?o a ?c }} }}"
nt = self.store.query(q, use_default_graph_as_union=True).serialize(format=ox.RdfFormat.N_TRIPLES) if nodes else b""
return rdflib.Graph().parse(data=nt, format="nt")
每个 signal 都记录其发生时间以及我们本应能够知道它的时间。09:33 的 agent 不能看到第二天才写入的 report：
# kg/store.py
AS_OF = """SELECT ?s WHERE {{ ?s a/rdfs:subClassOf* corr:Signal . ?s corr:at ?t . ?s corr:detectedAt ?d .
FILTER(?d <= {asof} && ?t >= {since}) }}"""
def visible(store: GraphStore, asof, since, graphs: list | None = None) -> set[str]:
# a fact written after `asof` does not exist yet
return {r["s"].value for r in store.ask(AS_OF.format(asof=dt_literal(asof).n3(), since=dt_literal(since).n3()), graphs)}
在完整运行中，4 月 incident 的 report 携带了一个 canary，即绝不能到达 agent 的事实：
#### OUTPUT ####
asked as of all graphs canary? agent graphs canary?
2024-04-17 09:33 UTC 105 no 105 no
2024-04-18 16:00 UTC 177 yes 175 no
rollback of a bad run: 2,067 quads before, 1,322 after
canary 只会在 report 存在后出现，并且永远不会出现在 agent 的 graphs 中：
通过 hour node，“前一个小时发生了什么变化？”所需时间从 85.55 ms 降至 0.50 ms。
Enricher Contract
事实通过 enrichers 到达。每个 enricher 都是读取一个来源的小程序：
每次运行都会记录其 coverage，因为“没有触发任何内容”和“没有人查看”是不同的答案：
"""enrich.py: the contract every enricher keeps, their order, and the checks on their work."""
import re
from collections import Counter
from dataclasses import dataclass
from graphlib import TopologicalSorter
import pandas as pd
import rdflib
from rdflib import RDF, Literal, URIRef
from kg.ontology import HARVEST, check, mint
from kg.store import GraphStore, dt_literal
@dataclass
class Coverage:
source: str
start: pd.Timestamp
end: pd.Timestamp
status: str # ok, partial or missing: 'nothing fired' is not 'nobody looked'
detail: str = ""
来源失效只会影响 coverage，不会影响 harvest：
# kg/enrich.py
class Enricher:
"""fetch() may fail. analyse() is pure and returns facts."""
name, requires, produces, source = "enricher", (), (), ""
def run(self, start, end) -> tuple[rdflib.Graph, Coverage]:
try:
raw, why = self.fetch(start, end), ""
except Exception as e: # a dead source costs coverage, not the harvest
raw, why = None, f"{type(e).__name__}: {e}"
facts = self.analyse(raw, start, end) if raw is not None else rdflib.Graph()
return facts, Coverage(self.source, start, end, "ok" if raw is not None else "missing", why)
每个 enricher 声明自己的输入和输出，标准库的 topological sort 会推导执行顺序：
# kg/enrich.py
def plan_order(enrichers: list, available: frozenset = frozenset()) -> list:
"""The run order, from what each enricher needs and makes."""
makers = {cls: e.name for e in enrichers for cls in e.produces}
for e in enrichers: # an unmet need stops the plan before any run
if missing := [c for c in e.requires if c not in makers and c not in available]:
raise ValueError(f"{e.name} requires {missing[0].rsplit('#', 1)[-1]}, which no enricher produces")
by_name = {e.name: e for e in sorted(enrichers, key=lambda e: e.name)} # sorted, so the plan never wobbles
deps = {n: sorted({makers[c] for c in e.requires if c in makers} - {n}) for n, e in by_name.items()}
return [by_name[n] for n in TopologicalSorter(deps).static_order()] # CycleError on a loop
即使以任意顺序列出，host enricher 仍会首先运行；如果某个 enricher 的输入没有任何生产者，它会在运行前被拒绝。pipeline 会使用已存储的 context 检查每个 delta，而被隔离的 facts 仍会留下 coverage record：
# kg/enrich.py
def run_pipeline(enrichers: list, start, end, store: GraphStore, tenant: str, run: str) -> list[Coverage]:
"""Each enricher in order. A delta that breaks a shape is quarantined."""
have = frozenset(URIRef(r["c"].value) for r in store.ask("SELECT DISTINCT ?c WHERE { ?s a ?c }"))
report = []
for e in plan_order(enrichers, have):
started, (facts, cov) = pd.Timestamp.now(tz="UTC"), e.run(start, end)
ok, rows = check(facts, store.context(facts)) if len(facts) else (True, [])
if not ok:
cov.status, cov.detail = "partial", f"{len(rows)} shape violations, first: {rows[0][2]} on {rows[0][1]}"
facts = rdflib.Graph()
node = mint("harvest", "coverage", f"{e.name}-{cov.source}-{start:%Y%m%dT%H%M}")
for p, o in ((RDF.type, HARVEST.Coverage), (HARVEST.source, Literal(cov.source)), (HARVEST.status,
Literal(cov.status)), (HARVEST.windowStart, dt_literal(start)), (HARVEST.windowEnd, dt_literal(end))):
facts.add((node, p, o)) # what was read, and how well, is a fact too
store.write(facts, tenant, e.name, run, [], started)
report.append(cov)
return report
Coverage 为一个问题提供三种答案：
# kg/enrich.py
def alert_answer(store: GraphStore, host: URIRef, t: pd.Timestamp, source: str) -> str:
# did anything alert on this host in the half hour before t? three answers, not two
a, b = dt_literal(t - pd.Timedelta(minutes=30)).n3(), dt_literal(t).n3()
if store.ask(f"SELECT ?f WHERE {{ ?f a alert:Firing . ?f corr:on <{host}> . ?f corr:at ?x . "
f"FILTER(?x >= {a} && ?x <= {b}) }} LIMIT 1"):
return "yes, an alert fired"
seen = store.ask(f'SELECT ?st WHERE {{ ?c harvest:source "{source}" . ?c harvest:status ?st . '
f"?c harvest:windowStart ?w0 . ?c harvest:windowEnd ?w1 . FILTER(?w0 <= {b} && ?w1 >= {b}) }}")
return "no, the source was read and nothing fired" if seen and seen[0]["st"].value == "ok" else \
"unknown, the alert source was not covered"
关闭 Loki 后，一个安静 host 的“no”变成了“unknown”。最后一道防线是 catalogue gate：
# kg/enrich.py
SERVICE_LIKE = re.compile(r"(?<![\w.-])([a-z][a-z0-9]+(?:-[a-z0-9]+)+)(?![\w-])")
def gate(texts, is_host, in_catalogue) -> Counter:
# service-looking tokens the catalogue refuses, with counts
seen = Counter(tok for t in texts for tok in SERVICE_LIKE.findall(t.lower()) if not is_host(tok))
return Counter({tok: n for tok, n in seen.items() if not in_catalogue(tok)})
在 Loki alerts 的 3,841 个此类字符串中，catalogue 知道其中 107 个，拒绝了 3,734 个：
没有这个 gate，每个字符串都会成为 graph 中没有 owner 的 application。
Extraction Ladder
Incident chat 是自由文本。Netflix 的文章使用五个层级，从成本最低的开始，并在第四层使用模型：
在 128 个 incidents 周围的 13,331 条人工聊天记录中，patterns、catalogue 和 identity 达到以下结果：
#### OUTPUT ####
after layer text typed lines linked
1 regular expressions 2.0% 8.8%
2 catalogue validation 2.7% 7.1%
不过，仍有 93% 的行无法关联，下一层需要模型。每次调用都经过同一个 client，包含 cache、ledger，且最多同时进行六次调用：
"""llm.py: one door for every model call, with a cache, a ledger and a budget checked before each call."""
import hashlib
import json
import re
import sqlite3
import threading
import time
from pathlib import Path
import httpx
from kg.config import API_KEY, BASE_URL, BUDGET, PRICE
Path("artifacts").mkdir(exist_ok=True)
CACHE = sqlite3.connect("artifacts/llm_cache.sqlite", check_same_thread=False)
CACHE.execute("CREATE TABLE IF NOT EXISTS calls (key TEXT PRIMARY KEY, reply TEXT, seconds REAL)")
LEDGER = Path("artifacts/llm_ledger.jsonl") # every paid call, so the budget survives restarts
HTTP = httpx.Client(base_url=BASE_URL, timeout=180, headers={"Authorization": f"Bearer {API_KEY}"})
GATE, LOCK, PHASE = threading.BoundedSemaphore(6), threading.Lock(), ["experiments"]
def spent(phase: str) -> float:
rows = [json.loads(x) for x in LEDGER.read_text().splitlines()] if LEDGER.exists() else []
return sum(r["credits"] for r in rows if r["phase"] == phase)
调用发送前会检查 budget；缓存中的问题免费，并且会重复返回相同答案：
# kg/llm.py
def chat(model: str, messages: list, json_mode: bool = True, tools: list | None = None, max_tokens: int = 800) -> dict:
body = {"model": model, "messages": messages, "max_tokens": max_tokens, "temperature": 0.0}
body |= ({"response_format": {"type": "json_object"}} if json_mode else {}) | ({"tools": tools} if tools else {})
key = hashlib.sha256(json.dumps(body, sort_keys=True).encode()).hexdigest() # same question, same key
if hit := CACHE.execute("SELECT reply, seconds FROM calls WHERE key = ?", (key,)).fetchone():
return {**json.loads(hit[0]), "seconds": hit[1], "cached": True} # free, and identical
if spent(PHASE[0]) + PRICE[model] > BUDGET[PHASE[0]]: # refused before it is made
raise RuntimeError(f"'{PHASE[0]}' would pass its cap of {BUDGET[PHASE[0]]} credits")
t0 = time.perf_counter()
with GATE: # at most six calls in flight
for attempt in range(1, 7):
r = HTTP.post("/chat/completions", json=body)
if r.status_code != 429 and r.status_code < 500:
break
time.sleep(min(30, 1.5 * 2 ** attempt))
r.raise_for_status()
msg, seconds = r.json()["choices"][0]["message"], time.perf_counter() - t0
reply = {"text": msg.get("content") or "", "tool_calls": msg.get("tool_calls")}
with LOCK: # every answered call is billed, so all are logged
LEDGER.open("a").write(json.dumps({"phase": PHASE[0], "model": model, "credits": PRICE[model]}) + "\n")
CACHE.execute("INSERT OR REPLACE INTO calls VALUES (?, ?, ?)", (key, json.dumps(reply), seconds))
CACHE.commit()
return {**reply, "seconds": seconds, "cached": False}
一个 helper 会从 code fence 中提取 JSON：
# kg/llm.py
def parse_json(text: str):
"""The first JSON object in a reply, even inside a code fence."""
m = re.search(r"\{.*\}", text or "", flags=re.S)
try:
return json.loads(m.group(0)) if m else None
except json.JSONDecodeError:
return None
模型不得自行发明 categories。它的词汇表从 shapes file 中读取：
# kg/ontology.py
def closed_list(shape: str, path: URIRef) -> list[str]:
# the sh:in list of one property
prop = next(p for p in SHAPES.objects(URIRef(f"{BASE}shapes#{shape}"), SH.property) if SHAPES.value(p, SH.path) == path)
return [str(v) for v in Collection(SHAPES, SHAPES.value(prop, SH["in"]))]
JSON mode 只能保证回复是有效 JSON，并不能保证其字段正确，因此回复会在代码中检查：
"""ladder.py: the model's rung of the extraction ladder, held to the shapes' closed lists."""
from kg.ontology import SLACK, closed_list
CATEGORIES = closed_list("MessageShape", SLACK.category) # the vocabulary comes from shapes.ttl
SENTIMENTS = closed_list("MessageShape", SLACK.sentiment)
L4_SYSTEM = ("You label on-call chat conversations. Answer with JSON only, with the keys category, sentiment, "
f"incident_related and mentions. category is one of {', '.join(CATEGORIES)}. sentiment is one of "
f"{', '.join(SENTIMENTS)}. incident_related is true or false. mentions lists candidate ids only.")
def l4_valid(data, candidates: list[str]) -> bool:
# checked in code before anything is written
return (isinstance(data, dict) and data.get("category") in CATEGORIES and data.get("sentiment") in SENTIMENTS
and isinstance(data.get("incident_related"), bool) and isinstance(data.get("mentions"), list)
and all(m in candidates for m in data["mentions"]))
这里，模型读取 4 月 incident 周围的十段对话：
from kg.config import MODELS
from kg.ladder import L4_SYSTEM, l4_valid
from kg.llm import PHASE, chat, parse_json
PHASE[0] = "chat classification"
for k, (text, ids) in enumerate(conversations): # scrubbed text, and the ids rungs 1 to 3 found
r = parse_json(chat(MODELS["flash"], [{"role": "system", "content": L4_SYSTEM},
{"role": "user", "content": f"candidates: {ids}\n\n{text}"}])["text"])
if l4_valid(r, ids) and r["incident_related"]:
print(f" {k:<3}{r['category']:<11}{r['sentiment']:<9}{r['mentions']}")
#### OUTPUT ####
1 incident urgent ['service:mw-web']
6 incident urgent ['host:ncredir2001', 'service:mw-api-int', 'service:mw-web']
7 discussion positive []
模型答案来自我在 2026 年 10 月 31 日和 11 月 1 日记录的运行结果，使用了缩短后的 prompts。在 temperature 0.7 下询问三次后，labels 是否会重复？
使用 closed vocabulary 时，Qwen/Qwen3.8-Flash-Next 在 90% 的对话中重复了自己的答案；使用自由 JSON 时，从未重复。使用 structured outputs 的 Claude 达到了 70%。
Correlation Engine
每次 incident review 都会询问刚刚发生了什么变化。Wikimedia 记录了 283,743 次变更，平均每 10.1 分钟一次，而 correlate.py 首先会计算某个时刻附近的变更次数：
"""correlate.py: how often a change is nearby by chance, and chains from a cause to a symptom."""
import numpy as np
def changes_before(moments: np.ndarray, minutes: float, times: np.ndarray) -> np.ndarray:
"""Changes in (t - minutes, t] for every moment t, by binary search."""
return np.searchsorted(times, moments, side="right") - np.searchsorted(times, moments - minutes * 60, side="right")
在 20,000 个随机时刻上运行后，它发现其中 58.9% 的时刻在前一个小时内存在变更：
在 30 分钟内，deploy 先于 40.4% 的 incidents 发生；而在其他周的相同时间段中，这个比例是 31.4%。仅凭时间主要测量的是 production 的繁忙程度。
detectors 将每个 OpenRCA window 与此前 60 分钟进行比较，并标记了 863,062 个分钟。每个 signal 都需要 answer key 中的 component names：
"""signals.py: signal minutes that know their component, and the call graph between components."""
import re
import numpy as np
import pandas as pd
def components_of(system: str, entity: str) -> list[str]:
"""The answer key's names for one measurement."""
e = str(entity)
if not system.startswith("market"):
return [e] # bank and telecom use the key's names already
if e.endswith("-grpc") or ".ts:" in e:
return [re.split(r"-grpc$|\.ts:", e)[0].rstrip("2")]
if "." in e:
node, pod = e.split(".", 1) # 'node-6.cartservice-0'
return [pod, re.sub(r"2?-\d+$", "", pod), node]
return [e] if re.fullmatch(r"node-\d+", e) else [e, re.sub(r"2?-\d+$", "", e)]
只有在以下情况下，一个 signal 才能解释另一个 signal：第二个 component 调用了第一个 component，无论是直接调用还是向上一级调用；或者第二个 component 运行在第一个 component 上：
# kg/signals.py
def related_in(callers: dict, node_of: dict):
# b may be an effect of a when b calls a, directly or one step up, or a is the node b runs on
def related(a: str, b: str) -> bool:
if re.sub(r"2?-\d+$", "", a) == re.sub(r"2?-\d+$", "", b) or node_of.get(b) == a:
return True
up = callers.get(a, set())
return b in up or any(b in callers.get(u, set()) for u in up)
return related
def chain_inputs(sig: pd.DataFrame, system: str, top: int = 3) -> list[dict]:
"""The strongest few signals per component and kind, severity in 0 to 1."""
if sig.empty:
return [] # six windows produced no signal at all
d = pd.concat([g.assign(comp=c) for e, g in sig.groupby("entity") for c in components_of(system, e)])
d["severity"] = np.clip(np.log1p(d.strength) / np.log1p(50), 0, 1) # z of 50 or more is fully severe
d = d.sort_values("severity", ascending=False).groupby(["comp", "kind"]).head(top)
return [{"sid": f"{r.comp}|{r.kind}|{r.minute:%H%M}|{i}", "type": r.kind, "comp": r.comp, "t": r.minute,
"severity": float(r.severity)} for i, r in enumerate(d.itertuples())]
在 Netflix 的文章中，已识别 cause 和 effect 之间的一跳得分为 0.9，其他跳转则在 10 分钟窗口内按“时间接近度的 0.6 倍 + severity 的 0.4 倍”计算。允许 lineage 延伸该时间窗口是我们自行加入的内容：
hop rules 被写入 correlate.py，但去掉了文章中的 0.6 floor，这不会改变下面的结果：
# kg/correlate.py
KNOWN_PAIRS = {("DEPLOYMENT", "LOG_ERROR"), ("CONFIG_CHANGE", "METRIC_ANOMALY"), ("ALERT", "INCIDENT"),
("TRACE_FAILED", "INCIDENT")}
def hop_score(a: str, b: str, minutes: float, severity: float, window: float = 10.0) -> float:
"""A known causal pair scores 0.9, any other hop 0.6 x closeness + 0.4 x severity."""
return 0.9 if (a, b) in KNOWN_PAIRS else 0.6 * max(0.0, 1 - minutes / window) + 0.4 * severity
def may_link(a: dict, b: dict, mode: str, related, window: float = 10.0, lineage: set = frozenset()) -> float | None:
"""Minutes from a to b if the hop is allowed, else None."""
gap = (b["t"] - a["t"]).total_seconds() / 60
if gap < 0:
return None
if (a["sid"], b["sid"]) in lineage:
return gap # the deploy ships the commit, whatever the clock says
if gap > window:
return None
return gap if mode == "timing" or a["comp"] == b["comp"] or related(a["comp"], b["comp"]) else None
depth-first search 会沿着五个最优的下一跳，向 symptoms 追踪：
# kg/correlate.py
def best_chains(signals: list, sinks: set, mode: str, related, max_hops: int = 4, lineage: set = frozenset()) -> dict:
"""Depth-first from every signal to the symptoms. Each keeps its best chain."""
by_time, best = sorted(signals, key=lambda s: s["t"]), {}
nxt = {s["sid"]: [] for s in by_time}
for i, a in enumerate(by_time):
for b in by_time[i + 1:]:
if (b["t"] - a["t"]).total_seconds() > 600 and not lineage:
break # sorted by time, so nothing later can link
if (gap := may_link(a, b, mode, related, lineage=lineage)) is not None:
score = None if (a["sid"], b["sid"]) in lineage else hop_score(a["type"], b["type"], gap, b["severity"])
nxt[a["sid"]].append((b, score))
nxt[a["sid"]] = sorted(nxt[a["sid"]], key=lambda x: -(x[1] if x[1] is not None else 1))[:5] # 5 best hops
def walk(root, node, scores, path):
if node["sid"] in sinks and scores and np.mean(scores) > best.get(root["sid"], (0, None))[0]:
best[root["sid"]] = (float(np.mean(scores)), list(path))
for b, sc in (nxt[node["sid"]] if len(path) <= max_hops else []):
walk(root, b, scores + ([sc] if sc is not None else []), path + [b["sid"]])
for s in by_time:
if s["sid"] not in sinks:
walk(s, s, [], [s["sid"]])
return best
components 按其最佳 chain 排名，未到达的 components 则按声音大小排序：
# kg/correlate.py
def rank_loudest(signals: list) -> list[str]:
"""The baseline: components by their loudest signal, ties by name."""
loud = {}
for s in signals:
loud[s["comp"]] = max(loud.get(s["comp"], 0), s["severity"])
return [c for c, _ in sorted(loud.items(), key=lambda x: (-x[1], x[0]))]
def rank_components(signals: list, mode: str, related) -> list[str]:
"""Components by best chain, then the unreached ones, loudest first."""
sinks = {s["sid"] for s in signals if s["type"] in ("TRACE_SLOW", "TRACE_FAILED")} or \
{max(signals, key=lambda s: s["t"])["sid"]} # the symptoms are what callers felt
comp, score = {s["sid"]: s["comp"] for s in signals}, {}
for sid, (s, _) in best_chains(signals, sinks, mode, related).items():
score[comp[sid]] = max(score.get(comp[sid], 0), s)
chained = [c for c, _ in sorted(score.items(), key=lambda x: (-x[1], x[0]))]
return chained + [c for c in rank_loudest(signals) if c not in score]
engine 首先必须复现文章中的示例 chain，其中 severity 为 0.825，得到 0.81 的 hop：
import pandas as pd
from kg.correlate import best_chains
t0, same = pd.Timestamp("2026-01-01 12:00", tz="UTC"), lambda a, b: a == b
chain = [{"sid": k, "type": k, "comp": "api-gateway", "t": t0 + pd.Timedelta(minutes=m), "severity": 0.825}
for k, m in (("COMMIT", 0), ("DEPLOYMENT", 14), ("LOG_ERROR", 17), ("ALERT", 19), ("INCIDENT", 24))]
score, path = best_chains(chain, {"INCIDENT"}, "topology", same, lineage={("COMMIT", "DEPLOYMENT")})["COMMIT"]
print(f" {score:.3f} via {path}, without lineage: {'COMMIT' in best_chains(chain, {'INCIDENT'}, 'topology', same)}")
#### OUTPUT ####
0.870 via ['COMMIT', 'DEPLOYMENT', 'LOG_ERROR', 'ALERT', 'INCIDENT'], without lineage: False
它复现了 0.87；没有 lineage 时，14 分钟的 hop 会被拒绝。在 41 个具有命名 component 的 development failures 中：
#### OUTPUT ####
ranking AC@1 AC@3
loudest signal 4.9% 4.9%
timing-only chains 4.9% 9.8%
topology chains 9.8% 26.8%
call graph 将 top-three accuracy 从 4.9% 提升到了 26.8%。但 score 到底意味着什么？
在得分高于 0.9 的 chains 中，只有 6% 指向了正确 component，Brier score 为 0.871。这个 score 用于排列 causes，并不是 probability。
Knowledge Flywheel
每个已解决的 failure 都带有答案，而每个答案都可以教会下一次 investigation 应该从哪里开始。完整运行学习到，每种 signal 出现在 root cause 上的频率比出现在旁观者上的频率高多少：
使用从其他 failures 中学习到的 lifts 进行评分时，它在 29.3% 的 failures 中将正确 component 排在第一位，而 loudness 的比例为 4.9%。在一个从未见过的 cluster 上，它在 7 个 failures 中胜过手写规则，在 8 个 failures 中落败。
Root-Cause Specialists
读取证据是 specialist agents 的工作：
specialist 永远不会读取 raw data。它的 typed tools 返回带有 id 的 rows，specialist 可以引用这些 id，并记住曾经提供过的每个 id：
"""agents.py: specialists with tools, a verifier for their evidence, a reconciler that may ask back."""
import inspect
import json
from collections import Counter
from concurrent.futures import ThreadPoolExecutor
import pandas as pd
from kg.config import MODELS
from kg.llm import chat, parse_json
def utc8(ts) -> str:
return pd.Timestamp(ts).tz_convert("Asia/Shanghai").strftime("%Y-%m-%d %H:%M:%S") # the answer key's clock
class LabTools:
"""What a specialist may ask about one window. Only ids handed out may be cited."""
def __init__(self, sig: pd.DataFrame, callers: dict):
self.sig = sig.assign(id=[f"s{i}" for i in range(len(sig))]) # one id per signal minute
self.callers, self.given = callers, set() # given: every id handed out
def _rows(self, d: pd.DataFrame) -> list[dict]:
d = d.sort_values("strength", ascending=False).head(12)
self.given |= set(d.id)
return [{"id": r.id, "entity": r.entity, "kind": r.kind, "family": r.family, "minute": utc8(r.minute),
"strength": round(float(r.strength), 1)} for r in d.itertuples()]
每个 tool 都是一个简短 method，模型可以读取它的 docstring：
# kg/agents.py
def top_anomalies(self, kind: str = "any") -> list[dict]:
"""The strongest anomalies in the window. kind: any, METRIC_ANOMALY, TRACE_SLOW, TRACE_FAILED or LOG_ERROR."""
d = self.sig if kind == "any" else self.sig[self.sig.kind == kind]
return self._rows(d.sort_values("strength", ascending=False).drop_duplicates(["entity", "family"]))
def component_signals(self, component: str) -> list[dict]:
"""Every anomaly on one component, strongest first."""
return self._rows(self.sig[self.sig.comps.map(lambda cs: component in cs)])
def first_anomaly(self, component: str) -> dict:
"""When the first anomaly on a component appeared."""
d = self.sig[self.sig.comps.map(lambda cs: component in cs)].sort_values("minute")
if d.empty:
return {"component": component, "first": None}
self.given.add(d.id.iloc[0])
return {"component": component, "first": utc8(d.minute.iloc[0]), "id": d.id.iloc[0], "family": d.family.iloc[0]}
def neighbours(self, component: str) -> dict:
"""Which components call this one and which it calls, from the traces."""
return {"component": component, "called_by": sorted(self.callers.get(component, set()))[:15],
"calls": sorted(c for c, up in self.callers.items() if component in up)[:15]}
tool spec 根据代码构建，错误参数会返回 agent 可以读取的 error：
# kg/agents.py
def schema(self, names: list[str]) -> list[dict]:
# docstrings become descriptions, so the code is the spec
return [{"type": "function", "function": {"name": n, "description": getattr(self, n).__doc__, "parameters": {
"type": "object", "properties": {a: {"type": "string"} for a in inspect.signature(getattr(self, n)).parameters}}}}
for n in names]
def call(self, name: str, args: dict):
try:
return getattr(self, name)(**args)
except Exception as e: # a bad argument is an answer the agent can read
return {"error": f"{type(e).__name__}: {e}"}
运行示例说明了为什么 tools 优于 raw telemetry：
该 window 包含 182 个 entities 的 7,465 个 signal minutes。最响亮的是 frontend disk measurements，而故障中的 cartservice 排名第 38 位。三个 specialists 分工处理：
# kg/agents.py
SPECIALISTS = { # role: (its tools, its brief)
"metrics": (["top_anomalies", "component_signals", "first_anomaly"], "You are the metrics specialist."),
"traces": (["top_anomalies", "neighbours", "first_anomaly"], "You are the traces specialist. A slow caller is often "
"the victim of a slow callee."),
"logs": (["top_anomalies", "component_signals"], "You are the logs specialist. Look at bursts of error lines."),
}
FINDING = ('Finish with JSON only: {"hypotheses": [{"component": ..., "reason": ..., "datetime": "YYYY-MM-DD HH:MM:SS" '
'in UTC+8, "evidence": [ids your tools returned], "confidence": 0 to 1}]}, best first, at most 3.')
每个 specialist 最多进行三轮 tool 调用，随后进行一轮不带 tools 的调用：
# kg/agents.py
def run_specialist(role: str, tools: LabTools, question: str, vocab: dict, rounds: int = 3) -> dict:
"""Three rounds of tools at most, then a last turn without them."""
names, brief = SPECIALISTS[role]
msgs = [{"role": "system", "content": f"{brief} Allowed components: {vocab['components']}. Allowed reasons: "
f"{vocab['reasons']}. Use the tools and cite only ids they return. {FINDING}"}, {"role": "user", "content": question}]
trail, turns, seconds = [], 0, 0.0
for step in range(rounds + 1):
last = step == rounds # no tools on the last turn, so it must answer
r = chat(MODELS["flash"], msgs + ([{"role": "user", "content": "No more tools. " + FINDING}] if last else []),
json_mode=False, tools=None if last else tools.schema(names), max_tokens=700)
turns, seconds = turns + 1, seconds + r["seconds"]
if not r["tool_calls"] or last:
hyps = (parse_json(r["text"]) or {}).get("hypotheses") or []
return {"hypotheses": hyps[:3], "tools": trail, "turns": turns, "seconds": seconds}
msgs.append({"role": "assistant", "content": r["text"], "tool_calls": r["tool_calls"]})
for tc in r["tool_calls"]: # one turn may ask for several tools
name, args = tc["function"]["name"], parse_json(tc["function"].get("arguments")) or {}
trail.append(name)
result = tools.call(name, args) if name in names else {"error": f"no tool {name}"}
msgs.append({"role": "tool", "tool_call_id": tc.get("id", name), "content": json.dumps(result, default=str)[:6000]})
三个 specialists 都运行在 Qwen/Qwen3.8-Flash-Next 上：
#### OUTPUT ####
metrics 4 turns, 7 tool calls, 11.3 s -> node-6, node memory consumption, 15:48
traces 4 turns, 5 tool calls, 9.1 s -> cartservice-0, container read I/O load, 15:48
logs 4 turns, 6 tool calls, 11.0 s -> cartservice-0, container read I/O load, 15:46
answer key -> cartservice, container packet loss, 15:45:31
两个 specialists 找到了故障 service，并且它们进行的每次调用都绘制在这里：
两者都将原因归咎于 disk reads，而 answer key 指向 lost packets。
Evidence Verifier and Reconciler
Agents 有时会引用任何 tool 都没有提供的 id，而一个伪造的 id 看起来就像真实 id。捕获它不需要模型：
# kg/agents.py
def verify(hypotheses: list, tools: LabTools, vocab: dict, ws, we) -> tuple[list, Counter]:
"""Mechanical and cheap. Nothing here asks a model whether an answer seems right."""
kept, why = [], Counter()
for h in hypotheses:
cited = set(map(str, h.get("evidence") or []))
t = pd.to_datetime(str(h.get("datetime")), errors="coerce")
if pd.notna(t):
t = (t if t.tzinfo else t.tz_localize("Asia/Shanghai")).tz_convert("UTC") # agents answer in UTC+8
if not cited:
why["no evidence"] += 1
elif not cited <= tools.given: # an id its own tools never handed out
why["cites evidence it was not given"] += 1
elif h.get("component") not in vocab["components"]:
why["component outside the list"] += 1
elif h.get("reason") not in vocab["reasons"]:
why["reason outside the list"] += 1
elif pd.isna(t) or not ws - pd.Timedelta(minutes=5) <= t <= we:
why["time outside the window"] += 1
else:
kept.append(h)
return kept, why
只有经过验证的 hypotheses 才会到达 reconciler，即 Qwen/Qwen3.8-Max；它可以向一个 specialist 追问一次：
# kg/agents.py
def reconcile(question: str, verified: dict, vocab: dict, allow_ask: bool = True) -> dict:
"""The strongest model reads verified hypotheses only, and may ask back once."""
system = (f"You lead a root-cause investigation. Allowed components: {vocab['components']}. Allowed reasons: "
f"{vocab['reasons']}. Prefer a cause that explains the other findings. Answer with JSON only: "
'{"answer": [{"datetime": ..., "component": ..., "reason": ...}], "confidence": 0 to 1, "ask": null or '
'{"specialist": ..., "question": ...}}.' + ("" if allow_ask else ' "ask" is null.'))
body = "\n".join(f"{role}: {json.dumps(kept, default=str)}" for role, kept in verified.items())
r = chat(MODELS["max"], [{"role": "system", "content": system}, {"role": "user", "content": f"{question}\n\nFindings:\n{body}"}])
d = parse_json(r["text"]) or {}
return {"answer": d.get("answer") or [], "confidence": d.get("confidence"), "ask": d.get("ask") if allow_ask else None}
一次 investigation 会并行运行三个 specialists，检查它们的主张，然后进行 reconciliation：
# kg/agents.py
def investigate(sig: pd.DataFrame, callers: dict, question: str, vocab: dict, ws, we) -> dict:
"""Specialists side by side, every finding verified, one answer."""
tools = {role: LabTools(sig, callers) for role in SPECIALISTS} # each keeps its own record of ids
with ThreadPoolExecutor(len(SPECIALISTS)) as ex:
found = dict(zip(SPECIALISTS, ex.map(lambda ro: run_specialist(ro, tools[ro], question, vocab), SPECIALISTS)))
checked = {role: verify(f["hypotheses"], tools[role], vocab, ws, we) for role, f in found.items()}
verified, rejected = {r: k for r, (k, _) in checked.items()}, sum((w for _, w in checked.values()), Counter())
rec, asked = reconcile(question, verified, vocab), None
if isinstance(rec["ask"], dict) and rec["ask"].get("specialist") in SPECIALISTS: # one question, never a loop
asked = rec["ask"]["specialist"]
found[asked + "+"] = run_specialist(asked, tools[asked], f"{question}\nThe lead asks: {rec['ask'].get('question')}", vocab, 2)
kept, why = verify(found[asked + "+"]["hypotheses"], tools[asked], vocab, ws, we)
verified[asked], rejected = kept + verified[asked], rejected + why
rec = reconcile(question, verified, vocab, allow_ask=False)
return {**rec, "asked": asked, "found": found, "rejected": sum(rejected.values()),
"calls": sum(f["turns"] for f in found.values()) + 1 + (asked is not None)}
这些部分共同构成一个 checkpoint：
为了测试 verifier，我添加了一个引用 s999999 的伪造 hypothesis，而任何 tool 都没有产生这个 id：
#### OUTPUT ####
verifier planted 1 rejected ('cites evidence it was not given'), 9 real hypotheses pass
answer cartservice-0, container read I/O load, 2022-03-21 15:46:00, confidence 0.7
asked back None, 13 model calls, 33.2 s one after another, 13.1 s side by side
score partial 0.33 against the answer key
伪造引用被捕获；答案正确给出了 pod 和 minute，但 reason 错了。在完整运行中，它会变成等待人工处理的 fix card。
Label Adjudication
Wikimedia 的 146 份 incident reports 没有 answer key，因此由模型进行 labelling。两个不同 family 的模型必须达成一致：
Claude Opus 5.5 和 Qwen/Qwen3.7-Max 读取了每份 report，Cohen's kappa 用于衡量两者超出随机猜测预期的一致程度：
#### OUTPUT ####
trigger kappa 0.73
detected by kappa 0.92
agreed on trigger and components: 100 of 146
对于其余 46 份，一份 Claude model 和 Qwen/Qwen3.8-Max 在 blind 条件下进行判断，其中 27 份达成一致。其余 19 份被移出测试。在打开任何测试数据之前，splits、floors 和 targets 都已通过 sha256 生成 fingerprint。
Graph Versus Raw Context
一个 SPARQL query 在全部 217 个有变更的 alert moments 上，跨五个 sources 匹配了五次 lookups。随后，四个 arms 读取了 24 个 test incidents，并将已知时间固定在 incident 前 15 分钟：
#### OUTPUT ####
arm trigger right cites unseen credits
graph, one prompt 42% 0% 0.20
raw, one prompt 25% 4% 0.20
graph, specialists 33% 0% 3.38
raw, specialists 33% 0% 3.36
在 single prompt 中，graph facts 击败了 raw text，比例为 42% 对 25%；只有 raw text 引用了 unseen evidence：
不过，样本只有 24 个 incidents，因此这里没有任何差距超过四个 incidents。
Benchmark Comparison
RCAEval 会告诉每个 method fault 被注入的时间。BARO 是 Train Ticket 上已发布的最强 baseline，它根据该时刻之后 metrics 的变化幅度为每个 service 评分：
"""exam.py: the rulers, a model-free ranker and exact tests."""
import pandas as pd
from scipy.stats import binomtest
def rank_rcaeval(metrics: pd.DataFrame, inject_time: int) -> list[str]:
"""Services by their most extreme metric after the fault, in normal spreads."""
t = pd.to_numeric(metrics.iloc[:, 0], errors="coerce") # the first column is the clock
normal, after, score = metrics[t < inject_time], metrics[t >= inject_time], {}
for col in metrics.columns[1:]:
x0 = pd.to_numeric(normal[col], errors="coerce").dropna()
x1 = pd.to_numeric(after[col], errors="coerce").dropna()
if "_" not in col or len(x0) < 10 or x1.empty:
continue # too little history to know normal
iqr = x0.quantile(0.75) - x0.quantile(0.25)
scale = iqr if iqr > 0 else (x0.std() if x0.std() > 0 else 1.0)
svc = col.rsplit("_", 1)[0] # 'cartservice_cpu' -> cartservice
score[svc] = max(score.get(svc, 0.0), float((x1 - x0.median()).abs().max() / scale))
return [s for s, _ in sorted(score.items(), key=lambda x: -x[1])]
在全部 360 个 cases 上，不使用模型，也不进行 training：
在全部 cases 中，91% 的时间里，被注入的 service 位于我们最初三个猜测之内。对于 Train Ticket，该比例为 80%，接近 BARO 的 82%；而我们的 first guess 正确率为 31%，BARO 为 69%。
OpenRCA 隐藏了 injection time。在保留用于测试的 cluster 上冻结并运行一次后，我们的 agents 获得了 16.9% 的 partial credit，以及严格评分下的 5.1%：
灰色 bars 是后续研究在该 cluster 上使用的自身 arms，OpenRCA authors 没有运行其中任何一个。它们全部胜过我们，因为我们的 detectors 在任何 agent 查看之前就隐藏了 cause。
Claim Verdicts
精确测试会决定十二项主张中的每一项。McNemar's test 只统计两种方法意见不一致的 cases，七项比较共享 Holm's correction**：**
# kg/exam.py
def mcnemar(a: list, b: list) -> tuple[int, int, float]:
"""Exact McNemar on paired outcomes. Only disagreements count."""
only_a, only_b = sum(bool(x) and not y for x, y in zip(a, b)), sum(bool(y) and not x for x, y in zip(a, b))
return only_a, only_b, (binomtest(only_a, only_a + only_b).pvalue if only_a + only_b else 1.0)
def holm(p: dict) -> dict:
"""The k-th smallest of m p-values times m - k + 1, never decreasing."""
adj, running = {}, 0.0
for k, name in enumerate(sorted(p, key=p.get)):
running = max(running, min(1.0, (len(p) - k) * p[name]))
adj[name] = running
return adj
一项主张必须满足：每个比较都支持它，并且经过校正的 p 小于 0.05：
#### OUTPUT ####
comparison n what we saw p Holm p
closed vocabulary repeats itself 10 only closed 9, only free 0 0.004 0.027
topology over timing 41 first: topology 4, timing 2 0.500 1.000
incidents over placebo 853 33/146 incidents, 103/707 0.018 0.111
learned over manual, held out 78 better on 7, worse on 8 1.000 1.000
three more with chat, raw text and coverage, each 1 incident against 0, Holm p 1.000
校正后，只有 closed vocabulary 存活：
"""replay.py: the whole OpenRCA corpus as one mixed stream, read by worker processes against the clock."""
import multiprocessing as mp
import random
import time
import duckdb
import pyarrow as pa
import pyarrow.parquet as pq
COLUMNS = {"spans": ["ts", "entity", "duration_ms"], "logs": ["ts", "entity", "message"], "metrics": ["ts", "entity", "kpi", "value"]}
REPLAY_SQL = { # the per-minute aggregates the detectors start from
"spans": "SELECT entity, date_trunc('minute', ts) AS m, count(*) AS n, avg(duration_ms) AS ms, max(duration_ms) AS worst "
"FROM batch GROUP BY ALL",
"logs": "SELECT entity, date_trunc('minute', ts) AS m, count(*) AS n, count(*) FILTER (WHERE "
"regexp_matches(message, '(?i)error|exception|fail|refused|timeout')) AS bad FROM batch GROUP BY ALL",
"metrics": "SELECT entity, kpi, date_trunc('minute', ts) AS m, avg(value) AS v, max(value) AS peak FROM batch GROUP BY ALL",
}
def micro_batches(files: dict, seed: int = 42) -> list[tuple]:
"""Every row group of every file, shuffled like a live stream."""
out = [(f, rg, kind) for kind, fs in files.items() for f in fs for rg in range(pq.ParquetFile(f).metadata.num_row_groups)]
return random.Random(seed).sample(out, len(out))
五项主张成立，其中一项 query 替代了三个 tools。两项失败，分别是 edge cap 和 flywheel；另外五项仍然没有结论，或无法测试。
Stream Replay and Fleet Sizing
Netflix 的 logging stream 峰值达到每秒 3,800 万个 events。为了测量一台机器，我们将全部 5.25 亿个 OpenRCA events 打乱为小 batches，并通过 minute-level aggregates 进行 stream replay：
每个 worker 会不断处理分配给自己的部分，直到时钟耗尽：
# kg/replay.py
def worker(spec: dict) -> dict:
"""One process: its share of the batches, until the clock runs out."""
pa.set_cpu_count(spec["threads"])
con = duckdb.connect(config={"threads": spec["threads"], "TimeZone": "UTC"})
files, done, rows, start = {}, [], [], time.time()
while time.time() < start + spec["seconds"]:
path, rg, kind = spec["batches"][len(done) % len(spec["batches"])]
if path not in files:
files[path] = pq.ParquetFile(path)
batch = files[path].read_row_group(rg, columns=COLUMNS[kind]) # one row group is one micro-batch
con.register("batch", batch)
con.sql(REPLAY_SQL[kind]).to_arrow_table()
con.unregister("batch")
done.append(time.time())
rows.append(batch.num_rows)
return {"start": start, "done": done, "rows": rows}
速率从第一个 start 开始计算，一直到最后一个 batch 完成：
# kg/replay.py
def replay(batches: list, processes: int, seconds: float, cores: int = 16) -> dict:
"""Events a second, from the first start to the last finished batch."""
specs = [{"batches": batches[k::processes], "seconds": seconds, "threads": max(1, cores // processes)}
for k in range(processes)]
with mp.get_context("spawn").Pool(processes) as pool:
res = pool.map(worker, specs)
t0, t1 = min(r["start"] for r in res), max(r["done"][-1] for r in res)
events = sum(sum(r["rows"]) for r in res)
return {"processes": processes, "seconds": round(t1 - t0, 1), "events": events, "events_per_s": events / (t1 - t0)}
我们分别运行了 1 到 16 个 workers，每次 45 秒；随后让 16 个 workers 运行十分钟：
from kg.replay import micro_batches, replay
batches = micro_batches(openrca_files) # {kind: Parquet files}, under if __name__ == "__main__" in a script
long = replay(batches, 16, 600)
print(f" ten minutes: {long['events']:,} events, {long['events_per_s'] / 1e6:.1f}M a second")
#### OUTPUT ####
ten minutes: 15,723,611,150 events, 26.1M a second
replay 在单个 worker 上以每秒 840 万个 events 运行，在十六个 workers 上达到每秒 2,640 万个。十分钟运行期间，十六个 workers 保持了每秒 2,610 万个 events：
Netflix 的 stream 需要 1.5 台这样的笔记本电脑；如果 events 以 JSON 到达，则需要 2.0 台。graph 只接收 signals：
一个 signal 压缩了 82,681 个 events，stream 变成每秒 2,753 个 triples。完整运行将 16,393,203 个 triples 加载到一个 store 中：
使用 hour buckets 后，无论 store 增长到多大，典型 query 都保持在 1.4 ms 以下。每个 batch 在 commit 前都经过 shapes 检查，实时写入达到每秒 39,871 个 triples，足以覆盖该 stream 的 14 倍。
My architecture Limits
本研究从未接触 Netflix 自己的 telemetry，图中列出了其他限制，其中 detectors 最弱。
repository 中仍有一些部分未包含：
left out：Wikimedia harvest client 和 pseudonyms、pageview converter 以及 detectors（添加 source 时再构建）。
left out：shape compiler 和 store 的 bulk writers（超过一百万个 triples 后再构建）。
left out：flywheel、alert episodes、change kinds、plan fingerprint 和 fix card（人们开始根据答案采取行动时再构建）。
全部代码都在 GitHub repository 中：
https://github.com/FareedKhan-dev/ontology-driven-observability