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 端点、流式客户端测试的全部闭环文件作用requirements.txt锁定全部依赖版本包含instructor、logfire、fastapi、openai、uvicorn等server.pyFastAPI 应用主体Logfire 配置 三个结构化输出端点test.py用requests流式消费/extract端点的测试脚本Readme.md三步快速启动指南即本篇文章依据的骨架此外examples/logfire 目录还提供了三个同主题的配套示例classify.py垃圾邮件分类、validate.pyllm_validator校验、image.pyGPT-4V 表格提取它们展示了 Logfire 在非 Web 场景下的插桩方式可与本示例互相印证。第一步环境准备与依赖安装按照 Readme.md 的说明先创建虚拟环境并安装 requirements.txt 中列出的全部依赖python -m venv .venv source .venv/bin/activate pip install -r examples/logfire-fastapi/requirements.txtrequirements.txt 中锁定的关键版本如下以当前仓库为准pydantic2.7.1 openai1.24.1 instructor1.0.3 logfire0.28.0 fastapi0.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. }, streamTrue, ) for chunk in response.iter_content(chunk_size1024): if chunk: print(str(chunk, encodingutf-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_pluginlogfire.PydanticPlugin(recordall)) logfire.instrument_fastapi(app) logfire.instrument_openai(openai_client) client instructor.from_openai(openai_client)逐行拆解这段骨架logfire.configure(pydantic_pluginlogfire.PydanticPlugin(recordall))启用 Pydantic 插件并设置recordall意味着每次 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 校验的完整调用链。端点一单次结构化提取/userapp.post(/user, response_modelUserDetail) async def endpoint_function(data: UserData) - UserDetail: user_detail await client.chat.completions.create( modelgpt-3.5-turbo, response_modelUserDetail, messages[ {role: user, content: fExtract: {data.query}}, ], ) logfire.info(/User returning, valueuser_detail) return user_detail该端点接收一个UserData仅含query字符串通过 Instructor 将自然语言文本结构化提取为UserDetailname: str、age: int并在返回前用logfire.info(/User returning, valueuser_detail)手动打一条业务日志把提取结果作为一个可搜索的结构化字段写入。调用该端点后Logfire 控制台会给出三组关键信息Pydantic 校验结果OpenAI 返回的原始内容经 Instructor 校验、解析为UserDetail的完整过程与结果OpenAI 调用详情发送的 prompt 载荷、模型、token 消耗与耗时端点入参调用/user时传入的原始请求体这对生产环境复现问题极有价值。下图为单次提取请求的 Pydantic 校验日志图片来源docs/blog/posts/img/logfire-sync-pydantic-validation.png端点二asyncio 并行批处理/many-users当需要同时从多条查询中提取信息时逐个串行调用会浪费大量等待时间。示例利用asyncio.gather并发执行app.post(/many-users, response_modellist[UserDetail]) async def extract_many_users(data: MultipleUserData): async def extract_user(query: str): user_detail await client.chat.completions.create( modelgpt-3.5-turbo, response_modelUserDetail, messages[ {role: user, content: fExtract: {query}}, ], ) logfire.info(/User returning, valueuser_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]streamTrue实现流式提取并借助logfire.span与logfire.info把每个产出对象单独记录下来app.post(/extract, response_classStreamingResponse) async def extract(data: UserData): supressed_client AsyncOpenAI() logfire.instrument_openai(supressed_client, suppress_other_instrumentationFalse) client instructor.from_openai(supressed_client) users await client.chat.completions.create( modelgpt-3.5-turbo, response_modelIterable[UserDetail], streamTrue, 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, valueresp_json) yield resp_json return StreamingResponse(generate(), media_typetext/event-stream)这里有两个容易忽略但至关重要的细节新建一个独立的AsyncOpenAI()客户端并对其插桩主客户端openai_client已经配置了recordall级别的日志而流式场景下 Instructor 会对 partial部分解析对象做大量中间处理会产生海量噪音日志。示例通过新客户端避免污染主日志流。suppress_other_instrumentationFalse如 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, valueresp_json)单独记录。下图展示了流式请求在 Logfire 控制台中的呈现方式——每个流对象都被独立记录并聚合在自定义 span 之下图片来源docs/blog/posts/img/logfire-stream.png可观测性视角三条端点的对比与取舍端点响应模型并发模型输出方式可观测重点/userUserDetail单次调用JSONPydantic 校验 OpenAI 载荷/many-userslist[UserDetail]asyncio.gather并行JSON 数组并行 span 时间线、单条提取耗时占比/extractIterable[UserDetail]流式生成text/event-stream自定义 span 聚合、逐对象产出日志三个端点恰好覆盖了 LLM 服务的三种典型形态同步单条、并行批处理与流式输出。在同一套 Logfire 配置下无需为每种形态写单独的日志代码插桩会自动跟随调用链生成对应的 span 树。从示例到生产可复用的可观测要点综合 server.py 与配套文档可以沉淀出几条直接可复用的经验四行配置建立全局可观测logfire.configure(pydantic_plugin...)instrument_fastapi(app)instrument_openai(client)instructor.from_openai(client)的顺序不可颠倒——必须先插桩 OpenAI 客户端再交给 Instructor 包装。用logfire.span聚合业务阶段把一组相关操作如流式生成包进命名 span控制台里就能一眼看出该阶段整体耗时。用logfire.info记录结构化业务事件如logfire.info(/User returning, valueuser_detail)字段化的日志便于后续按值检索与关联。流式场景注意日志噪音为流式请求单独创建客户端并理解suppress_other_instrumentation对 partial 解析的影响。控制台级别调为 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 
长河君 · 2026/9/15 15:28:47 · 阅读 1.2k 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 端点、流式客户端测试的全部闭环文件作用requirements.txt锁定全部依赖版本包含instructor、logfire、fastapi、openai、uvicorn等server.pyFastAPI 应用主体Logfire 配置 三个结构化输出端点test.py用requests流式消费/extract端点的测试脚本Readme.md三步快速启动指南即本篇文章依据的骨架此外examples/logfire 目录还提供了三个同主题的配套示例classify.py垃圾邮件分类、validate.pyllm_validator校验、image.pyGPT-4V 表格提取它们展示了 Logfire 在非 Web 场景下的插桩方式可与本示例互相印证。第一步环境准备与依赖安装按照 Readme.md 的说明先创建虚拟环境并安装 requirements.txt 中列出的全部依赖python -m venv .venv source .venv/bin/activate pip install -r examples/logfire-fastapi/requirements.txtrequirements.txt 中锁定的关键版本如下以当前仓库为准pydantic2.7.1 openai1.24.1 instructor1.0.3 logfire0.28.0 fastapi0.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. }, streamTrue, ) for chunk in response.iter_content(chunk_size1024): if chunk: print(str(chunk, encodingutf-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_pluginlogfire.PydanticPlugin(recordall)) logfire.instrument_fastapi(app) logfire.instrument_openai(openai_client) client instructor.from_openai(openai_client)逐行拆解这段骨架logfire.configure(pydantic_pluginlogfire.PydanticPlugin(recordall))启用 Pydantic 插件并设置recordall意味着每次 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 校验的完整调用链。端点一单次结构化提取/userapp.post(/user, response_modelUserDetail) async def endpoint_function(data: UserData) - UserDetail: user_detail await client.chat.completions.create( modelgpt-3.5-turbo, response_modelUserDetail, messages[ {role: user, content: fExtract: {data.query}}, ], ) logfire.info(/User returning, valueuser_detail) return user_detail该端点接收一个UserData仅含query字符串通过 Instructor 将自然语言文本结构化提取为UserDetailname: str、age: int并在返回前用logfire.info(/User returning, valueuser_detail)手动打一条业务日志把提取结果作为一个可搜索的结构化字段写入。调用该端点后Logfire 控制台会给出三组关键信息Pydantic 校验结果OpenAI 返回的原始内容经 Instructor 校验、解析为UserDetail的完整过程与结果OpenAI 调用详情发送的 prompt 载荷、模型、token 消耗与耗时端点入参调用/user时传入的原始请求体这对生产环境复现问题极有价值。下图为单次提取请求的 Pydantic 校验日志图片来源docs/blog/posts/img/logfire-sync-pydantic-validation.png端点二asyncio 并行批处理/many-users当需要同时从多条查询中提取信息时逐个串行调用会浪费大量等待时间。示例利用asyncio.gather并发执行app.post(/many-users, response_modellist[UserDetail]) async def extract_many_users(data: MultipleUserData): async def extract_user(query: str): user_detail await client.chat.completions.create( modelgpt-3.5-turbo, response_modelUserDetail, messages[ {role: user, content: fExtract: {query}}, ], ) logfire.info(/User returning, valueuser_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]streamTrue实现流式提取并借助logfire.span与logfire.info把每个产出对象单独记录下来app.post(/extract, response_classStreamingResponse) async def extract(data: UserData): supressed_client AsyncOpenAI() logfire.instrument_openai(supressed_client, suppress_other_instrumentationFalse) client instructor.from_openai(supressed_client) users await client.chat.completions.create( modelgpt-3.5-turbo, response_modelIterable[UserDetail], streamTrue, 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, valueresp_json) yield resp_json return StreamingResponse(generate(), media_typetext/event-stream)这里有两个容易忽略但至关重要的细节新建一个独立的AsyncOpenAI()客户端并对其插桩主客户端openai_client已经配置了recordall级别的日志而流式场景下 Instructor 会对 partial部分解析对象做大量中间处理会产生海量噪音日志。示例通过新客户端避免污染主日志流。suppress_other_instrumentationFalse如 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, valueresp_json)单独记录。下图展示了流式请求在 Logfire 控制台中的呈现方式——每个流对象都被独立记录并聚合在自定义 span 之下图片来源docs/blog/posts/img/logfire-stream.png可观测性视角三条端点的对比与取舍端点响应模型并发模型输出方式可观测重点/userUserDetail单次调用JSONPydantic 校验 OpenAI 载荷/many-userslist[UserDetail]asyncio.gather并行JSON 数组并行 span 时间线、单条提取耗时占比/extractIterable[UserDetail]流式生成text/event-stream自定义 span 聚合、逐对象产出日志三个端点恰好覆盖了 LLM 服务的三种典型形态同步单条、并行批处理与流式输出。在同一套 Logfire 配置下无需为每种形态写单独的日志代码插桩会自动跟随调用链生成对应的 span 树。从示例到生产可复用的可观测要点综合 server.py 与配套文档可以沉淀出几条直接可复用的经验四行配置建立全局可观测logfire.configure(pydantic_plugin...)instrument_fastapi(app)instrument_openai(client)instructor.from_openai(client)的顺序不可颠倒——必须先插桩 OpenAI 客户端再交给 Instructor 包装。用logfire.span聚合业务阶段把一组相关操作如流式生成包进命名 span控制台里就能一眼看出该阶段整体耗时。用logfire.info记录结构化业务事件如logfire.info(/User returning, valueuser_detail)字段化的日志便于后续按值检索与关联。流式场景注意日志噪音为流式请求单独创建客户端并理解suppress_other_instrumentation对 partial 解析的影响。控制台级别调为 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考