跳到主要内容

Arize Phoenix — 架构与原理

30 秒导读: Phoenix 是一个开源、可自托管的 AI 可观测平台。你给 LLM 应用装上 OpenTelemetry 探针,它把每一次调用(模型、检索、工具、agent 步骤)作为 span 发到 Phoenix;Phoenix 异步批量存进 SQLite/Postgres,再提供 UI、GraphQL、REST 来看轨迹,并能用 evals(LLM 当裁判打分)和 experiments(在数据集上跑变体并对比)帮你回答"这次为什么变差了"。

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

一句话定义: Phoenix 是给 LLM 应用用的"飞行记录仪 + 体检中心"——记录每一步发生了什么,再给每一步打分。

解决什么问题 / 给谁用:

假设你在写一个 RAG 问答 agent:用户提问 → 检索文档 → 拼 prompt → 调 GPT → 返回。某天用户抱怨"答非所问"。你想知道:

  • 这次到底检索到了哪些文档?(检索错了?)
  • 喂给模型的 prompt 长什么样?花了多少 token、多少钱?
  • 模型这步耗时多久?报错了吗?
  • 跨多次请求,答案质量在变好还是变差?

Phoenix 就是来回答这些的。给谁用:做 LLM / agent / RAG 应用的工程师,本地 notebook 调试、或部署成团队共享的服务都行。

它能做什么(对外能力):

能力白话
Tracing收集并可视化 LLM 应用的运行轨迹(基于 OpenTelemetry)
Evaluation用 LLM 或代码规则给轨迹/回答打分(相关性、幻觉、毒性…)
Datasets把样例存成带版本的数据集,供实验和评估用
Experiments在数据集上跑"prompt/模型/检索"的变体并对比
Playground调 prompt、比模型、重放被追踪的 LLM 调用
Prompt Management给 prompt 做版本管理、打标签

(能力清单见仓库 README.md 顶部。)

用起来什么样: 最小的本地用法——一行启动,浏览器看轨迹:

# 示意,基于 src/phoenix/__init__.py 的导出
import phoenix as px

session = px.launch_app()   # 起一个本地 Phoenix(默认 127.0.0.1:6006)
# ...你的 LLM 应用此时通过 OpenTelemetry 把 span 发到 Phoenix...
print(session.url)          # 打开它看轨迹

launch_app 真实签名见 src/phoenix/session/session.py:268,它能在 notebook 里起线程、也能用临时目录的 SQLite 存数据。

一句话直觉 / 类比:

把 LLM 应用想成一条生产流水线,Phoenix 是装在流水线上方的摄像头网络 + 质检站:摄像头(OTel 探针)拍下每个工位(span)的动作,录像存进仓库(SQL),质检站(evals)给每段录像打分,实验台(experiments)让你换个零件再录一遍对比。

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

Phoenix 是个写多读多的系统:海量 span 高速写入,UI 大量读。它的核心设计就是把这两条路分开。

2.1 一张顶层图​

这张图从左到右是写路径(数据进来),右侧分叉出读路径(UI/API 查出去)。看的时候抓一条主线:span 字节 → 队列 → SQL。

  你的 LLM 应用
  (OpenTelemetry 探针)
        │  OTLP (protobuf)
        │  gRPC :4317  /  HTTP :6006/v1/traces
        ▼
  ┌─────────────────────┐
  │ ① 接收层            │  decode_otlp_span: 把 protobuf 解成 Python Span
  │  grpc_server.py     │  (扁平 key → 嵌套 dict,见 ch.01)
  └──────────┬──────────┘
             │ enqueue_span(span, project_name)   ← 只是塞进内存队列,立刻返回
             ▼
  ┌─────────────────────┐
  │ ② 写缓冲            │  BulkInserter:一个后台 asyncio 任务
  │  bulk_inserter.py   │  攒一批 → 一个事务批量写 → 算累计/成本
  └──────────┬──────────┘
             │ insert_span(...)            │ put(DmlEvent)
             ▼                             ▼
  ┌─────────────────────┐      ┌────────────────────────┐
  │ ③ 存储 (SQL)        │      │ ④ 变更事件总线          │
  │  models.py          │      │  DmlEventHandler        │
  │  SQLite / Postgres  │      │  (失效缓存 / 推订阅)    │
  └──────────┬──────────┘      └────────────────────────┘
             │ 读
             ▼
  ┌──────────────────────────────────────────────┐
  │ ⑤ 读 / 应用层  (FastAPI app.py)               │
  │   • GraphQL  /graphql   (UI 主用,带 dataloaders)│
  │   • REST     /v1/...    (SDK / 程序化访问)     │
  │   • 守护进程: 实验运行、成本计算、保留期清理…   │
  └──────────────────────────────────────────────┘

2.2 部件一句话职责​

部件干什么在哪
OTLP 接收收 protobuf,解码成 Span 对象src/phoenix/server/grpc_server.py、src/phoenix/trace/otel.py
BulkInserter内存队列 + 后台任务,批量写 span/标注src/phoenix/db/bulk_inserter.py
insert_span一条 span 落库的全部 SQL 逻辑src/phoenix/db/insertion/span.py
ORM 模型Project/Trace/Span/Session/Annotation 表src/phoenix/db/models.py
DML 事件span 写入后通知"谁的缓存该失效/谁该收到推送"src/phoenix/server/dml_event.py、dml_event_handler.py
GraphQL/REST读侧 API(UI 走 GraphQL)src/phoenix/server/api/、app.py
span filter DSL把用户输入的过滤表达式安全编译成 SQLsrc/phoenix/trace/dsl/filter.py
evals 子包打分库(LLM/启发式)packages/phoenix-evals/
experiments数据集上跑变体 + 评估src/phoenix/server/daemons/experiment_runner.py

2.3 主线走一遍(高层)​

  1. 你的应用产生一个 LLM 调用 → OTel 探针把它编码成 OTLP span,通过 gRPC(:4317)或 HTTP(/v1/traces)发给 Phoenix。
  2. 接收层 Servicer.Export(grpc_server.py:41)逐个解码:decode_otlp_span 把扁平的 llm.token_count.prompt=12 这类 key 还原成嵌套 dict,产出一个 Span 数据类。
  3. 解码后只做一件事:enqueue_span(span, project_name) 把它塞进 BulkInserter 的内存 deque,RPC 立即返回——接收和落盘解耦。
  4. 后台 BulkInserter._bulk_insert 循环:每隔约 0.1s 醒来,从队列取一批,在一个事务里逐条 insert_span,同时累计父链 token、算成本。
  5. 写完后往 event_queue 放 SpanInsertEvent,通知 DML 事件处理器去失效相关缓存 / 推 GraphQL 订阅。
  6. UI 通过 GraphQL 读数据,经 dataloaders 批量化数据库查询;程序化访问走 /v1 REST。
  7. 想打分就用 evals;想系统性对比变体就建数据集跑 experiment。

3. 阅读地图(各章讲什么、建议顺序)​

按"数据怎么进来 → 进来后长什么样、怎么读 → 怎么评价它"的顺序读:

  1. 01-ingestion-pipeline.md — 先读这章。一条 span 从 OTLP 字节到 SQL 行的完整旅程:OTLP 解码、扁平 key 还原成嵌套结构(unflatten)、为什么要用内存队列 + 后台批量写、累计 token 怎么沿父链传播、成本怎么算。这是整个平台的心脏。
  2. 02-data-model-and-read.md — 落盘后的数据模型(Project / Trace / Span / ProjectSession / Annotation 的关系),以及读侧两个关键机制:span filter DSL(把用户写的 llm.token_count.prompt > 100 安全编译成 SQL,而不 eval 任意代码)和 dataloaders(解决 GraphQL 的 N+1 查询)。
  3. 03-evals.md — phoenix-evals 子包(可独立 pip 安装)。Evaluator/Score 抽象、input_mapping 怎么把任意数据形状"绑"到评估器要的字段、ClassificationEvaluator 如何用一个 prompt + choices 产出结构化的 label/score/explanation。
  4. 04-experiments.md — 把数据集、任务(task)、评估器组合成实验;服务端 ExperimentRunner 守护进程如何调度"跑任务"和"打分"两类工作项。

4. 横向对比(同 shelf 的兄弟)​

Phoenix 属于 evals-observability 区。和它常被一起比较的:

维度Phoenix典型 tracing-only 工具典型 evals-only 库
数据进来的协议标准 OTLP / OpenTelemetry(厂商无关)各家私有 SDK 居多不收 trace
是否自托管是(一行 launch_app 或 Docker)多为 SaaS库,无服务
tracing + evals + experiments三合一,共享同一份 span 数据通常只做 tracing只做打分
评估器能否独立用能(pip install arize-phoenix-evals)—是其全部

Phoenix 的取舍很清楚:赌 OpenTelemetry 这个开放标准(解码逻辑全在 trace/otel.py,配套 OpenInference 语义约定),而不是发明私有协议——代价是要做 OTLP↔嵌套结构的来回转换(见 ch.01 的 unflatten),收益是任何会发 OTLP 的框架都能直接接。

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

主题文件关键符号
对外入口(脚本)pyproject.tomlarize-phoenix = "phoenix.server.main:main"
本地启动src/phoenix/session/session.pylaunch_app
OTLP gRPC 接收src/phoenix/server/grpc_server.pyServicer.Export、GrpcServer
OTLP 解码src/phoenix/trace/otel.pydecode_otlp_span
扁平↔嵌套还原src/phoenix/trace/attributes.pyunflatten、has_mapping、load_json_strings
异步批量写src/phoenix/db/bulk_inserter.pyBulkInserter._bulk_insert、_insert_spans
span 落库src/phoenix/db/insertion/span.pyinsert_span、SpanInsertionEvent
ORM 模型src/phoenix/db/models.pySpan、Trace、Project、ProjectSession
变更事件src/phoenix/server/dml_event.py、dml_event_handler.pyDmlEvent、SpanInsertEvent、DmlEventHandler
应用装配src/phoenix/server/app.pycreate_app、_lifespan、create_graphql_router
过滤 DSLsrc/phoenix/trace/dsl/filter.pySpanFilter、_FilterTranslator
evals 抽象packages/phoenix-evals/src/phoenix/evals/evaluators.pyEvaluator、Score、ClassificationEvaluator、create_classifier
实验运行src/phoenix/server/daemons/experiment_runner.pyExperimentRunner