使用 Instructor 与 Burr 构建 YouTube 闪卡生成应用：从结构化输出到可观测的 LLM 工作流

使用 Instructor 与 Burr 构建 YouTube 闪卡生成应用：从结构化输出到可观测的 LLM 工作流

【免费下载链接】instructorstructured outputs for llms项目地址: https://gitcode.com/GitHub_Trending/in/instructor

本文基于 Instructor 项目的官方实战指南，演示如何用instructor从 YouTube 字幕中可靠地提取结构化问答对（flashcards），再借助 Burr 将整个流程组织为易于理解和调试的 LLM 应用。读完本文，你将掌握 Pydantic 响应模型的约束设计、create_iterable流式多对象提取，以及用 Burr 的actions/transitions/State搭建带用户交互循环的完整应用，并学会接入 Burr UI 实现遥测与数据标注。完整可运行代码见 examples/youtube-flashcards/run.py，本文是对 docs/blog/posts/youtube-flashcards.md 的深度展开。

为什么要用 Instructor + Burr 的组合

闪卡（flashcards）能帮我们把复杂主题拆解成可消化的小块，无论是学生物、学外语还是背剧本台词都很有效。用 LLM 生成闪卡的难点有两个：

• 输出不可靠：LLM 返回的是自由文本，无法保证每次都能产出格式一致的题目、选项、答案索引和难度评分；
• 流程不可控：从「用户输入 URL」到「获取字幕」再到「生成题目」，多步骤串联后一旦出错难以定位。

Instructor 解决前者——它用 Pydantic 模型约束 LLM 的输出结构；Burr 解决后者——它用actions与transitions定义应用流程，并自带 Burr UI 提供本地优先、免费开源的观测、标注与调试能力。

本文是 Analyzing Youtube Transcripts with Instructor 的进阶篇：上一篇展示了如何将字幕切分为章节，本篇进一步把字幕转换为可用于自测的问答对，并用 Burr 把脚本包装成真正的交互式应用。

环境准备

安装所需的依赖包：

pip install openai instructor pydantic youtube_transcript_api "burr[start]"

其中"burr[start]"会一并安装 Burr 的可观测性相关组件。另外请确保已配置OPENAI_API_KEY环境变量（或采用你所用 provider 对应的认证方式）。

1. 用 Instructor 生成闪卡

1.1 定义 LLM 响应模型

Instructor 的核心用法是定义 Pydantic 模型作为 LLM 填写的模板。没有默认值的属性将由 LLM 生成：

import uuid from pydantic import BaseModel, Field from pydantic.json_schema import SkipJsonSchema class QuestionAnswer(BaseModel): question: str = Field(description="Question about the topic") options: list[str] = Field( description="Potential answers to the question.", min_items=3, max_items=5 ) answer_index: int = Field( description="Index of the correct answer options (starting from 0).", ge=0, lt=5 ) difficulty: int = Field( description="Difficulty of this question from 1 to 5, 5 being the most difficult.", gt=0, le=5, ) youtube_url: SkipJsonSchema[str | None] = None id: uuid.UUID = Field(description="Unique identifier", default_factory=uuid.uuid4)

这个例子集中展示了 Instructor 的几项关键能力：

• 用default/default_factory阻止 LLM 幻觉：id字段通过default_factory=uuid.uuid4在本地生成唯一标识，不需要（也不应该）让模型凭空捏造一个 UUID。
• 用SkipJsonSchema将字段排除在生成的 JSON Schema 之外：youtube_url由应用程序在运行时回填（见后文generate_question_and_answers中qna.youtube_url = youtube_url），我们不希望 LLM 猜测或虚构它。SkipJsonSchema在类型层面标注该字段不进入发给模型的 schema，同时保留在最终 Pydantic 对象上。
• 用Field约束生成内容：min_items=3, max_items=5限制选项数量在 3~5 个之间；ge=0, lt=5限制answer_index为合法的选项下标；gt=0, le=5将难度限制在 1~5 区间。这些约束会被编码进发给 LLM 的 JSON Schema，让模型在生成阶段就遵循规则，而非事后靠校验兜底。

补充说明：约束校验在生成阶段就生效，是因为 Instructor 会把 Pydantic 模型序列化为 schema 并作为工具参数（或 JSON Schema）传给模型；即便个别模型偶尔违反约束，Instructor 的校验层也能捕获并自动重试修正，相关机制可参考 docs/concepts/retrying.md。

1.2 获取 YouTube 字幕

使用youtube-transcript-api拉取视频的官方字幕。从 URL 中解析出 video id 后直接取回逐条字幕片段并拼成整段文本：

from youtube_transcript_api import YouTubeTranscriptApi youtube_url = "https://www.youtube.com/watch?v=hqutVJyd3TI" _, _, video_id = youtube_url.partition("?v=") segments = YouTubeTranscriptApi.get_transcript(video_id) transcript = " ".join([s["text"] for s in segments])

partition("?v=")会把 URL 在?v=处切成三段，中间的video_id即视频 ID。字幕片段可能带有时间戳等信息，这里仅拼接text字段作为 LLM 的输入文本。

1.3 用create_iterable生成问答对

一次调用即可从整段字幕中产出多条QuestionAnswer：

import instructor instructor_client = instructor.from_provider("openai/gpt-5-nano") system_prompt = """Analyze the given YouTube transcript and generate question-answer pairs to help study and understand the topic better. Please rate all questions from 1 to 5 based on their difficulty.""" response = instructor_client.create_iterable( model="gpt-4o-mini", response_model=QuestionAnswer, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": transcript}, ], )

关键点拆解：

1. 创建 Instructor 客户端：instructor.from_provider("openai/gpt-5-nano")采用provider/model的统一路由格式，底层实现在 instructor/v2/auto_client.py：它会按 provider 解析并包装对应厂商的 SDK（如 OpenAI 的openai.OpenAI），默认模式为Mode.TOOLS；还支持async_client=True返回异步客户端、cache=接入缓存等参数。也可以退而求其次使用instructor.from_openai(openai.OpenAI())显式包装（examples/youtube-flashcards/run.py 中即采用这种写法）。
2. .create_iterable()一次生成多个对象：该方法定义于 instructor/v2/core/client.py，它会内部构造一个包含tasks列表的动态模型（底层由 instructor/v2/dsl/iterable.py 的IterableModel生成），把流式响应逐个切分成独立对象，官方文档推荐优先用它而非手写Iterable[...] + stream=True，详见 docs/concepts/iterable.md。
3. 指定response_model=QuestionAnswer：保证每个输出都是校验过的QuestionAnswer实例。
4. 通过messages传递指令与输入：system消息给出任务说明（分析字幕、生成问答对、按 1~5 评分难度），user消息放入整段字幕文本。

返回值是一个生成器（generator），逐个迭代即可拿到QuestionAnswer对象：

print("Preview:\n") count = 0 for qna in response: if count > 2: break print(qna.question) print(qna.options) print() count += 1 """ Preview: What is the primary purpose of the new OpenTelemetry instrumentation released with Burr? ['To reduce code complexity', 'To provide full instrumentation without changing code', 'To couple the project with OpenAI', 'To enhance customer support'] What do you need to install to use the OpenTelemetry instrumentation with Burr applications? ['Only OpenAI package', 'Specific OpenTelemetry instrumentation module', 'All available packages', 'No installation needed'] What advantage does OpenTelemetry provide in the context of instrumentation? ['It is vendor agnostic', 'It requires complex integration', 'It relies on specific vendors', 'It makes applications slower'] """

注意：create_iterable返回的是惰性生成器，题目在迭代时才被逐个解析出来，因此非常适合「先出几条、边看边用」的场景。

2. 用 Burr 组装闪卡应用

前文的脚本是线性的；一旦要支持「用户反复输入 URL、逐轮生成」的交互，就需要把流程组织成状态机。Burr 用actions（应用能做的事）和transitions（动作间的流转）定义应用，同时保持流程图般的直观性，方便理解与调试。

2.1 定义actions

@action装饰器声明一个动作可读/可写哪些State字段；被装饰的函数以State为第一参数，返回更新后的State对象。这里定义三个动作，它们只是对前文代码片段的轻量重构：

from burr.core import action, State @action(reads=[], writes=["youtube_url"]) def process_user_input(state: State, user_input: str) -> State: """Process user input and update the YouTube URL.""" youtube_url = ( user_input # In practice, we would have more complex validation logic. ) return state.update(youtube_url=youtube_url) @action(reads=["youtube_url"], writes=["transcript"]) def get_youtube_transcript(state: State) -> State: """Get the official YouTube transcript for a video given it's URL""" youtube_url = state["youtube_url"] _, _, video_id = youtube_url.partition("?v=") transcript = YouTubeTranscriptApi.get_transcript(video_id) full_transcript = " ".join([entry["text"] for entry in transcript]) # store the transcript in state return state.update(transcript=full_transcript, youtube_url=youtube_url) @action(reads=["transcript", "youtube_url"], writes=["question_answers"]) def generate_question_and_answers(state: State) -> State: """Generate `QuestionAnswer` from a YouTube transcript using an LLM.""" # read the transcript from state transcript = state["transcript"] youtube_url = state["youtube_url"] # create the instructor client instructor_client = instructor.from_provider("openai/gpt-5-nano") system_prompt = ( "Analyze the given YouTube transcript and generate question-answer pairs" " to help study and understand the topic better. Please rate all questions from 1 to 5" " based on their difficulty." ) response = instructor_client.create_iterable( model="gpt-4o-mini", response_model=QuestionAnswer, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": transcript}, ], ) # iterate over QuestionAnswer, add the `youtube_url`, and append to state for qna in response: qna.youtube_url = youtube_url # `State` is immutable, so `.append()` returns a new object with the appended value state = state.append(question_answers=qna) return state

几个值得注意的细节：

• reads/writes声明让 Burr 能静态推断每个动作的数据依赖，这在可视化和调试时非常有用；
• State是不可变对象，state.update(...)与state.append(...)都返回新对象而非原地修改；
• question_answers在循环中被逐条追加：每条题目生成后立即写入 state，天然支持「边生成边查看」。

2.2 构建Application

使用ApplicationBuilder组装应用。最小配置需要三步：

• .with_actions()传入所有@action装饰过的函数；
• .with_transitions()用(from_action, to_action)元组定义动作间的流转；
• .with_entrypoint()指定第一个执行的动作。

from burr.core import ApplicationBuilder app = ( ApplicationBuilder() .with_actions( process_user_input, get_youtube_transcript, generate_question_and_answers, ) .with_transitions( ("process_user_input", "get_youtube_transcript"), ("get_youtube_transcript", "generate_question_and_answers"), ("generate_question_and_answers", "process_user_input"), ) .with_entrypoint("process_user_input") .build() ) app.visualize()

注意最后一个 transition("generate_question_and_answers", "process_user_input")：它让应用在生成完一轮题目后回到入口，等待下一次用户输入，从而形成可持续交互的循环。app.visualize()可随时输出应用图来理解逻辑流向（即文首那张流程图）。

2.3 运行应用并交互

Application.run()会执行动作直到命中停止条件。这里在进入process_user_input之前暂停，以便接收用户输入的 YouTube URL。run()返回三元组(action_name, result, state)，本例只关心state中的生成结果：

action_name, result, state = app.run( halt_before=["process_user_input"], inputs={"user_input": "https://www.youtube.com/watch?v=hqutVJyd3TI"}, ) print(state["question_answers"][0])

用while循环即可搭建一个简单的本地交互体验：

while True: user_input = input("Enter a YouTube URL (q to quit): ") if user_input.lower() == "q": break action_name, result, state = app.run( halt_before=["process_user_input"], inputs={"user_input": user_input}, ) print(f"{len(state['question_answers'])} question-answer pairs generated")

每轮输入一个视频 URL，应用就会依次完成「记录 URL → 拉取字幕 → 生成问答对」并在下一次循环回到等待输入的状态。

3. 进阶：把应用生产化

Burr 的Application本质是一个轻量 Python 对象，可以跑在 notebook、脚本、Streamlit/Gradio 等 Web 前端，甚至作为 Web 服务（如 FastAPI）对外提供。ApplicationBuilder提供了多个面向生产的特性：状态持久化（保存/恢复State，例如存放对话历史）、可观测性（记录 LLM 调用、token 用量、错误与重试）、流式与异步执行（让 UI 更跟手）。

3.1 接入 Burr UI 遥测

只需几行代码就能把遥测写入 Burr UI。关键顺序是：先对 OpenAI 库插桩，再创建 Instructor 客户端，否则无法捕获底层调用；随后在 builder 上调用.with_tracker()并指定项目名，开启use_otel_tracing=True启用 OpenTelemetry 追踪：

from burr.core import ApplicationBuilder from opentelemetry.instrumentation.openai import OpenAIApiInstrumentor # instrument before importing instructor or creating the OpenAI client OpenAIApiInstrumentor().instrument() app = ( ApplicationBuilder() .with_actions( process_user_input, get_youtube_transcript, generate_question_and_answers, ) .with_transitions( ("process_user_input", "get_youtube_transcript"), ("get_youtube_transcript", "generate_question_and_answers"), ("generate_question_and_answers", "process_user_input"), ) .with_tracker(project="youtube-qna", use_otel_tracing=True) .with_entrypoint("process_user_input") .build() )

开启后，Burr UI 中就能看到每次 OpenAI API 调用的完整信息——包括 prompt、响应模型以及响应内容，方便逐轮核对 Instructor 的结构化输出是否符合预期。

3.2 用标注工具沉淀评测数据

Burr UI 内置标注工具，可以对记录的运行数据（用户输入、LLM 响应、RAG 检索到的内容等）进行打标签、评分或评论。这对构建测试用例和评测数据集尤其有用——例如在多次运行后，把效果好的问答对挑出来作为回归测试样本。

4. 下一步：构建更复杂的 Agent

掌握「Instructor 保证可靠输出 + Burr 组织应用结构」后，可以按目标继续深入：

构建复杂 Agent：Instructor 通过结构提升 LLM 的推理质量。嵌套模型并叠加约束，可以在几行代码内实现 带引用的事实提取 或 知识图谱抽取；重试机制 则让 LLM 能在校验失败时自我修正。Burr 这边可以在transitions上加Condition条件，构建复杂却依然易于推理的工作流。

融入你的产品：把Application作为轻量组件嵌入 notebook、脚本或 Web 应用（如 FastAPI 服务）即可上线；结合状态持久化、可观测性与流式/异步能力，就能支撑真实用户场景。

总结

本文完整演示了一条可复用的技术路线：用 Instructor 的 Pydantic 模型约束 LLM 生成结构化闪卡，用youtube-transcript-api取回字幕作为输入，用create_iterable一次产出多条问答对，再用 Burr 的actions/transitions/State把它组装成可持续交互的应用，最后通过 Burr UI 实现遥测观测与数据标注。完整可运行的参考实现就在 examples/youtube-flashcards/run.py，现在就可以动手把它改造成自己的学习工具。

【免费下载链接】instructorstructured outputs for llms项目地址: https://gitcode.com/GitHub_Trending/in/instructor