初始化客户端
from siflow import SiFlow
client = SiFlow(region="cn-beijing", cluster="auriga")
inference = client.inference
client.inference 管理。服务 ID 通常为整数,例如 123。
查询创建前可用选项
创建服务前,建议先查询当前集群支持的引擎、存储类型、KV Cache 选项和资源包。engine_options = inference.list_engine_options()
storage_types = inference.list_storage_types()
kv_cache_options = inference.list_kv_cache_options()
packages = inference.list_resource_packages(resource_pool="<RESOURCE_POOL>")
print("engine options:", engine_options)
print("storage types:", storage_types)
print("kv cache options:", kv_cache_options)
for package in packages:
print(package.instance_name, package.cpu, package.memory, package.gpu, package.gpu_type)
创建 vLLM 推理服务
以下示例使用模型仓库作为模型来源,并创建一个单节点 vLLM 推理服务。请按实际环境替换资源池、模型名称、模型版本、实例规格、服务名称和网关配置。from siflow.types import ServiceCreateParams
payload = {
"name": "qwen2-5b-vllm",
"description": "Qwen2.5 vLLM inference service",
"resourcePool": "<RESOURCE_POOL>",
"resourcePoolType": "dedicated",
"resourcePoolQosType": "reserved",
"gatewayConfigs": [
{
"type": "Shared",
"authToken": "",
"enableHashRoute": False,
}
],
"metricsConfig": {
"metricsType": "prometheus",
"metricsPath": "/metrics",
"metricsPort": 8000,
},
"modelConfig": {
"modelSource": {
"storageType": "model-repo",
"storage": {
"modelRepo": {
"name": "Qwen/Qwen2.5-0.5B-Instruct",
"version": "v1.0",
}
},
}
},
"monitoringConfig": {
"enableAlarmPush": True,
},
"roleConfig": {
"server": {
"ports": [
{
"containerPort": 8000,
"name": "router",
}
],
"replicas": 1,
"resourceConfig": {
"instanceName": "sci.g20-3",
"instanceQuantity": 1,
},
"routerConfig": {
"routerPolicy": "round_robin",
"resourcePoolQosType": "ondemand",
"resourcePoolType": "shared",
"resourcePool": "<ROUTER_RESOURCE_POOL>",
},
},
"worker": {
"replicas": 1,
"resourceConfig": {
"instanceName": "sci.g20-3",
"instanceQuantity": 1,
},
},
},
"serviceConfig": {
"autoScaleConfig": {
"autoScaleType": "timeseries",
"timeseriesConfig": {
"timePoints": [],
},
},
"registerConfig": {
"servedModelName": "qwen2.5",
},
"replicas": 0,
"servicePort": {
"name": "server",
"port": 8000,
},
},
"servingEngineConfig": {
"engineParams": [
{"key": "--served-model-name", "value": "qwen2.5"},
{"key": "--host", "value": "0.0.0.0"},
{"key": "--port", "value": "8000"},
],
"engineType": "vllm",
"engineVersion": "v0.8.5",
"executeType": "single-node",
},
"env": {
"MODEL_PATH": "/mnt/model/Qwen/Qwen2.5-0.5B-Instruct",
"TZ": "Asia/Shanghai",
},
"storageConfig": {
"fileSystemVolumes": [],
},
}
params = ServiceCreateParams(**payload)
print(params.model_dump(by_alias=True, exclude_none=True))
service_id = inference.create_service(service_params=params)
print(service_id)
| 配置 | 说明 |
|---|---|
serviceConfig | 副本数、服务端口、健康检查、网关注册、自动扩缩容和更新策略。 |
servingEngineConfig | 推理引擎类型、版本、执行方式、引擎参数和 KV Cache 配置。 |
modelConfig | 模型来源配置,可使用模型仓库、Volume、PVC、OSS、HostPath、Hugging Face 等来源;具体可用类型以 list_storage_types() 返回为准。 |
roleConfig | 角色配置,例如 worker、server、router 等角色的副本数、资源规格、端口、镜像、命令和环境变量。 |
gatewayConfigs | 外部访问网关配置。 |
metricsConfig | Prometheus 指标暴露端口和路径。 |
monitoringConfig | 告警推送等监控配置。 |
storageConfig | 额外挂载的文件系统卷。 |
env | 服务级环境变量。 |
ServiceCreateParams 校验。提交前可通过 model_dump(by_alias=True, exclude_none=True) 打印最终请求体,确认字段名和嵌套结构符合预期。
配置不同模型来源
以下片段用于替换上一节payload 中的 modelConfig。完成替换后,重新构造 ServiceCreateParams 并调用 create_service() 提交服务。
使用 Volume
payload["modelConfig"] = {
"modelSource": {
"storageType": "volume",
"storage": {
"volume": {
"volumeId": 123,
"mountPath": "/models",
"modelPath": "/models/Qwen2.5-7B-Instruct",
"subPath": "team-a/qwen",
"readOnly": True,
}
},
}
}
subPath;subPath 为卷内相对路径。
使用 PVC
payload["modelConfig"] = {
"modelSource": {
"storageType": "pvc",
"storage": {
"pvc": {
"persistentVolumeClaimName": "<PVC_NAME>",
"mountPath": "/mnt/models",
"modelPath": "/mnt/models/Qwen2.5-7B-Instruct",
"subPath": "models/qwen",
"readOnly": True,
}
},
}
}
使用 OSS
OSS 密钥不建议硬编码到代码仓库。示例中用环境变量读取,生产环境可结合密钥管理系统注入。import os
payload["modelConfig"] = {
"modelSource": {
"storageType": "oss",
"storage": {
"oss": {
"endpoint": "https://oss-cn-beijing.aliyuncs.com",
"bucket": "<MODEL_BUCKET>",
"region": "cn-beijing",
"modelPath": "models/Qwen2.5-7B-Instruct",
"accessKey": os.environ["MODEL_OSS_ACCESS_KEY"],
"secretKey": os.environ["MODEL_OSS_SECRET_KEY"],
"enableCache": True,
}
},
}
}
创建自定义引擎服务
使用自定义引擎时,通常需要在roleConfig.worker 中指定镜像、启动命令、参数、端口和资源规格。
from siflow.types import ServiceCreateParams
payload = {
"name": "custom-openai-server",
"resourcePool": "<RESOURCE_POOL>",
"serviceConfig": {
"replicas": 1,
"servicePort": {"name": "http", "port": 8000},
},
"servingEngineConfig": {
"engineType": "custom",
"engineVersion": "custom",
"executeType": "single-node",
},
"modelConfig": {
"modelSource": {
"storageType": "volume",
"storage": {
"volume": {
"volumeId": 123,
"mountPath": "/models",
"modelPath": "/models/my-model",
"readOnly": True,
}
},
}
},
"workloadConfig": {"type": "deployment"},
"roleConfig": {
"worker": {
"replicas": 1,
"image": "registry-cn-shanghai.siflow.cn/ai-infra/custom-llm-server:v1.0.0-a1b2c3d",
"command": "python",
"args": "-m my_server --model /models/my-model --host 0.0.0.0 --port 8000",
"ports": [{"name": "http", "containerPort": 8000}],
"resourceConfig": {
"instanceName": "sci.g22-1",
"instanceQuantity": 1,
},
"env": [
{"name": "PYTHONUNBUFFERED", "value": "1"},
],
}
},
}
service_id = inference.create_service(
service_params=ServiceCreateParams(**payload),
)
print(service_id)
创建多角色或多机服务
多机推理通常会拆分为worker、router、server 等角色。具体角色名和资源约束取决于引擎版本的 executeTypeRoles 以及服务端支持能力,可先查询引擎版本确认。
engines = inference.list_engine_versions(engine="vllm", page=1, page_size=20)
for engine in engines:
print(engine.id, engine.version, engine.execute_type_roles)
from siflow.types import ServiceCreateParams
payload = {
"name": "qwen2-72b-multi-node",
"resourcePool": "<RESOURCE_POOL>",
"serviceConfig": {
"replicas": 1,
"servicePort": {"name": "http", "port": 8000},
},
"servingEngineConfig": {
"engineType": "vllm",
"engineVersion": "0.8.5",
"executeType": "multi-node",
"enableGang": {"worker": True},
},
"modelConfig": {
"modelSource": {
"storageType": "volume",
"storage": {
"volume": {
"volumeId": 123,
"mountPath": "/models",
"modelPath": "/models/Qwen2.5-72B-Instruct",
"subPath": "team-a/qwen-72b",
"readOnly": True,
}
},
}
},
"workloadConfig": {
"type": "deployment",
"enableGang": True,
},
"roleConfig": {
"worker": {
"replicas": 2,
"size": 2,
"resourceConfig": {
"instanceName": "sci.g22-8",
"instanceQuantity": 1,
},
},
"router": {
"replicas": 1,
"resourceConfig": {
"resources": {
"cpu": "4",
"memory": "8Gi",
}
},
},
},
}
service_id = inference.create_service(
service_params=ServiceCreateParams(**payload),
)
print(service_id)
查询和调用服务
查询服务详情
service = inference.get_service(service_id=123)
print("name:", service.name)
print("resource pool:", service.resource_pool)
print("status:", service.status.status if service.status else None)
print("current version:", service.status.current_version if service.status else None)
if service.endpoints:
for endpoint in service.endpoints:
print(endpoint.name, endpoint.url, endpoint.status)
查询服务列表
services = inference.list_services(
search="qwen",
status="online",
resource_pool="<RESOURCE_POOL>",
page=1,
page_size=20,
)
for service in services:
status = service.status.status if service.status else "unknown"
print(service.id, service.name, status)
| 参数 | 说明 |
|---|---|
id | 按服务 ID 过滤,类型为字符串。 |
tenant | 按租户过滤。 |
owner | 按服务所有者过滤。 |
status | 按状态过滤。 |
name / search | 按名称精确过滤或关键词模糊搜索。 |
is_share | 是否查询共享给我的服务。 |
resource_pool / resource_pool_id | 按资源池名称或资源池 ID 过滤。 |
project_group_id | 按项目组 ID 过滤。 |
page / page_size | 分页参数。 |
测试网关
gateway_test 适合在正式接入前验证网关是否能访问。
from siflow.types import GatewayTestRequest
resp = inference.gateway_test(
request=GatewayTestRequest(
url="https://console.siflow.cn/siflow/cn-beijing/auriga/demo/qwen2-5b-vllm/v1/models",
method="GET",
headers={"Authorization": "Bearer <TOKEN>"},
)
)
print(resp.status_code)
print(resp.latency_ms)
print(resp.body)
使用 OpenAI 兼容接口调用
如果服务暴露的是 OpenAI 兼容接口,可以使用 OpenAI Python SDK 调用。base_url 需要按实际网关地址和路径设置。
pip install openai
from openai import OpenAI
service = inference.get_service(service_id=123)
if service.status is None:
raise RuntimeError("service has no status yet")
endpoint = service.status.url_external_public or service.status.url_external
if not endpoint:
raise RuntimeError("service has no external endpoint")
base_url = endpoint.rstrip("/")
if not base_url.endswith("/v1"):
base_url += "/v1"
llm = OpenAI(
base_url=base_url,
api_key="<GATEWAY_AUTH_TOKEN>",
)
completion = llm.chat.completions.create(
model="qwen2.5",
messages=[
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "用三句话介绍人工智能平台推理服务。"},
],
temperature=0.7,
)
if not completion.choices:
raise RuntimeError("推理服务没有返回候选结果")
print(completion.choices[0].message.content)
stream = llm.chat.completions.create(
model="qwen2.5",
messages=[{"role": "user", "content": "写一个 Python 快排示例"}],
stream=True,
)
for chunk in stream:
if not chunk.choices:
continue
delta = chunk.choices[0].delta.content
if delta:
print(delta, end="", flush=True)
更新、扩缩容和生命周期管理
更新服务配置
更新服务时,建议先读取当前服务配置,再在此基础上修改目标字段,避免遗漏需要保留的配置。from siflow.types import ServiceUpdateParams
service = inference.get_service(service_id=123)
payload = service.model_dump(by_alias=True, exclude_none=True)
payload["description"] = "updated by SDK"
payload.setdefault("serviceConfig", {})["replicas"] = 2
payload.setdefault("roleConfig", {}).setdefault("worker", {})["replicas"] = 2
updated_id = inference.update_service(
service_id=123,
service_params=ServiceUpdateParams(**payload),
)
print(updated_id)
update_service_register:
service = inference.get_service(service_id=123)
payload = service.model_dump(by_alias=True, exclude_none=True)
payload.setdefault("serviceConfig", {}).setdefault("registerConfig", {})[
"servedModelName"
] = "qwen2-5b-prod"
inference.update_service_register(
service_id=123,
service_params=ServiceUpdateParams(**payload),
)
inference.cancel_update_service(service_id=123)
扩缩容
scale_service 用于调整角色副本数,最常见的是调整 worker.replicas。
from siflow.types import ServiceScaleParams
resp = inference.scale_service(
service_id=123,
scale_params=ServiceScaleParams(
**{
"roleConfig": {
"worker": {"replicas": 3},
}
}
),
)
print(resp)
inference.stop_scale_service(service_id=123)
上线、下线和删除
上线服务:inference.online_service(service_id=123)
inference.offline_service(service_id=123)
inference.delete_service(service_id=123)
inference.batch_online_service(service_ids=[123, 124])
inference.batch_online_service_by_name(names=["qwen-prod-a", "qwen-prod-b"])
inference.batch_offline_service(service_ids=[123, 124])
inference.batch_offline_service_by_name(names=["qwen-prod-a", "qwen-prod-b"])
inference.batch_delete_service(service_ids=[123, 124])
inference.batch_delete_service_by_name(names=["qwen-prod-a", "qwen-prod-b"])
共享和公开服务
共享给指定用户:inference.share_service(
service_id=123,
shared_users=["alice", "bob"],
)
inference.public_service(service_id=123)
inference.unpublic_service(service_id=123)
灰度发布和版本回滚
查看服务版本
versions = inference.list_service_versions(service_id=123)
for version in versions:
print(version.version, version.status, version.created_at)
回滚服务版本
回滚到上一版本:inference.rollback_service(service_id=123)
inference.rollback_service_to_version(service_id=123, version=3)
发起灰度
from siflow.types import CanaryPublishRequest
rollout = inference.canary_publish(
service_id=123,
request=CanaryPublishRequest(
**{
"mode": "simple",
"rollout": {
"totalReplicas": 4,
"initialProgress": 10,
"maxProgressStep": 30,
},
}
),
)
print(rollout.state, rollout.progress, rollout.new_version)
调整和完成灰度
from siflow.types import CanaryPatchRequest
rollout = inference.patch_rollout(
service_id=123,
request=CanaryPatchRequest(
**{
"progress": 50,
"maxProgressStep": 30,
}
),
)
diff = inference.get_rollout_diff(service_id=123)
print(diff)
inference.complete_rollout(service_id=123)
complete_rollout() 提交全量发布后,需要等待灰度状态变为 completed,再执行收尾:
rollout = inference.get_rollout(service_id=123)
print(rollout.state)
if rollout.state == "completed":
inference.finalize_rollout(service_id=123)
inference.rollback_rollout(service_id=123)
查询 Pod、日志和指标
查询服务 Pod:instances = inference.list_service_instances(service_id=123)
for role, pods in instances.items():
print("role:", role)
for pod in pods:
print(pod.name, pod.status, pod.pod_ip, pod.containers)
instances = inference.list_offline_service_instances(service_id=123)
inference.recreate_instance(
service_id=123,
pod_name="qwen2-5b-vllm-worker-0",
)
resp = inference.query_logs(
123,
limit=100,
offset=0,
sort_order="asc",
)
print("total:", resp.total)
for item in resp.logs:
print(f"[{item.time}] [{item.pod_name}] {item.content}")
logs = inference.get_offline_pod_container_logs(
service_id=123,
pod_name="qwen2-5b-vllm-worker-0",
container_name="worker",
lines=200,
)
print(logs)
siflow[websocket]:
pip install "siflow[websocket]"
for message in inference.stream_pod_container_logs(
service_id=123,
pod_name="qwen2-5b-vllm-worker-0",
container_name="worker",
lines=20,
timeout=30,
):
if isinstance(message, bytes):
print(message.decode("utf-8", errors="replace"), end="")
else:
print(message, end="")
metrics = inference.get_service_metrics(
service_id=123,
role="worker",
timeout_seconds=3,
)
for role, rows in metrics.items():
print(role, rows[:3])
rows = inference.dashboard_llm_services(
status="online",
details_limit=5,
resource_pool="<RESOURCE_POOL>",
)
for row in rows:
print(row.name, row.status, row.resources)
管理模板和引擎
使用模板创建服务
from siflow.types import ServiceCreateParams
templates = inference.list_templates(model_name="qwen", page_size=10)
if not templates:
raise RuntimeError("没有找到可用的推理服务模板")
template = inference.get_template(template_id=templates[0].id)
payload = template.model_dump(by_alias=True, exclude_none=True)
payload["name"] = "qwen-from-template"
payload["resourcePool"] = "<RESOURCE_POOL>"
payload.setdefault("serviceConfig", {})["replicas"] = 1
service_id = inference.create_service(
service_params=ServiceCreateParams(**payload),
)
print(service_id)
从已有服务创建模板
template_id = inference.create_template_from_service(
service_id=123,
request={"modelName": "qwen2-5b-prod-template"},
)
print(template_id)
创建、更新和删除模板
template = inference.get_template(template_id=69)
payload = template.model_dump(by_alias=True, exclude_none=True)
payload["modelName"] = "qwen2-5b-new-template"
new_template_id = inference.create_template(request=payload)
payload["modelName"] = "qwen2-5b-new-template-v2"
inference.update_template(template_id=new_template_id, request=payload)
new_template_id = 70
inference.delete_template(template_id=new_template_id)
查询引擎版本
engines = inference.list_engine_versions(engine="vllm", page=1, page_size=20)
for engine in engines:
print(engine.id, engine.engine or engine.engine_type, engine.version or engine.engine_version, engine.image)
if not engines:
raise RuntimeError("当前集群没有可用的 vLLM 引擎版本")
engine = inference.get_engine_version(engine_id=engines[0].id)
print(engine.model_dump(by_alias=True, exclude_none=True))
压测和接口一致性测试
创建测试任务前,建议先查询服务当前启用的测试类型;测试所需的资源规格、数据源和目标端点以当前服务端能力及集群配置为准。创建压测任务
from siflow.types import (
LoadTestCreateRequest,
LoadTestDataSource,
LoadTestModeParams,
LoadTestResourceConfig,
)
service_id = 123
capabilities = inference.get_load_test_capabilities(service_id=service_id)
print(capabilities.enabled_test_types)
enabled_test_types = capabilities.enabled_test_types or []
if "fixed_concurrency" not in enabled_test_types:
raise RuntimeError("当前服务未启用固定并发压测")
request = LoadTestCreateRequest(
test_type="fixed_concurrency",
api_type="chat_completions",
mode_params=LoadTestModeParams(
mode="fixed_concurrency",
concurrency=4,
duration_sec=60,
warmup_sec=10,
timeout_sec=120,
),
data_source=LoadTestDataSource(
type="inline_single",
prompt="请简要解释张量并行。",
),
resource_config=LoadTestResourceConfig(
resource_pool="<RESOURCE_POOL>",
instance="sci.c22-2",
instance_quantity=1,
),
)
task = inference.create_load_test_task(
service_id=service_id,
request=request,
)
print(task.id, task.test_type, task.status)
创建接口一致性测试
from siflow.types import (
ConformanceParams,
LoadTestCreateRequest,
LoadTestResourceConfig,
)
request = LoadTestCreateRequest(
test_type="conformance_test",
api_type="chat_completions",
mode_params=ConformanceParams(
interface_types=["openai"],
timeout_seconds=120,
skip_long_context=False,
skip_attack=False,
skip_sampling_params=False,
),
resource_config=LoadTestResourceConfig(
resource_pool="<RESOURCE_POOL>",
instance="sci.c22-2",
instance_quantity=1,
),
)
task = inference.create_load_test_task(
service_id=123,
request=request,
)
print(task.id, task.status)
查询、停止、删除任务和获取报告
service_id = 123
tasks = inference.list_load_test_tasks(service_id=service_id)
for item in tasks:
print(item.id, item.test_type, item.status, item.error_message)
if not tasks:
raise RuntimeError("当前服务没有测试任务")
task = inference.get_load_test_task(
service_id=service_id,
task_id=tasks[0].id,
)
print(task.id, task.test_type, task.status)
report = inference.get_load_test_report(
service_id=123,
task_id=456,
)
print(type(report).__name__)
inference.stop_load_test_task(
service_id=123,
task_id=456,
)
inference.delete_load_test_task(
service_id=123,
task_id=456,
)
get_load_test_report 会先读取任务的 test_type,压测任务返回 LoadTestReport,一致性测试返回 ConformanceReport。
驱逐 Pod
evict_pod 用于优雅驱逐一个推理 Pod,由其控制器重新创建。该接口属于运维能力,需要相应权限;生产环境建议先使用 dryRun=True 校验目标和权限,再执行实际驱逐。
from siflow.types import EvictPodRequest
request = EvictPodRequest(
nodeName="node-a",
podName="qwen2-5b-vllm-worker-0",
workloadUUID="123",
source="sdk",
reason="recreate unhealthy pod",
operator="oncall@example.com",
dryRun=True,
)
preview = inference.evict_pod(request=request)
print(preview.action, preview.detail)
request.dryRun = False
result = inference.evict_pod(request=request)
print(result.action, result.detail)
nodeName 和 podName 为必填字段;workloadUUID、source、action、reason、operator 和 workloadName 为可选上下文。返回值包含 action 和 detail。
常见问题
创建或更新失败时,先检查什么? 先打印最终请求体,确认字段名、资源池、模型路径和角色配置是否符合服务端预期。params = ServiceCreateParams(**payload)
print(params.model_dump(by_alias=True, exclude_none=True))
get_service()?
update_service() 通常需要完整或接近完整的服务配置。先读取当前配置,再修改目标字段,可以减少误删已有配置的风险。
日志流式读取提示缺少依赖怎么办?
安装 WebSocket 依赖后重试:
pip install "siflow[websocket]"
