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 / 超时 / 默认调度策略 / 传输层重试次数
# max_retries 默认 3，走 httpx transport 层（连接失败与瞬时错误自动重试）
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, m.modal_type)   # 模型 ID / 拥有方 / 模态类型

# modal_type 由 UIG /v1/models 直接返回，可按模态筛选
llm_models = [m for m in models if m.modal_type == "llm"]
# 另有 node_kind（local / remote）与 engine（umr / vllm / ollama）

对话会话（Chat Sessions）

UIP 侧 chat_sessions / chat_messages 表提供持久化对话能力。chat_in_session() 是推荐入口 —— 服务端 /v1/sessions/send 只存用户消息、不触发推理，完整一轮对话 需要 send → chat → save_response 三步，该方法把它们合成一次调用，并自动把会话历史 作为上下文带上。

sess = client.create_session(title="体测咨询", model="qwen2.5:7b")
sid = sess.id

# 一步完成：存用户消息 + 带历史上下文推理 + 存助手回复
resp = client.chat_in_session(sid, "什么是 BMI？")
print(resp.text)

# 流式同样支持，流结束后自动把完整回复落库
for chunk in client.chat_in_session(sid, "详细说说", stream=True):
    print(chunk.text, end="")

# 会话管理
sessions = client.list_sessions(limit=20)   # -> list[Session]
detail = client.get_session(sid)            # -> SessionDetail，含全部消息
client.update_session_title(sid, "新标题")
client.archive_session(sid)
client.delete_session(sid)

需要细粒度控制时可拆开用 send_message() / save_response()。

异步客户端（chat_in_session 按 stream 分别返回协程与 async generator）：

resp = await client.chat_in_session(sid, "你好")

async for chunk in client.chat_in_session(sid, "你好", stream=True):
    print(chunk.text, end="")

非对话类单次推理（embedding / rerank / ocr）另有独立历史：

client.save_inference_history(modal_type="embedding", model_id="bge-m3:567m",
                              input_data={"text": "..."}, output_data={"dim": 1024})
history = client.get_inference_history(modal_type="embedding", limit=10)

指定调度策略

通过 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"])

### 训练配方（推荐使用）

训练模板的正确形态是**结构化配方**而非源码文本。一个 recipe 同时确定
`framework` + `task_type` + 一组经过验证的默认超参，调用方只给 `model` / `dataset`
即可开跑，无需理解 DeepSpeed ZeRO 之类的框架细节
（对标 OpenAI `hyperparameters`、Vertex AI trainer 类、HF TRL `SFTConfig`）。

```python
# 查看全部配方
for r in client.list_recipes():
    print(r.name, "->", r.framework, r.task_type, "lr=", r.learning_rate)

# 用配方提交：一行开跑 LoRA 微调
job = client.submit_job(model="qwen2.5:0.5b", dataset="my-sft", recipe="lora_sft")

# 显式传入的参数优先于配方默认值
job = client.submit_job(model="qwen2.5:0.5b", dataset="my-sft",
                        recipe="lora_sft", epochs=10, learning_rate=5e-5)

提交前先校验参数，避免无效任务占用 GPU 队列：

v = client.validate_job(model="qwen2.5:0.5b", dataset="demo", recipe="lora_sft")
print(v.valid)      # 是否通过
print(v.errors)     # 硬错误（会阻断提交）
print(v.warnings)   # 软警告

支持的 8 种训练框架

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

framework 与 task_type 是两个维度：上表 8 项是训练框架； task_type（sft / text_classification / image_classification / audio_classification）是独立的任务类型维度。UMT 的 templates/ 目录里 除框架模板外还有 image_classification.py.j2 / text_classification.py.j2 / audio_classification.py.j2 三个任务模板 —— 它们按 task_type 选取，不是 framework。 推荐直接用 recipe= 配方，避免手工拼装这两个维度。

逐框架提交示例：

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

# 训练配方（推荐：结构化训练模板）
for r in client.list_recipes():
    print(r.name, "->", r.framework, r.task_type)

# 框架模板源码（escape hatch：仅供需要改写训练脚本本体的高级场景）
templates = client.list_templates()               # -> list[TemplateItem]
tpl = client.get_template(framework="transformers")  # -> TemplateItem
print(tpl.name, tpl.description)
print(tpl.code)   # 源码改完以 submit_job(script=...) 提交

GPU 资源与模型部署

训练完成后把检查点部署为可推理模型，打通「训练 → 推理」闭环：

# GPU 资源总览（逐卡显存 / 利用率 / 状态，提交训练前先看资源）
overview = client.gpu_overview()

# 查看某作业的检查点，挑一个 step 部署
ckpts = client.list_checkpoints("job-xxx")
step = ckpts[-1].step

# 部署为可推理模型：导出 LoRA → 分发推理主机 → 验证推理 → 登记
client.register_model(job_id="job-xxx", checkpoint_step=step,
                      prompt="介绍一下人工智能")

# 已部署模型列表（可按作业 ID 过滤）
deployments = client.list_deployments(job_id="job-xxx")

# 下载检查点权重（zip 二进制）
data = client.download_checkpoint("job-xxx", step)
open("ckpt.zip", "wb").write(data)

# UMR 注册表（经 UIF 联邦路由获取已注册模型 / 能力）
registry = client.umr_registry()

# 当前登录用户信息
me = client.get_me()

计费与余额

from uip_sdk import BalanceResponse

# 查询余额（UIP 人民币单轨; credits 字段保留兼容, 恒为 0）
bal: BalanceResponse = client.get_balance()
print(f"余额: ¥{bal.balance_rmb:.2f} | 累计消费: ¥{bal.total_spent:.2f}")
print(f"状态: {bal.status}  # active / suspended — chargeback 欠费停用")

# 账单流水（UIP record_type: recharge / consume / refund）
txns = client.get_transactions(page=1, page_size=20)
for item in txns.items:
    print(f"{item.created_at} {item.tx_type}: {item.amount} 元 — {item.description}")

# 用量明细（每次推理请求的 tokens / 计量 / 费用; 0.20.0 新增）
usage: list[UsageItem] = client.list_usage(days=7, limit=20)
for u in usage:
    print(f"{u.created_at} {u.model_name} in={u.input_tokens} out={u.output_tokens} "
          f"units={u.units} cost=¥{u.cost}")

# 管理员充值（需 admin 权限; UIP 人民币单轨）
resp = client.recharge(user_id=42, amount=10.0, description="充值")
print(f"充值成功: 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 侧访问资源。

异步客户端（与同步客户端方法级对齐）

AsyncUIPClient 与 UIPClient 方法级完全对齐（零缺口）：会话、训练、 模板、数据集、计费、OAuth、文件上传全部可用。

from uip_sdk import AsyncUIPClient

async with AsyncUIPClient(api_key="ggw-xxx") as client:
    sess = await client.create_session(model="qwen2.5:7b")
    resp = await client.chat_in_session(sess.id, "你好")
    jobs = await client.list_jobs(status="running")   # -> list[TrainingJob]
    ckpt = await client.get_latest_checkpoint(jobs[0].job_id)  # -> Checkpoint | None

关于返回类型：绝大多数方法返回 dataclass，且每个 dataclass 都带 raw 字段保留服务端原始响应 —— 服务端后续新增字段也不会丢信息。少数方法仍返回 裸 dict：get_job_logs / get_qos / preview_dataset / send_message / submit_job / ping / OAuth 系列，因为它们的返回结构随模型或端点而异， 强类型化反而会造成字段丢失。

chat_in_session 在 async 下是普通 def（非 async def），按 stream 分别返回协程与 async generator：

resp = await client.chat_in_session(sid, "你好")

async for chunk in client.chat_in_session(sid, "你好", stream=True):
    print(chunk.text, end="")

异步流式（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 边收边出。