跳到主要内容

数据截至 (上游 commit 0261ea4f33d4)

Opik — 架构与原理

30 秒导读: Opik 是一套自托管的 LLM 应用可观测 + 评测平台。你在 Python 函数上加一个 @track 装饰器,它就把这次调用录成一棵 trace/span 树,异步批量送进后端的 ClickHouse;之后你可以拿数据集离线跑实验打分、给生产流量挂在线评估规则、甚至让优化器自动改写你的 prompt。一句话:给 LLM 应用装上「行车记录仪 + 考官 + 调参师」。


1. 这是什么(零基础也能懂)

一句话定义: Opik 是「LLM 应用的记录仪与考卷系统」——把每次模型调用的输入输出录下来,再用一套指标反复给这些记录打分。

它解决的真实痛点。 传统程序出错会抛异常,你看堆栈就行。LLM 应用不会抛异常——它只是答得不好。链路里有检索、有工具调用、有三层 agent 嵌套,你想知道「到底哪一步开始跑偏」,光看日志是看不出来的;就算看出来了,你改完 prompt 也无法证明「这次是真变好了,不是碰巧」。

Opik 把这件事拆成两个可工程化的动作:

  • 记录:把一次请求的完整调用树(谁调了谁、各自的输入输出、耗时、token、成本)结构化存下来。
  • 打分:用指标给这些记录打分——离线用固定数据集比版本,在线按采样率给生产流量打分。

给谁用: 在搭 RAG、agent、聊天机器人的工程师,以及要给这些系统做质量守门的团队。

它能做什么(功能速览):

能力说明
链路追踪@track 装饰器 + 17 个框架的自动埋点(OpenAI / LangChain / LlamaIndex / CrewAI / ADK / OTel…),见 sdks/python/src/opik/integrations/
数据集与实验把评测样本存成 dataset,一次 evaluate() 跑成一个 experiment,版本间可比
指标库启发式(BLEU / ROUGE / Levenshtein / 正则…)与 LLM 裁判(幻觉、审核、G-Eval、轨迹准确性…)两大类
在线评估给生产项目挂规则,按采样率自动给新 trace 打分,打分器可以是 LLM 裁判或你写的 Python 代码
Prompt 优化独立 SDK opik_optimizer,用 meta-prompt / 进化 / 贝叶斯少样本等算法自动改写 prompt
自托管一条 ./opik.sh 起全套:MySQL + Redis + ClickHouse + MinIO + Java 后端 + React 前端

用起来什么样。 最小的一次追踪,就是给函数加一行装饰器:

# 示意,非源码:Opik 最小追踪用法
import opik

@opik.track # 这层会建一个 span;最外层的那个还会建 trace
def retrieve(question: str) -> list[str]:
return search_index(question)

@opik.track
def answer(question: str) -> str:
docs = retrieve(question) # 嵌套调用自动变成子 span
return llm(f"{docs}\n\n{question}")

answer("Opik 是什么?") # 一次调用 = 一棵 trace 树,异步送到后端

重点看:你什么都不用手工传——父子关系由装饰器自己从上下文里推出来(sdks/python/src/opik/decorator/span_creation_handler.py:38create_span_respecting_context)。

一句话直觉: 把 Opik 当成「LLM 版的 APM + 单元测试」。APM 那半负责「这次请求经过了哪些环节、各花了多少」;单元测试那半负责「这次输出到底合不合格」。Opik 的全部价值,在于让这两半共用同一份数据结构(trace/span)——录下来的痕迹,直接就是考卷。


2. 顶层全景(它大概怎么转)

2.1 一条 trace 的一生

先看数据怎么从你的进程流到数据库。从左到右是时间顺序,中间不阻塞你的业务代码。

① 装饰函数 ② 进程内组树 ③ 异步上报 ④ 服务端落库
┌────────────┐ ┌───────────────┐ ┌────────────────┐ ┌──────────────┐
│ @track │───►│ contextvars │───►│ 批处理器 → 队列 │───►│ ClickHouse │
│ 包住你的函数│ │ span 栈 │ │ 4 个后台消费线程│ │ Replacing │
│ │ │ → trace/span 树│ │ 满 1000 条即发 │ │ MergeTree │
└────────────┘ └───────────────┘ └────────────────┘ └──────────────┘

连不上后端时

┌──────────────┐
│ SQLite 暂存 │
│ 恢复后自动重放 │
└──────────────┘

怎么读这张图: ①②发生在你的调用栈里(微秒级,只是往栈上压一个对象);③在独立线程里跑,业务代码不等它;④是 Java 后端把消息写进 ClickHouse。断网只影响③④,不影响你的程序继续跑。

2.2 部件一句话职责

部件干什么在哪
@track / OpikTrackDecorator包住函数,调用前建 span、调用后收尾sdks/python/src/opik/decorator/tracker.py:11
OpikContextStorage用 contextvars 存「当前 trace + span 栈」,线程/协程各自隔离sdks/python/src/opik/context_storage.py:11
Streamer + MessageQueue进程内的上报管道:预处理 → 入队 → 后台消费sdks/python/src/opik/message_processing/streamer.py:20
BaseBatcher 家族按「条数 / 时间 / 载荷 MB」三重条件切批sdks/python/src/opik/message_processing/batching/base_batcher.py:11
ReplayManager + SQLite后端不可达时落盘,连上后重放sdks/python/src/opik/message_processing/replay/db_manager.py:97
Java 后端 SpanDAO / TraceDAO把消息写进 ClickHouse,处理部分更新的字段级合并apps/opik-backend/src/main/java/com/comet/opik/domain/SpanDAO.java:174
evaluate() + EvaluationEngine离线跑「数据集 × 任务 × 指标」,产出 experimentsdks/python/src/opik/evaluation/evaluator.py:137
BaseMetric 家族所有打分器的统一契约:score() 返回 ScoreResultsdks/python/src/opik/evaluation/metrics/base_metric.py:9
在线评分器消费 Redis 流,按规则给生产 trace 打分apps/opik-backend/src/main/java/com/comet/opik/api/resources/v1/events/OnlineScoringBaseScorer.java:45
Python 沙箱在受限 Docker 容器里执行用户自定义指标代码apps/opik-python-backend/src/opik_backend/executor_docker.py:103
opik_optimizer反复跑实验、改写 prompt、留下分最高的那版sdks/opik_optimizer/src/opik_optimizer/base_optimizer.py:1618

2.3 打分的三条回路

记录只是原料。Opik 用同一份 trace 数据喂三条不同的打分回路:

回路什么时候触发打分器产物
离线实验你手动跑 evaluate()任意 BaseMetricexperiment + experiment items + feedback scores
在线评估生产 trace 落库后,按采样率自动触发LLM 裁判 或 用户 Python 代码挂在原 trace 上的 feedback scores
自动优化你跑 optimize_prompt()复用离线实验一版分更高的 prompt

在线这条最值得看结构——它把「打分」做成了后端里的一条异步流:

新 trace 落库 ──► 规则匹配 ──► 随机采样 ──► Redis 流 ──► 打分器 ──► feedback_scores
(filters) (samplingRate) LLM 裁判
或 Python 沙箱

采样那一步就是一行 secureRandom.nextFloat() >= evaluator.getSamplingRate() 直接丢弃(apps/opik-backend/src/main/java/com/comet/opik/api/resources/v1/events/OnlineScoringSampler.java:357)——生产量大时不可能每条都送去问 LLM。

2.4 主线走一遍(不进代码)

  1. 你的函数被 @track 包住。调用进来时,装饰器看当前 contextvars:栈空 → 建一个 trace + 根 span;栈非空 → 拿栈顶当父,建子 span
  2. 函数返回后,装饰器给 span 填上 output/耗时/错误,弹栈,把 CreateSpanMessage 交给 Streamer
  3. Streamer 先做附件抽取,再交给对应类型的批处理器攒着;攒够 1000 条或过了 2 秒,打成一个批消息塞进队列。
  4. 4 个后台消费线程从队列取批消息,走 REST 发给 Java 后端;被限流就按 retry_after 推迟并放回队列;后端彻底连不上就落 SQLite 等重放。
  5. Java 后端把批写进 ClickHouse。因为 span 的「开始」和「结束」是两条独立消息、还可能乱序到达,写入 SQL 里逐字段做 multiIf 合并(老值非空就保老值)。
  6. 之后:你跑 evaluate() 拿数据集重放任务并打分;或者给项目挂在线规则,让新来的 trace 被自动抽样打分;或者让 optimizer 拿实验分数当目标函数去改 prompt。

3. 阅读地图

建议按顺序读——前三章是「数据怎么进来」,后三章是「数据怎么被用」。

顺序章节读完你会知道
1追踪层:@track 怎么把一次函数调用变成 span 树contextvars 栈怎么撑起父子关系、生成器/异步函数怎么处理、分布式 trace 头怎么跨进程接上
2上报层:从内存对象到后端的异步管道(批量、限流、断线续传)三重切批条件、限流退避、有界队列的丢弃策略、SQLite 重放
3服务端存储:ClickHouse 怎么吞下乱序到达的 trace 与 spanReplacingMergeTree 选型、字段级 multiIf 合并、冲突字段用星号填充的处理
4评估层:dataset × task × metric 怎么跑成一次 experimentevaluate() 的参数语义、并发执行器、scoring_key_mapping、断点续跑
5打分器:启发式指标与 LLM 裁判的两套工程范式BaseMetric 契约、G-Eval 的 logprobs 加权、结构化输出解析
6生产闭环:在线评估规则、Python 沙箱与 prompt 自动优化Redis 流式打分、Docker 沙箱的资源与网络限制、优化器的模板方法骨架

如果你只想读一章: 做可观测选 01/02;做后端选 03;做评测选 04/05;做 prompt 工程选 06。


4. 巧妙之处(可借鉴的技术)

这一节是精华。每条先说「妙在哪」,再给源码位置。

4.1 用不可变元组存 contextvars 栈

妙在哪: contextvars 在协程/线程间是「拷贝引用」语义。如果栈用 list,子协程 append 会污染父上下文。Opik 全程用 tuple,压栈是 stack + (new,)、弹栈是 stack[:-1]——每次都产生新对象,天然隔离。

类里甚至把这条规矩写进了 docstring 当纪律(sdks/python/src/opik/context_storage.py:11OpikContextStorage)。配套还给了 trim_span_data_stack_to_certain_span(同文件 :59)——回调式集成里容易漏掉 pop,这个方法能把「挂住的 span」一次性削掉。

4.2 攒批时顺手做「开始/结束消息去重」

妙在哪: 一个 span 会先发「创建」再发「结束更新」两条消息。如果两条都还在同一个批里,发两条纯属浪费。批处理器在 add() 时检查:新消息带 end_time,就把批里同 span_id 的旧消息删掉,只发合并后的那条。

sdks/python/src/opik/message_processing/batching/batchers.py:39-42CreateSpanMessageBatcher.add_remove_matching_messages),trace 侧同理(同文件 :76-79)。

4.3 估算 JSON 体积而不真去序列化

妙在哪: 切批要按载荷大小(默认 50 MB 上限)算,但对每条消息 json.dumps() 一遍只为量体积,CPU 和内存都吃不消。Opik 写了个递归的尺寸估算器:字符串按 UTF-8 字节数 +2(引号),数字按字符串长度,dict/list 按结构加分隔符——不产生任何中间字符串。

sdks/python/src/opik/message_processing/batching/sequence_splitter.py:25_get_json_size)。估算失败时返回 float("inf"),让这条消息单独成批——保守但安全(同文件 :53-56)。

4.4 有界队列的记账修正

妙在哪: 队列用 collections.deque(maxlen=N),满了 appendleft静默丢掉另一端的元素。这会让「未完成任务计数」永远归不了零,flush() 就会卡死。Opik 在 put() 里比较入队前后长度,只有真的变长才 +1;put_back() 里反过来,发现长度没变就 -1,因为被挤掉的那条消息永远不会 task_done() 了。

sdks/python/src/opik/message_processing/message_queue.py:35-64MessageQueue.put / put_back),注释把这个坑写得很清楚。

4.5 限流不是重试,是「给消息盖一个投递时间戳」

妙在哪: 遇到 429 时,常见做法是当场 sleep(retry_after)——但那会把整个消费线程堵住。Opik 的做法是给消息设一个 delivery_time,放回队列尾部;消费线程每次取出消息先看 delivery_time <= now,没到点就再放回去。这样限流期间线程还能处理别的消息,且保持了顺序

sdks/python/src/opik/message_processing/queue_consumer.py:57-70QueueConsumer._loopOpikCloudRequestsRateLimited 分支)。

4.6 ClickHouse 里做字段级「先到先得」合并

妙在哪: ClickHouse 没有 UPDATE。span 的开始和结束是两次写入,还可能乱序。Opik 选了 ReplacingMergeTree(last_updated_at)apps/opik-backend/src/main/resources/liquibase/db-app-analytics/migrations/000001_init_script.sql:24),并且每次插入都先 LEFT JOIN 查一遍旧行,然后逐字段写 multiIf:老值非空就用老值,否则用新值。

-- 真实源码片段,SpanDAO.java:234-237
multiIf(
LENGTH(old_span.input) > 0, old_span.input,
new_span.input
) as input,

为什么是「老值优先」而不是「新值优先」: 因为「结束消息」通常不携带 input,如果新值优先就会把 input 抹成空。老值优先等于「只补空洞、不覆盖已有信息」。

更狠的是冲突检测:如果新旧 project_id/trace_id 不一致(说明有人复用了 id),直接写成一串星号 leftPad('', 40, '*') 而不是二选一——把脏数据标记出来,而不是悄悄猜(apps/opik-backend/src/main/java/com/comet/opik/domain/SpanDAO.java:207-217)。

4.7 G-Eval 用 top_logprobs 把离散分变成连续分

妙在哪: 让 LLM 打 1–10 分,输出是离散的,而且很跳(同一段文本这次 7 分下次 8 分)。G-Eval 的做法是请求 logprobs=True, top_logprobs=20,拿到「模型认为下一个 token 是 1..10 各自的概率」,做概率加权平均,再除以 10 归一到 [0,1]。分数因此是连续的、也更稳。

sdks/python/src/opik/evaluation/metrics/llm_judges/g_eval/metric.py:227-229(请求侧)和 sdks/python/src/opik/evaluation/metrics/llm_judges/g_eval/parser.py:71-93linear_probs_sum / weighted_score_sum 加权)。模型不支持 logprobs 时自动降级为直接取分数(metric.py:146-149 探测 supported_params)。

4.8 指标自己也被追踪

妙在哪: BaseMetric.__init__ 里,如果 track=True,就用 opik.track(...) 把自己的 score / ascore 方法包一层。于是「裁判 LLM 调了什么、花了多少钱」也变成 span,跟被评的那条 trace 一起可见。

sdks/python/src/opik/evaluation/metrics/base_metric.py:50-53。这是个很好的自举例子:观测系统用自己观测自己

4.9 沙箱里给 opik 包做假导入

妙在哪: 用户自定义指标常写 from opik.evaluation.metrics import BaseMetric。但真的 import opik 要 2.5 秒(pydantic-settings、REST client、40 多个指标类),而沙箱的默认执行超时只有 3 秒(apps/opik-python-backend/src/opik_backend/executor.py:47,compose 与 Dockerfile 里也都是 3)——光导入就把整次执行拖死。

Opik 的解法:在 sys.modules 里注册一批桩模块,让上面那行导入命中纯 stdlib 的轻量版(约 8 ms)。桩模块同时把自己注册进 sys.meta_path 当 finder——因为 Python 解析带点的子模块路径走的是 meta_path 而不是父模块的 __getattr__。一旦用户真的碰了别的东西,桩就自我卸载并触发一次真正的 import opik

apps/opik-sandbox-executor-python/scoring_runner.py:33-68_load_real_opik / _FallbackModule)。注意该文件开头的注释里写的是「9s 执行超时」,那是过时说法,以 executor.py:52 的默认值为准。

4.10 沙箱容器预热成池

妙在哪: 每次打分现起一个 Docker 容器,冷启动就赢不了延迟。Opik 维护一个预热容器池,配合默认 network_disabled=True、内存与 CPU 限额、3 秒执行超时。

apps/opik-python-backend/src/opik_backend/executor_docker.py:110-115(限制项)、:214_pre_warm_container_pool)、:380get_container 阻塞取用)。


5. 代码地图(导航索引)

按主题跳源码。优先用符号名 grep——行号会随上游更新漂移,符号名一般还在。

5.1 追踪层

主题文件路径符号名
对外入口 opik.tracksdks/python/src/opik/decorator/tracker.pyOpikTrackDecoratortrack
装饰器骨架(同步/异步/生成器四条包装路径)sdks/python/src/opik/decorator/base_track_decorator.pyBaseTrackDecorator.track_tracked_sync_tracked_async
调用前后的建/收 span同上__before_call_unsafe__after_call_unsafe
父子关系决策(栈空建 trace、栈非空挂父)sdks/python/src/opik/decorator/span_creation_handler.pycreate_span_respecting_contextSpanCreationResult
contextvars 栈sdks/python/src/opik/context_storage.pyOpikContextStoragetrim_span_data_stack_to_certain_span
手工读写当前 span/tracesdks/python/src/opik/opik_context.pyget_current_span_dataupdate_current_traceget_distributed_trace_headers
框架自动埋点(17 个)sdks/python/src/opik/integrations/各子包(openailangchainadkotel…)

5.2 上报层

主题文件路径符号名
管道总装sdks/python/src/opik/message_processing/streamer_constructors.pyconstruct_online_streamerconstruct_streamer
入队与关闭sdks/python/src/opik/message_processing/streamer.pyStreamer.putStreamer.close
有界双端队列 + 任务记账sdks/python/src/opik/message_processing/message_queue.pyMessageQueue.putput_backcalculate_max_queue_size
消费线程 / 限流退避sdks/python/src/opik/message_processing/queue_consumer.pyQueueConsumer._loop_push_message_back
切批三重条件sdks/python/src/opik/message_processing/batching/base_batcher.pyBaseBatcher.addis_ready_to_flush
各类型批处理器 + 去重sdks/python/src/opik/message_processing/batching/batchers.pyCreateSpanMessageBatcher.add
批参数常量(1000 条 / 2 秒等)sdks/python/src/opik/message_processing/batching/batch_manager_constuctors.pyCREATE_SPANS_MESSAGE_BATCHER_MAX_BATCH_SIZEcreate_batch_manager
载荷体积估算sdks/python/src/opik/message_processing/batching/sequence_splitter.pysplit_into_batches_get_json_size
断线落盘与重放sdks/python/src/opik/message_processing/replay/db_manager.pyDBManagerreplay_failed_messages
重放调度线程sdks/python/src/opik/message_processing/replay/replay_manager.pyReplayManager
消息类型定义sdks/python/src/opik/message_processing/messages.pyCreateSpanMessageCreateSpansBatchMessageAddTraceFeedbackScoresBatchMessage
客户端装配与配置sdks/python/src/opik/api_objects/opik_client.pysdks/python/src/opik/config.pyOpik._initialize_streamerOpikConfig.background_workersmaximal_queue_size

5.3 服务端存储

主题文件路径符号名
ClickHouse 建表(表引擎与排序键)apps/opik-backend/src/main/resources/liquibase/db-app-analytics/migrations/000001_init_script.sqlspanstracesfeedback_scoresReplacingMergeTree
span 写入与字段级合并apps/opik-backend/src/main/java/com/comet/opik/domain/SpanDAO.javaINSERTUPDATEBATCH_INSERT 常量
trace 写入与查询apps/opik-backend/src/main/java/com/comet/opik/domain/TraceDAO.javaTraceDAO
反馈分数存储apps/opik-backend/src/main/java/com/comet/opik/domain/FeedbackScoreDAO.javaFeedbackScoreDAO
实验与实验项apps/opik-backend/src/main/java/com/comet/opik/domain/ExperimentDAO.javaExperimentItemDAO.javaExperimentDAOExperimentItemDAO
后续 schema 演进同目录 000002_*.sql逐个 changeset

5.4 评估层

主题文件路径符号名
对外入口sdks/python/src/opik/evaluation/evaluator.pyevaluateevaluate_experimentevaluate_promptevaluate_resume
执行引擎sdks/python/src/opik/evaluation/engine/engine.pyEvaluationEngine_compute_test_result_for_llm_task
并发执行器sdks/python/src/opik/evaluation/engine/evaluation_tasks_executor.pyStreamingExecutor
指标编排sdks/python/src/opik/evaluation/engine/metrics_evaluator.pymetrics_evaluator
结果模型sdks/python/src/opik/evaluation/test_result.pyevaluation_result.pyTestResultEvaluationResult
数据集与实验对象sdks/python/src/opik/api_objects/dataset/dataset.pyapi_objects/experiment/experiment.pyDatasetDatasetVersionExperiment

5.5 打分器

主题文件路径符号名
指标契约(并把自己也 track 起来)sdks/python/src/opik/evaluation/metrics/base_metric.pyBaseMetric
分数结果sdks/python/src/opik/evaluation/metrics/score_result.pyScoreResult
启发式指标sdks/python/src/opik/evaluation/metrics/heuristics/LevenshteinRatioBLEUROUGERegexMatch
G-Eval(logprobs 加权)sdks/python/src/opik/evaluation/metrics/llm_judges/g_eval/metric.pyparser.pyGEvalGEvalPresetparse_litellm_model_output
幻觉检测sdks/python/src/opik/evaluation/metrics/llm_judges/hallucination/metric.pyHallucinationHallucinationResponseFormat
其余 LLM 裁判sdks/python/src/opik/evaluation/metrics/llm_judges/answer_relevancecontext_precisionmoderationtrajectory_accuracyllm_juries
裁判模型抽象sdks/python/src/opik/evaluation/models/LiteLLMChatModelbase_model

5.6 生产闭环

主题文件路径符号名
规则匹配与采样apps/opik-backend/.../v1/events/OnlineScoringSampler.javaOnlineScoringSamplershouldSampleTrace
打分器基类(消费 Redis 流)apps/opik-backend/.../v1/events/OnlineScoringBaseScorer.javaOnlineScoringBaseScorer.score
LLM 裁判打分器apps/opik-backend/.../v1/events/OnlineScoringLlmAsJudgeScorer.javaOnlineScoringLlmAsJudgeScorer
模板渲染与响应解析apps/opik-backend/.../v1/events/OnlineScoringEngine.javaprepareLlmRequesttoReplacementstoFeedbackScoresbuildSpanTree
用户 Python 指标打分器apps/opik-backend/.../v1/events/OnlineScoringUserDefinedMetricPythonScorer.javaOnlineScoringUserDefinedMetricPythonScorer
沙箱执行策略(Docker / 进程)apps/opik-python-backend/src/opik_backend/executor_docker.pyexecutor_process.pyCodeExecutorBaserun_scoring_pre_warm_container_pool
沙箱内的运行脚本apps/opik-sandbox-executor-python/scoring_runner.py_FallbackModule_load_real_opik
优化器骨架(模板方法)sdks/opik_optimizer/src/opik_optimizer/base_optimizer.pyBaseOptimizer.optimize_prompt_run_algorithm_and_finalize
各优化算法sdks/opik_optimizer/src/opik_optimizer/algorithms/MetaPromptOptimizerfew_shot_bayesian_optimizerevolutionary_optimizergepa_optimizer

5.7 部署与其他

主题文件路径说明
本地全栈编排deployment/docker-compose/docker-compose.yamlMySQL 8.4 / Redis 7.2 / ClickHouse 25.8 / ZooKeeper / MinIO / 后端 / 前端
一键启动脚本opik.shopik.ps1包装 docker compose
TypeScript SDKsdks/typescript/JS/TS 侧的同构客户端
护栏服务apps/opik-guardrails-backend/PII / 越权等前置检查