Logfire 观测与生产部署

一、为什么 Logfire 和 Pydantic AI 是一对

Logfire 是 Pydantic 团队出品的可观测性平台,底层基于 OpenTelemetry。它和 Pydantic AI 的关系就像 FastAPI 和 Starlette——同家出品,零配置就对齐。

二、五分钟接入 Logfire

1. 安装 + 登录

pip install "pydantic-ai[logfire]"
logfire auth                # 浏览器打开,OAuth 登录
logfire projects new my-agent
logfire projects use my-agent

执行完后 ~/.logfire/default.toml 会存下 token,本地 python 脚本直接可发数据。

2. 一行代码打点

import logfire
from pydantic_ai import Agent

logfire.configure()                          # 读取默认 token
logfire.instrument_pydantic_ai()             # ★ 关键一行

agent = Agent("openai:gpt-4o-mini", system_prompt="你是助理")
r = agent.run_sync("什么是古法编程?")
print(r.output)

跑完打开 https://logfire.pydantic.dev,你会看到一条完整的 trace 树:

3. 把整个应用全量打点

多个库时用 instrument_* 系列,或直接 instrument_all:

logfire.configure(service_name="my-agent", environment="prod")
logfire.instrument_pydantic_ai()
logfire.instrument_fastapi(app)               # FastAPI 端点自动 trace
logfire.instrument_httpx()                    # 外部 HTTP 请求
logfire.instrument_sqlalchemy(engine=engine)  # 数据库
logfire.instrument_redis()                    # Redis

现在一条请求从 HTTP 进来,到 Agent → LLM → 工具里的 SQL → Redis 写回,所有 span 都在一张 trace 上。出问题时把 trace id 甩给后端开发,5 分钟定位。

三、Logfire 帮你看到什么

四、手动 span:给业务节点加标签

Pydantic AI 自动打点的是框架层。你的业务代码想查也能查,得手动加 span:

import logfire

async def handle_chat(user_id: str, msg: str):
    with logfire.span("chat_turn", user_id=user_id, msg_preview=msg[:50]):
        history = await load_history(user_id)
        r = await agent.run(msg, message_history=history)
        await save_history(user_id, r.all_messages())
        logfire.info("chat completed", tokens=r.usage().total_tokens)
        return r.output

几个用法要点:

• logfire.span(name, **kwargs) 里的 kwargs 自动变成 span attributes,可以在 UI 里筛选
• logfire.info/warn/error(msg, **kwargs) 发结构化日志,自动挂到当前 span
• 避免把用户原文整段塞进去——Logfire 付费按数据量,敏感信息也别上报

五、成本与 token 追踪

Logfire 自动从每次 model request 里扒 input_tokens / output_tokens,结合 provider 的定价表算美元成本。你能直接在 SQL Explorer 查:

-- 最近 24h 每个 Agent 的总成本
SELECT
  attributes->>'agent_name' AS agent,
  sum(attributes->>'total_cost_usd')::float AS cost
FROM spans
WHERE start_timestamp > now() - '24h'::interval
  AND span_name = 'agent run'
GROUP BY agent
ORDER BY cost DESC;

生产期建议:

• 在代码里给 agent 命名:Agent(..., name="customer-support"),Logfire 会把 name 放进 span attributes,查表时一眼认出
• 每个请求带 user_id、session_id 到 span,支持按用户维度算成本、追异常用户
• 配合第 8 章的 UsageLimits,单会话超限制抛异常——双保险

六、FastAPI 生产模板

把前面 10 章的东西串起来,给一个能直接 docker run 的最小生产模板:

# app.py
import os
from contextlib import asynccontextmanager
import logfire
import httpx
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from pydantic_ai import Agent, UsageLimits
from pydantic_ai.exceptions import UsageLimitExceeded, ModelHTTPError

# ── Logfire ──
logfire.configure(
    service_name="qa-agent",
    environment=os.getenv("ENV", "dev"),
    send_to_logfire="if-token-present",  # 本地开发无 token 时不发
)
logfire.instrument_pydantic_ai()
logfire.instrument_httpx()

# ── Agent ──
agent = Agent(
    "openai:gpt-4o-mini",
    name="qa-agent",
    system_prompt="你是技术问答助理,回答准确简洁。",
    model_settings={"timeout": 20, "temperature": 0.2},
)

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动
    app.state.http = httpx.AsyncClient(timeout=10.0)
    yield
    # 关闭
    await app.state.http.aclose()

app = FastAPI(lifespan=lifespan)
logfire.instrument_fastapi(app)

# ── 路由 ──
class ChatIn(BaseModel):
    user_id: str
    message: str

@app.post("/chat")
async def chat(req: ChatIn):
    with logfire.span("chat", user_id=req.user_id):
        try:
            r = await agent.run(
                req.message,
                usage_limits=UsageLimits(output_tokens_limit=1500, request_limit=6),
            )
            return {"output": r.output, "usage": r.usage().model_dump()}
        except UsageLimitExceeded as e:
            raise HTTPException(429, detail=f"超出预算: {e}")
        except ModelHTTPError as e:
            logfire.error("model upstream failed", status=e.status_code)
            raise HTTPException(502, detail="模型服务不可用,稍后重试")

@app.post("/chat-stream")
async def chat_stream(req: ChatIn):
    async def gen():
        with logfire.span("chat_stream", user_id=req.user_id):
            async with agent.run_stream(req.message) as resp:
                async for chunk in resp.stream_text(delta=True):
                    yield f"data: {chunk}\n\n"
            yield "event: done\ndata: \n\n"
    return StreamingResponse(gen(), media_type="text/event-stream")

@app.get("/healthz")
async def healthz():
    return {"ok": True}

这个模板把前面章节的"基础功"都带上了:

• Logfire 全量观测,跑起来就能看 trace
• Agent 做了超时 + 温度配置
• 每次请求挂 UsageLimits,防止某个用户把账单打爆
• 错误分层:上游模型错返 502、预算超返 429、业务错默认 500
• FastAPI lifespan 管理共享 httpx client(如果有工具要调外部 API)
• 健康检查 /healthz 给 k8s 探针用

七、超时、重试、限流

多层超时

生产环境三层超时都要配:

# 1. 模型层 - model_settings.timeout
agent = Agent("openai:gpt-4o", model_settings={"timeout": 30})

# 2. HTTP 层 - httpx client(Pydantic AI 底层走 httpx)
from pydantic_ai.providers.openai import OpenAIProvider
import httpx

provider = OpenAIProvider(http_client=httpx.AsyncClient(timeout=30.0))

# 3. 请求级 - asyncio.wait_for 顶层兜底
import asyncio

try:
    r = await asyncio.wait_for(agent.run("..."), timeout=60)
except asyncio.TimeoutError:
    raise HTTPException(504, "处理超时")

429 限流:优雅重试

OpenAI / Anthropic 压力大时会 429。Pydantic AI 不自动重试 429(retries 只重试校验错),需手动:

import asyncio
from pydantic_ai.exceptions import ModelHTTPError

async def run_with_backoff(agent, prompt, max_retries=3):
    for attempt in range(max_retries):
        try:
            return await agent.run(prompt)
        except ModelHTTPError as e:
            if e.status_code == 429 and attempt < max_retries - 1:
                wait = 2 ** attempt + 0.5
                logfire.warn("rate limited, backoff", attempt=attempt, wait=wait)
                await asyncio.sleep(wait)
                continue
            raise

八、错误分类:告诉调用方"哪一种失败"

Agent 可能以几种完全不同的原因失败。把它们映射到 HTTP 状态码,前端/上游才能做对的处理:

from pydantic import ValidationError
from pydantic_ai.exceptions import (
    UsageLimitExceeded, ModelHTTPError, UnexpectedModelBehavior,
)

@app.exception_handler(Exception)
async def global_err(request: Request, exc: Exception):
    logfire.exception("unhandled", exc_type=type(exc).__name__)
    match exc:
        case ValidationError():
            return JSONResponse(400, {"error": "input invalid"})
        case UsageLimitExceeded():
            return JSONResponse(429, {"error": "quota exceeded"})
        case ModelHTTPError() as e if 500 <= e.status_code < 600:
            return JSONResponse(503, {"error": "upstream unavailable"})
        case ModelHTTPError():
            return JSONResponse(502, {"error": "provider rejected"})
        case UnexpectedModelBehavior():
            return JSONResponse(500, {"error": "model output invalid"})
        case _:
            return JSONResponse(500, {"error": "internal"})

九、Docker 部署

最小可用的生产 Dockerfile(基于 uv,第 88 门技艺那个):

# syntax=docker/dockerfile:1.7
FROM python:3.12-slim AS base
ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1 \
    UV_SYSTEM_PYTHON=1 UV_LINK_MODE=copy

# 非 root 用户,降风险
RUN useradd -m -u 10001 app
WORKDIR /app

# 装 uv
COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv

# 依赖层,利用缓存
COPY pyproject.toml uv.lock ./
RUN --mount=type=cache,target=/root/.cache/uv \
    uv sync --frozen --no-dev

# 代码层
COPY --chown=app:app . .
USER app

EXPOSE 8000
HEALTHCHECK --interval=30s --timeout=5s --retries=3 \
  CMD python -c "import httpx; httpx.get('http://127.0.0.1:8000/healthz').raise_for_status()"

CMD ["uv", "run", "uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2"]

配套 docker-compose.yml:

services:
  agent:
    build: .
    ports: ["8000:8000"]
    environment:
      OPENAI_API_KEY: ${OPENAI_API_KEY}
      LOGFIRE_TOKEN: ${LOGFIRE_TOKEN}
      ENV: prod
    restart: unless-stopped
    deploy:
      resources:
        limits:
          memory: 512M

十、上线 checklist

把它打印贴墙上,每次上线对一遍:

十一、本章小结

十二、十章总览

Pydantic AI 从 Agent("gpt-4o") 到生产上线,十章走下来:

这一整套下来,你已经能用 Pydantic AI 写出可 IDE 补全、可 mypy 校验、可 pytest 断言、可 Logfire 观测的 Agent 应用——不再是魔法黑盒,就是一段普通的 Python 代码。

接下来建议:

• 结合《Langfuse 实战》:如果不用 Logfire,自托管一个 Langfuse 是开源替代
• 结合《LiteLLM 实战》:把多 provider 路由 + 限流 + 观测收敛到网关
• 结合《LangGraph》:如果流程真的复杂到需要可视化 DAG 和 checkpoint
• 去翻 ai.pydantic.dev 官方文档,跟进最新更新

祝你在 LLM 时代,写出好读、好测、好改的 Agent。古法编程,精神不变——工具再变,能被理解的代码永远是最珍贵的。