UltraFlow

轻量级 LLM 应用可观测性 SDK — 调用追踪、提示词工程、批量处理。

特性

• Prompty — 将 .prompty 文件作为无状态函数执行（同步 / 异步 / 流式）
• Tracing — @trace 装饰器自动采集调用树，业务代码零侵入
• 批量处理 — FlowProcessor 并发执行批量任务
• CLI — uf init、uf new、uf run，快速上手

安装

pip install UltraFlow

快速开始

1. 初始化项目

uf init my-project
cd my-project

2. 执行 Prompty

from ultraflow import Prompty

translate = Prompty("translate.prompty")
result = translate(text="Hello, world!")

3. 添加调用追踪

from ultraflow import Prompty, configure_tracing, shutdown_tracing, trace

configure_tracing(endpoint="http://localhost:8031")

@trace(kind="flow")
def translate_flow(text: str) -> str:
    prompty = Prompty("translate.prompty")
    return prompty(text=text)

result = translate_flow(text="Hello!")
shutdown_tracing()

4. 批量处理

from ultraflow import Prompty, FlowProcessor

translate = Prompty("translate.prompty")
processor = FlowProcessor(translate, "data.json", max_workers=5)
results = processor.run()

CLI 命令

License

MIT