RAG SDK 中的观测系统应该如何设计

在前面的内容里，我们已经陆续讨论了 indexing、adapters、runtime 以及生成阶段的设计。

这些内容解决的是一个 RAG SDK 如何把流程跑起来：文档如何进入索引，查询如何被处理，候选内容如何被召回和筛选，最后又如何交给模型生成答案。

但是，当一条 RAG 链路真正跑起来之后，还会出现一个新的问题：

我们怎么知道它跑得好不好？

这个问题看起来像是运维问题，但在 RAG 系统里，它其实也是一个很重要的 SDK 设计问题。因为 RAG 不是一个简单函数调用，它是一条由多个阶段组成的链路。链路越长，问题定位就越依赖中间过程。如果 SDK 只返回一个最终答案，而不暴露中间发生了什么，那么一旦效果不好，开发者就只能靠猜。

这篇文章就来讨论，在一个面向 SDK 的 RAG 体系中，观测系统应该解决什么问题，它和普通日志有什么区别，哪些能力应该由 SDK 提供，哪些指标又应该交给业务系统自己定义。

1. 为什么 RAG SDK 需要观测系统

先从一个非常常见的场景说起。

用户问了一个问题，系统也返回了答案。但用户反馈说：

这个回答不对。

这时候，开发者真正要排查的并不是“模型为什么回答错了”这么简单。因为在一个 RAG 系统里，最终答案只是链路末端的结果。在答案生成之前，前面已经发生了很多事情。

例如：

1. 用户输入的问题有没有被正确理解
2. query rewrite 有没有把原始问题改坏
3. 检索阶段有没有召回到正确文档
4. 召回候选的数量是否足够
5. rerank 或 filter 有没有把正确内容丢掉
6. prompt 里面最终放进去的上下文是否合适
7. 模型生成时有没有脱离上下文
8. 整个请求是不是因为某一步太慢导致超时

也就是说，一个“回答不对”的问题，背后可能对应很多不同原因。

如果没有观测能力，开发者看到的可能只有这样的结果：

const result = await runtime.run({
  query: "公司年假政策是什么？",
});

console.log(result.answer);

这段代码当然能拿到答案，但它没有告诉我们：

1. 原始 query 是否被改写
2. 检索阶段召回了多少候选
3. 哪些 chunk 最终进入了上下文
4. 哪些候选被丢弃了
5. 每个阶段分别耗时多久
6. 如果失败，失败发生在哪个阶段

在 demo 阶段，这可能还可以接受。因为数据少、链路短，开发者可以直接加几个 console.log 临时看一下。

但一旦进入真实项目，这种方式就会很快失效。

首先，RAG 链路通常不是一个单点函数，而是一条流水线。检索前处理、检索、检索后处理、生成，每一步都有自己的输入输出。如果只在最终结果处打印日志，就无法还原中间过程。

其次，RAG 的问题很多时候不是稳定复现的。某些 query 会错，某些文档会错，某些过滤条件会错，某些上下文组合会错。如果没有 traceId 把一次请求里的所有事件串起来，后续就很难知道当时到底发生了什么。

再次，RAG 系统的问题不只包括“答错”。它还包括：

1. 慢
2. 超时
3. 召回为空
4. 候选过多
5. 上下文过长
6. 答案没有引用来源
7. 用户继续追问
8. 用户给出负反馈

这些问题有些属于系统性能，有些属于链路调试，有些属于业务效果。它们不能都靠一个最终返回值表达。

所以，对 RAG SDK 来说，观测系统不是锦上添花的外围功能，而是让链路可解释的基础能力。

观测系统并不等于简单打印日志。

日志通常是面向开发者临时排查的文本输出，它可以很自由，但也很容易变得不结构化。比如今天打印一句 "retrieval done"，明天打印一个对象，后天又换成另一种格式。时间一长，这些日志很难被统一收集、查询和分析。

观测系统关注的是结构化事件。

例如，一次 runtime 查询中，SDK 可以在不同阶段发出这样的事件：

{
  traceId: "trace-001",
  scope: "runtime",
  stage: "retrieval",
  name: "runtime.retrieval.complete",
  durationMs: 32,
  attributes: {
    candidateCount: 8,
    emptyRetrieval: false,
  },
}

这类事件和普通日志最大的区别在于：它是有结构、有阶段、有 traceId 的。

有了这些结构化数据，开发者就可以进一步回答很多具体问题：

1. 这次请求总共耗时多久
2. 最慢的是检索阶段还是生成阶段
3. 检索是否为空
4. 召回了多少候选
5. 最终选择了多少 chunk
6. 失败发生在哪个阶段
7. 同一类问题是否经常在某个阶段失败

从 SDK 设计角度看，这件事尤其重要。

因为 SDK 本身最了解 RAG 链路的内部阶段。业务方当然也可以自己埋点，但业务方通常只能在外层看到“调用前”和“调用后”。至于 runtime 内部到底经历了哪些步骤，每一步的输入输出是什么，哪个阶段耗时更长，这些信息天然应该由 SDK 暴露出来。

换句话说，SDK 不一定负责建设完整的监控平台，但它应该负责把可观测的数据发出来。

如果把职责说得更准确一点：

1. SDK 负责在关键阶段产生结构化事件
2. SDK 负责把同一次请求里的事件串成 trace
3. SDK 负责记录阶段耗时、错误、候选数量等技术事实
4. SDK 负责提供 observer/exporter，让数据可以被导出
5. 业务系统负责根据自己的口径解释这些数据

这也是为什么在一个 RAG SDK 里，观测系统应该作为独立能力存在。

它不是某个 runtime 函数里的几行日志，也不是某个 demo 脚本里的调试输出，而是一套横跨 runtime、indexing，甚至后续 eval 和业务反馈的基础协议。

所以，RAG 观测的核心不是“记录日志”，而是把一次回答拆成可解释的链路，让开发者知道慢在哪里、错在哪里，以及后续应该从哪里开始排查。

2. 先用四层理解观测系统

在继续拆指标之前，可以先用一个更稳定的四层模型来理解 RAG SDK 的观测系统。

这四层分别是：

Trace      一次完整 RAG 调用的链路骨架
Events     链路中每个阶段发生的结构化事实
Metrics    可以被聚合统计的数值信号
Evaluation 从 trace 中提取问题样本，进入质量改进闭环

它们关注的问题不一样。

Trace 回答的是：这一次请求从哪里开始，经过了哪些阶段，最后是成功还是失败。

Events 回答的是：在 query、retrieval、post-retrieval、generation、indexing 等阶段分别发生了什么。

Metrics 回答的是：这些事实能不能被统计，比如耗时、错误率、空召回率、token 数、候选数量、丢弃比例。

Evaluation 回答的是：哪些 trace 值得拿出来复盘，哪些样本暴露了召回、筛选或生成质量问题。

这四层不是互相替代的关系，而是逐层递进的关系。

没有 Trace，就串不起一次请求。
没有 Events，就不知道中间发生了什么。
没有 Metrics，就无法从单次请求走向系统统计。
没有 Evaluation，就很难把观测数据变成质量改进样本。

当前 SDK 的实现也是沿着这个方向展开的：先用 RAGTrace 和 RAGEvent 把单次链路记录清楚，再通过 observer / exporter 把 trace 导出去；随后补充 redaction、sampling、HTTP exporter 和 eval trace sample，让观测数据可以进入更长期的分析与复盘。

接下来再看性能类、调试类和业务类指标时，可以把它们放回这个四层模型里理解：性能类更多落在 Metrics，调试类更多落在 Trace 和 Events，业务类通常需要通过 traceId 和外部业务事件关联，最后进入 Evaluation 或数据平台分析。

3. 观测系统主要看什么

明确了为什么需要观测系统之后，接下来就要看一个更具体的问题：

RAG 观测系统到底应该看哪些东西？

这些指标大致可以分成三类：

1. 性能类
2. 调试类
3. 业务类

这三类指标的关注点并不一样。性能类关注系统是否稳定，调试类关注一次请求内部发生了什么，业务类关注用户是否真的觉得有用。

3-1. 性能类指标

性能类指标主要回答的是：系统跑得稳不稳。

常见的性能类指标包括：

1. Average Latency
2. P95 / P99
3. QPS
4. 错误率
5. 超时率

这些指标更接近平台侧或运维侧视角。

例如，Average Latency 看的是平均响应时间，P95 / P99 看的是尾部延迟，QPS 看的是单位时间内系统承载了多少请求，错误率和超时率则反映系统稳定性。

在当前 SDK 里，已经具备一部分性能类指标的原始数据。

一次 runtime 运行结束时，trace 里会有本次请求的总耗时：

type RAGTrace = {
  traceId: string;
  scope: "runtime" | "indexing";
  startedAt: string;
  endedAt?: string;
  durationMs?: number;
  status: "ok" | "error";
  events: RAGEvent[];
  errors?: RAGErrorRecord[];
};

这里的 durationMs 可以用于计算平均延迟，status 可以用于计算错误率。

SDK 记录的是单次 trace 的事实数据，并不直接计算 P95、P99、QPS 这类聚合指标。因为这些指标需要基于一段时间内的大量请求统计，通常应该交给外部监控系统或数据平台完成。

也就是说，当前 SDK 更像是提供原始材料：

{
  traceId: "trace-001",
  scope: "runtime",
  durationMs: 1280,
  status: "ok",
  events: []
}

而不是直接在 SDK 内部维护这样的统计结果：

{
  avgLatency: 860,
  p95Latency: 2100,
  qps: 32,
  errorRate: 0.01
}

这种边界是合理的。SDK 应该尽量保持轻量，不应该在运行时内部维护复杂的时间窗口、滑动统计和聚合状态。

3-2. 调试类指标

调试类指标主要回答的是：一次请求内部到底发生了什么。

这部分是 RAG SDK 最应该重点覆盖的内容。原因很简单，SDK 最了解 runtime 和 indexing 的内部阶段。

对于 runtime 来说，常见的调试类指标包括：

1. 每一步耗时
2. query rewrite 后的内容
3. 召回候选数
4. rerank 或筛选前后的变化
5. prompt 上下文和返回结果摘要

当前 SDK 已经覆盖了其中的大部分。

为突出事件内容，示意代码会省略 traceId、scope、timestamp 这类固定字段，只保留和当前 SDK 事件结构相关的核心信息。

例如，在检索前处理完成后，可以记录原始 query 和最终参与检索的 query：

emitEvent({
  stage: "query",
  name: "runtime.query.preprocess",
  durationMs: 12,
  attributes: {
    originalQuery: "公司年假怎么算？",
    effectiveQuery: "公司年假政策 计算规则",
    rewriteReason: "normalize_policy_query",
  },
});

在检索完成后，可以记录召回候选数量：

emitEvent({
  stage: "retrieval",
  name: "runtime.retrieval.complete",
  durationMs: 36,
  attributes: {
    candidateCount: 8,
    emptyRetrieval: false,
  },
});

在检索后处理完成后，可以记录候选筛选结果：

emitEvent({
  stage: "post_retrieval",
  name: "runtime.post_retrieval.select",
  durationMs: 9,
  attributes: {
    inputCandidateCount: 8,
    selected: 3,
    dropped: 5,
    selectedChunkIds: ["chunk-1", "chunk-3", "chunk-7"],
  },
});

在生成完成后，可以记录上下文长度和答案摘要：

emitEvent({
  stage: "generation",
  name: "runtime.generation.complete",
  durationMs: 820,
  attributes: {
    contextLength: 2400,
    answerPreview: "根据公司当前年假政策，员工年假天数...",
  },
});

这些示意代码和当前 SDK 的设计是一致的：runtime 在关键阶段发出 RAGEvent，每个事件里包含 stage、name、durationMs 和 attributes。

对于 indexing 来说，调试类指标则主要围绕离线索引流程：

1. 加载了多少文档
2. 过滤掉多少文档
3. 切出了多少 chunks
4. 生成了多少 vectors
5. 哪个阶段失败
6. 哪个文档或 chunk 失败

例如 indexing 在 chunk 阶段完成后，可以记录：

emitEvent({
  stage: "chunk",
  name: "indexing.chunk.complete",
  durationMs: 18,
  attributes: {
    documentId: "doc-1",
    chunkCount: 12,
  },
});

在 embedding 阶段完成后，可以记录：

emitEvent({
  stage: "embed",
  name: "indexing.embed.complete",
  durationMs: 46,
  attributes: {
    documentId: "doc-1",
    chunkCount: 12,
    vectorCount: 12,
  },
});

所以，当前 SDK 最完整覆盖的是调试类指标。它不是直接告诉你“系统效果好不好”，而是先告诉你“这次链路里每一步发生了什么”。

3-3. 业务类指标

业务类指标主要回答的是：用户觉得这次回答有没有用。

常见的业务类指标包括：

1. 用户满意度
2. 追问率
3. 引用点击率
4. 反馈分布

这些指标和性能类、调试类不一样。它们通常不是 SDK 自己能够判断的。

例如，同样是用户继续追问，在不同产品里含义可能完全不同：

1. 有可能是答案没有解决问题
2. 有可能是用户想继续深入了解
3. 有可能是产品本来就设计成多轮问答

再比如引用点击率。SDK 可以记录生成阶段使用了哪些 chunk，也可以把 citation 信息返回给业务系统。但用户最终有没有点击引用，这个行为发生在前端或业务应用里，不发生在 SDK 的 runtime 内部。

所以，业务类指标更合理的分工是：

1. SDK 提供 traceId、requestId 等关联字段
2. SDK 提供 observer/event/exporter 这类基础设施
3. 业务系统负责上报满意度、点击、追问、反馈等业务事件
4. 数据平台根据业务口径计算最终指标

例如，业务系统可以在用户点踩时上报一个和 trace 关联的事件：

observer.onEvent?.({
  traceId: "trace-001",
  scope: "runtime",
  stage: "feedback",
  name: "runtime.feedback.submit",
  timestamp: new Date().toISOString(),
  attributes: {
    rating: "thumbs_down",
    reason: "answer_not_helpful",
  },
});

这段代码表达的是一种后续演进方向。当前 SDK 的事件协议已经有 traceId、stage、attributes 这些基础字段，但事件命名还主要围绕 runtime 和 indexing 的内部阶段，业务反馈事件还没有正式沉淀成稳定协议。

所以，业务类指标的计算口径应该留在业务系统里，SDK 只需要保证这些业务事件能够和 RAG trace 关联起来。

这样一来，观测系统就不会被设计成一个大而全的指标系统，而是会先成为一套稳定的数据采集和关联基础设施。

4. SDK 的职责

在观测系统里，最容易混淆的是 SDK 和业务系统的职责边界。

如果边界不清楚，SDK 很容易走向两个极端。

一种极端是只提供几个日志开关，让使用者自己去拼接所有过程数据。这样虽然简单，但 SDK 没有真正承担起链路观测的职责。

另一种极端是 SDK 试图内置所有指标计算，甚至直接判断“这次回答是否成功”“用户是否满意”。这样看起来功能完整，但实际上会把业务口径写死在 SDK 里。

更合理的边界是：

SDK 负责记录技术事实，业务系统负责解释业务含义。

所谓技术事实，是 SDK 在执行链路时天然可以知道的信息。

例如 runtime 在一次查询过程中，可以知道：

1. 原始 query 是什么
2. effective query 是什么
3. 检索阶段召回了多少候选
4. 检索后处理阶段选择了哪些 chunk
5. 生成阶段耗时多久
6. 最终 trace 是成功还是失败
7. 如果失败，失败发生在哪个阶段

这些信息不需要业务系统额外推断。因为它们本来就发生在 SDK 内部。

所以，SDK 应该主动把这些信息记录成结构化事件：

emitEvent({
  stage: "retrieval",
  name: "runtime.retrieval.complete",
  durationMs: 36,
  attributes: {
    candidateCount: 8,
    emptyRetrieval: false,
  },
});

这类事件表达的是事实：检索完成了，耗时 36ms，召回了 8 个候选，结果不是空的。

但 SDK 不应该基于这个事件直接判断“这次回答质量好”。因为候选数量足够，并不代表候选内容正确；检索不为空，也不代表最终回答能解决用户问题。

业务含义通常要由业务系统根据具体场景判断。

例如，同样是一次 runtime.generation.complete：

emitEvent({
  stage: "generation",
  name: "runtime.generation.complete",
  durationMs: 820,
  attributes: {
    contextLength: 2400,
    answerPreview: "根据公司当前年假政策...",
  },
});

SDK 能记录生成耗时、上下文长度和答案摘要，但它无法只凭这些字段判断：

1. 答案是否让用户满意
2. 用户是否还需要继续追问
3. 答案是否满足某个业务审核标准
4. 这次回答是否应该计入业务成功率

这些判断都依赖业务口径。

比如在企业知识库里，用户点踩可能代表答案没有解决问题；但在学习系统里，用户继续追问不一定是坏事，它可能代表用户正在深入学习。SDK 如果把这些行为直接解释成固定指标，就会限制业务系统自己的分析方式。

因此，SDK 更适合提供“关联能力”，而不是替业务系统定义所有指标。

一次 runtime 请求可以由 SDK 生成或接收一个 traceId：

const result = await runtime.run(
  { query: "公司年假政策是什么？" },
  {
    trace: {
      traceId: "trace-001",
      tags: {
        app: "internal-kb",
      },
    },
  },
);

业务系统后续产生的反馈、点击、追问等事件，都可以通过同一个 traceId 和这次 RAG 链路关联起来：

observer.onEvent?.({
  traceId: "trace-001",
  scope: "runtime",
  stage: "feedback",
  name: "runtime.feedback.submit",
  timestamp: new Date().toISOString(),
  attributes: {
    rating: "thumbs_down",
    reason: "missing_policy_detail",
  },
});

这样设计之后，SDK 不需要知道 thumbs_down 在业务上如何计分，也不需要知道 missing_policy_detail 应该归到哪一类反馈。它只需要保证这条业务事件能够和原来的 RAG trace 放在同一条链路里。

从职责上看，SDK 应该覆盖这些能力：

1. 定义统一的 trace、event、error 协议
2. 在 runtime 和 indexing 的关键阶段发出事件
3. 记录阶段耗时、候选数量、筛选结果、错误信息
4. 提供 observer，让外部可以接收这些事件
5. 提供 exporter，把 trace 输出到 console、memory、JSONL 或后续其他系统
6. 保证观测失败不会影响主流程

业务系统则应该负责：

1. 定义满意度、追问率、点击率等业务指标口径
2. 在前端或服务端上报业务行为
3. 决定哪些反馈属于正向反馈，哪些属于负向反馈
4. 在数据平台里把业务事件和 RAG trace 进行分析
5. 根据业务目标设计 dashboard 和报警规则

这样的分工可以让 SDK 保持稳定。

SDK 的事件协议越稳定，业务系统越容易在不同项目里复用它；业务指标的计算口径留在业务层，也不会因为某个产品的规则变化而迫使 SDK 修改核心设计。

所以，在 RAG SDK 里设计观测系统时，重点不是把所有指标都算出来，而是先把链路内部的事实数据稳定地暴露出来，并为业务事件留下可以关联的位置。

5. SDK 实现

明确职责之后，就可以进入 SDK 内部实现。

从整体思路上看，观测系统不应该侵入每一个业务实现。它更适合被放在 runtime 和 indexing 这类流程编排层里，由编排层在关键阶段发出事件。

也就是说，具体的 retriever、generator、embedder 不需要自己知道观测系统怎么工作。它们只要完成自己的职责，外层 pipeline 负责记录阶段开始、阶段完成、阶段失败，以及最终 trace。

这和当前 SDK 的设计是一致的。

5-1. 事件协议

SDK 首先需要定义统一的事件结构。

简化之后，核心字段大概是这样：

type RAGEvent = {
  traceId: string;
  scope: "runtime" | "indexing";
  stage: string;
  name: string;
  timestamp: string;
  durationMs?: number;
  attributes?: Record<string, unknown>;
};

这里的几个字段各有用途：

1. traceId 用来串起同一次链路
2. scope 表示事件来自 runtime 还是 indexing
3. stage 表示事件发生在哪个阶段
4. name 表示具体事件名称
5. durationMs 表示阶段耗时
6. attributes 放阶段相关的扩展信息

最终 trace 则负责汇总一次链路：

type RAGTrace = {
  traceId: string;
  scope: "runtime" | "indexing";
  startedAt: string;
  endedAt?: string;
  durationMs?: number;
  status: "ok" | "error";
  events: RAGEvent[];
  errors?: RAGErrorRecord[];
};

这两个结构决定了观测数据的基本形态：运行过程中不断产生 event，运行结束后形成一个完整 trace。

5-2. runtime 埋点

runtime 的主流程本身已经是一条清晰的链路：

query receive
  -> query preprocess
  -> retrieval
  -> post retrieval
  -> generation
  -> run complete

因此，runtime 观测的实现方式就是在这些阶段附近发出事件。

一个大幅简化后的示意代码如下：

async function runRuntime(input, options) {
  const trace = createRuntimeTrace(input, options);

  await emitEvent(trace, {
    stage: "query",
    name: "runtime.query.receive",
    attributes: {
      query: input.query,
    },
  });

  const request = await measureStage("pre-retrieval", async () => {
    return preprocessor.preprocess(input);
  });

  await emitEvent(trace, {
    stage: "query",
    name: "runtime.query.preprocess",
    durationMs: trace.timings["pre-retrieval"],
    attributes: {
      originalQuery: request.originalQuery.query,
      effectiveQuery: request.effectiveQuery.query,
      rewriteReason: request.rewriteReason,
    },
  });

  const retrievalResult = await measureStage("retrieval", async () => {
    return retriever.retrieve(request);
  });

  await emitEvent(trace, {
    stage: "retrieval",
    name: "runtime.retrieval.complete",
    durationMs: trace.timings.retrieval,
    attributes: {
      candidateCount: retrievalResult.candidates.length,
      emptyRetrieval: retrievalResult.candidates.length === 0,
    },
  });

  const postResult = await measureStage("post-retrieval", async () => {
    return postprocessor.postprocess({
      request,
      candidates: retrievalResult.candidates,
    });
  });

  await emitEvent(trace, {
    stage: "post_retrieval",
    name: "runtime.post_retrieval.select",
    durationMs: trace.timings["post-retrieval"],
    attributes: {
      inputCandidateCount: retrievalResult.candidates.length,
      selected: postResult.chunks.length,
      selectedChunkIds: postResult.chunks.map((chunk) => chunk.id),
    },
  });

  const generationResult = await measureStage("generation", async () => {
    return generator.generate({
      request,
      chunks: postResult.chunks,
      promptContext: postResult.promptContext,
    });
  });

  await emitEvent(trace, {
    stage: "generation",
    name: "runtime.generation.complete",
    durationMs: trace.timings.generation,
    attributes: {
      contextLength: postResult.promptContext?.length ?? 0,
      answerPreview: generationResult.answer.slice(0, 200),
    },
  });

  await endTrace(trace, "ok");

  return generationResult;
}

这段代码不是当前 SDK 的完整实现，但行为和当前 SDK 是一致的。

它表达了三个关键点：

1. 每个阶段先执行原本的业务逻辑
2. 阶段完成后把关键事实写入 event
3. 整条链路结束后通过 endTrace() 产出最终 trace

当前 SDK 里的 measureStage() 只负责记录耗时，不负责自动发事件。事件是在阶段完成后由 emitEvent() 显式发出的。这样做的好处是，每个阶段可以按自己的结果补充不同的 attributes。

5-3. indexing 埋点

indexing 的观测方式和 runtime 类似，只是阶段换成了离线索引流程。

简化后的索引链路大致如下：

run start
  -> load
  -> transform
  -> filter
  -> chunk
  -> metadata
  -> embed
  -> store
  -> run complete

由于 indexing 阶段更多，当前 SDK 把阶段执行封装成了类似 runStage() 的结构。

简化示意如下：

async function runStage(stage, task, context) {
  const startedAt = Date.now();

  await emitEvent({
    stage,
    name: `indexing.${stage}.start`,
    attributes: context,
  });

  try {
    const result = await task();

    await emitEvent({
      stage,
      name: `indexing.${stage}.complete`,
      durationMs: Date.now() - startedAt,
      attributes: buildStageAttributes(stage, result, context),
    });

    return result;
  } catch (error) {
    await emitEvent({
      stage,
      name: `indexing.${stage}.fail`,
      durationMs: Date.now() - startedAt,
      attributes: {
        documentId: context.documentId,
        chunkId: context.chunkId,
      },
    });

    await emitError(stage, error, context);

    throw error;
  }
}

这样一来，主流程可以保持相对干净：

const documents = await runStage("load", () => loader.load(), {
  mode,
});

const chunks = await runStage("chunk", () => chunker.chunk(document), {
  documentId: document.id,
  mode,
});

const vectors = await runStage("embed", () => embedder.embed(chunks), {
  documentId: document.id,
  chunkCount: chunks.length,
});

await runStage("store", () => store.upsert(vectors), {
  documentId: document.id,
  vectorCount: vectors.length,
});

当前 SDK 的实际实现比这更完整，会处理 document transformer、chunk transformer、metadata extractor、chunk filter、批量 embedding、错误上下文等细节。但核心思路不变：每个阶段执行前后都可以发事件，失败时发 fail event 和 error record。

5-4. observer

事件产生之后，还需要有一个统一的接收入口。

这个入口就是 RAGObserver：

type RAGObserver = {
  onEvent?(event: RAGEvent): void | Promise<void>;
  onError?(error: RAGErrorRecord): void | Promise<void>;
  onTraceEnd?(trace: RAGTrace): void | Promise<void>;
  flush?(): void | Promise<void>;
  shutdown?(): void | Promise<void>;
};

runtime 和 indexing 不需要关心事件最终写到哪里。它们只需要在合适的位置调用 observer：

async function emitEvent(trace, event) {
  const resolvedEvent = {
    traceId: trace.traceId,
    scope: trace.scope,
    timestamp: new Date().toISOString(),
    ...event,
  };

  trace.events.push(resolvedEvent);

  await notifyObserver(() => {
    return observer.onEvent?.(resolvedEvent);
  });
}

这样，SDK 内部的采集逻辑和外部的导出逻辑就解耦了。

调用方可以不传 observer，此时观测不会产生额外行为；也可以传入 console、memory、JSONL 或其他自定义 observer。

5-5. 失败隔离

观测系统不能影响主流程。

如果一次 RAG 请求本来已经成功，但因为写 trace 失败导致请求失败，这个设计就是不合理的。

所以，当前 SDK 会把 observer 调用包在安全执行逻辑里：

async function notifyObserver(callback) {
  try {
    await callback?.();
  } catch {
    // observer 失败不影响 runtime / indexing 主流程
  }
}

这也是 SDK 观测实现里很重要的一条工程原则：观测失败可以丢失一部分诊断数据，但不能改变业务链路本身的结果。

6. 包结构

观测能力最终被收敛到独立的 observability 包里。

这个包不负责执行 RAG 流程，也不负责接入具体业务系统。它主要提供协议、observer、exporter 和少量工具函数。

当前结构可以简化理解为：

packages/observability/
  src/
    types/
      rag-event.ts
      rag-trace.ts
      rag-metric.ts
      rag-error-record.ts
      trace-context.ts
    observer/
      rag-observer.ts
      noop-observer.ts
      console-observer.ts
      create-rag-observer.ts
    exporters/
      trace-exporter.ts
      console-exporter.ts
      memory-trace-exporter.ts
      jsonl-trace-exporter.ts
      http-trace-exporter.ts
    redaction/
      redact-trace.ts
      redaction-options.ts
    sampling/
      should-sample-trace.ts
      sampling-options.ts
    formatters/
      console-output.ts
    utils/
      invoke-observer-safely.ts
    errors/
      observability-internal-error.ts
    index.ts

6-1. types

types 是观测协议层。

这里定义的是跨包共享的基础结构，例如：

1. RAGEvent
2. RAGTrace
3. RAGMetric
4. RAGErrorRecord
5. TraceContext

runtime 和 indexing 发出的事件都依赖这些结构。这样做可以保证不同包产出的观测数据形态一致。

其中 RAGMetric 当前更多是协议预留：

type RAGMetric = {
  traceId: string;
  name: string;
  value: number;
  unit?: "ms" | "count" | "tokens" | "ratio" | "bytes";
  scope?: "runtime" | "indexing";
  stage?: string;
  attributes?: Record<string, unknown>;
};

当前 SDK 的主要数据还是通过 event 和 trace 输出，metric 流还没有成为主路径。

6-2. observer

observer 是采集入口层。

它包括三类能力：

1. RAGObserver：统一接口
2. NoopObserver、createConsoleObserver()：基础 observer
3. createRAGObserver()：组合 observer

其中 createRAGObserver() 是当前更重要的实现。它会接收 event 和 error，并在 trace 结束时把最终 trace 交给 exporter。

示意代码如下：

const observer = createRAGObserver({
  serviceName: "rag-demo",
  environment: "local",
  exporters: [
    createConsoleExporter(),
    createJsonlTraceExporter({
      filePath: "./rag-trace.jsonl",
    }),
  ],
});

这个 observer 可以同时对接多个 exporter。runtime 或 indexing 只需要把事件交给 observer，不需要感知具体的导出方式。

6-3. exporters

exporters 是导出层。

当前已经有四个基础 exporter：

1. createConsoleExporter()：把 trace 摘要输出到终端
2. createMemoryTraceExporter()：把 trace 存在内存里，适合测试
3. createJsonlTraceExporter()：把 trace 按 JSONL 写入本地文件
4. createHttpTraceExporter()：通过 HTTP 把 trace 发送到自建后端或网关

JSONL exporter 适合本地调试、demo 记录和离线分析。HTTP exporter 则让 trace 可以离开本地文件，进入后续集中化收集链路。

简化使用方式如下：

const observer = createRAGObserver({
  exporters: [
    createJsonlTraceExporter({
      filePath: "./.artifacts/runtime-trace.jsonl",
      append: false,
    }),
  ],
});

运行结束后，trace 会以一行一个 JSON 对象的形式写入文件：

{"traceId":"trace-001","scope":"runtime","durationMs":1280,"status":"ok","events":[]}

这种格式不复杂，但很适合本地调试、demo 记录和后续离线分析。

6-4. redaction 和 sampling

当 trace 可以被导出之后，还需要考虑两个工程问题：哪些内容能导出，以及导出多少。

redaction 解决的是敏感信息问题。RAG trace 里可能包含 query、answerPreview、promptContext、chunk content、metadata 等字段。如果这些内容直接写入文件或发送到远程端点，就可能带来数据风险。

当前 SDK 已经提供了 redactTrace()，并且 createRAGObserver() 可以在 exporter 之前执行脱敏：

const observer = createRAGObserver({
  exporters: [createJsonlTraceExporter({ filePath })],
  redact: {
    fields: ["events[].attributes.candidates[].metadata.secret"],
    maskContent: true,
    contentPreviewLength: 200,
  },
});

sampling 解决的是数据量问题。生产环境里不一定要导出每一条成功 trace，但错误 trace 通常应该保留。

当前 SDK 的 sampling 是 trace 级别的：

const observer = createRAGObserver({
  exporters: [createHttpTraceExporter({ url })],
  sampling: {
    rate: 0.1,
    alwaysSampleOnError: true,
  },
});

这两个能力不改变 runtime 或 indexing 的主流程，但会让观测数据更适合长期运行和外部导出。

6-5. utils 和 errors

utils 里主要放观测系统自己的辅助逻辑，例如安全调用 observer 或 exporter。

这类逻辑虽然不复杂，但很关键。因为 observability 包本身要遵守前面提到的原则：观测失败不能影响主流程。

errors 里则放包内错误类型。当前只保留了最小实现，避免过早设计复杂错误体系。

6-6. 对外入口

src/index.ts 负责统一导出公开能力：

export * from "./types/index.js";
export * from "./observer/index.js";
export * from "./exporters/index.js";
export * from "./redaction/index.js";
export * from "./sampling/index.js";
export * from "./errors/index.js";

对使用者来说，最小接入通常只需要从 @rag-sdk/observability 导入 observer 和 exporter：

import {
  createHttpTraceExporter,
  createJsonlTraceExporter,
  createRAGObserver,
} from "@rag-sdk/observability";

这样的包结构让 observability 保持在一个相对稳定的位置：它不参与 runtime 或 indexing 的具体流程，但为这些流程提供统一观测协议和导出能力。

7. 当前状态

到目前为止，当前 SDK 的观测系统已经完成了第一阶段最核心的事情：让一次 runtime 或 indexing 流程变得可追踪。

7-1. 当前能力

具体来说，当前已经具备这些能力：

1. 定义了统一的观测协议，包括 RAGEvent、RAGTrace、RAGMetric、RAGErrorRecord
2. 定义了 RAGObserver 接口，用来接收 event、error 和 trace
3. 提供了 createRAGObserver()，可以把事件缓存起来，并在 trace 结束时交给 exporter
4. 提供了 console、memory、JSONL、HTTP 四种基础 exporter
5. 提供了 redaction 和 sampling，让 trace 可以在导出前被脱敏和采样
6. runtime 已经能在 query、retrieval、post retrieval、generation 等阶段发出事件
7. indexing 已经能在 load、chunk、embed、store 等阶段发出事件
8. eval 已经能从 JSONL trace 中提取 error、empty_retrieval、high_drop_ratio 等最小质量信号
9. observer 或 exporter 失败不会打断主流程

所以，当前观测系统最适合解决的是“单次链路如何解释”的问题。

例如一次 runtime 请求结束后，可以得到类似这样的 trace：

{
  traceId: "trace-001",
  scope: "runtime",
  durationMs: 1280,
  status: "ok",
  events: [
    {
      name: "runtime.query.preprocess",
      stage: "query",
      durationMs: 12,
      attributes: {
        originalQuery: "公司年假怎么算？",
        effectiveQuery: "公司年假政策 计算规则",
      },
    },
    {
      name: "runtime.retrieval.complete",
      stage: "retrieval",
      durationMs: 36,
      attributes: {
        candidateCount: 8,
      },
    },
    {
      name: "runtime.post_retrieval.select",
      stage: "post_retrieval",
      durationMs: 9,
      attributes: {
        selected: 3,
        selectedChunkIds: ["chunk-1", "chunk-3", "chunk-7"],
      },
    },
    {
      name: "runtime.generation.complete",
      stage: "generation",
      durationMs: 820,
      attributes: {
        answerPreview: "根据公司当前年假政策...",
      },
    },
  ],
}

这类 trace 已经可以帮助开发者回答很多问题：

1. query 是否发生过改写
2. 检索是否为空
3. 召回候选数是否合理
4. 最终哪些 chunk 进入了上下文
5. 哪个阶段耗时最长
6. 失败时错误发生在哪个阶段

但这还不是一个完整的生产级观测系统。

7-2. 当前缺口

当前缺口主要集中在以下几类能力上：

能力	当前状态	后续方向
metrics 采集层	已有 RAGMetric 类型，但主链路主要还是 event / trace	增加标准 metrics 汇总，例如 token、耗时、drop ratio、空召回率
retrieval 证据细节	已记录候选数量和空召回	补候选摘要、score、sourceId、metadata preview、rerank 前后变化
generation 细节	已记录 contextLength 和 answerPreview	补 model、promptTokens、completionTokens、totalTokens、finishReason、cost
eval 质量评估	已能提取 error、empty_retrieval、high_drop_ratio 样本	补低置信、source coverage、引用覆盖率、忠实性等评测信号
生产集成	已有 JSONL / HTTP exporter、redaction、sampling	后续接入 OpenTelemetry、日志平台、APM、dashboard、告警和 retention
业务事件	已有 traceId、requestId、attributes 这类关联基础	沉淀反馈、引用点击、追问等业务事件协议

7-3. 演进路径

后续演进可以按阶段推进：

阶段	目标	关键能力
第一阶段：看清单次请求	让每一次 runtime / indexing 链路可复盘	trace 完整、阶段 event、错误阶段、JSONL 样本
第二阶段：看清系统状态	从单次 trace 进入系统级指标	平均耗时、P95、P99、QPS、错误率、超时率、dashboard、报警
第三阶段：接入业务反馈	把用户行为和 RAG 链路关联起来	点赞、点踩、引用点击、继续追问、人工纠错、traceId 关联
第四阶段：补齐生产治理	让观测系统适合长期运行在真实业务环境	redaction、sampling、retention、HTTP exporter、权限控制、数据合规
第五阶段：进入质量评估	从 trace 中沉淀可复盘样本，形成优化闭环	empty retrieval、high drop ratio、low confidence、source coverage、人工评测

从这个角度看，当前实现选择先做 event、trace、observer 和 exporter 是合理的。

这些能力是观测系统的地基：

1. 没有稳定的事件协议，后续无法做聚合
2. 没有 traceId，业务事件无法关联
3. 没有 exporter，观测数据无法离开 SDK
4. 没有失败隔离，观测系统反而会影响主链路

所以，RAG SDK 的观测系统不需要一开始就追求大而全。更稳妥的路径是先让每一次请求可解释，再逐步扩展到系统级指标、业务效果分析和生产治理。

8. 写在最后

到这里，RAG SDK 中观测系统的基本设计就比较清楚了。

它的核心目标不是替业务系统计算所有指标，也不是在 SDK 里内置一套完整监控平台，而是先把 RAG 链路内部发生的事情结构化地暴露出来。

对于 runtime 来说，观测系统要让一次在线查询变得可解释：query 有没有改写，检索有没有召回，检索后处理选了哪些 chunk，生成阶段耗时多久，失败发生在哪一步。

对于 indexing 来说，观测系统要让一次离线索引变得可复盘：加载了多少文档，切出了多少 chunk，生成了多少向量，写入是否成功，失败时关联到哪个 document 或 chunk。

这些信息本质上都是 SDK 执行链路时产生的技术事实。只有这些事实被稳定记录下来，后续的性能分析、问题排查、业务反馈关联和生产治理才有基础。

所以，一个面向 SDK 的观测系统，最重要的不是一开始就做多少 exporter，或者立刻支持多少 dashboard，而是先把几个基础问题解决好：

1. 是否有统一的事件协议
2. 是否能用 traceId 串起一次链路
3. 是否能记录关键阶段的耗时和结果
4. 是否能在失败时保留错误上下文
5. 是否能把观测数据交给外部系统
6. 是否能保证观测失败不影响主流程

当前实现已经完成了这条路径上的第一步，并且已经向后续阶段推进：通过 event、trace、observer 和 exporter，让 runtime 与 indexing 的核心流程具备了最小可用的观测能力；通过 redaction、sampling、HTTP exporter 和 eval trace sample，让观测数据具备了继续进入分析闭环的基础。

后续无论是接入监控平台、补全 metrics 聚合，还是扩展业务反馈事件和质量评测，本质上都是在这套基础协议之上继续增强。

这也是 RAG SDK 设计里一个很重要的原则：先把链路抽象稳定下来，再逐步增强能力。观测系统同样如此。只有先让每一次请求可解释，整个 RAG 系统才有继续优化的依据。

---

-EOF-