UltraFlow 1.0.3

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

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 命令

命令	说明
uf init [PROJECT]	初始化项目，生成配置和示例 prompty
uf new FLOW_NAME	创建 .prompty + .json 模板
uf run FLOW_NAME	运行流程（自动查找同名 .json）
uf run FLOW -d DATA -w N	指定数据文件和并发数
uf connection init	生成连接配置模板
uf connection list	显示当前生效的连接配置

License

MIT