SDK de Python de Lumen En desarrollo
Tras ejecutar pip install lumen-ai, el SDK proporciona tres clases principales que cubren la recolección, la persistencia y la agregación de costes:
LumenTracer
Recolección de Trace + Cost en forma de callback.
LumenCheckpointer
Capa de persistencia que reemplaza a SqliteSaver.
CostTracker
Consulta de agregación de costes entre agentes.
LumenTracer
python
from lumen_ai import LumenTracer
tracer = LumenTracer(
project="my-agent",
tags={"env": "prod", "team": "core"},
redact=("api_key", "password"), # 自动脱敏字段
)
graph = workflow.compile(callbacks=[tracer])Parámetros principales:
| Parámetro | Valor por defecto | Descripción |
|---|---|---|
project | "default" | Identificador del proyecto; determina la pertenencia de los datos |
tags | {} | Etiquetas clave-valor añadidas a cada Trace |
redact | () | Tupla de nombres de campos a ofuscar automáticamente |
backend | "local" | local / lurus / URL personalizada |
LumenCheckpointer
Reemplaza al SqliteSaver de LangGraph:
python
from lumen_ai import LumenCheckpointer
graph = workflow.compile(checkpointer=LumenCheckpointer())En comparación con SqliteSaver:
| Dimensión | SqliteSaver | LumenCheckpointer |
|---|---|---|
| Velocidad de recuperación | Nivel de milisegundos | Nivel de microsegundos (basado en el motor WAL de Kova) |
| Multiproceso | Requiere bloqueo | Soporte nativo |
| Persistencia remota | Requiere implementación propia | Integrada |
| Asociación de costes | Ninguna | Automática |
Reemplazo directo, compatible con la API
LumenCheckpointer es compatible con la interfaz BaseCheckpointSaver de LangGraph; basta con sustituir el SqliteSaver existente en una sola línea, sin necesidad de modificar la estructura del grafo. Consulta Integración del ecosistema · LangGraph.CostTracker
Consulta de agregación de costes entre agentes:
python
from lumen_ai import CostTracker
# 过去 24h
CostTracker.summary(hours=24)
# 按 Agent 拆分
CostTracker.by_agent(hours=24)
# 按 Graph 节点拆分
CostTracker.by_node(graph_id="research-graph")
# 预算告警
CostTracker.set_budget(daily_rmb=50.0, on_exceed=my_handler)Backend personalizado
Por defecto, los datos se escriben en ~/.lumen/. Para enviarlos a un destino remoto: LumenTracer(backend="https://lumen.lurus.cn/ingest"). O bien personaliza:
python
from lumen_ai.backend import Backend
class MyBackend(Backend):
def send(self, event): ...
LumenTracer(backend=MyBackend())Soporte para Async
Todas las operaciones de E/S del SDK son compatibles con asincronía:
python
import asyncio
from lumen_ai import LumenTracer
async def main():
tracer = LumenTracer()
async with tracer:
await graph.ainvoke({"query": "hi"})
asyncio.run(main())