Skill

Generate and Debug Logfire FastAPI Code

Name: Generate and Debug Logfire FastAPI Code
Availability: OnlineOnly
Author: Instructor

Example integration demonstrating how to use Logfire observability with FastAPI applications in the Instructor library for structured LLM output validation.

Get skill

Works with fastapilogfire

Instructor

Maintainer?

Spark score

out of 100

Updated 9 days ago

Version 1.15.3

Add to Favorites

Why it matters

Streamline your FastAPI application development by automatically generating and debugging code with Logfire integration. This asset helps ensure your logging is robust and your application runs smoothly.

Outcomes

What it gets done

Generate boilerplate code for Logfire-enabled FastAPI applications.

Assist in debugging issues related to logging and application errors.

Provide code examples for integrating Logfire with FastAPI.

Facilitate code reviews for Logfire and FastAPI implementations.

Install

Add it to your toolbox

Run in your project directory:

curl -fsSL https://spark.entire.vc/get/inst-logfire-fastapi | bash

Capabilities

What this skill does

Generate code

Writes source code or scripts from a description.

Debug

Traces errors to their root cause and suggests fixes.

Review code

Analyzes code for bugs, style issues, and improvements.

Overview

Logfire Fastapi

What it does

This is a reference example that demonstrates integrating Logfire observability tooling with FastAPI web applications that use the Instructor library. It provides a working implementation showing how to configure and instrument the stack for monitoring structured LLM output validation in production.

How it connects

Use this example when you're building FastAPI services with Instructor and need to add observability, tracing, or monitoring capabilities using Logfire. It's particularly relevant when you want to understand request flows, debug issues, or monitor performance in AI applications that validate and structure language model responses.

Source README

--- server.py ---

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)

@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

@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)

@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")

--- test.py ---

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")

Discussion