logfire

Logfire

使用 Pydantic Logfire 为 Python 提供结构化可观测性——快速设置、强大功能，并兼容 OpenTelemetry。

快速开始

uv pip install logfire

import logfire

logfire.configure(service_name="my-api", service_version="1.0.0")
logfire.info("Application started")

核心模式

1. 服务配置

始终在启动时设置服务元数据：

import logfire

logfire.configure(
    service_name="backend",
    service_version="1.0.0",
    environment="production",
    console=False,           # Disable console output in production
    send_to_logfire=True,    # Send to Logfire platform
)

2. 框架插桩

创建客户端／应用之前，先对框架进行插桩：

import logfire
from fastapi import FastAPI

# Configure FIRST
logfire.configure(service_name="backend")

# Then instrument
logfire.instrument_fastapi()
logfire.instrument_httpx()
logfire.instrument_sqlalchemy()

# Then create app
app = FastAPI()

3. 日志级别和结构化日志

# All log levels (trace → fatal)
logfire.trace("Detailed trace", step=1)
logfire.debug("Debug context", variable=locals())
logfire.info("User action", action="login", success=True)
logfire.notice("Important event", event_type="milestone")
logfire.warn("Potential issue", threshold_exceeded=True)
logfire.error("Operation failed", error_code=500)
logfire.fatal("Critical failure", component="database")

# Python 3.11+ f-string magic (auto-extracts variables)
user_id = 123
status = "active"
logfire.info(f"User {user_id} status: {status}")
# Equivalent to: logfire.info("User {user_id}...", user_id=user_id, status=status)

# Exception logging with automatic traceback
try:
    risky_operation()
except Exception:
    logfire.exception("Operation failed", context="extra_info")

4. 手动 Span

# Spans for tracing operations
with logfire.span("Process order {order_id}", order_id="ORD-123"):
    logfire.info("Validating cart")
    # ... processing logic
    logfire.info("Order complete")

# Dynamic span attributes
with logfire.span("Database query") as span:
    results = execute_query()
    span.set_attribute("result_count", len(results))
    span.message = f"Query returned {len(results)} results"

5. 自定义指标

# Counter - monotonically increasing
request_counter = logfire.metric_counter("http.requests", unit="1")
request_counter.add(1, {"endpoint": "/api/users", "method": "GET"})

# Gauge - current value
temperature = logfire.metric_gauge("temperature", unit="°C")
temperature.set(23.5)

# Histogram - distribution of values
latency = logfire.metric_histogram("request.duration", unit="ms")
latency.record(45.2, {"endpoint": "/api/data"})

6. 大模型可观测性

import logfire
from pydantic_ai import Agent

logfire.configure()
logfire.instrument_pydantic_ai()  # Traces all agent interactions

agent = Agent("openai:gpt-4o", system_prompt="You are helpful.")
result = agent.run_sync("Hello!")

7. 抑制嘈杂的插桩

# Suppress entire scope (e.g., noisy library)
logfire.suppress_scopes("google.cloud.bigquery.opentelemetry_tracing")

# Suppress specific code block
with logfire.suppress_instrumentation():
    client.get("https://internal-healthcheck.local")  # Not traced

8. 敏感数据清理

import logfire

# Add custom patterns to scrub
logfire.configure(
    scrubbing=logfire.ScrubbingOptions(
        extra_patterns=["api_key", "secret", "token"]
    )
)

# Custom callback for fine-grained control
def scrubbing_callback(match: logfire.ScrubMatch):
    if match.path == ("attributes", "safe_field"):
        return match.value  # Don't scrub this field
    return None  # Use default scrubbing

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

9. 高流量服务的采样

import logfire

# Sample 50% of traces
logfire.configure(sampling=logfire.SamplingOptions(head=0.5))

# Disable metrics to reduce volume
logfire.configure(metrics=False)

10. 测试

import logfire
from logfire.testing import CaptureLogfire

def test_user_creation(capfire: CaptureLogfire):
    create_user("Alice", "[email protected]")
    
    spans = capfire.exporter.exported_spans
    assert len(spans) >= 1
    assert spans[0].attributes["user_name"] == "Alice"
    
    capfire.exporter.clear()  # Clean up for next test

可用集成

类别	集成	方法
Web	FastAPI	logfire.instrument_fastapi(app)
	Starlette	logfire.instrument_starlette(app)
	Django	logfire.instrument_django()
	Flask	logfire.instrument_flask(app)
	AIOHTTP Server	logfire.instrument_aiohttp_server()
	ASGI	logfire.instrument_asgi(app)
	WSGI	logfire.instrument_wsgi(app)
HTTP	HTTPX	logfire.instrument_httpx()
	Requests	logfire.instrument_requests()
	AIOHTTP Client	logfire.instrument_aiohttp_client()
数据库	SQLAlchemy	logfire.instrument_sqlalchemy(engine)
	Asyncpg	logfire.instrument_asyncpg()
	Psycopg	logfire.instrument_psycopg()
	Redis	logfire.instrument_redis()
	PyMongo	logfire.instrument_pymongo()
大模型	Pydantic AI	logfire.instrument_pydantic_ai()
	OpenAI	logfire.instrument_openai()
	Anthropic	logfire.instrument_anthropic()
	MCP	logfire.instrument_mcp()
任务	Celery	logfire.instrument_celery()
	AWS Lambda	logfire.instrument_aws_lambda()
日志记录	标准日志记录	logfire.instrument_logging()
	Structlog	logfire.instrument_structlog()
	Loguru	logfire.instrument_loguru()
	Print	logfire.instrument_print()
其他	Pydantic	logfire.instrument_pydantic()
	系统指标	logfire.instrument_system_metrics()

常见陷阱

问题	症状	修复方法
缺少服务名称	在界面中难以找到 Span	在 configure() 中设置 service_name
插桩太晚	未捕获任何 Span	创建客户端之前调用 configure()
高基数属性	存储量激增	使用 ID，而不要将完整载荷用作属性
控制台噪声	日志污染 stdout	在生产环境中设置 console=False

参考资料

• 配置选项 - 所有 configure() 参数
• 集成指南 - 特定框架的设置方法
• 指标指南 - 计数器、仪表、直方图、系统指标
• 高级模式 - 采样、清理、抑制、测试
• 陷阱与故障排查 - 常见问题及解决方案
• 官方文档