Skip to content

Lumen Python SDK 开发中 ​

pip install lumen-ai 后,SDK 提供三大核心类,覆盖采集、持久化与成本聚合:

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])

主要参数:

参数默认值说明
project"default"项目标识,决定数据归属
tags{}每条 Trace 附加的键值对标签
redact()自动脱敏的字段名元组
backend"local"local / lurus / 自定义 URL

LumenCheckpointer ​

替代 LangGraph 的 SqliteSaver:

python
from lumen_ai import LumenCheckpointer

graph = workflow.compile(checkpointer=LumenCheckpointer())

相比 SqliteSaver:

维度SqliteSaverLumenCheckpointer
恢复速度毫秒级微秒级(基于 Kova WAL 引擎)
多进程需加锁原生支持
远程持久化需自实现内置
成本关联无自动

直接替换,API 兼容

LumenCheckpointer 兼容 LangGraph 的 BaseCheckpointSaver 接口,把原有 SqliteSaver 一行换掉即可,无需改动图结构。详见 生态集成 · LangGraph。

CostTracker ​

跨 Agent 成本聚合查询:

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 ​

默认数据写入 ~/.lumen/。发送远端:LumenTracer(backend="https://lumen.lurus.cn/ingest")。或自定义:

python
from lumen_ai.backend import Backend
class MyBackend(Backend):
    def send(self, event): ...
LumenTracer(backend=MyBackend())

Async 支持 ​

SDK 所有 I/O 异步友好:

python
import asyncio
from lumen_ai import LumenTracer

async def main():
    tracer = LumenTracer()
    async with tracer:
        await graph.ainvoke({"query": "hi"})

asyncio.run(main())

下一步 ​

下一步

Powered by Lurus Technologies