LLM Zoomcamp 2026 工作坊用 dltHub AI Workbench 把 Agent 日志变成可查询的数据管道与仪表盘【免费下载链接】llm-zoomcampLLM Zoomcamp - a free online course about real-life applications of LLMs. In 10 weeks you will learn how to build an AI system that answers questions about your knowledge base. Register here 项目地址: https://gitcode.com/GitHub_Trending/ll/llm-zoomcamp在 LLM Zoomcamp 2026 的 dlt 工作坊中由 dltHub 的 Alena Astrakhantseva 主讲你将学习如何把编码 AgentClaude Code、Codex、Copilot 等在本机产生的 JSONL 会话日志以及云端监控服务暴露的 REST API 追踪数据通过自然语言提示词驱动 dltHub AI Workbench 构建成 dlt 数据管道最终落入 DuckDB、产出 marimo 仪表盘并部署到 dltHub Platform 上实现定时刷新与团队共享。读完本文你将掌握用uvx dlthub-init脚手架搭建 AI 工作区、用 filesystem 与 rest_api 两类 dlt 源加载数据、理解 dlt 的规范化normalization机制、用 marimo 反应式笔记本做 SQL 优先的可视化以及如何在云端部署与调度这些管道。本文基于仓库 cohorts/2026/workshops/dlt 下的 7 篇课程文档与 4 个源码文件code/、lessons/编写全程以仓库内真实代码与配置为据。总览与课程索引见 dlt.md配套作业见 homework.md。工作坊总览你要构建什么每当你使用 Claude Code、Codex 或 Copilot 这类编码 Agent 时它都会在笔记本电脑上记录每个会话的元数据。日志存放在类似~/.claude/projects/的目录中以 JSONL 文件形式存在每行一个 JSON 对象包含 usage 数据、token 计数、模型名称、工具调用等信息——这些宝贵的数据被困在一种难以手工查询的嵌套格式里。这个工作坊的目标就是把这些日志转化为结构化数据表与仪表盘核心工具是 dlt 与 dltHub AI Workbench后者允许编码 Agent 根据自然语言提示词直接构建管道。完成全部课程后你将拥有一个将本地 Claude Code 日志加载进 DuckDB 的 dlt 管道一个基于该数据的 marimo 仪表盘展示活动、模型、token 与项目维度一个从托管 API 拉取 Agent 追踪数据的 REST API 管道一个部署在 dltHub Platform 上、带可分享仪表盘的定时任务。整体架构如下见 lessons/01-overview.md两条来源线本地 JSONL 日志、REST API 追踪经 dlt 管道汇入 DuckDB再向上供给 marimo 仪表盘与 dltHub Platform 的云端部署。前置条件开始前需要准备以下账号与工具Python 3.11 或更高版本uv包管理器一个编码 AgentClaude Code、Codex 或 Copilot一个免费的 dltHub Platform 账号app.dlthub.com本地有一些 Agent 日志使~/.claude/projects/下存在可加载的 JSONL 文件如果还没有先用你的 Agent 干一会儿活再回来继续。说明工作坊原文提供的外部链接uv 文档、dltHub 平台等在此不再重复列出请以工作坊文档与仓库内代码为准。脚手架工作区dltHub AI Workbench 自带脚手架命令在一个空文件夹中运行uvx dlthub-initlatest该命令会创建一个包含pyproject.toml、.dlt/配置目录、.claude/skills、.mcp.jsonMCP 服务器配置的工作区还会生成用于云端部署的__deployment__.py并创建虚拟环境。当它询问是否创建虚拟环境并安装依赖时选择“是”它会为你执行uv sync。在 Agent 中打开工作区把脚手架生成的文件夹用你的编码 Agent 打开。Agent 会读取 router skill当你让它构建管道时它会按数据源分派到对应的 toolkit工具包。确认 Workbench 正在运行uv run dlthub ai status本地开发阶段的目标库是 DuckDB——一种进程内分析型数据库无需启动服务dlt 直接写入磁盘上的.duckdb文件。由于 DuckDB 是 dlt 的依赖项无需额外配置即可使用。Part 1本地日志到管道filesystem 管道观察原始日志打开~/.claude/projects/并挑选一个.jsonl文件。每个会话是一个文件每行一个 JSON 对象。各行type值各不相同常见的有userassistantattachmentfile-history-snapshot数据嵌套很深usage 对象包含 token 计数message 对象包含 content 数组。这些是真实且有价值的数据——用过的模型、消耗的 token、调用的工具——但困在难以手工查询的格式里。用提示词构建管道告诉 Agent 为本地日志构建 dlt 管道build a dlt pipeline, load data from local Claude logs as raw JSONs into DuckDBAgent 会从 dltHub router skill 开始判断数据存在于磁盘文件中然后按需安装 filesystem-pipeline toolkit该项目开始时并不存在该 toolkitrouter 根据数据源把它拉进来。该 toolkit 会引导你走标准工作流确认计划 → 脚手架管道 → 配置凭据 → 运行。Agent 构建的管道管道使用 dlt 的filesystem源与read_jsonl读取器。源按 glob 列出文件读取器逐个打开并产出解析后的 JSON 记录。dlt 用管道操作符把它们连接起来from dlt.sources.filesystem import filesystem, read_jsonl reader ( filesystem(file_glob**/*.jsonl) | read_jsonl() ).with_name(messages)完整的管道见 code/filesystem_pipeline.py定义了一个load函数来创建并运行管道pipeline dlt.pipeline( pipeline_nameagent_logs, destinationduckdb, dataset_nameagent_logs, dev_modeTrue, ) load_info pipeline.run(reader, write_dispositionreplace)几个值得注意的点dev_modeTrue每次运行都会给 dataset 名加时间戳每次运行都从全新状态开始。开发时很方便但生产环境很浪费——后面会关掉它。write_dispositionreplace每次都会先删表再重载。仓库中的生产级版本课程文档中的示例是概念性的最小管道而仓库里实际随工作坊发布的 filesystem_pipeline.py 要完整得多它把 4 个 Agent 源统一成一张log_records表。从源码结构看其设计要点如下多源统一。SOURCES字典把 4 个 Agent 映射到各自的目录与 globClaude 风格布局放在projects/下claude、zlaudeCodex 风格布局放在sessions/YYYY/MM/DD/下codex、zodexSOURCES { claude: (ffile://{HOME}/.claude, projects/**/*.jsonl), zlaude: (ffile://{HOME}/.zlaude, projects/**/*.jsonl), codex: (ffile://{HOME}/.codex, sessions/**/*.jsonl), zodex: (ffile://{HOME}/.zodex, sessions/**/*.jsonl), }自定义 transformer 逐行解析。文档中的read_jsonl()示例会直接解析 JSON而仓库版本用一个dlt.transformerraw_reader逐行读取把每条原始 JSONL 行原样保留在data列并顺手抽出agent、session_id、source_file、line_no、type、timestamp等轻量字段便于后续分析rows.append( { agent: agent, session_id: session_id, source_file: rel_path, line_no: line_no, type: rec_type, timestamp: ts, data: line, } )这种做法把加载原始行与后续用 DuckDB 的 JSON 函数建模分离——文件注释里明确写着 Model later with DuckDBs JSON functions。解码时使用errorsreplace容忍坏 UTF-8并对type、timestamp的解析做了容错处理。采样模式。load(sampleTrue)命令行传--sample时每个源只读 1 个文件files.add_limit(1)、files_per_page1并用dev_modeTrue创建一次性 dataset便于快速验证全量加载时保持数据持久。管道运行后会打印info与pipeline.last_trace.last_normalize_info供检查。规范化78 张表管道运行时dlt 不只是把原始 JSON 倒进一张表。它会推断类型、扁平化嵌套对象、为嵌套数组创建子表并通过_dlt_id与_dlt_parent_id关联。第一次运行创建了 78 张表——Agent 日志嵌套很深dlt 把每个数组都拆成了子表。这是正确行为但 78 张表用起来太繁琐下一步会修复。本地查看数据dlt 自带内置仪表盘。从命令行而不是 Agent 会话中运行它。先确保管道已成功运行uv run dlthub local show这会打开一个从本地 DuckDB 文件读取数据的 marimo 仪表盘。你可以浏览 schema、查看表数量、查看每张表的数据、运行 SQL 查询——这是验证管道是否按预期加载数据的地方。Part 2调试与构建仪表盘管道运行并加载了数据但它创建了 78 张表。在构建仪表盘之前要确保管道正确、schema 可控。调试管道告诉 Agentdebug my pipelineAgent 会在 rest-api-pipeline toolkit 中找到 debug-pipeline skill 并安装它。调试意味着 Agent 运行管道、检查 trace、排查错误并修复直到可用。正如 Alena 在工作坊中解释的调试能确保管道不失败、能加载一些数据但它无法告诉你数据是否正确——那需要你亲自查看输出来判断。修复 schema 污染调试过程中Agent 发现 78 张表过多将其识别为 schema 污染——深层嵌套的 JSON 爆炸成了过多子表。修复方式是把部分列设为 JSON 数据类型而不是拆成子表。并非所有数据都被拆开管道现在只创建 40 张表深层嵌套字段保留为 JSON 列之后可用 DuckDB 的 JSON 函数查询。构建 marimo 报告告诉 Agent 构建仪表盘build a marimo report with detailed information about my Claude Code usageAgent 会安装>pipeline dlt.attach(agent_logs) dataset pipeline.dataset() df dataset( SELECT agent, COUNT(*) AS records FROM log_records GROUP BY 1 ORDER BY records DESC ).df()每个图表由一个数据 cell运行 SQL 并返回 DataFrame加一个图表 cell构建 Altair 可视化配对组成chart alt.Chart(df).mark_bar().encode( xalt.X(agent:N, sort-y), yrecords:Q, coloragent:N, tooltip[agent:N, records:Q], ).properties(titleTotal Log Records by Agent)仪表盘展示随时间变化的活动、按类型分布的消息、模型、token 用量与热门项目。在直播演示中Alena 能看到自己度假那一周在活动曲线上出现的缺口。Opus 是她最常用的模型。运行仪表盘uv run marimo edit code/claude_logs_dashboard.py仓库中的完整仪表盘仓库里随工作坊发布的 claude_logs_dashboard.py 是一个由app.cell装饰器组织的完整 marimo 笔记本可以看到若干值得注意的实现细节数据集连接。仓库版本不是简单dlt.attach(agent_logs)而是显式指定目标文件与 dataset并注释说明 catalog 名agent_logs_store与 schema 名agent_logs故意不同以规避 DuckDB 对 catalog/schema 的歧义解析pipeline dlt.pipeline( pipeline_nameagent_logs, destinationdlt.destinations.duckdb(agent_logs_store.duckdb), dataset_nameagent_logs, ) dataset pipeline.dataset()SQL 优先的查询风格。每个数据 cell 都通过dataset(...).df()直接跑 SQL例如用窗口函数ROW_NUMBER() OVER (PARTITION BY agent ...)取每个 Agent 排名前 8 的 record type、把其余归入otherRecord-Type Composition by Agent 图用COUNT(*) * 1.0 / COUNT(DISTINCT session_id)计算每会话平均记录数Records per Day by Type 图甚至对 y 轴使用scalealt.Scale(typelog)对数刻度因为 codex 的高频遥测event_msg/response_item会让会话类记录assistant、user在普通刻度下不可见——这些细节都写在 marimo 的 markdown cell 注释里。Part 3从托管 API 摄取REST API 管道Part 1 从磁盘加载本地日志。但在真实组织中Agent 运行在云端日志藏在 API 后面——例如 Logfire、Langfuse、Datadog 或 Anthropic API。你无法读取磁盘文件只能通过 HTTP 请求数据。为此工作坊准备了一个测试 API提供一百万条与真实日志结构相同的模拟 Claude Code 追踪数据。无需认证数据可安全分享。Base URL 为https://test-agent-traces-api-xt2e7ottma-ew.a.run.app云端日志器构建 Agent 时其日志存放在云端日志器中。Logfire、Langfuse 这类服务收集的元数据与本地 Claude 追踪类似usage、模型、工具调用、使用的 skills。要分析这些数据需要通过日志器的 REST API 请求再加载进数据库。每个日志器产出不同的追踪格式——不同的键、不同的嵌套、不同的字段顺序。当你同时使用多个 Agent 时这总是个问题dlt 会替你完成规范化。构建管道继续在同一个仓库中工作告诉 Agentbuild a dlt pipeline for https://test-agent-traces-api-xt2e7ottma-ew.a.run.app/docs, for /logs endpoint, load 20k logs into DuckDB, and build a similar marimo reportAgent 会安装 rest-api-pipeline toolkit其中包含创建管道、调试、探索数据与应用增量加载等 skills。Agent 会检查/docs地址的 OpenAPI spec弄清 base URL、分页类型与数据选择器然后写出管道。如果你曾手工为 API 构建过数据管道就会知道这省了多少事——你需要知道如何请求数据、如何分页、输出长什么样。REST API 源管道见 code/rest_api_pipeline.py把 API 描述成一个配置字典config: RESTAPIConfig { client: { base_url: base_url, paginator: { type: offset, limit: page_size, offset: 0, limit_param: limit, offset_param: offset, total_path: total, }, }, resources: [ { name: logs, endpoint: { path: /logs, data_selector: logs, }, primary_key: index, }, ], }data_selector告诉 dlt 记录位于响应信封的logs键下而非顶层。分页器使用基于 offset 的分页每次请求发送limit与offset查询参数dlt 从total键读取总数以决定何时停止。maximum_offset为 20000 时无需手写分页循环即可把加载量限制在 2 万行。仓库中的 REST 管道实现仓库版 rest_api_pipeline.py 把上述配置封装进dlt.source(nameagent_logs_api)装饰的源函数dlt.source(nameagent_logs_api) def agent_logs_source(base_url: str dlt.config.value, page_size: int 1000): ... yield from rest_api_resources(config)值得注意的实现细节base_url使用dlt.config.value作为默认值意味着它可以通过.dlt/config.toml的[sources.agent_logs_api]段自动注入注释明确说明这一点page_size默认 1000即每页记录数。resource_defaults里设置了write_disposition: replace与 Part 1 的加载语义保持一致。管道名agent_traces与 dataset 名traces刻意不同文件注释说明是为了避免 DuckDB 的 catalog/schema 歧义解析错误——与仪表盘中的处理方式一脉相承。--full标志不带参数默认source.add_limit(1)只拉 1 页1000 条传--full才全量拉取 100 万条。运行先用样本运行再做全量加载uv run python code/rest_api_pipeline.py # one page, 1000 records uv run python code/rest_api_pipeline.py --full # all 1 million records同样的规范化会发生dlt 推断类型、把message.content这类嵌套对象扁平化为子表通过_dlt_parent_id关联。嵌套的usage对象会变成usage__output_tokens这类双下划线分隔的列。Part 4部署到云端两条管道都能在本地运行但本地仪表盘无法与团队分享。dltHub Platform 允许你把管道与仪表盘部署到云端、调度运行并与同事共享。登录把本地工作区连接到 dltHub Platformuv run dlthub login # device-code OAuth in the browser uv run dlthub workspace connect # pick or create a workspace连接后打开平台 UIuv run dlthub show每个新账号都有一个 playground 工作区本地工作区会自动连接到它因此你在本地运行的任何内容都会同步到平台。部署管道告诉 Agent 部署 REST API 管道deploy this on the dlthub platform, use duckdb as destinationAgent 会安装 dlthub-platform toolkit部署前走完一个五步检查清单然后把管道注册进__deployment__.py并部署。也可以手动完成uv run dlthub deploy # ship the current project as a new version uv run dlthub run # run the pipeline on the cloud每次代码变更后重复这个部署-运行循环让云端始终反映最新版本。临时存储以 DuckDB 为目的地部署时数据进入临时存储平台在容器里运行管道任务结束后本地文件被清除数据不会跨运行持久化。切换到 Playground 目的地要持久化数据把duckdb换成playground目的地——这是一个托管的 S3 lake跨运行保留数据。在rest_api_pipeline.py中修改目的地# was: # destinationduckdb # now: destinationplaygroundplayground目的地需要deltalake包所以修改目的地后需要重新部署并运行uv run dlthub deploy uv run dlthub run如果运行因缺少deltalake而失败deploy 步骤会自动把依赖加进pyproject.toml重新部署再运行即可。Part 5部署仪表盘并调度管道已部署并写入 playground lake。现在把 marimo 仪表盘一起部署并设置调度。把仪表盘加入部署在__deployment__.py中导入仪表盘模块并加入__all__from agent_traces_dashboard import app as agent_traces_dashboard这把仪表盘注册为交互式 job。平台可以运行管道和交互式应用如 marimo 笔记本或 Streamlit 应用。仪表盘还需要指向 playground 目的地而不是 DuckDB。更新连接dlt.attach(agent_traces, destinationplayground, dataset_nameagent_logs)部署笔记本时必须显式把destination和dataset_name传给dlt.attach()。然后部署并运行uv run dlthub deploy uv run dlthub run运行模式在平台 UI 中打开笔记本。它以运行模式而非编辑模式运行所有代码被隐藏只显示报告与可视化——这正是你与团队分享的视图。数据存放在 playground 目的地它同样可以是 MotherDuck、BigQuery、Snowflake 或 LanceDB 这类向量数据库dlt 用同一套管道代码写入所有这些目标。分享发布仪表盘以获得公开 URLuv run dlthub job publish agent_traces_dashboard或者通过平台的 Users and Roles 在工作区内共享。调度为保持数据新鲜用 cron 触发器调度管道。把它加到__deployment__.py的装饰器上from dlt.hub.run import trigger run.pipeline(agent_traces, triggertrigger.schedule(0 12 * * *)) def ingest_agent_logs(): ...确认调度uv run dlthub job list还可以创建后续链followup chains先运行摄取管道成功后运行仪表盘刷新报告。平台支持job.success触发器把 job 串联起来。你也可以在平台 UI 中管理 job——启动/取消运行、管理调度。收尾与下一步回顾我们构建了两条 dlt 管道与两个仪表盘并部署到了云端一个把本地 Claude Code 日志加载进 DuckDB 的 filesystem 管道一个基于该数据的 marimo 仪表盘一个从托管 API 拉取追踪数据的 REST API 管道一个部署在 dltHub Platform、带可分享仪表盘的定时任务。你用自然语言描述需求编码 Agent 用 dltHub AI workbench 写出管道。Workbench 的 toolkits、skills 与 MCP 工具替你处理了 dlt 相关的专业知识。增量加载两条管道目前都使用write_dispositionreplace配合dev_modeTrue意味着每次运行都删掉并重载全部数据。开发阶段没问题但一旦达到百万行级别就不可扩展。dlt 会跟踪游标列最后加载的值并在后续运行中用它作为过滤条件游标存放在管道状态中跨运行持久化。对 filesystem 管道按文件修改日期过滤files filesystem( bucket_url..., file_glob..., incrementaldlt.sources.incremental(modification_date), )对 REST API 管道使用顺序 idresources: [ { name: logs, endpoint: {path: /logs, data_selector: logs}, primary_key: index, incremental: dlt.sources.incremental(index), }, ]把write_disposition切换为merge使已有行更新、新行插入而不删除任何数据并从管道中移除dev_mode让数据持久化。其他源与目的地dlt 内置源不止 filesystem 与 REST APIsql_database——从 Postgres、MySQL 等增量加载google_sheets——直接从电子表格拉数据notion——加载 Notion 页面与数据库hubspot、salesforce、stripe——厂商特定源认证已替你处理。工作流始终一致配置源 → 创建管道 → 运行。同一套管道代码也适用于 Postgres、BigQuery、Snowflake 与 Redshift——只需更换目的地字符串与凭据。关键概念速查以下是各概念在工作坊中的落点Toolkitsfilesystem、rest-api、data-exploration、dlthub-platform——按需安装每个都是一套引导式工作流MCP 工具查看管道、schema、行数与预览dlt 规范化嵌套 JSON 变成带类型的表与子表REST API 源offset 分页、data_selector、maximum_offset命名目的地playground在开发时是 duckdb、生产时是 S3 lake一套代码路径marimo反应式笔记本SQL 优先的数据 cell 加 Altair 图表 cell调度与触发器cronschedule与job.success后续链用装饰器声明。你产出的文件整个工作坊结束后你会得到这些文件均在 code/ 目录下有仓库版本可对照filesystem_pipeline.py # Part 1: local Claude logs - DuckDB claude_logs_dashboard.py # Part 1: usage report rest_api_pipeline.py # Part 2: agent_traces API - lake agent_traces_dashboard.py # Part 2: agent_traces report __deployment__.py # deployment manifest (jobs triggers)延伸学习想深入每个主题仓库内的配套资料包括dlt.md工作坊总览与视频链接、7 篇课程文档lessons/以及 homework.md配套作业。此外LLM Zoomcamp 主课程中还有监控模块05-monitoring与本工作坊的主题互补可对照学习如何用 Grafana 等工具观测 RAG 应用。【免费下载链接】llm-zoomcampLLM Zoomcamp - a free online course about real-life applications of LLMs. In 10 weeks you will learn how to build an AI system that answers questions about your knowledge base. Register here 项目地址: https://gitcode.com/GitHub_Trending/ll/llm-zoomcamp创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考