Instructor + FastAPI + Logfire：结构化输出服务的全链路可观测实战指南

Instructor + FastAPI + Logfire：结构化输出服务的全链路可观测实战指南

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

日志埋点与调试打印只能告诉你"程序跑到了哪"，却很难回答"一次 LLM 调用到底花了多少钱、耗时多久、模型返回了什么、Pydantic 校验是否通过"。本指南以仓库中 examples/logfire-fastapi 示例为主体，演示如何用 Logfire 对 Instructor + FastAPI 服务做全链路可观测：从单个结构化提取接口、asyncio并行批处理，到基于Iterable的流式响应，你将掌握如何用几行代码看清整条调用链的耗时、载荷与校验结果，并能在自己的服务里直接复用这套可观测方案。

示例目录结构与核心文件

该示例位于仓库的 examples/logfire-fastapi 目录，仅包含 4 个文件，却完整覆盖了"安装、启动、三个 API 端点、流式客户端测试"的全部闭环：

此外，examples/logfire 目录还提供了三个同主题的配套示例：classify.py（垃圾邮件分类）、validate.py（llm_validator校验）、image.py（GPT-4V 表格提取），它们展示了 Logfire 在非 Web 场景下的插桩方式，可与本示例互相印证。

第一步：环境准备与依赖安装

按照 Readme.md 的说明，先创建虚拟环境并安装 requirements.txt 中列出的全部依赖：

python -m venv .venv source .venv/bin/activate pip install -r examples/logfire-fastapi/requirements.txt

requirements.txt 中锁定的关键版本如下（以当前仓库为准）：

pydantic==2.7.1 openai==1.24.1 instructor==1.0.3 logfire==0.28.0 fastapi==0.110.3 uvicorn[standard] logfire[fastapi]

需要注意两点：一是logfire[fastapi]这个 extra 会额外安装 FastAPI 集成所需的依赖；二是这些版本是针对该示例编写时期的固定组合，若在你自己的项目中使用，建议参考 docs/blog/posts/full-fastapi-visibility.md 中的说明，先注册 Logfire 账号并创建一个项目，然后通过logfire auth完成命令行认证，后续日志才能正常上报到云端控制台。

第二步：启动服务与交互式文档

安装完成后，在examples/logfire-fastapi目录下用 uvicorn 启动服务（--reload便于开发调试时热重载）：

uvicorn server:app --reload

服务启动后，打开自动生成的交互式 API 文档：

http://127.0.0.1:8000/docs

你可以在 Swagger UI 中直接尝试三个 POST 端点。仓库中配套的 test.py 则提供了流式端点的命令行测试方式——它向/extract发送一段文本并逐块打印返回结果：

import requests response = requests.post( "http://127.0.0.1:3000/extract", json={ "query": "Alice and Bob are best friends. They are currently 32 and 43 respectively. " }, stream=True, ) for chunk in response.iter_content(chunk_size=1024): if chunk: print(str(chunk, encoding="utf-8"), end="\n")

运行后应看到两个独立的 JSON 对象逐块流出：

{"name":"Alice","age":32} {"name":"Bob","age":43}

注意：原脚本中的地址使用了端口3000，而 uvicorn 默认监听8000。若你的服务跑在 8000 端口，请把脚本中的端口改为8000，或将 uvicorn 显式指定为--port 3000。

第三步：理解 server.py 的可观测骨架

start 的精髓在于：用四行配置就让整个 FastAPI 应用、OpenAI 客户端和 Pydantic 校验全部纳入 Logfire 的追踪范围。

from pydantic import BaseModel from fastapi import FastAPI from openai import AsyncOpenAI import instructor import logfire import asyncio from collections.abc import Iterable from fastapi.responses import StreamingResponse class UserData(BaseModel): query: str class MultipleUserData(BaseModel): queries: list[str] class UserDetail(BaseModel): name: str age: int app = FastAPI() openai_client = AsyncOpenAI() logfire.configure(pydantic_plugin=logfire.PydanticPlugin(record="all")) logfire.instrument_fastapi(app) logfire.instrument_openai(openai_client) client = instructor.from_openai(openai_client)

逐行拆解这段骨架：

• logfire.configure(pydantic_plugin=logfire.PydanticPlugin(record="all"))：启用 Pydantic 插件并设置record="all"，意味着每次 Pydantic 模型校验的完整过程都会被记录。从 docs/blog/posts/full-fastapi-visibility.md 的截图可以看到，日志中不仅包含 OpenAI 调用的返回结果，还会记录校验的输入、输出与耗时。你也可以根据场景把record调成更细的粒度以控制日志量。
• logfire.instrument_fastapi(app)：对 FastAPI 应用整体插桩，每个请求的路径、请求参数、响应状态与耗时都会被记录为 span。若想看到请求体的具体内容，还需要像博客文章提示的那样，把控制台日志级别从默认的info调到debug。
• logfire.instrument_openai(openai_client)：对 OpenAI 异步客户端插桩，捕获每次 chat/completions 调用的完整请求载荷、模型名称、tokens 用量与响应。
• instructor.from_openai(openai_client)：在已插桩的客户端之上叠加 Instructor，让response_model结构化解析、重试与校验逻辑同样处于可观测范围内。

从实现看，Logfire 的插桩基于 OpenTelemetry 的 span 模型（docs/blog/posts/full-fastapi-visibility.md 中明确说明其通过 OpenTelemetry 提供关键洞察），因此instrument_fastapi与instrument_openai产生的子 span 会自动嵌套在请求的父 span 之下，形成一条从"HTTP 请求 → OpenAI 调用 → Pydantic 校验"的完整调用链。

端点一：单次结构化提取/user

@app.post("/user", response_model=UserDetail) async def endpoint_function(data: UserData) -> UserDetail: user_detail = await client.chat.completions.create( model="gpt-3.5-turbo", response_model=UserDetail, messages=[ {"role": "user", "content": f"Extract: `{data.query}`"}, ], ) logfire.info("/User returning", value=user_detail) return user_detail

该端点接收一个UserData（仅含query字符串），通过 Instructor 将自然语言文本结构化提取为UserDetail（name: str、age: int），并在返回前用logfire.info("/User returning", value=user_detail)手动打一条业务日志，把提取结果作为一个可搜索的结构化字段写入。

调用该端点后，Logfire 控制台会给出三组关键信息：

1. Pydantic 校验结果：OpenAI 返回的原始内容经 Instructor 校验、解析为UserDetail的完整过程与结果；
2. OpenAI 调用详情：发送的 prompt 载荷、模型、token 消耗与耗时；
3. 端点入参：调用/user时传入的原始请求体，这对生产环境复现问题极有价值。

下图为单次提取请求的 Pydantic 校验日志（图片来源：docs/blog/posts/img/logfire-sync-pydantic-validation.png）：

端点二：asyncio 并行批处理/many-users

当需要同时从多条查询中提取信息时，逐个串行调用会浪费大量等待时间。示例利用asyncio.gather并发执行：

@app.post("/many-users", response_model=list[UserDetail]) async def extract_many_users(data: MultipleUserData): async def extract_user(query: str): user_detail = await client.chat.completions.create( model="gpt-3.5-turbo", response_model=UserDetail, messages=[ {"role": "user", "content": f"Extract: `{query}`"}, ], ) logfire.info("/User returning", value=user_detail) return user_detail coros = [extract_user(query) for query in data.queries] return await asyncio.gather(*coros)

MultipleUserData接收queries: list[str]，内部为每条查询构建一个协程，再通过asyncio.gather(*coros)并行执行并汇总结果。结合 Logfire 的 span 树，可以清晰看到总耗时由最慢的那次 OpenAI 调用决定，其余调用在时间线上重叠执行——这正是判断"并行是否生效、瓶颈在哪里"的直接证据。若希望把每次extract_user拆成更细粒度的独立 span，可以在函数内部再包一层logfire.span(...)上下文管理器（如/extract端点所示）。

端点三：Iterable 流式提取/extract

流式场景对可观测性的要求更高：对象是一个个边生成边返回的，传统日志难以定位"第几个对象出了问题"。示例用Iterable[UserDetail]+stream=True实现流式提取，并借助logfire.span与logfire.info把每个产出对象单独记录下来：

@app.post("/extract", response_class=StreamingResponse) async def extract(data: UserData): supressed_client = AsyncOpenAI() logfire.instrument_openai(supressed_client, suppress_other_instrumentation=False) client = instructor.from_openai(supressed_client) users = await client.chat.completions.create( model="gpt-3.5-turbo", response_model=Iterable[UserDetail], stream=True, messages=[ {"role": "user", "content": data.query}, ], ) async def generate(): with logfire.span("Generating User Response Objects"): async for user in users: resp_json = user.model_dump_json() logfire.info("Returning user object", value=resp_json) yield resp_json return StreamingResponse(generate(), media_type="text/event-stream")

这里有两个容易忽略但至关重要的细节：

• 新建一个独立的AsyncOpenAI()客户端并对其插桩：主客户端openai_client已经配置了record="all"级别的日志，而流式场景下 Instructor 会对 partial（部分解析对象）做大量中间处理，会产生海量噪音日志。示例通过新客户端避免污染主日志流。
• suppress_other_instrumentation=False：如 docs/blog/posts/full-fastapi-visibility.md 中注释所指出的，这与 Instructor 解析 partial 的机制有关——保持该项关闭以确保流式传输过程中的对象解析过程本身不被二次插桩干扰，最终只记录完整的产出对象。

响应通过StreamingResponse以text/event-stream媒体类型逐块下发，generate()生成器内部用with logfire.span("Generating User Response Objects")将整个生成阶段聚合为一个自定义 span，每个user.model_dump_json()的产出对象则通过logfire.info("Returning user object", value=resp_json)单独记录。

下图展示了流式请求在 Logfire 控制台中的呈现方式——每个流对象都被独立记录，并聚合在自定义 span 之下（图片来源：docs/blog/posts/img/logfire-stream.png）：

可观测性视角：三条端点的对比与取舍

三个端点恰好覆盖了 LLM 服务的三种典型形态：同步单条、并行批处理与流式输出。在同一套 Logfire 配置下，无需为每种形态写单独的日志代码，插桩会自动跟随调用链生成对应的 span 树。

从示例到生产：可复用的可观测要点

综合 server.py 与配套文档，可以沉淀出几条直接可复用的经验：

1. 四行配置建立全局可观测：logfire.configure(pydantic_plugin=...)+instrument_fastapi(app)+instrument_openai(client)+instructor.from_openai(client)的顺序不可颠倒——必须先插桩 OpenAI 客户端，再交给 Instructor 包装。
2. 用logfire.span聚合业务阶段：把一组相关操作（如流式生成）包进命名 span，控制台里就能一眼看出该阶段整体耗时。
3. 用logfire.info记录结构化业务事件：如logfire.info("/User returning", value=user_detail)，字段化的日志便于后续按值检索与关联。
4. 流式场景注意日志噪音：为流式请求单独创建客户端，并理解suppress_other_instrumentation对 partial 解析的影响。
5. 控制台级别调为 debug 可看请求体：需要复现线上问题时，请求参数的可视化依赖更细的日志级别。

若想了解 Logfire 在非 Web 场景（分类、LLM 校验器、多模态提取）的更多用法，可以参考 examples/logfire 下的 classify.py、validate.py、image.py，以及仓库中的两篇配套博文：docs/blog/posts/logfire.md 与 docs/blog/posts/full-fastapi-visibility.md。此外，Instructor 本身关于并行与流式的底层实现可进一步查阅 docs/concepts/parallel.md、docs/concepts/iterable.md 与 docs/concepts/fastapi.md。

总结

examples/logfire-fastapi 这个不到百行的示例，完整演示了一条"FastAPI 接收请求 → Instructor 结构化提取 → Logfire 全链路追踪 → 流式返回"的现代 LLM 服务链路。它回答了三个实践问题：结构化输出服务如何快速搭建、异步与流式如何优雅落地、以及最关键的——当服务出问题时，如何用 span 树而非 print 语句快速定位瓶颈。将这套可观测骨架照搬到自己的服务中，只需替换 Pydantic 模型与 prompt，即可获得同等粒度的生产级追踪能力。

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