Langfuse 队列事件回填脚本refill-queue-event实战指南从本地 JSONL 批量注入 BullMQ 队列【免费下载链接】langfuse Open source AI engineering platform: LLM evals, observability, metrics, prompt management, playground, datasets. Integrates with OpenTelemetry, LangChain, OpenAI SDK, LiteLLM, and more. YC W23项目地址: https://gitcode.com/GitHub_Trending/la/langfuse导读本指南围绕 Langfuse 开源仓库中 worker/src/scripts/refillQueueEvent/README.md 所描述的工具脚本展开讲解如何将本地机器上的事件以 JSONL 文件的形式回填backfill到任意 Langfuse 队列中。文章将完整覆盖该脚本的环境准备、文件格式、执行命令等实操步骤并结合 worker/src/scripts/refillQueueEvent/index.ts 的源码深入剖析其读取 → Schema 校验 → 分批入队的处理管线以及底层 BullMQ 队列体系QueueName、getQueue、projectDeleteProcessor是如何协同工作的。读完本文你将能够在自建 Langfuse 环境中安全、高效地回填队列事件并具备扩展脚本支持其他队列类型的能力。一、脚本定位为什么需要回填队列事件Langfuse 是一个开源 AI 工程平台其 worker 进程通过 BullMQ基于 Redis驱动大量异步任务trace 删除、project 删除、eval 执行、数据导出、告警等。在开发、调试或数据迁移场景下开发者经常需要将本地机器上构造好的一批事件重新注入线上/测试环境的队列以复现问题或验证消费端逻辑——这就是refill-queue-event脚本的用武之地。它的核心能力由两点构成见 worker/src/scripts/refillQueueEvent/index.tsSchema 校验每个事件在入队前都会用目标队列对应的 Zod Schema 做合法性校验不合规的事件被拦截并报告避免脏数据进入生产队列批量高效入队事件按每批 1000 条BATCH_SIZE 1000分组通过 BullMQ 的queue.addBulk()批量写入避免逐条入队带来的网络往返开销。当前脚本默认也是唯一支持的目标队列是QueueName.ProjectDeleteproject-delete 队列事件结构为{ projectId, orgId }。二、快速开始完整操作步骤2.1 创建事件文件在工作区中创建事件文件./worker/events.jsonl每行一个 JSON 事件例如与原文档示例一致{projectId: project-123, orgId: org-456} {projectId: project-789, orgId: org-101}关键约束每个事件必须匹配目标队列期望的 Schema。脚本会逐个事件校验并报告 Schema 违规项。对于默认的project-delete队列Schema 定义于 packages/shared/src/server/queues.tsexport const ProjectQueueEventSchema z.object({ projectId: z.string(), orgId: z.string(), });即每个事件必须同时包含字符串类型的projectId与orgId字段缺失、类型不符如数字 ID或多余字段都会导致该校验失败——Zod 默认开启 strip 模式多余字段会被剔除但缺字段会直接报错。2.2 设置环境变量在仓库根目录创建.env文件内容如下# Required: Redis connection REDIS_CONNECTION_STRINGredis://:myredissecret127.0.0.1:6379 # Required: Supporting services for worker initialization LANGFUSE_S3_EVENT_UPLOAD_BUCKETlangfuse CLICKHOUSE_URLhttp://localhost:8123 CLICKHOUSE_USERclickhouse CLICKHOUSE_PASSWORDclickhouse这些变量之所以必需是因为脚本运行在完整的 worker 初始化环境中见下文 3.2 节的环境解析原理。其中REDIS_CONNECTION_STRING队列后端连接串。格式为redis://[:password]host:port[/db]脚本通过它建立与目标队列实例的连接LANGFUSE_S3_EVENT_UPLOAD_BUCKET事件上传用的 S3 bucket 名称worker 初始化时会读取它构造 S3 客户端CLICKHOUSE_URL/CLICKHOUSE_USER/CLICKHOUSE_PASSWORDClickHouse 连接信息worker 初始化及后续消费端处理依赖。2.3 确保连通性确保本地机器能够访问 Redis 实例——例如通过 SSH 隧道转发端口或在/etc/hosts中映射主机名。可先用redis-cli -u $REDIS_CONNECTION_STRING ping验证连通再运行脚本。2.4 运行脚本pnpm run --filterworker refill-queue-event该命令对应的 npm script 定义在 worker/package.jsonrefill-queue-event: dotenv -e ../.env -- tsx src/scripts/refillQueueEvent/index.ts这条命令揭示了三个重要事实环境加载方式通过dotenv -e ../.env显式加载仓库根目录的.env相对 worker 包目录的上级与文档 2.2 节在仓库根目录创建.env的要求严格对应运行载体使用tsx直接执行 TypeScript 源码无需预先编译工作目录命令在worker包目录内执行因此脚本内部用相对路径events.jsonl读取文件时实际定位到./worker/events.jsonl与 2.1 节的文件位置一致。三、源码级原理剖析3.1 脚本整体执行管线从 worker/src/scripts/refillQueueEvent/index.ts 的main()可以还原完整流程配置校验validateConfig第 43-59 行检查目标队列是否在QUEUE_SCHEMA_MAP中注册、events.jsonl文件是否存在并打印当前配置队列名、文件路径、批大小读取事件readEventsFromFile第 64-88 行逐行读取文件跳过空行对每行执行JSON.parse某行解析失败仅打印警告含行号与错误信息并跳过不中断整体执行Schema 校验validateEvents第 93-121 行对每个事件执行schema.parse()捕获z.ZodError并输出详细校验错误将事件分为valid与invalid两组分批入队processEvents第 126-189 行对合法事件按BATCH_SIZE分批调用queue.addBulk()记录进度与首条样本事件统计输出printStats第 194-212 行打印总事件数、合法/非法事件数、已处理数、耗时与吞吐量events/second异常兜底任何异常都会被捕获并以非零退出码结束第 251-254 行方便 CI 判断成败。值得注意的是脚本末尾的if (require.main module) main();守卫第 258-260 行使其既可作为 CLI 直接运行也可被其他模块以 import 方式复用内部函数而不触发副作用。3.2 队列注册表与类型安全设计脚本采用注册表 映射的设计集中在 worker/src/scripts/refillQueueEvent/index.tsconst QUEUE_SCHEMA_MAP { [QueueName.ProjectDelete]: ProjectQueueEventSchema, } as const; const QUEUE_JOB_MAP { [QueueName.ProjectDelete]: QueueJobs.ProjectDelete, } as const;QUEUE_SCHEMA_MAP队列名 → 事件校验 SchemaQUEUE_JOB_MAP队列名 → 入队时使用的 Job 名称isSupportedQueue第 25-29 行TypeScript 类型守卫同时承担运行时与编译期的双重保护——只有注册过的队列名才能通过校验编译期也会因keyof typeof QUEUE_SCHEMA_MAP的类型约束而报错。QueueName与QueueJobs枚举定义于 packages/shared/src/server/queues.ts覆盖了 Langfuse 的全部队列如trace-upsert、trace-delete、project-delete、evaluation-execution-queue、ingestion-queue等。扩展新队列时只需在两张映射表中各追加一行并在QUEUE_SCHEMA_MAP中引入对应 Schema——例如TraceQueueEventSchema{ projectId, traceId, exactTimestamp?, traceEnvironment? }见 packages/shared/src/server/queues.ts即可支持TraceDelete队列。3.3 队列实例的获取与限制脚本通过getQueue(queueName)获取 BullMQQueue实例第 143 行其实现位于 packages/shared/src/server/redis/getQueue.ts。这是一个大型 switch 分发为每个队列返回对应的单例实例如ProjectDeleteQueue.getInstance()。重要限制从源码结构可以推断getQueue的入参类型显式排除了分片队列sharded queues包括IngestionQueue、EvaluationExecution、LLMAsJudgeExecution、CodeEvalExecution、TraceUpsert、OtelIngestionQueue及其 Secondary 变体。这些高吞吐队列需要shardingKey才能实例化见 getQueue.ts 的注释说明应改用队列类自身的getInstance({ shardingKey })直接构造。因此若要回填这些分片队列需要在脚本中绕过getQueue的通用路径这一前提当前脚本尚未覆盖。3.4 批量入队Job 结构与 addBulkprocessEvents的核心入队逻辑第 152-182 行为每个事件构造 BullMQ Job 并批量提交const jobs batch.map((event) ({ name: jobName, data: { payload: event, id: randomUUID(), timestamp: new Date(), name: jobName, }, })); await queue.addBulk(jobs);每个 Job 包含字段类型说明nameQueueJobs枚举值如project-delete消费端据此路由处理器data.payload事件本体即 JSONL 中的一行经 Schema 校验后原样保留data.idstring由randomUUID()生成任务唯一标识data.timestampDate入队时间戳data.name同name冗余记录 Job 名便于消费端读取addBulk以流水线方式批量写入 Redis相比逐条add显著降低了网络开销批大小为 1000兼顾内存占用与吞吐。每个批次完成后打印进度Processed batch 1/2 (1000/1500 events)首个批次还会打印一条格式化后的样本事件便于人工核对。3.5 消费端视角project-delete 处理器回填的事件最终由 worker 的队列处理器消费。以默认的project-delete队列为例其消费逻辑位于 worker/src/queues/projectDelete.ts 的projectDeleteProcessor从job.data.payload解构orgId与projectId并写入 OpenTelemetry span 属性用于链路追踪依次执行媒体清理deleteMediaLinkRowsByProjectId、S3 媒体文件删除、ClickHouse 数据删除traces/observations/scores/events、数据集运行项异步删除最后通过 Prisma 删除 PG 中的 project 行对P2025/P2016记录不存在等幂等场景做容错不会因重复回填同一项目而报错——这从侧面印证了该脚本适用于幂等性较强的删除类队列。四、运行输出解读与统计指标脚本正常运行后的输出分为三个阶段。以 2 条合法事件为例Starting refill queue event script... Configuration: Queue: project-delete Events file: events.jsonl Batch size: 1000 Reading events from events.jsonl... Read 2 events from 2 lines Validating events for queue: project-delete... Valid events: 2 Invalid events: 0 Processing 2 events for queue: project-delete... Connected to queue: project-delete (job: project-delete) Processed batch 1/1 (2/2 events) Sample event: { projectId: project-123, orgId: org-456 } Processing Complete Total events: 2 Valid events: 2 Invalid events: 0 Processed events: 2 Processing time: 0.1 seconds Average speed: 20 events/second其中Average speedevents/second是脚本内建的吞吐量指标见 index.ts可直接用于评估入队性能。若存在非法事件则会在校验阶段打印Invalid event:及 Zod 校验错误详情最终统计中的Invalid events会体现该数量但非法事件不会被入队。五、典型使用场景与注意事项5.1 典型场景故障恢复某批队列事件在线上丢失或未入队用本地备份的 JSONL 重新注入消费端验证构造特定projectId/orgId的事件验证删除链路的正确性与幂等性数据迁移跨环境如 staging → production复制删除请求保证数据一致性压测利用events/second指标评估队列吞吐寻找 Redis 或消费端的瓶颈。5.2 注意事项Schema 严格匹配事件字段必须与目标队列 Schema 完全兼容否则被静默拦截仅计入Invalid events统计事件文件位置必须是./worker/events.jsonl脚本以 worker 目录为基准的相对路径文件不存在会直接抛错退出分片队列限制当前脚本不能通过getQueue访问分片队列ingestion/eval/trace-upsert 等扩展时需特殊处理 sharding key环境变量完整性.env缺失任一必需变量都会导致 worker 初始化失败务必按 2.2 节补齐目标队列的消费端语义回填前确认该队列的处理器是否具备幂等性如 project-delete 对已删除记录做了容错避免重复入队引发副作用。六、延伸相关回填脚本本仓库的 worker/package.json 中还提供了两个同族脚本可互为参照refill-ingestion-events第 27 行通过replayIngestionEvents/s3-ingestion-event-replay.ts从 S3 重放 ingestion 事件对应目录 worker/src/scripts/replayIngestionEventsrefill-billing-event第 28 行运行refill-billing-event.ts回填计费事件。三者共享同一套dotenv 加载根目录 .env tsx 执行的运行模式可见 Langfuse 将这类一次性运维工具统一收敛在 worker 包的scripts目录下worker/src/scripts 是浏览全部运维脚本的入口。参考文件索引使用文档worker/src/scripts/refillQueueEvent/README.md脚本实现worker/src/scripts/refillQueueEvent/index.ts命令定义worker/package.json队列与 Schema 定义packages/shared/src/server/queues.ts队列实例工厂packages/shared/src/server/redis/getQueue.ts消费端处理器worker/src/queues/projectDelete.ts【免费下载链接】langfuse Open source AI engineering platform: LLM evals, observability, metrics, prompt management, playground, datasets. Integrates with OpenTelemetry, LangChain, OpenAI SDK, LiteLLM, and more. YC W23项目地址: https://gitcode.com/GitHub_Trending/la/langfuse创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考