> ## Documentation Index
> Fetch the complete documentation index at: https://docs.siflow.cn/llms.txt
> Use this file to discover all available pages before exploring further.

# 使用 Python SDK 管理大模型推理服务

> 使用人工智能平台 Python SDK 创建、查询、调用、上线、下线、删除大模型推理服务，并管理模板、引擎、灰度、压测和运行信息。

Python SDK 可用于自动化管理大模型推理服务。使用本文示例前，请先完成 [Python SDK 快速开始](./python-sdk-quickstart)，并准备模型来源、推理引擎、资源池、实例规格、端口和访问配置。

## 初始化客户端

```python theme={null}
from siflow import SiFlow

client = SiFlow(region="cn-beijing", cluster="auriga")
inference = client.inference
```

大模型推理服务通过 `client.inference` 管理。服务 ID 通常为整数，例如 `123`。

## 查询创建前可用选项

创建服务前，建议先查询当前集群支持的引擎、存储类型、KV Cache 选项和资源包。

```python theme={null}
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 推理服务。请按实际环境替换资源池、模型名称、模型版本、实例规格、服务名称和网关配置。

```python theme={null}
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

```python theme={null}
payload["modelConfig"] = {
    "modelSource": {
        "storageType": "volume",
        "storage": {
            "volume": {
                "volumeId": 123,
                "mountPath": "/models",
                "modelPath": "/models/Qwen2.5-7B-Instruct",
                "subPath": "team-a/qwen",
                "readOnly": True,
            }
        },
    }
}
```

开启目录级权限的 Volume 通常需要配置 `subPath`；`subPath` 为卷内相对路径。

### 使用 PVC

```python theme={null}
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 密钥不建议硬编码到代码仓库。示例中用环境变量读取，生产环境可结合密钥管理系统注入。

```python theme={null}
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` 中指定镜像、启动命令、参数、端口和资源规格。

```python theme={null}
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` 以及服务端支持能力，可先查询引擎版本确认。

```python theme={null}
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)
```

示例：

```python theme={null}
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)
```

## 查询和调用服务

### 查询服务详情

```python theme={null}
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)
```

### 查询服务列表

```python theme={null}
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` 适合在正式接入前验证网关是否能访问。

```python theme={null}
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` 需要按实际网关地址和路径设置。

```bash theme={null}
pip install openai
```

```python theme={null}
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)
```

流式调用：

```python theme={null}
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)
```

## 更新、扩缩容和生命周期管理

### 更新服务配置

更新服务时，建议先读取当前服务配置，再在此基础上修改目标字段，避免遗漏需要保留的配置。

```python theme={null}
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`：

```python theme={null}
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),
)
```

如果更新过程需要中止：

```python theme={null}
inference.cancel_update_service(service_id=123)
```

### 扩缩容

`scale_service` 用于调整角色副本数，最常见的是调整 `worker.replicas`。

```python theme={null}
from siflow.types import ServiceScaleParams

resp = inference.scale_service(
    service_id=123,
    scale_params=ServiceScaleParams(
        **{
            "roleConfig": {
                "worker": {"replicas": 3},
            }
        }
    ),
)
print(resp)
```

停止正在进行的扩缩容：

```python theme={null}
inference.stop_scale_service(service_id=123)
```

### 上线、下线和删除

上线服务：

```python theme={null}
inference.online_service(service_id=123)
```

下线服务：

```python theme={null}
inference.offline_service(service_id=123)
```

确认服务不再使用后再删除：

```python theme={null}
inference.delete_service(service_id=123)
```

批量上线服务：

```python theme={null}
inference.batch_online_service(service_ids=[123, 124])
```

也可以按服务名称批量上线：

```python theme={null}
inference.batch_online_service_by_name(names=["qwen-prod-a", "qwen-prod-b"])
```

批量下线服务：

```python theme={null}
inference.batch_offline_service(service_ids=[123, 124])
```

也可以按服务名称批量下线：

```python theme={null}
inference.batch_offline_service_by_name(names=["qwen-prod-a", "qwen-prod-b"])
```

确认目标服务不再使用后再批量删除：

```python theme={null}
inference.batch_delete_service(service_ids=[123, 124])
```

也可以按服务名称批量删除：

```python theme={null}
inference.batch_delete_service_by_name(names=["qwen-prod-a", "qwen-prod-b"])
```

上线、下线、扩缩容和删除都会影响服务可用性或资源占用。执行前，请确认业务流量、资源配额和变更窗口。

## 共享和公开服务

共享给指定用户：

```python theme={null}
inference.share_service(
    service_id=123,
    shared_users=["alice", "bob"],
)
```

需要向所有符合权限条件的用户公开服务时：

```python theme={null}
inference.public_service(service_id=123)
```

需要取消公开时：

```python theme={null}
inference.unpublic_service(service_id=123)
```

## 灰度发布和版本回滚

### 查看服务版本

```python theme={null}
versions = inference.list_service_versions(service_id=123)

for version in versions:
    print(version.version, version.status, version.created_at)
```

### 回滚服务版本

回滚到上一版本：

```python theme={null}
inference.rollback_service(service_id=123)
```

需要回滚到指定历史版本时：

```python theme={null}
inference.rollback_service_to_version(service_id=123, version=3)
```

### 发起灰度

```python theme={null}
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)
```

### 调整和完成灰度

```python theme={null}
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`，再执行收尾：

```python theme={null}
rollout = inference.get_rollout(service_id=123)
print(rollout.state)

if rollout.state == "completed":
    inference.finalize_rollout(service_id=123)
```

灰度异常时回滚：

```python theme={null}
inference.rollback_rollout(service_id=123)
```

## 查询 Pod、日志和指标

查询服务 Pod：

```python theme={null}
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)
```

下线状态或需要读取 offline endpoint 时：

```python theme={null}
instances = inference.list_offline_service_instances(service_id=123)
```

重建单个 Pod：

```python theme={null}
inference.recreate_instance(
    service_id=123,
    pod_name="qwen2-5b-vllm-worker-0",
)
```

按服务 ID 查询日志：

```python theme={null}
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}")
```

读取指定 Pod 容器日志：

```python theme={null}
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]`：

```bash theme={null}
pip install "siflow[websocket]"
```

```python theme={null}
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：

```python theme={null}
metrics = inference.get_service_metrics(
    service_id=123,
    role="worker",
    timeout_seconds=3,
)

for role, rows in metrics.items():
    print(role, rows[:3])
```

查询大模型服务 Dashboard 汇总：

```python theme={null}
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)
```

更多日志分页、下载和系统日志示例，请参考 [使用 Python SDK 查询日志和指标](./query-logs-and-metrics-with-python-sdk)。

## 管理模板和引擎

### 使用模板创建服务

```python theme={null}
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)
```

### 从已有服务创建模板

```python theme={null}
template_id = inference.create_template_from_service(
    service_id=123,
    request={"modelName": "qwen2-5b-prod-template"},
)

print(template_id)
```

### 创建、更新和删除模板

```python theme={null}
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)
```

确认不再需要模板后再删除：

```python theme={null}
new_template_id = 70
inference.delete_template(template_id=new_template_id)
```

### 查询引擎版本

```python theme={null}
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))
```

创建、更新或删除引擎版本通常属于平台管理操作。业务调用方一般只需要查询当前可用引擎和版本。

## 压测和接口一致性测试

创建测试任务前，建议先查询服务当前启用的测试类型；测试所需的资源规格、数据源和目标端点以当前服务端能力及集群配置为准。

### 创建压测任务

```python theme={null}
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)
```

### 创建接口一致性测试

```python theme={null}
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)
```

### 查询、停止、删除任务和获取报告

```python theme={null}
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)
```

测试任务完成后，获取测试报告：

```python theme={null}
report = inference.get_load_test_report(
    service_id=123,
    task_id=456,
)
print(type(report).__name__)
```

需要中止仍在运行的测试任务时：

```python theme={null}
inference.stop_load_test_task(
    service_id=123,
    task_id=456,
)
```

确认不再需要测试任务及其记录后再删除：

```python theme={null}
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` 校验目标和权限，再执行实际驱逐。

```python theme={null}
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`。

## 常见问题

**创建或更新失败时，先检查什么？**

先打印最终请求体，确认字段名、资源池、模型路径和角色配置是否符合服务端预期。

```python theme={null}
params = ServiceCreateParams(**payload)
print(params.model_dump(by_alias=True, exclude_none=True))
```

**为什么建议更新前先 `get_service()`？**

`update_service()` 通常需要完整或接近完整的服务配置。先读取当前配置，再修改目标字段，可以减少误删已有配置的风险。

**日志流式读取提示缺少依赖怎么办？**

安装 WebSocket 依赖后重试：

```bash theme={null}
pip install "siflow[websocket]"
```

## 相关文档

* [Python SDK 快速开始](./python-sdk-quickstart)
* [使用 Python SDK 查询日志和指标](./query-logs-and-metrics-with-python-sdk)
* [准备大模型推理配置](../inference/prepare-llm-inference-configuration)
* [创建大模型推理服务](../inference/create-llm-inference-services)
* [管理大模型推理服务](../inference/manage-llm-inference-services)