Lumen Python SDK 開発中
pip install lumen-ai の後、SDK は3つのコアクラスを提供し、収集・永続化・コスト集計をカバーします:
LumenTracer
コールバック形式の Trace + Cost 収集。
LumenCheckpointer
SqliteSaver を置き換える永続化レイヤー。
CostTracker
エージェント横断のコスト集計クエリ。
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 との比較:
| 観点 | SqliteSaver | LumenCheckpointer |
|---|---|---|
| 復元速度 | ミリ秒級 | マイクロ秒級(Kova WAL エンジンベース) |
| マルチプロセス | ロックが必要 | ネイティブ対応 |
| リモート永続化 | 自前実装が必要 | 組み込み |
| コスト関連付け | なし | 自動 |
そのまま置き換え可能、API 互換
LumenCheckpointer は LangGraph の BaseCheckpointSaver インターフェースと互換性があり、既存の SqliteSaver を1行差し替えるだけで済み、グラフ構造を変更する必要はありません。詳細は エコシステム統合 · LangGraph を参照してください。CostTracker
エージェント横断のコスト集計クエリ:
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())