路由质量评测系统 —— 从 video-agent 的 Langfuse 链路里抽取 Gatekeeper 路由决策和 Expert 间 handoff 调用,用 LLM Judge 自动打分,Web 面板展示结果并支持人工复核。

Langfuse API / ClickHouse  →  Extractor  →  PostgreSQL  →  Judge (LiteLLM)  →  Web 面板

• 后端: FastAPI + SQLAlchemy 2.0 + APScheduler
• 前端: React 19 + Vite + TypeScript + Tailwind v4 + recharts
• DB: PostgreSQL (test/生产用外部实例)
• LLM Judge: 经 LiteLLM 代理调 Gemini 3 Flash

核心代码组织:

路径	说明
api/	FastAPI app 与路由
core/sources/	两种数据源实现:langfuse_api / clickhouse
core/extractor.py	源无关的编排层,按天切分 + 写库
core/judge.py	并发 LLM Judge
core/judge_config.py	评分 prompt / scoring 维度 / expert 描述
db/	ORM 模型 + Repository
config/	pydantic-settings + TOML 配置
web/	前端 SPA
scripts/	本地脚本(dev-server、CH/QPS 探测)
deploy/	部署 Dockerfile
tests/	单元测试 (40 个)
docs/	设计与研究文档

• Python 3.13 + uv
• Node 20+
• 可访问的 PostgreSQL 实例

# 后端依赖
uv sync

# 前端依赖
cd web && npm install && cd ..

# 配置文件 (从 .env.example.toml 复制,填入 Langfuse / LLM / DB 凭证)
cp .env.example.toml .env.toml

# 一个脚本启动前后端 (端口 8080 + 5173)
bash scripts/dev-server.sh

# 或分别启动
.venv/bin/uvicorn api.app:app --reload --port 8080
cd web && npm run dev

.venv/bin/python -m pytest tests/ -v

一个 env 可以同时配置 API 和 ClickHouse 两种源,通过前端触发抽取时选择,或改 default_source:

场景	推荐源	原因
每天增量拉取	api	Langfuse REST,限流 40 RPS
全量历史/按月拉取	clickhouse	直连 CH,单日秒级,无限流
跑 QPS 探测	scripts/probe_langfuse_qps.py	逐级加压测 API 的安全上限
验证 CH schema	scripts/probe_clickhouse.py	DESCRIBE 表、样例数据查询

详见 docs/extraction-pipeline.md。

配置按优先级由低到高合并: config/default.toml → .env.toml → 环境变量。

pydantic-settings 把配置树扁平化成环境变量,用 __ 分隔层级,前缀 ROUTING_EVAL_:

环境变量	作用
ROUTING_EVAL_DATABASE__URL	PG 连接串,如 postgresql://user:pw@host:5432/routing_eval
ROUTING_EVAL_LANGFUSE__PROD__PUBLIC_KEY	Langfuse prod public key
ROUTING_EVAL_LANGFUSE__PROD__SECRET_KEY	Langfuse prod secret key
ROUTING_EVAL_LANGFUSE__PROD__CLICKHOUSE__URL	ClickHouse 地址
ROUTING_EVAL_LANGFUSE__PROD__CLICKHOUSE__PASSWORD	ClickHouse 密码
ROUTING_EVAL_LLM__API_KEY	LiteLLM API key
ROUTING_EVAL_SCHEDULER__ENABLED	true 启用定时抽取

.env.toml 里的敏感字段(public_key / secret_key / password / api_key)不要打进镜像。

方法	路径	作用
GET	/api/health /healthz /health	存活探测(CMDB 探针用 /api/health)
GET	/api/readyz /readyz	就绪探测(含 DB 连通性)
GET	/api/summary	总览指标
GET	/api/intent	意图样本列表(含 total / stats,服务端分页)
GET	/api/handoff	Handoff 样本列表
GET	/api/intent.csv /api/handoff.csv	导出 CSV
GET	/api/intent-confusion	错误路由对矩阵
GET	/api/expert-breakdown	按 Expert 的路由准确率
GET	/api/trend	近 N 天每日指标趋势
POST	/api/extract	触发抽取 {env, from_date, to_date, source, max_sessions, auto_judge}
POST	/api/extract/resume/{run_id}	继续中断的抽取
POST	/api/judge	全量评测
POST	/api/judge/single/{type}/{id}	评测单条
POST	/api/judge/batch	批量评测选中样本
GET/PUT	/api/judge-config[/...]	读/改评测 prompt、model、expert 描述

主要表:

• eval_runs — 每次抽取批次
• daily_tasks — 每次批次按天拆的子任务(用于进度与断点续跑)
• sessions / traces — Langfuse 原始元数据
• intent_samples / handoff_samples — 评测样本 + judge 打分
• judge_profiles / app_config — 可编辑的评测配置
• schema_migrations — 迁移跟踪表(runner 维护)

裸 SQL 文件,启动自动应用:

migrations/
├── 0001_create_config_tables.up.sql
├── 0001_create_config_tables.down.sql
├── 0002_create_run_tables.up.sql
├── 0002_create_run_tables.down.sql
└── ...

• 文件名 NNNN_<description>.up.sql / .down.sql,按字典序应用
• up.sql 在应用启动时自动跑(db/migrate.py),每个文件一个事务,记录进 schema_migrations(version, applied_at)
• down.sql 不会自动跑,只用于手动回滚(参考)
• 写新迁移请用幂等 DDL(CREATE TABLE IF NOT EXISTS、ALTER TABLE ... ADD COLUMN IF NOT EXISTS), 这样同一 DB 重跑或与已有表并存都安全

# 假设要加个字段
cat > migrations/0004_add_user_review_status.up.sql <<'SQL'
ALTER TABLE intent_samples
    ADD COLUMN IF NOT EXISTS user_reviewed BOOLEAN DEFAULT FALSE;
SQL

cat > migrations/0004_add_user_review_status.down.sql <<'SQL'
ALTER TABLE intent_samples DROP COLUMN IF EXISTS user_reviewed;
SQL

下次应用启动就会自动跑 0004。

test 环境提供的 PG 实例,创建 database:

CREATE DATABASE routing_eval;

docker build -f deploy/test/Dockerfile -t routing-eval:test .

环境变量通过 secret / CMDB 注入(清单见上一节"环境变量"):

docker run -d --name routing-eval -p 8080:8080 \
  -e ROUTING_EVAL_DATABASE__URL='postgresql://...' \
  -e ROUTING_EVAL_LANGFUSE__PROD__PUBLIC_KEY='...' \
  -e ROUTING_EVAL_LANGFUSE__PROD__SECRET_KEY='...' \
  -e ROUTING_EVAL_LANGFUSE__PROD__CLICKHOUSE__URL='http://clickhouse-langfuse-test.internal-jl.com' \
  -e ROUTING_EVAL_LANGFUSE__PROD__CLICKHOUSE__PASSWORD='...' \
  -e ROUTING_EVAL_LLM__API_KEY='...' \
  -e ROUTING_EVAL_SCHEDULER__ENABLED='true' \
  routing-eval:test

• GET /healthz → {"status":"ok"}(存活)
• GET /readyz → {"status":"ready"} 或 503(DB 故障时)
• 启动日志里应看到:Recovery: marked N runs as interrupted(如有残留任务), 以及 Seeded default profiles。

• 单实例 1C/2G 够用
• 全量拉 1 个月 prod 数据约 15 分钟(CH 源),中间 CPU 峰值 ~1 核
• Judge 阶段吃 LLM 配额(默认 8 并发),按样本量估算成本

容器 HEALTHCHECK 已内置,每 30s 检 /api/health。CMDB 探活建议(与面板默认对齐):

• liveness: /api/health,initial_delay=30s,period=10s,timeout=10s,failure_threshold=5
• readiness: /api/readyz,initial_delay=30s,period=10s(先确保 DB 连通再接流量)

1. 定时增量: scheduler.enabled=true 时进程内 APScheduler 按 extract_cron(默认 02:00 UTC) 每天抽取 T-1 的数据,走对应 env 的 default_source,auto_judge=true 会自动评测。 多副本通过 pg_try_advisory_lock 防重 —— 第一个拿到锁的 pod 跑,其他跳过。 pod 全挂时会漏那一次(不补跑),要强保证可以改成平台 CronJob 打 /api/extract。
2. 中断恢复: 进程重启后,启动时自动把 extracting/judging 状态的 run 标为 interrupted, daily_task 回退 pending,UI 上点"继续抽取"即可从断点续。
3. Judge 配置热更新: /api/judge-config 支持 PUT,修改模型 / prompt / expert 描述无需 重启(落到 PG,下次评测读取新值)。
4. CSV 导出上限: 当前导出最多 2 万条,超过需要分段拉。
5. 慎重删表: create_all 不会删任何东西。手动 DROP COLUMN 后下次启动会重建 —— 别用 DROP。