Use Pydantic AI + Logfire as an alternative observability stack

1import os
2import logfire
3from pydantic_ai import Agent
4from pydantic import BaseModel, Field
5from typing import Optional
6
7class LogfireInstrumentor:
8    """Encapsulates Logfire setup for Pydantic AI agents."""
9
10    def __init__(
11        self,
12        service_name: str = "genai-app",
13        token: Optional[str] = None,
14        environment: str = "development",
15    ):
16        self.service_name = service_name
17        self.token = token or os.getenv("LOGFIRE_TOKEN")
18        self.environment = environment
19        self._configured = False
20
21    def setup_logfire(self) -> bool:
22        if self._configured:
23            return True
24        if self.token is None:
25            logfire.configure(
26                send_to_logfire=False,
27                service_name=self.service_name,
28            )
29        else:
30            logfire.configure(
31                token=self.token,
32                service_name=self.service_name,
33                environment=self.environment,
34            )
35        logfire.instrument_pydantic_ai()
36        self._configured = True
37        return True
38
39    def create_traced_agent(
40        self,
41        model: str,
42        result_type: type,
43        system_prompt: str,
44    ) -> Agent:
45        if not self._configured:
46            self.setup_logfire()
47        return Agent(
48            model=model,
49            result_type=result_type,
50            system_prompt=system_prompt,
51        )
52
53class SummaryResult(BaseModel):
54    title: str = Field(description="Summary title")
55    key_points: list[str] = Field(description="Extracted key points")
56    confidence: float = Field(ge=0.0, le=1.0)
57
58instrumentor = LogfireInstrumentor(
59    service_name="summarization-service",
60    environment="staging",
61)
62agent = instrumentor.create_traced_agent(
63    model="openai:gpt-4o",
64    result_type=SummaryResult,
65    system_prompt="Extract structured summaries from text.",
66)

• Lines 1–4: Import the four critical dependencies—logfire for telemetry, Agent from pydantic_ai for the agent runtime, BaseModel for typed results, and Optional for nullable token handling.
• Lines 7–15: The LogfireInstrumentor.__init__ method accepts a service name, optional token, and environment string. When token is not passed explicitly, it falls back to the LOGFIRE_TOKEN environment variable, which may resolve to None if unset.
• Lines 17–19: The _configured guard prevents double-initialization, which would raise a Logfire SDK error in production.
• Lines 21–25: When self.token is None, Logfire is configured with send_to_logfire=False, activating console-only export mode. This ensures local development produces visible trace output without requiring a Logfire cloud account.
• Lines 26–31: The production path passes the token, service name, and environment to logfire.configure(), which establishes the HTTPS connection to Logfire's ingestion endpoint.
• Lines 32–33: The logfire.instrument_pydantic_ai() call is the single most important line—it monkey-patches the Pydantic AI agent runtime to emit spans for every agent.run() call, tool invocation, and result validation step.
• Lines 36–46: The create_traced_agent factory method ensures Logfire is configured before creating any agent, preventing the common bug of emitting untraced agent calls.
• Lines 49–52: The SummaryResult Pydantic model defines the structured output schema. Logfire automatically serializes this schema into trace metadata, so every trace includes the expected output structure.
• Lines 55–62: Instantiation wires everything together—the instrumentor configures Logfire once, and the agent is created with full tracing enabled.

Loading diagram...

• Line 1: Declares a Mermaid flowchart with top-down (TD) layout direction.
• Lines 2-8: Defines a subgraph labeled "Langfuse Trace (Manual Instrumentation)" containing a manually instrumented observability trace structure.
• Line 3: Creates a node LT (Trace: agent-run) with a solid arrow to LG1 (Generation: llm-call), representing the first LLM generation span within the trace.
• Line 4: Links the trace node LT to LS1 (Span: tool-execution), representing a tool execution span as a child of the trace.
• Line 5: Links the trace node LT to LG2 (Generation: llm-call-2), representing a second LLM generation span.
• Line 6: Attaches metadata (model, tokens, cost) to the first generation span LG1 using a dotted line (-.-), indicating associated but non-hierarchical information.
• Line 7: Attaches metadata (tool_name, duration) to the tool-execution span LS1 via a dotted line.
• Line 8: Closes the Langfuse subgraph block.
• Lines 10-22: Defines a subgraph labeled "Logfire Trace (Automatic Instrumentation)" containing a more granular, automatically instrumented trace structure.
• Line 11: Creates a trace node FT (Trace: agent.run) linking to FM (Span: model-request), the first model request span.
• Lines 12-13: Breaks down the first model request FM into two child spans: FV1 (request-validation) and FL1 (llm-api-call), showing automatic sub-span decomposition.
• Line 14: Links the trace FT to FTC (Span: tool-call), representing an automatically instrumented tool call span.
• Lines 15-17: Decomposes the tool-call span FTC into three child spans: FTV (arg-validation), FTE (tool-execution), and FTR (return-validation), showing Logfire's automatic validation wrapping around tool execution.
• Lines 18-20: Creates a second model request FM2 as a child of the trace, with its own child spans FL2 (llm-api-call-2) and FRV (result-validation), representing the agent's second LLM call with output validation.
• Line 21: Attaches a dotted-line annotation to FRV showing the structured output schema (SummaryResult{title, key_points, confidence}), illustrating that Logfire captures Pydantic validation details.
• Line 22: Closes the Logfire subgraph block.
• Line 24: Styles the Langfuse subgraph with a dark background (#1a1a2e), red border (#e94560), and light text.
• Line 25: Styles the Logfire subgraph with the same dark background, a blue border (#0f3460), and light text, visually distinguishing it from Langfuse.

1import logfire
2from langfuse.decorators import observe, langfuse_context
3from pydantic_ai import Agent
4from pydantic import BaseModel, Field
5from dataclasses import dataclass, field
6from typing import Any
7
8@dataclass
9class TraceComparison:
10    langfuse_span_count: int = 0
11    logfire_span_count: int = 0
12    langfuse_fields: set = field(default_factory=set)
13    logfire_fields: set = field(default_factory=set)
14    coverage_gap: set = field(default_factory=set)
15
16    def compute_gap(self) -> None:
17        self.coverage_gap = self.langfuse_fields - self.logfire_fields
18
19class DualInstrumentor:
20    def __init__(self, agent: Agent, langfuse_trace_name: str):
21        self.agent = agent
22        self.trace_name = langfuse_trace_name
23
24    @observe(name="dual-instrumented-call")
25    async def run_with_comparison(
26        self, prompt: str
27    ) -> tuple[Any, TraceComparison]:
28        langfuse_context.update_current_trace(
29            metadata={"comparison_mode": True}
30        )
31
32        with logfire.span("comparison-agent-run") as lf_span:
33            result = await self.agent.run(prompt)
34
35            comparison = TraceComparison(
36                langfuse_span_count=3,
37                logfire_span_count=lf_span.attributes.get(
38                    "logfire.span_count", 0
39                ),
40                langfuse_fields={"model", "tokens", "latency", "cost"},
41                logfire_fields=set(
42                    result.all_messages_json()[:50]
43                ),
44            )
45            comparison.compute_gap()
46
47        langfuse_context.update_current_observation(
48            metadata={
49                "logfire_span_count": comparison.logfire_span_count,
50                "coverage_gap": list(comparison.coverage_gap),
51            },
52        )
53        return result.data, comparison

• Lines 1–6: Both logfire and langfuse.decorators are imported side-by-side. This dual-import pattern is safe because each SDK maintains independent span contexts; they do not interfere with each other's trace propagation.
• Lines 9–19: The TraceComparison dataclass holds the comparison metrics. The coverage_gap field computed by compute_gap() identifies Langfuse fields that Logfire does not capture, which tells you exactly what custom metadata you need to add if migrating.
• Lines 22–25: The DualInstrumentor wraps a Pydantic AI Agent instance and a Langfuse trace name. The agent is already instrumented by Logfire via logfire.instrument_pydantic_ai(), so no additional decoration is needed on the Logfire side.
• Lines 27–29: The @observe() decorator creates the Langfuse trace. The method is async because Pydantic AI's agent.run() is an async operation that performs actual LLM API calls.
• Lines 30–33: The langfuse_context.update_current_trace() call tags this trace as a comparison run, making it filterable in the Langfuse dashboard.
• Lines 35–36: The logfire.span("comparison-agent-run") context manager creates a parent Logfire span. Inside this span, agent.run() automatically creates child spans for model requests, tool calls, and validations.
• Lines 38–48: The TraceComparison is populated with span counts and field sets from both systems. The langfuse_fields are hardcoded because you know what your @observe() decorators capture. The logfire_fields are dynamically extracted from the result's message JSON.
• Lines 47–53: After the comparison is computed, the gap analysis results are pushed back into the Langfuse trace as metadata. This creates a self-documenting trace that tells you exactly what information you would lose (or gain) by switching to Logfire.