LLM Zoomcamp 2026 dlt 与 Logfire 作业实战：为 Pydantic AI 智能体接入可观测性与数据管道

LLM Zoomcamp 2026 dlt 与 Logfire 作业实战：为 Pydantic AI 智能体接入可观测性与数据管道

【免费下载链接】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 工作坊的 homework/README.md 为骨架，完整讲解如何将一个基于 Pydantic AI 的 FAQ 智能体接入 Logfire 可观测性、再通过 dlt 将 trace 数据落入 DuckDB 并分析。读者将掌握：用uv初始化项目、用依赖注入方式给智能体装配检索工具、用两行代码完成 Logfire 埋点、以及用 dlt 自动将深层嵌套的 JSON trace 规范化成多张关联表并执行 SQL 查询——这套"智能体 + 可观测性 + 数据管道"的组合正是生产环境 RAG 应用监控的标准范式。

背景：为什么是 dlt + Logfire

本次作业的代码存放在 homework/ 目录，其任务链条非常清晰：把 Module 1 中手写的 FAQ 智能体重写为Pydantic AI框架，接入Pydantic Logfire做可观测性（observability），再用dlt把 trace 数据拉回本地DuckDB进行分析。

之所以这样组合，原因有三：

  • Pydantic AI 与 Logfire 同为 Pydantic 生态，二者配合开箱即用，logfire.instrument_pydantic_ai()一行即可自动记录智能体运行轨迹；
  • Module 5 中学过自建监控方案，Logfire 正是它的托管替代品，无需自建存储与查询层；
  • dlt 擅长把"深层嵌套、字段不固定"的 JSON（正是 trace 数据的形态）自动规范化为关系表，省去手写 schema 与展开逻辑。

先理解代码：作业里的三件套

作业目录 cohorts/2026/workshops/dlt/homework 下有三个核心文件，任务说明清晰：

  • ingest.py—— 下载课程 FAQ 并构建检索索引；
  • agent.py—— 用 Pydantic AI 构建 FAQ 智能体（系统提示词 + 搜索工具）；
  • main.py—— 入口，把索引、依赖与智能体串起来运行。

在开始之前，先读懂这三份代码，因为它们揭示了"手写 agentic loop"与"框架化 agent"之间的映射关系。

ingest.py：FAQ 数据加载与 minsearch 索引

ingest.py 定义了两个函数：

  • load_faq_data()：先从https://datatalks.club/faq/json/courses.json拉取课程列表，再逐个请求每个课程的 FAQ JSON，合并成文档列表；
  • build_index(documents)：用minsearch.Index构建检索索引，声明text_fields=['question', 'section', 'answer']与keyword_fields=['course']，然后index.fit(documents)。

这与 Module 1 的检索方案完全一致，FAQ 文档被切分为 question / section / answer 三个可检索文本字段，并保留 course 作为过滤字段。

agent.py：Pydantic AI 对"手写智能体"的三合一替换

agent.py 是本次作业的精华。代码注释明确说明：Module 1 手写智能体需要三样东西——系统提示词（developer message）、手写 JSON schema 的工具定义、以及手动实现的 while 循环 agentic loop。Pydantic AI 把三者全部替代：

  1. instructions 直接传入构造器：
faq_agent = Agent( 'openai:gpt-5.4-mini', deps_type=SearchDeps, instructions=INSTRUCTIONS, )
  1. 工具通过装饰器注册，类型注解自动生成 schema：
@faq_agent.tool def search(ctx: RunContext[SearchDeps], query: str) -> str: """Search the FAQ database for entries matching the given query.""" boost_dict = {'question': 3.0, 'section': 0.5} filter_dict = {'course': 'llm-zoomcamp'} results = ctx.deps.index.search( query, num_results=5, boost_dict=boost_dict, filter_dict=filter_dict ) return results

函数签名中的query: str自动成为模型可见的 JSON schema，docstring 成为工具描述，参数解析与调用由框架接管。检索时对 question 字段加权 3.0、section 加权 0.5，并按course == 'llm-zoomcamp'过滤，返回前 5 条。

  1. agentic loop 内置在run_sync()中——调用一次即返回最终答案，模型可在内部多次调用 search。

系统提示词 INSTRUCTIONS 规定了智能体行为：只能回答课程相关问题、优先用问题中的关键词发起首次检索、多次搜索并分析结果、FAQ 查不到就回答不了，结尾询问用户是否还有其他问题。

main.py：依赖注入与运行入口

main.py 展示了 Pydantic AI 的依赖注入模式：

deps = SearchDeps(index=index) result = faq_agent.run_sync(question, deps=deps)

SearchDeps是一个 dataclass 容器（见 agent.py），把索引显式传入，工具内通过ctx.deps.index访问。相比 Module 1 用全局变量的写法，这消除了隐藏的全局依赖，也让后续测试与替换更干净。

环境搭建：uv 初始化与依赖安装

作业 README 给出的第一步是用 uv 初始化项目并安装依赖：

uv init --no-readme uv add openai minsearch requests python-dotenv pydantic-ai

对应 pyproject.toml 中的依赖清单，实际项目还声明了logfire>=4.37.0（README 中未列出但作业正文要求安装，参考 homework.md 的uv add openai minsearch requests python-dotenv pydantic-ai logfire与uv add "dlt[duckdb]"）：

dependencies = [ "openai>=1.0.0", "minsearch>=0.0.11", "requests>=2.28.1", "python-dotenv>=1.0.0", "pydantic-ai>=0.0.30", "logfire>=4.37.0", ]

项目要求 Python 3.11+。接着复制环境变量模板并填入OPENAI_API_KEY：

cp .env.example .env

.env中写入OPENAI_API_KEY=sk-YOUR_KEY_HERE，并确保.env已被加入.gitignore。模型不限于 OpenAI——Pydantic AI 支持多种 provider，可查阅其文档切换。验证智能体可运行：

uv run python main.py

问题一：用 Logfire 埋点智能体

注册免费 Logfire 账号、创建项目并生成 write token，写入.env的LOGFIRE_TOKEN。埋点只需两行：

logfire.configure() logfire.instrument_pydantic_ai()

configure()读取环境变量完成初始化，instrument_pydantic_ai()自动为 Pydantic AI 的每次运行、每次 LLM 调用、每次工具调用生成 span。用不同问题运行几次后，到 Logfire 控制台即可看到完整 trace。

作业要求观察的查询是：

How do I run Ollama locally?

需要回答：单次智能体运行会产生多少个 span？每个 span 要么是智能体运行本身、要么是一次 LLM 调用、要么是一次工具调用。由于模型自行决定调用几次搜索，span 数量会在不同运行间浮动——这也是理解"LLM 应用输出不确定性"的一个直观样例。

问题二：用 dlt 把 traces 载入 DuckDB

先生成 Logfire 项目的 read token，写入.env的LOGFIRE_READ_TOKEN。然后参照工作坊（lessons/01-overview.md）初始化 dltHub 项目，用uvx dlthub-init@latest在工作目录生成脚手架，其中包括pyproject.toml、.dlt/配置目录、.claude/skills 与.mcp.json，之后用uv run dlthub ai status确认 workbench 可用。

接着让 coding agent 从 Logfire 拉取数据写入 DuckDB。作业建议使用 dltHub AI workbench 中现成的 Logfire 上下文；没有 coding agent 时可用 OpenCode 免费额度完成，或用 ChatGPT/搜索引擎按传统方式手写。

为什么必须用 dlt：深层嵌套 JSON 的自动规范化

Logfire traces 是深度嵌套的 JSON：span 属性里包含 LLM 消息、工具调用、token 用量等。dlt 会自动把它规范化成多张表——主表存放主记录，嵌套的每一层生成子表，通过_dlt_id与_dlt_parent_id关联（机制见 lessons/02-filesystem-pipeline.md 中对 78 张表的讨论，以及 lessons/04-rest-api-pipeline.md 中_dlt_parent_id与usage__output_tokens双下划线展开列的描述）。

作业要求用如下 SQL 统计 dlt 创建的表数量：

SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = 'agent_traces';

如果使用 dlt 的filesystemsource 加载本地 JSONL（工作坊 Part 1 的做法，见 code/filesystem_pipeline.py），标准写法是：

reader = ( filesystem(file_glob="**/*.jsonl") | read_jsonl() ).with_name("messages") pipeline = dlt.pipeline( pipeline_name="agent_logs", destination="duckdb", dataset_name="agent_logs", dev_mode=True, ) load_info = pipeline.run(reader, write_disposition="replace")
  • dev_mode=True每次运行给数据集名追加时间戳，开发期方便，生产环境建议关闭；
  • write_disposition="replace"每次重建表；
  • DuckDB 是 dlt 的内置依赖，进程内运行、无需独立服务，直接落盘为.duckdb文件。

若加载的是 REST API 数据（工作坊 Part 2 的思路），dlt 的 REST API source 以配置字典描述接口，例如：

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指定记录位于响应封套的logs键下，paginator 用 offset 分页并依据total键判断终止条件。完整示例见 code/rest_api_pipeline.py。

本地验证数据

管道跑通后，可用 dlt 自带仪表盘本地查看：

uv run dlthub local show

这会打开一个读取本地 DuckDB 的 marimo 仪表盘，可浏览 schema、查看各表数据、执行 SQL，用来确认管道加载是否符合预期（lessons/02-filesystem-pipeline.md）。

关于表数量的预期

注意一点：Logfire traces 的嵌套程度低于 Claude 本地 JSONL 日志。工作坊加载 Claude 日志时，首轮产生了 78 张表（随后通过把部分列保留为 JSON 类型压到 40 张，见 lessons/03-debug-and-dashboard.md）；而作业针对 Logfire traces 的表数量需要你实际运行information_schema.tables查询确认，从源码结构推断应处于一个远小于 78 的量级——这正是作业考察点，不要凭猜测作答。

问题三：用智能体查询 trace 中的 token 用量

最后一步是分析数据：找出问题一中那次智能体运行的input token 用量。

token 计数存放在 span 属性中，键为gen_ai.usage.input_tokens。需要把 trace 内所有 LLM 调用的该值求和。由于求和结果取决于智能体实际执行了几次搜索，作业要求报告其落入的区间（100–500 / 1500–5000 / 10000–20000 / 50000–100000）。

用 coding agent 完成此任务时，可以给出类似指令：

查询 agent_traces 数据集，找到 Q1 那次运行的 span，把gen_ai.usage.input_tokens属性求和

dlt 已将嵌套的 usage 对象展开为带双下划线的列（如gen_ai.usage.input_tokens对应展开列），因此 DuckDB 中可以直接按列名聚合，无需手工解析 JSON。这正是"dlt 规范化 + 数据库 SQL 分析"链路的价值所在：原本藏在多层 JSON 里的指标，现在变成了可查询的关系表。

小结：从手写循环到可观测管道

本次作业完整走通了三条主线：

  1. 框架化改造：Pydantic AI 用构造器参数、装饰器工具与内置循环，替换了 Module 1 手写的指令、JSON schema 与 while 循环，且保持了"检索 + 加权 + 过滤"的搜索逻辑不变（agent.py）；
  2. 可观测性接入：logfire.configure()+logfire.instrument_pydantic_ai()两行代码即获得完整的 span 级 trace，可观测每次 LLM 调用与工具调用；
  3. 数据管道落地：dlt 将深层嵌套的 trace JSON 自动规范化为多张关联表落入 DuckDB，随后用 SQL 对gen_ai.usage.input_tokens等指标做聚合分析。

这套"框架化智能体 + 托管可观测性 + 自动化数据管道"的组合，与 Module 5 自建监控互为补充，也直接复用了工作坊中 dlt 对嵌套 JSON 的规范化能力。完成三个问题后，到作业提交页提交答案即可。

延伸阅读：工作坊完整讲义见 lessons/ 目录（从 01-overview 到 07-where-to-go），本地 Claude 日志管道、调试与 marimo 仪表盘、REST API 管道与云端部署均有对应 lesson 与 code/ 下的可运行示例。

【免费下载链接】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），仅供参考