UIP Python SDK

Universal Inference Platform 的 Python 客户端库。

本文档与 uip_sdk/client.py 的 UIPClient 接口保持同步。所有公开方法、参数与返回类型均以源码为准。

安装

pip install uip-sdk

快速开始

from uip_sdk import UIPClient

# 方式 1: API Key
client = UIPClient(api_key="ggw-xxx...")

# 方式 2: JWT Token
client = UIPClient(token="eyJhbGciOiJIUzI1NiIs...")

# 方式 3: 环境变量 (UIP_API_KEY 或 UIP_TOKEN)
client = UIPClient()

# 指定 base_url / 超时 / 默认调度策略
client = UIPClient(
    base_url="http://10.0.1.115:51438",
    api_key="ggw-xxx...",
    timeout=300,
    strategy="least_queue",
)

# 作为上下文管理器使用（自动释放连接）
with UIPClient(api_key="ggw-xxx...") as client:
    print(client.chat(messages=[{"role": "user", "content": "你好"}]).text)

初始化参数：

• 三者皆为空时，自动读取环境变量 UIP_API_KEY / UIP_TOKEN；仍为空则抛出 AuthenticationError。
• user_id / unit_id 为只读属性，构造时自动调用 /v1/users/me 解析。

推理

对话 (Chat Completions)

OpenAI 兼容接口。

resp = client.chat(
    messages=[{"role": "user", "content": "你好"}],
    model="qwen2.5:7b",
    temperature=0.7,
    max_tokens=512,
    top_p=1.0,
)
print(resp.text)              # 回复文本
print(resp.tokens.total)      # 总 token 数

流式：

for chunk in client.chat(messages=[...], model="qwen2.5:7b", stream=True):
    print(chunk.text, end="", flush=True)

流式生成 (Ollama 兼容)

for chunk in client.generate("写一首关于春天的诗", stream=True):
    print(chunk.text, end="", flush=True)

# 非流式
resp = client.generate("介绍一下你自己", model="qwen2.5:7b", system="你是助手")
print(resp.text)

参数：prompt、model、system、stream、options（Ollama 选项，如 {"num_predict": 100}）。

嵌入向量

resp = client.embed(input="需要向量化的文本", model="bge-m3:567m")
print(len(resp.embedding))   # 768
print(resp.tokens.input)      # 输入 token 数

input 也支持 list[str]（批量向量化）。

Rerank (文档重排序)

results = client.rerank(
    query="CBA季后赛战术分析",
    documents=["CBA联赛采用胜率决定排名", "篮球三分线距离为6.75米", "广东队采用全场紧逼战术"],
    model="Qwen3-Reranker-0.6B",
    top_n=2,
)
for r in results.results:
    print(f"#{r.index}: {r.document[:30]}... score={r.relevance_score:.2f}")

批量推理

batch = client.batch(prompts=["你好", "介绍你自己"], model="qwen2.5:7b")
for item in batch.results:
    print(f"[{item.index}] {item.response[:50]}")

文件上传

up = client.upload("/path/to/file.pdf")
print(up.filename, up.size, up.url, up.content_type)

模型列表

models = client.list_models()
for m in models:
    print(m.id, m.owned_by)   # 模型 ID / 拥有方

指定调度策略

通过 with_strategy 链式设置本次请求的调度策略（写入请求头 X-UIP-Strategy），返回 self 可继续调用。

client.with_strategy("least_queue").generate("hi")

可选策略：model_first（按模型匹配）/ least_queue（最少队列优先）/ weighted_rr（加权轮询）/ affinity（亲和性复用）。

健康检查

print(client.ping())   # 返回服务状态 dict

UMR 通用推理（17 种模态全覆盖）

client.infer(model_id, payload) 是 UMR 统一运行时入口，一个接口覆盖全部 17 种 AI 模态。 model_id 采用 family:size 命名，payload 字段随模态不同；返回的 InferResponse.modal_type 为下表「模态类型」列的字符串值（与 umr/core/models.ModalType 枚举一致）。

注：Video 模态在 ModalType 中细分为 video_understanding（视频理解）与 video_generation（文生视频，如 Wan2.1），本表按论文口径统一归为第 11 项「视频理解」。其余枚举值（video、video_generation）调用方式相同，仅 modal_type 字符串不同。

逐模态调用示例：

from uip_sdk import UIPClient

client = UIPClient(api_key="ggw-xxx...")

# 1) 大语言模型 LLM — 智能问答
r = client.infer("qwen2.5:7b", {"prompt": "解释牛顿第二定律"})
print(r.modal_type, r.result)                       # llm / 回复文本

# 2) 语音识别 ASR — 课堂录音转写
r = client.infer("whisper:large-v3", {"audio": "lecture.wav"})
print(r.modal_type, r.result)                       # asr / 转写文本

# 3) 语音合成 TTS — 发音示范
r = client.infer("speecht5:base", {"text": "Hello, world", "speaker": 0})
print(r.modal_type, r.result)                       # tts / 音频 base64 或路径

# 4) 视觉理解 Vision — 实训操作评估
r = client.infer("qwen2.5-vl:7b", {"image": "lab.jpg", "prompt": "描述画面中的操作"})
print(r.modal_type, r.result)                       # vision / 描述文本

# 5) 向量嵌入 Embedding — 知识库检索
r = client.infer("bge-m3:567m", {"text": "GPU 显存调度"})
print(r.modal_type, len(r.result))                  # embedding / 向量维度（如 1024）

# 6) 重排序 Rerank — RAG 精排
r = client.infer("bge-reranker:0.6b",
                 {"query": "VAHS 是什么", "documents": ["显存感知调度", "天气报告"]})
print(r.modal_type, r.result)                       # rerank / 排序后文档列表

# 7) 图像生成 ImageGen — 教学素材
r = client.infer("sdxl:base", {"prompt": "a chemistry lab, flat illustration"})
print(r.modal_type, r.result)                       # image_generation / 图片 url 或 base64

# 8) 目标检测 ObjDetection — 体育姿态分析
r = client.infer("yolov8:m", {"image": "student.jpg"})
print(r.modal_type, r.result)                       # object_detection / 检测框列表

# 9) 图像分割 Segmentation — 医学影像
r = client.infer("segformer:b0", {"image": "xray.png"})
print(r.modal_type, r.result)                       # segmentation / mask 数据

# 10) 文字识别 OCR — 试卷批改
r = client.infer("pp-ocr:v3", {"image": "exam.png"})
print(r.modal_type, r.result)                       # ocr / 识别文本

# 11) 视频理解 Video — 动作分析
r = client.infer("videomae:base", {"video": "action.mp4"})
print(r.modal_type, r.result)                       # video_understanding / 标签或描述

# 12) 音频处理 Audio — 音乐分析
r = client.infer("wav2vec2:base", {"audio": "music.wav"})
print(r.modal_type, r.result)                       # audio / 特征或分类

# 13) 图神经网络 GNN — 课程推荐
r = client.infer("gcn:base", {"graph": {"nodes": [...], "edges": [...]}})
print(r.modal_type, r.result)                       # gnn / 节点表示或预测

# 14) 编码器 EncoderOnly — 情感分析
r = client.infer("bert:base", {"text": "这部电影太精彩了"})
print(r.modal_type, r.result)                       # encoder_only / 向量或分类

# 15) 时间序列 TimeSeries — 体能趋势预测
r = client.infer("prophet:base", {"series": [120, 128, 131, 135], "periods": 3})
print(r.modal_type, r.result)                       # time_series / 预测值

# 16) 科学计算 SciCompute — 蛋白质结构
r = client.infer("esm2:base", {"sequence": "MKTAYIAKQR..."})
print(r.modal_type, r.result)                       # sci_compute / 结构或表示

# 17) 传统机器学习 ML — 统计建模
r = client.infer("xgboost:base", {"features": [0.1, 2.3, 5.0]})
print(r.modal_type, r.result)                       # ml / 预测值

说明：payload 字段取决于具体模型适配器实现，上例为典型用法；所有 InferResponse 均带 raw 字段可读取原始返回。modal_type 由 UMR 适配器自动回写，可用于按模态做路由、计费与历史过滤（如 get_inference_history(modal_type="llm")）。

联邦推理

UIP→UIF→UIS→UIG 跨机构推理链路，需方通过 SDK 向供方发起推理请求。

# 自动选择最优供方
result = client.federated_infer(
    model_id="qwen2.5:7b",
    payload={"prompt": "介绍中国体育教育发展"},
)
print(result.supplier_unit)    # 供方单位 ID
print(result.supplier_node)    # 供方 GPU 节点名
print(result.cost_rmb)         # 人民币费用
print(result.cost_credits)     # 积分费用

# 指定供方 + GPU 类型偏好 + 双重价格上限
result = client.federated_infer(
    model_id="qwen2.5:7b",
    payload={"prompt": "你好"},
    target_unit="tsinghua-sports",    # 指定供方单位
    prefer_gpu_type="rtx_5090",       # GPU 类型偏好
    max_price_rmb=1.0,                # 人民币价格上限
    max_price_credits=10.0,           # 积分价格上限
)

训练管理

# 提交训练任务（含 gpu_type 和 unit_id 多租户隔离）
job = client.submit_job(
    model="qwen2.5:7b",
    name="fine-tune-v1",
    framework="transformers",     # 训练框架
    dataset="my-data",            # 数据集名
    script="train.py",            # 训练脚本（可选）
    n_gpus=1,
    epochs=3,
    learning_rate=1e-5,
    batch_size=4,
    priority=0,
    timeout=3600,
    gpu_type="rtx_5090",          # GPU 类型追踪
    unit_id="tsinghua-sports",    # 多租户命名空间隔离
)
print(job["job_id"])

### 支持的 8 种训练框架

`submit_job(framework=...)` 的 `framework` 对应 UMT 的模板文件 `<framework>.py.j2`，共支持 **8 种训练环境**。传入未匹配的框架名时，UMT 自动回退到 `transformers` 模板。

| # | framework 值 | 模板文件 | 适用场景 |
|:--:|:--|:--|:--|
| 1 | `transformers` | transformers.py.j2 | HuggingFace Transformers / PyTorch 原生，标准 SFT、预训练 |
| 2 | `deepspeed` | deepspeed.py.j2 | DeepSpeed 分布式训练，多卡大模型训练 |
| 3 | `llama-factory` | llama-factory.py.j2 | LLaMA-Factory 低代码微调框架 |
| 4 | `axolotl` | axolotl.py.j2 | Axolotl 指令 / 多模态微调 |
| 5 | `unsloth` | unsloth.py.j2 | Unsloth 量化高效微调（低显存） |
| 6 | `torchtune` | torchtune.py.j2 | PyTorch 官方 recipes（TorchTune） |
| 7 | `train` | train.py.j2 | 通用 PyTorch 训练脚本（自定义训练循环） |
| 8 | `custom` | custom.py.j2 | 完全自定义训练模板，由 `script` 指定入口 |

逐框架提交示例：

```python
# 1) HuggingFace Transformers / PyTorch 原生
job = client.submit_job(model="qwen2.5:7b", name="sft-v1", framework="transformers",
                        dataset="my-sft", epochs=3, learning_rate=1e-5, batch_size=4)

# 2) DeepSpeed 分布式训练（多卡）
job = client.submit_job(model="qwen2.5:14b", name="ds-v1", framework="deepspeed",
                        dataset="my-sft", n_gpus=4, gpu_type="rtx_5090")

# 3) LLaMA-Factory 微调
job = client.submit_job(model="llama3:8b", name="lf-v1", framework="llama-factory",
                        dataset="my-sft", batch_size=8)

# 4) Axolotl 指令微调
job = client.submit_job(model="qwen2.5:7b", name="axolotl-v1", framework="axolotl",
                        dataset="instruct-en", batch_size=2)

# 5) Unsloth 量化高效微调（低显存）
job = client.submit_job(model="qwen2.5:7b", name="unsloth-v1", framework="unsloth",
                        dataset="my-sft", n_gpus=1, gpu_type="rtx_4070")

# 6) TorchTune
job = client.submit_job(model="llama3:8b", name="tt-v1", framework="torchtune",
                        dataset="my-sft", epochs=1)

# 7) 通用 PyTorch 训练脚本
job = client.submit_job(model="resnet50", name="train-v1", framework="train",
                        dataset="imagenet-mini", script="train.py", n_gpus=2)

# 8) 自定义模板
job = client.submit_job(model="custom-model", name="custom-v1", framework="custom",
                        dataset="my-data", script="my_train.py", unit_id="tsinghua-sports")

提示：可用 client.list_templates() 查看服务端实际支持的框架模板清单，并用 client.get_template(framework="deepspeed") 拉取对应训练脚本模板源码；框架特有的超参（如 DeepSpeed 的 ZeRO 配置、Unsloth 的 4bit 量化）通过在 dataset/script 或模板内指定，而非 submit_job 的顶层参数。

任务列表 / 详情 / 取消

jobs = client.list_jobs(status="running", limit=50) detail = client.get_job("job-xxx") client.cancel_job("job-xxx")

训练日志（长轮询 / SSE 流式）

logs = client.get_job_logs("job-xxx", tail=50) for line in client.stream_job_logs("job-xxx"): print(line, end="")

断点 / 指标

ckpts = client.list_checkpoints("job-xxx") latest = client.get_latest_checkpoint("job-xxx") metrics = client.get_training_metrics("job-xxx")

### 数据集与模板

```python
# 数据集列表
datasets = client.list_datasets()
for d in datasets.get("datasets", []):
    print(d["name"], d["size"])

# 上传数据集（base64 内联，支持多租户隔离）
client.upload_dataset(name="my-data", file_path="/path/to/data.jsonl", unit_id="tsinghua-sports")

# 预览数据集前 N 行
preview = client.preview_dataset(name="my-data", lines=20)

# 训练框架模板
templates = client.list_templates()
code = client.get_template(framework="transformers")
print(code.get("code"))

计费与余额

from uip_sdk import BalanceResponse

# 查询余额（双轨：人民币 + 积分）
bal: BalanceResponse = client.get_balance()
print(f"人民币: ¥{bal.balance_rmb:.2f}")
print(f"积分:   {bal.balance_credits:.0f} credits")
print(f"冻结:   ¥{bal.frozen_rmb:.2f} / {bal.frozen_credits:.0f} credits")

# 快速查询积分
credits = client.get_credits()

# 交易记录（支持币种 / 类型过滤，分页）
txns = client.get_transactions(page=1, page_size=20, currency="credits", tx_type="consume")
for item in txns.items:
    print(f"{item.created_at} {item.tx_type}: {item.amount} {item.currency} — {item.description}")

# 管理员充值（需 admin 权限）
resp = client.recharge(user_id=42, amount=100.0, currency="credits", description="赠送积分")
print(f"充值成功: tx_id={resp.tx_id}, balance_after={resp.balance_after}")

历史与 QoS

# QoS 信息（限流/配额状态）
qos = client.get_qos()
print(qos)

# 推理历史（按模态过滤，默认 llm）
history = client.get_inference_history(modal_type="llm", limit=30)
for h in history.get("items", []):
    print(h)

# 删除某条推理历史
client.delete_inference_history(history_id=123)

返回对象字段速查

所有响应对象均带 raw 字段（原始响应 dict）以便扩展读取。

异常处理

所有异常继承自 UIPError，捕获基类即可统一处理。

from uip_sdk import UIPClient
from uip_sdk.errors import UIPError, InsufficientBalanceError

try:
    client.generate("hi")
except InsufficientBalanceError as e:
    print("余额不足:", e)
except UIPError as e:
    print("请求失败:", e.status_code, e.detail)

注：UIPError / AuthenticationError / InsufficientBalanceError / UIGOfflineError / TimeoutError / RateLimitError 已在 uip_sdk 顶层直接导出；NotFoundError / ServerError 可通过 from uip_sdk.errors import NotFoundError, ServerError 使用。

资源释放

client.close()   # 关闭底层 httpx 连接

# 或使用 with 上下文管理器（推荐）
with UIPClient(api_key="...") as client:
    ...

License

Apache License 2.0. Copyright (c) 2026 Zhu Wenbo (zwb.2002@tsinghua.org.cn).

UMRClient — 引擎层直连

UIPClient 走平台（认证/计费/审计）；UMRClient 直连 UMR 引擎（59:51450），覆盖引擎级管理（load/unload/vram/registry/部署注册）与全部 20 个 UMR 端点。

from uip_sdk import UMRClient

# X-API-Key 认证 (UIR_INTERNAL_KEY)
umr = UMRClient(base_url="http://10.0.1.59:51450", api_key="<internal-key>")

# 引擎级管理
umr.list_models()          # by_modal 模型清单
umr.vram_status()          # 显存状态
umr.registry_status()      # 注册表状态
umr.load_model("qwen2.5:7b") / umr.unload_model("qwen2.5:7b")
umr.register_model(model_id="deploy-xxx", base_model="/path/to/base",
                   adapter_path="/path/to/lora")   # 部署 LoRA 注册

# 引擎级推理
umr.infer("qwen2.5:7b", {"messages": [{"role": "user", "content": "hi"}]})
umr.chat_completions({"model": "qwen2.5:7b", "messages": [...]})  # OpenAI 兼容

OAuth2 应用与 SSO（AI 实训室等应用接入）

UIP 作为平台底座为应用（如 AI 实训室）提供 OAuth2 授权码登录与应用级 JWT：

from uip_sdk import UIPClient, exchange_oauth_token

client = UIPClient(base_url="http://10.0.1.115:51438", api_key="<your-api-key>")

# 1. 注册应用（client_secret 仅此一次可见，妥善保存）
app = client.oauth_create_app(
    name="AI 实训室",
    scope="lab inference training",
    redirect_uris="http://<ai-lab-host>/callback",
)
cid, secret = app["client_id"], app["client_secret"]

# 2. 用户已登录时签发一次性授权码（10min）
auth = client.oauth_authorize(client_id=cid, scope="lab inference")
code = auth["code"]

# 3. 授权码换应用级 JWT（实训室后端主入口）
tok = client.oauth_exchange_token(code=code, client_id=cid, client_secret=secret)
# 等价便捷函数（无需 UIPClient 实例）：
# tok = exchange_oauth_token(base_url="http://10.0.1.115:51438", code=code,
#                            client_id=cid, client_secret=secret)

# 4. 应用管理
client.oauth_list_apps()
client.oauth_revoke_app(client_id=cid)

返回的 JWT 携带 scope（权限范围）与 azp（来源应用），实训室据此代表用户在 UIP 侧访问资源。

异步流式（FastAPI 后端推荐）

实训室等 async 后端集成推理工作台流式时，使用 AsyncUIPClient（对标 openai-python 的 AsyncOpenAI 模式），不占线程、不阻塞事件循环：

from uip_sdk import AsyncUIPClient

async with AsyncUIPClient(base_url="http://10.0.1.115:51438", api_key="<your-api-key>") as client:
    # 流式：打字机效果
    async for chunk in client.chat_stream([{"role": "user", "content": "你好"}]):
        print(chunk.text, end="", flush=True)

    # 非流式
    resp = await client.chat([{"role": "user", "content": "hi"}])
    print(resp.text)

    # Ollama 兼容 generate 流式
    async for chunk in client.generate_stream("你好"):
        print(chunk.text, end="", flush=True)

    # embedding
    emb = await client.embed("AI 实训室")
    print(emb.dimensions)

流式链路：实训室 FastAPI -> UIP 网关(SSE 透传) -> UIG -> UMR，SDK async 边收边出。