跳转至

Pydantic Logfire：Python 应用可观测性平台

为什么要学 Logfire

"代码能跑"和"代码在生产环境可靠运行"之间的差距，往往就在于可观测性（Observability）。当你的 AI Agent 调用了哪个模型、花了多少 token、某个请求为什么慢、数据库查询是否有 N+1 问题——这些问题如果没有可观测性工具，就只能靠 print 和猜测。

Pydantic Logfire 是 Pydantic 团队推出的可观测性平台，专为 Python 开发者设计。它基于 OpenTelemetry 标准，但极大简化了接入成本。核心理念是：一行代码接入，自动采集你需要的一切。

为什么选 Logfire 而不是 Datadog/Grafana： - 对 Python 生态原生支持（FastAPI、Django、SQLAlchemy、httpx 自动探针） - 对 AI/LLM 调用的特殊支持（PydanticAI、LangChain、OpenAI SDK 集成） - 极低的接入成本（不需要配置 Collector、Exporter） - 免费 tier 足够个人和小团队使用

核心概念

白话解释

概念白话说明
Span一次操作的记录（如一个API请求、一次DB查询、一次LLM调用）
Trace一组相关 Span 的链路（从用户请求到所有下游调用）
Metric数值型度量（请求数、延迟P99、错误率等）
Instrumentation自动探针，无需改代码即可采集数据
Scrubbing自动脱敏，防止敏感数据进入日志
Live View实时查看正在发生的请求和链路
SQL Explorer用 SQL 查询你的可观测性数据
Alerts基于条件自动告警

Logfire vs 其他方案

特性LogfireDatadogGrafana+TempoLangfuse
Python集成一行代码SDK较重需配置较多仅LLM
AI/LLM支持原生有限无原生
OpenTelemetry基于支持基于不支持
自托管计划中不支持支持支持
定价慷慨免费tier昂贵免费免费tier
学习曲线极低中高低

安装配置

安装

pip install logfire

初始化配置

# 第一次使用，登录并创建项目
logfire auth

# 这会在浏览器打开认证页面，完成后本地会保存 token

代码集成

import logfire

# 一行初始化（自动读取本地配置）
logfire.configure()

# 或显式指定 token（CI/CD环境）
logfire.configure(token="your-write-token")

验证

import logfire

logfire.configure()
logfire.info("Hello from Logfire! 应用启动成功")

在 https://logfire.pydantic.dev 的 Live View 中应该能看到这条日志。

快速上手

基础日志

import logfire

logfire.configure()

# 不同级别
logfire.trace("极细粒度调试信息")
logfire.debug("调试信息")
logfire.info("一般信息，user_id={user_id}", user_id=42)
logfire.warn("警告：重试次数过多")
logfire.error("错误：连接超时", url="https://api.example.com")

使用 Span 追踪操作

import logfire
import time

logfire.configure()

# 使用装饰器
@logfire.instrument("处理订单")
def process_order(order_id: int):
    time.sleep(0.1)  # 模拟处理
    return {"status": "completed"}

# 使用上下文管理器
def complex_operation():
    with logfire.span("复杂操作"):
        with logfire.span("步骤1: 数据验证"):
            time.sleep(0.05)
        with logfire.span("步骤2: 业务逻辑"):
            time.sleep(0.1)
        with logfire.span("步骤3: 持久化"):
            time.sleep(0.05)

process_order(123)
complex_operation()

FastAPI 自动追踪

from fastapi import FastAPI
import logfire

app = FastAPI()
logfire.configure()

# 一行代码：自动追踪所有请求
logfire.instrument_fastapi(app)

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # 这个请求会自动出现在 Logfire 中
    # 包含：路径、方法、状态码、耗时、请求体
    return {"user_id": user_id, "name": "张三"}

数据库查询追踪

import logfire
from sqlalchemy import create_engine

logfire.configure()

engine = create_engine("sqlite:///app.db")

# 自动追踪所有 SQL 查询（含执行时间和参数）
logfire.instrument_sqlalchemy(engine)

进阶用法

1. AI/LLM 调用追踪

import logfire
from openai import OpenAI

logfire.configure()

client = OpenAI()

# 自动追踪所有 OpenAI 调用
logfire.instrument_openai(client)

# 现在每次调用都会记录：模型、token用量、耗时、输入输出
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "什么是可观测性？"}]
)

2. PydanticAI 集成

import logfire
from pydantic_ai import Agent

# PydanticAI 自动集成 Logfire，无需额外配置
logfire.configure()

agent = Agent("openai:gpt-4o-mini")
result = agent.run_sync("解释微服务")
# Logfire 中会显示完整的 Agent 执行链路：
# - 系统提示构建
# - LLM 调用
# - 工具调用（如有）
# - 重试（如有）
# - 最终结果

3. httpx 请求追踪

import logfire
import httpx

logfire.configure()

client = httpx.AsyncClient()
logfire.instrument_httpx(client)

# 所有外部 HTTP 请求自动追踪
async def fetch_data():
    resp = await client.get("https://api.github.com/users/octocat")
    return resp.json()

4. 自定义指标

import logfire

logfire.configure()

# 记录业务指标
def process_payment(amount: float, currency: str):
    with logfire.span("处理支付", amount=amount, currency=currency):
        # 业务逻辑...
        logfire.metric_counter("payments_processed").add(1, {"currency": currency})
        logfire.metric_histogram("payment_amount").record(amount, {"currency": currency})

5. 异常追踪

import logfire

logfire.configure()

@logfire.instrument("危险操作")
def risky_operation():
    try:
        result = 1 / 0
    except ZeroDivisionError:
        logfire.exception("除零错误")  # 自动附带堆栈信息
        raise

# 或使用 span
def another_operation():
    with logfire.span("数据处理") as span:
        try:
            data = process()
        except Exception as e:
            span.set_attribute("error", True)
            span.record_exception(e)
            raise

6. 数据脱敏

import logfire
from logfire import ScrubMatch

# 自定义脱敏规则
def custom_scrubber(match: ScrubMatch):
    if "password" in match.path:
        return "[REDACTED]"
    if "api_key" in match.path:
        return "[API_KEY]"
    return None  # 使用默认规则

logfire.configure(scrubbing=logfire.ScrubbingOptions(callback=custom_scrubber))

# 密码等敏感字段会被自动脱敏
logfire.info("用户登录", username="alice", password="secret123")
# 在 Logfire 中看到的是: password="[REDACTED]"

7. 与 Django 集成

# settings.py
import logfire

logfire.configure()
logfire.instrument_django()

# 自动追踪：
# - 所有视图请求
# - ORM 查询
# - 中间件执行时间
# - 模板渲染时间

8. 分布式链路追踪

import logfire
from fastapi import FastAPI, Request

app = FastAPI()
logfire.configure(service_name="order-service")
logfire.instrument_fastapi(app)

# 如果上游服务也用了 Logfire/OpenTelemetry
# trace context 会通过 HTTP headers 自动传播
# 在 Logfire UI 中可以看到跨服务的完整链路

@app.post("/orders")
async def create_order(request: Request):
    with logfire.span("创建订单"):
        # 调用下游服务时 trace context 自动传播
        async with httpx.AsyncClient() as client:
            await client.post("http://payment-service/charge", ...)

9. 使用 SQL Explorer 查询数据

Logfire 支持用 SQL 查询你的可观测性数据：

-- 找到最慢的 10 个 API 请求
SELECT 
    span_name,
    duration_ms,
    attributes->>'http.method' as method,
    attributes->>'http.url' as url
FROM spans
WHERE span_name LIKE 'GET %' OR span_name LIKE 'POST %'
ORDER BY duration_ms DESC
LIMIT 10;

-- 统计每日 LLM token 用量
SELECT 
    date_trunc('day', start_timestamp) as day,
    SUM((attributes->>'gen_ai.usage.total_tokens')::int) as total_tokens
FROM spans
WHERE attributes->>'gen_ai.system' IS NOT NULL
GROUP BY day
ORDER BY day;

常见问题

Q1: Logfire 免费版够用吗

免费 tier 包含： - 每月 50M spans - 30 天数据保留 - 最多 5 个项目 - 对于个人和小团队足够

Q2: 会影响应用性能吗

Logfire 使用异步批量上传，对性能影响极小（通常 < 1ms 开销/请求）。SDK 内部做了采样和缓冲。

Q3: 如何在生产环境使用

import logfire

# 通过环境变量控制
# LOGFIRE_TOKEN=xxx
# LOGFIRE_ENVIRONMENT=production

logfire.configure(
    environment="production",
    # 生产环境可以降低采样率
    sampling=logfire.SamplingOptions(head=0.1)  # 只采集 10%
)

Q4: 和 OpenTelemetry 的关系

Logfire SDK 底层就是 OpenTelemetry。所以： - 已有 OTel instrumentation 可以直接用 - Logfire 可以作为 OTel 的 exporter 目标 - 数据格式完全兼容 OTel 标准

Q5: 能只用来做日志吗

可以。即使不用 trace/span，Logfire 也可以作为结构化日志平台替代 print/logging：

import logfire
logfire.configure()

# 比 print 和 logging 都好的结构化日志
logfire.info("用户注册", email="a@b.com", source="google", plan="free")

参考资源

资源链接
官方网站https://pydantic.dev/logfire
文档https://logfire.pydantic.dev/docs
GitHubhttps://github.com/pydantic/logfire
PyPIhttps://pypi.org/project/logfire
集成列表https://logfire.pydantic.dev/docs/integrations
OpenTelemetryhttps://opentelemetry.io

小结： Logfire 做对了一件事——把可观测性的门槛从"需要运维团队配置"降到"开发者一行代码搞定"。特别是对于 AI/LLM 应用，能清楚看到每次模型调用的延迟、token 消耗和成本，是从玩具项目迈向生产级应用的关键一步。