Lumen Python SDK In development
After pip install lumen-ai, the SDK provides three core classes covering collection, persistence, and cost aggregation:
LumenTracer
Callback-based Trace + Cost collection.
LumenCheckpointer
A persistence layer that replaces SqliteSaver.
CostTracker
Cross-agent cost aggregation queries.
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])Key parameters:
| Parameter | Default | Description |
|---|---|---|
project | "default" | Project identifier; determines data ownership |
tags | {} | Key-value tags attached to every Trace |
redact | () | Tuple of field names to redact automatically |
backend | "local" | local / lurus / custom URL |
LumenCheckpointer
Replaces LangGraph’s SqliteSaver:
python
from lumen_ai import LumenCheckpointer
graph = workflow.compile(checkpointer=LumenCheckpointer())Compared to SqliteSaver:
| Dimension | SqliteSaver | LumenCheckpointer |
|---|---|---|
| Recovery speed | Millisecond | Microsecond (powered by the Kova WAL engine) |
| Multi-process | Requires locking | Native support |
| Remote persistence | Self-implemented | Built-in |
| Cost correlation | None | Automatic |
Drop-in replacement, API-compatible
LumenCheckpointer is compatible with LangGraph’s BaseCheckpointSaver interface — just swap your existing SqliteSaver in a single line, with no changes to the graph structure. See Ecosystem Integration · LangGraph.CostTracker
Cross-agent cost aggregation queries:
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)Custom Backend
By default, data is written to ~/.lumen/. To send to a remote endpoint: LumenTracer(backend="https://lumen.lurus.cn/ingest"). Or customize:
python
from lumen_ai.backend import Backend
class MyBackend(Backend):
def send(self, event): ...
LumenTracer(backend=MyBackend())Async Support
All SDK I/O is async-friendly:
python
import asyncio
from lumen_ai import LumenTracer
async def main():
tracer = LumenTracer()
async with tracer:
await graph.ainvoke({"query": "hi"})
asyncio.run(main())