使用 dlt 將 Langfuse 可觀測(cè)性數(shù)據(jù)導(dǎo)出到數(shù)據(jù)湖倉(cāng)【免費(fèi)下載鏈接】dltdata load tool (dlt) is an open source Python library that makes data loading easy ?項(xiàng)目地址: https://gitcode.com/GitHub_Trending/dl/dlt導(dǎo)讀Langfuse 是開(kāi)源的 LLM 可觀測(cè)性與評(píng)估平臺(tái)它把 traces、評(píng)估數(shù)據(jù)集、評(píng)分與成本數(shù)據(jù)全部持久化在自托管的 PostgreSQL 中。本指南講解如何借助 dltdata load tool內(nèi)置的sql_database()源將 Langfuse 的 Postgres 后端一鍵導(dǎo)出到 DuckDB 等數(shù)據(jù)湖倉(cāng)/數(shù)倉(cāng)從而支撐離線分析、報(bào)表與批量評(píng)估。讀完本文你將掌握連接憑據(jù)配置、resolve_foreign_keys外鍵關(guān)系自動(dòng)關(guān)聯(lián)、table_names選擇性加載等核心能力并能獨(dú)立搭建一條 Langfuse → 目標(biāo)存儲(chǔ)的完整數(shù)據(jù)管道。Langfuse 數(shù)據(jù)為什么要導(dǎo)出Langfuse 是一個(gè)開(kāi)源 LLM 可觀測(cè)性與評(píng)估平臺(tái)它會(huì)捕獲 AI 應(yīng)用的 traces 與 observations管理評(píng)估數(shù)據(jù)集并跟蹤 LLM 調(diào)用成本——所有這些數(shù)據(jù)都存放在一個(gè)可自托管的、基于 PostgreSQL 的存儲(chǔ)中自托管場(chǎng)景下。它的 Web UI 適合在線查看與人工評(píng)估但要支持以下場(chǎng)景把數(shù)據(jù)搬出到數(shù)據(jù)湖倉(cāng)/數(shù)倉(cāng)是必經(jīng)之路分析與報(bào)表跨項(xiàng)目統(tǒng)計(jì) token 消耗、成本趨勢(shì)、模型分布離線評(píng)估把datasets/dataset_items與評(píng)分結(jié)果批量拉入分析引擎做批量評(píng)測(cè)數(shù)據(jù)治理與審計(jì)audit_logs、comments等元數(shù)據(jù)進(jìn)入統(tǒng)一數(shù)倉(cāng)與業(yè)務(wù)數(shù)據(jù)打通。Langfuse 的關(guān)系型元數(shù)據(jù)全部持久化在PostgreSQL中而 dlt 內(nèi)置的sql_database()源讓抽取變得非常直接用連接字符串直連 Langfuse 的 Postgres 后端設(shè)置resolve_foreign_keysTrue讓 dlt 自動(dòng)把子表如datasets → projects關(guān)聯(lián)起來(lái)加載到DuckDB實(shí)現(xiàn)零配置的本地分析或者換成任意其他 dlt 目標(biāo)。憑據(jù)配置連接 Langfuse 的 Postgres 后端在項(xiàng)目目錄下新建.dlt/secrets.toml填入你部署實(shí)例對(duì)應(yīng)的值# .dlt/secrets.toml [sources.sql_database.credentials] drivername postgresql host localhost port 5432 database langfuse username langfuse password langfuse這些字段與倉(cāng)庫(kù)中ConnectionStringCredentials的配置規(guī)范一一對(duì)應(yīng)drivername、host、port、database、username、password、querypassword會(huì)被 dlt 作為密文處理。[sources.sql_database.credentials]這一節(jié)名與sql_database()默認(rèn)參數(shù)credentialsdlt.secrets.value的解析路徑一致你也可以改用完整連接字符串如postgresql://langfuse:langfuselocalhost:5432/langfuse通過(guò)dlt.secrets注入。另外還需要安裝 psycopg2 驅(qū)動(dòng)pip install psycopg2-binary各部署形態(tài)下這些值從哪里找Docker Compose查看docker-compose.yml中的db服務(wù)。憑據(jù)由POSTGRES_USER、POSTGRES_PASSWORD、POSTGRES_DB環(huán)境變量設(shè)定host 是服務(wù)名若端口已發(fā)布到宿主機(jī)則用localhost。Helm chart查看values.yaml中postgresql.auth配置或 Langfuse 部署上的DATABASE_URL環(huán)境變量。Langfuse Cloud云托管無(wú)法直接訪問(wèn)數(shù)據(jù)庫(kù)請(qǐng)改用 Langfuse API 進(jìn)行導(dǎo)出而不是直連 Postgres。核心原理sql_database()是如何工作的本文方案的核心是 dlt 內(nèi)置的 SQL 數(shù)據(jù)庫(kù)源sql_database()其實(shí)現(xiàn)位于dlt/sources/sql_database/__init__.py。調(diào)用它時(shí)dlt 會(huì)反射Reflect數(shù)據(jù)庫(kù)結(jié)構(gòu)通過(guò) SQLAlchemy 連接數(shù)據(jù)庫(kù)讀取 schema 中全部表/視圖的列、主鍵、外鍵約束無(wú)需手寫(xiě)任何建表或映射代碼對(duì)應(yīng)源碼中的metadata.reflect(...)調(diào)用__init__.py生成資源Resources為每個(gè)表生成一個(gè)獨(dú)立的DltResource逐表執(zhí)行SELECT并按chunk_size默認(rèn) 50000 行分批產(chǎn)出數(shù)據(jù)推斷并固化 schema把 SQLAlchemy 列類(lèi)型映射為 dlt 目標(biāo) schema 類(lèi)型見(jiàn)下文“類(lèi)型推斷”在抽取的同時(shí)完成類(lèi)型轉(zhuǎn)換。關(guān)鍵參數(shù)reflection_level 與 resolve_foreign_keysLangfuse 示例中使用了兩個(gè)值得深入理解的參數(shù)reflection_levelminimal只反射表名、可空性和主鍵列的數(shù)據(jù)類(lèi)型從實(shí)際數(shù)據(jù)中推斷?？蛇x值見(jiàn)ReflectionLevel定義minimal/full默認(rèn)在 minimal 基礎(chǔ)上反射類(lèi)型并做必要強(qiáng)轉(zhuǎn)/full_with_precision額外攜帶 decimal、text、binary 的精度與 scale。Langfuse 的input、metadata等 JSON 字段在minimal下會(huì)被 dlt 規(guī)范化拆分為子表保持目標(biāo) schema 的穩(wěn)定與可讀。resolve_foreign_keysTrue把同 schema 內(nèi)的外鍵翻譯成 dlt 的references表提示hint從而在目標(biāo)端生成可查詢的表關(guān)系。其底層實(shí)現(xiàn)是get_table_references()遍歷每個(gè)表的foreign_key_constraints解析被引用表名與列合并指向同一張表的多個(gè)外鍵最終形成{referenced_table: ..., referenced_columns: [...], columns: [...]}結(jié)構(gòu)。需要留意的是該開(kāi)關(guān)會(huì)觸發(fā)額外數(shù)據(jù)庫(kù)調(diào)用去反射所有被引用表因此會(huì)略微增加啟動(dòng)開(kāi)銷(xiāo)。類(lèi)型推斷在非minimal反射級(jí)別下sqla_col_to_column_schema()負(fù)責(zé)把 SQLAlchemy 類(lèi)型映射到 dlt 類(lèi)型Numeric→decimal/double、Integer→bigint、String→text、DateTime→timestamp并透?jìng)?timezone 提示、JSON/ARRAY→json、Boolean→bool、UUID→text。即使使用minimal級(jí)別目標(biāo)端建表時(shí) dlt 仍會(huì)以這些推斷結(jié)果為準(zhǔn)保證 DuckDB 等目標(biāo)的列類(lèi)型貼合源庫(kù)。Langfuse 數(shù)據(jù)模型自動(dòng)發(fā)現(xiàn)與關(guān)鍵表dlt 會(huì)自動(dòng)發(fā)現(xiàn) Langfuse 數(shù)據(jù)庫(kù)中的每一張表并從外鍵推斷它們之間的關(guān)系無(wú)需任何 schema 配置。實(shí)際得到的表集合取決于你使用了哪些 Langfuse 功能——只有被使用過(guò)的功能其對(duì)應(yīng)的表才會(huì)有數(shù)據(jù)被寫(xiě)入。與 LLM 可觀測(cè)性與評(píng)估最相關(guān)的表organizations— 頂層多租戶容器每個(gè) org 擁有自己的 projects 與 members。projects— 一個(gè) project 劃定某個(gè)應(yīng)用的所有 traces、scores 與 datasets。trace_sessions— 把相關(guān) traces 歸組為一個(gè) session如多輪對(duì)話帶environment字段用于區(qū)分 staging 與 production。datasets— 用于評(píng)估運(yùn)行的精選示例集合可通過(guò) Langfuse Web UI 或 SDK 創(chuàng)建。dataset_items— dataset 內(nèi)的單條記錄每條攜帶input、可選的expected_output以及回指到產(chǎn)生它的 source trace/observation 的鏈接。score_configs— 命名評(píng)分類(lèi)型定義數(shù)值或分類(lèi)可帶min_value/max_value邊界子表score_configs__categories存放分類(lèi)標(biāo)簽。eval_templates— 帶版本的 LLM-as-judge 模板包含prompt、model、provider和結(jié)構(gòu)化輸出 schema子表eval_templates__vars存放變量名。models— LLM 模型定義含input_price、output_price與 tokenizer 配置用于計(jì)算每條 trace 的成本。prices/pricing_tiers— 關(guān)聯(lián)到模型定義的分層定價(jià)規(guī)則。annotation_queues— 人工標(biāo)注工作流條目記錄在annotation_queue_items指派人記錄在annotation_queue_assignments。comments— 用戶附加到任意 Langfuse 對(duì)象trace、dataset 等上的評(píng)論。dashboards/dashboard_widgets— 已保存的分析儀表盤(pán)及其圖表配置。audit_logs— 平臺(tái)內(nèi)所有 create/update/delete 動(dòng)作的完整審計(jì)軌跡。部分表屬于 Langfuse 內(nèi)部實(shí)現(xiàn)如_prisma_migrations、background_migrations可以通過(guò)sql_database(..., table_names...)顯式選擇要加載的表子集把它們排除在外。目標(biāo) schema 實(shí)體關(guān)系圖dlt 在minimal反射級(jí)別下加載后會(huì)在目標(biāo)端如 DuckDB生成如下 ER 圖所示的表結(jié)構(gòu)包含 dlt 自身的_dlt_loads、_dlt_version、_dlt_pipeline_state元數(shù)據(jù)表以及由 JSON 列規(guī)范化拆分出的*__*子表與_dlt_parent_id/_dlt_list_idx關(guān)聯(lián)列從上圖可以直觀看到projects與organizations、datasets、score_configs、trace_sessions、eval_templates、comments等表之間存在外鍵關(guān)聯(lián)——這正是resolve_foreign_keysTrue發(fā)揮作用的地方dataset_items的input、expected_output、metadata等 JSON 字段則被展開(kāi)為dataset_items__input__messages、dataset_items__expected_output__parts等多層子表。完整源碼Langfuse → DuckDB 導(dǎo)出管道以下腳本是完整的可運(yùn)行示例。它定義了一個(gè)langfuse_source源把整庫(kù)導(dǎo)出到本地 DuckDBimport dlt from dlt.sources.sql_database import sql_database dlt.source def langfuse_source(credentialsdlt.secrets.value): return sql_database( credentialscredentials, reflection_levelminimal, resolve_foreign_keysTrue, ) if __name__ __main__: pipeline dlt.pipeline( pipeline_namelangfuse, # can be configure to any dlt destination destinationduckdb, ) load_info pipeline.run(langfuse_source) print(load_info)關(guān)鍵點(diǎn)解讀dlt.source裝飾器把langfuse_source聲明為 dlt 源credentialsdlt.secrets.value讓它從.dlt/secrets.toml自動(dòng)讀取[sources.sql_database.credentials]一節(jié)。reflection_levelminimal只做輕量反射、類(lèi)型由數(shù)據(jù)推斷配合 JSON 列自動(dòng)拆分適合 Langfuse 這種含大量嵌套input/metadata字段的庫(kù)。resolve_foreign_keysTrue讓datasets → projects、prices → models等外鍵關(guān)系在目標(biāo)端以references提示保留。destinationduckdb零配置即可本地運(yùn)行注釋已標(biāo)明可以換成任意 dlt 目標(biāo)如postgres、filesystem、bigquery、snowflake等。運(yùn)行后load_info會(huì)打印每個(gè)資源表的加載狀態(tài)與行數(shù)。管道默認(rèn)把數(shù)據(jù)寫(xiě)入 DuckDB 的langfusedatasetdlt 會(huì)自動(dòng)建好目標(biāo)表與_dlt_loads等元數(shù)據(jù)表。進(jìn)階配置與常見(jiàn)優(yōu)化1. 只加載關(guān)鍵表跳過(guò)內(nèi)部表Langfuse 庫(kù)中包含_prisma_migrations、background_migrations等內(nèi)部表。通過(guò)table_names顯式指定子集即可只加載與可觀測(cè)性相關(guān)的表source sql_database( credentialsdlt.secrets.value, reflection_levelminimal, resolve_foreign_keysTrue, table_names[ organizations, projects, trace_sessions, datasets, dataset_items, score_configs, eval_templates, models, prices, pricing_tiers, comments, audit_logs, ], )從源碼看sql_database()在table_names給定后只會(huì)反射并生成這些表的資源而如果使用.with_resources(...)則會(huì)先反射全庫(kù)再做過(guò)濾對(duì)大 schema 而言顯式傳table_names能顯著提升性能。2. 按增量方式持續(xù)同步Langfuse 各表普遍帶有created_at/updated_at時(shí)間戳列。使用sql_table資源配合dlt.sources.incremental即可實(shí)現(xiàn)增量抽取例如只加載增量更新的dataset_itemsimport dlt from dlt.sources.sql_database import sql_table table sql_table( tabledataset_items, incrementaldlt.sources.incremental(updated_at), reflection_levelfull, resolve_foreign_keysTrue, )dlt 會(huì)在狀態(tài)state中記錄游標(biāo)最大值后續(xù)運(yùn)行自動(dòng)生成WHERE updated_at ...的查詢避免全表重復(fù)抽取。3. 切換后端以提升吞吐sql_database()支持四種后端backend參數(shù)見(jiàn) 源碼定義sqlalchemy默認(rèn)產(chǎn)出字典批次無(wú)額外依賴、pyarrow產(chǎn)出 Arrow 表類(lèi)型保真、性能更好、pandas產(chǎn)出 DataFrame、connectorxRust 實(shí)現(xiàn)通常最快但會(huì)忽略chunk_size。對(duì)包含decimal價(jià)格的models/prices表PyArrow 后端能避免精度損失source sql_database( credentialsdlt.secrets.value, reflection_levelfull_with_precision, resolve_foreign_keysTrue, backendpyarrow, )4. 在配置文件中管理表與列選擇除了在代碼中傳參也可以在config.toml中聲明要加載的表與列表名列名大小寫(xiě)敏感須與數(shù)據(jù)庫(kù)中完全一致# 選擇要加載的表 [sources.sql_database] table_names [ projects, dataset_items, ] # 只加載 dataset_items 的部分列 [sources.sql_database.dataset_items] included_columns [ id, dataset_id, input, expected_output, status, updated_at, ]5. 擴(kuò)展到其他目標(biāo)DuckDB 適合本地零配置分析生產(chǎn)環(huán)境中把pipeline的destination換成postgres、bigquery、snowflake、filesystemparquet/iceberg/delta 湖等即可源端代碼無(wú)需改動(dòng)。若 Langfuse 部署在遠(yuǎn)程且數(shù)據(jù)庫(kù)不可直連可參考倉(cāng)庫(kù)中sql_database官方源文檔 里的 SSH 隧道方案sshtunnel 自定義 SQLAlchemy engine來(lái)建立安全通道。總結(jié)Langfuse 的 Postgres 后端包含了 traces、scores、datasets、成本模型等完整的 LLM 可觀測(cè)性數(shù)據(jù)通過(guò) dlt 內(nèi)置的sql_database()源可以在零手工建表的前提下把這些數(shù)據(jù)整庫(kù)或按需導(dǎo)入 DuckDB 及其他數(shù)據(jù)湖倉(cāng)/數(shù)倉(cāng)。核心要點(diǎn)可歸納為三條在.dlt/secrets.toml配置postgresql憑據(jù)并安裝psycopg2-binary用reflection_levelminimal獲得穩(wěn)定的嵌套 JSON 規(guī)范化、用resolve_foreign_keysTrue保留表間關(guān)系用table_names或incremental控制加載范圍與增量同步。相關(guān)源碼與測(cè)試位于倉(cāng)庫(kù)的dlt/sources/sql_database/與tests/sources/sql_database/官方源的完整用法可進(jìn)一步查閱sql_database使用與配置文檔?！久赓M(fèi)下載鏈接】dltdata load tool (dlt) is an open source Python library that makes data loading easy ?項(xiàng)目地址: https://gitcode.com/GitHub_Trending/dl/dlt創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考