AI 推理 API

AI Inference Client

class neoruntime_ipc_sdk.inference.InferenceClient(endpoint=None)[source]

Bases: GenAiMixin

AI Inference Client

Usage:

inf = InferenceClient()

# Single inference
result = inf.infer(image, model_id="person_v1")

# Stream inference
for frame, res in inf.subscribe(stream="cam0_main", model="person_v1", fps=10):
    print(f"Detected {len(res.objects)} objects")

__init__(endpoint=None)[source]

connect()[source]

property connected: bool

close()[source]

infer(image, model_id, timeout_ms=5000, priority=4, session_id='')[source]

Run one inference.

image is an ndarray (pixels shipped as bytes) or a zero-copy input: a keep-fd Frame/FrameHandle (NV12 only — its dma-bufs are imported and referenced by buffer id) or a device-resident DspBufferRef (NV12 only — its daemon buffer id is sent directly, no import, no copy; the call blocks, so simply passing the ref keeps it alive). In both buffer-id cases no pixel copy crosses the transport.

infer_async(image, model_id, timeout_ms=5000, priority=4, session_id='')[source]

Non-blocking infer: returns a concurrent.futures.Future that resolves to a parsed InferenceResult. The caller MUST call fut.result(timeout=…) to obtain the result (or propagate the error).

image may be an ndarray or a zero-copy input, same contract as infer(): a keep-fd NV12 Frame/FrameHandle (import released when the RPC settles) or a NV12 DspBufferRef — the ref’s id is borrowed, so keep the ref alive until the future resolves.

Enables depth-N pipelines: submit frame N+1 while still awaiting frame N so the NPU stays busy across the host-side gap between jobs. Schedules onto the same background asyncio loop infer() already uses. The existing blocking infer() is unchanged.

infer_batch(items, timeout_ms=10000)[source]

Submit multiple model inferences in a single batch RPC.

ai-runtime runs them in parallel on the NPU via shared VDevice ROUND_ROBIN scheduling, returning all results together. For models registered with batch_size=B, consecutive same-model items are further aggregated into single NPU jobs (B frames through the array in parallel per job); a partial tail chunk is padded with the last item’s image and the padded results are discarded server-side.

Parameters:

items (list[BatchInferItem]) – List of (image, model_id, …) tuples.
timeout_ms (int) – Overall wall-clock timeout for the entire batch.

Returns:

List of InferenceResult, one per item, in the same order.

Return type:

list[InferenceResult]

infer_batch_async(items, timeout_ms=10000)[source]

Non-blocking infer_batch: returns a concurrent.futures.Future that resolves to List[InferenceResult] (one per item, in submission order). The caller MUST call fut.result(timeout=…). Symmetric to infer_batch(); enables depth-N pipelines on the dual-model (pose+detect) path. The existing blocking infer_batch() is unchanged.

infer_with_tensors(model_id, inputs, input_names=None, timeout_ms=5000)[source]

subscribe(stream, model, fps=10, session_id='', raw_output_only=False, max_consecutive_failures=10, queue_size=100)[source]

Yield (frame_sequence, InferenceResult) for a camera stream subscription.

Failed frames are skipped with a warning. If max_consecutive_failures frames fail in a row (default 10), a RuntimeError is raised instead of yielding nothing forever. Pass 0 or None to disable the limit.

queue_size bounds the client-side result queue. A consumer slower than the stream drops its oldest queued results (the newest always gets through); each drop is counted on the returned iterator’s dropped attribute and reported in a throttled warning — instead of the queue growing memory without bound. Pass 0 or None for an unbounded queue. A stream error is the queue’s one terminal item (offered exactly once, after every result) so backpressure can delay but never drop it: the consumer always sees the exception.

register_model(model_path, model_id=None, owner_id=None, model_type=None, model_variant=None, inputs=None, outputs=None, batch_size=1)[source]

Register a model with ai-runtime.

Parameters:: batch_size (int) – NPU batch for the session (>1 requires a batch-capable HEF; registration fails otherwise). Batched models must use infer_batch(): consecutive same-model items are aggregated into a single NPU job, so B frames traverse the array in parallel. Single-image infer() is rejected for batch>1 models.

unregister_model(model_id)[source]

list_models()[source]

get_model_info(model_id)[source]

get_stats(sampling_window_ms=None)[source]

System and per-model stats.

sampling_window_ms bounds the blocking HAL sampling window behind device/CPU/DSP utilization (they are measured, not read — the call blocks roughly this long). None sends the server-default request (500 ms); the server clamps explicit values to [1, 5000]. Cheap periodic snapshots should ask for a small window (1-50 ms).

create_session(session_id, app_id='', allowed_models=None, max_qps=0, max_concurrent=0, priority=4)[source]

destroy_session(session_id)[source]

update_postprocess_config(model_id, config_json)[source]

Update postprocess configuration for a model at runtime.

For CLIP models, config_json can contain:

{“prompts”: [“a person”, “a car”], “score_threshold”: 0.3}

For detection models, the numeric postprocess keys are accepted:

{“detection_threshold”: 0.38, “iou_threshold”: 0.45,: “max_boxes”: 80}

Applicability, verified on-device (hailo15, 2026-09):

detection_threshold is honored at runtime only when the model’s postprocess resolves to a family function (hailo_yolov8n/hailo_yolov8s/hailo_yolov8m — the default for detection models registered without a backend_function in their variant JSON). Generic plugin exports (e.g. yolov5m_vehicles) hardcode their thresholds and ignore JSON tuning entirely.
iou_threshold / max_boxes are accepted by the chain but have no behavioral effect: suppression and box capping happen in the HEF’s compile-time integrated NMS on the accelerator, so they cannot be moved after compilation.

Unknown keys are rejected server-side (the RPC raises) — push only keys the model’s postprocess schema knows.

Returns True on success.

InferenceClient

class neoruntime_ipc_sdk.InferenceClient(endpoint=None)[source]

Bases: GenAiMixin

AI Inference Client

Usage:

inf = InferenceClient()

# Single inference
result = inf.infer(image, model_id="person_v1")

# Stream inference
for frame, res in inf.subscribe(stream="cam0_main", model="person_v1", fps=10):
    print(f"Detected {len(res.objects)} objects")

__init__(endpoint=None)[source]

connect()[source]

property connected: bool

close()[source]

infer(image, model_id, timeout_ms=5000, priority=4, session_id='')[source]

Run one inference.

image is an ndarray (pixels shipped as bytes) or a zero-copy input: a keep-fd Frame/FrameHandle (NV12 only — its dma-bufs are imported and referenced by buffer id) or a device-resident DspBufferRef (NV12 only — its daemon buffer id is sent directly, no import, no copy; the call blocks, so simply passing the ref keeps it alive). In both buffer-id cases no pixel copy crosses the transport.

infer_async(image, model_id, timeout_ms=5000, priority=4, session_id='')[source]

Non-blocking infer: returns a concurrent.futures.Future that resolves to a parsed InferenceResult. The caller MUST call fut.result(timeout=…) to obtain the result (or propagate the error).

image may be an ndarray or a zero-copy input, same contract as infer(): a keep-fd NV12 Frame/FrameHandle (import released when the RPC settles) or a NV12 DspBufferRef — the ref’s id is borrowed, so keep the ref alive until the future resolves.

Enables depth-N pipelines: submit frame N+1 while still awaiting frame N so the NPU stays busy across the host-side gap between jobs. Schedules onto the same background asyncio loop infer() already uses. The existing blocking infer() is unchanged.

infer_batch(items, timeout_ms=10000)[source]

Submit multiple model inferences in a single batch RPC.

ai-runtime runs them in parallel on the NPU via shared VDevice ROUND_ROBIN scheduling, returning all results together. For models registered with batch_size=B, consecutive same-model items are further aggregated into single NPU jobs (B frames through the array in parallel per job); a partial tail chunk is padded with the last item’s image and the padded results are discarded server-side.

Parameters:

items (list[BatchInferItem]) – List of (image, model_id, …) tuples.
timeout_ms (int) – Overall wall-clock timeout for the entire batch.

Returns:

List of InferenceResult, one per item, in the same order.

Return type:

list[InferenceResult]

infer_batch_async(items, timeout_ms=10000)[source]

Non-blocking infer_batch: returns a concurrent.futures.Future that resolves to List[InferenceResult] (one per item, in submission order). The caller MUST call fut.result(timeout=…). Symmetric to infer_batch(); enables depth-N pipelines on the dual-model (pose+detect) path. The existing blocking infer_batch() is unchanged.

infer_with_tensors(model_id, inputs, input_names=None, timeout_ms=5000)[source]

subscribe(stream, model, fps=10, session_id='', raw_output_only=False, max_consecutive_failures=10, queue_size=100)[source]

Yield (frame_sequence, InferenceResult) for a camera stream subscription.

Failed frames are skipped with a warning. If max_consecutive_failures frames fail in a row (default 10), a RuntimeError is raised instead of yielding nothing forever. Pass 0 or None to disable the limit.

queue_size bounds the client-side result queue. A consumer slower than the stream drops its oldest queued results (the newest always gets through); each drop is counted on the returned iterator’s dropped attribute and reported in a throttled warning — instead of the queue growing memory without bound. Pass 0 or None for an unbounded queue. A stream error is the queue’s one terminal item (offered exactly once, after every result) so backpressure can delay but never drop it: the consumer always sees the exception.

register_model(model_path, model_id=None, owner_id=None, model_type=None, model_variant=None, inputs=None, outputs=None, batch_size=1)[source]

Register a model with ai-runtime.

Parameters:: batch_size (int) – NPU batch for the session (>1 requires a batch-capable HEF; registration fails otherwise). Batched models must use infer_batch(): consecutive same-model items are aggregated into a single NPU job, so B frames traverse the array in parallel. Single-image infer() is rejected for batch>1 models.

unregister_model(model_id)[source]

list_models()[source]

get_model_info(model_id)[source]

get_stats(sampling_window_ms=None)[source]

System and per-model stats.

sampling_window_ms bounds the blocking HAL sampling window behind device/CPU/DSP utilization (they are measured, not read — the call blocks roughly this long). None sends the server-default request (500 ms); the server clamps explicit values to [1, 5000]. Cheap periodic snapshots should ask for a small window (1-50 ms).

create_session(session_id, app_id='', allowed_models=None, max_qps=0, max_concurrent=0, priority=4)[source]

destroy_session(session_id)[source]

update_postprocess_config(model_id, config_json)[source]

Update postprocess configuration for a model at runtime.

For CLIP models, config_json can contain:

{“prompts”: [“a person”, “a car”], “score_threshold”: 0.3}

For detection models, the numeric postprocess keys are accepted:

{“detection_threshold”: 0.38, “iou_threshold”: 0.45,: “max_boxes”: 80}

Applicability, verified on-device (hailo15, 2026-09):

detection_threshold is honored at runtime only when the model’s postprocess resolves to a family function (hailo_yolov8n/hailo_yolov8s/hailo_yolov8m — the default for detection models registered without a backend_function in their variant JSON). Generic plugin exports (e.g. yolov5m_vehicles) hardcode their thresholds and ignore JSON tuning entirely.
iou_threshold / max_boxes are accepted by the chain but have no behavioral effect: suppression and box capping happen in the HEF’s compile-time integrated NMS on the accelerator, so they cannot be moved after compilation.

Unknown keys are rejected server-side (the RPC raises) — push only keys the model’s postprocess schema knows.

Returns True on success.

数据类型

InferenceResult

class neoruntime_ipc_sdk.InferenceResult(frame_sequence: 'int', timestamp_ns: 'int', objects: 'list[DetectedObject]' = <factory>, classifications: 'list[Classification]' = <factory>, landmarks: 'list[LandmarkSet]' = <factory>, masks: 'list[SegmentationMask]' = <factory>, ocr_lines: 'list[OcrLine]' = <factory>, embeddings: 'list[Embedding]' = <factory>, depth_maps: 'list[DepthMap]' = <factory>, raw_outputs: 'list[np.ndarray] | None' = None, infer_time_us: 'int' = 0, queue_time_us: 'int' = 0, hw_infer_time_us: 'int' = 0, status_message: 'str' = '', skew_us: 'int' = 0)[source]

frame_sequence: int

timestamp_ns: int

objects: list[DetectedObject]

classifications: list[Classification]

landmarks: list[LandmarkSet]

masks: list[SegmentationMask]

ocr_lines: list[OcrLine]

embeddings: list[Embedding]

depth_maps: list[DepthMap]

raw_outputs: list[ndarray] | None = None

infer_time_us: int = 0

queue_time_us: int = 0

hw_infer_time_us: int = 0

status_message: str = ''

skew_us: int = 0

has_person()[source]

count_by_label(label)[source]

get_objects_by_label(label)[source]

__init__(frame_sequence, timestamp_ns, objects=<factory>, classifications=<factory>, landmarks=<factory>, masks=<factory>, ocr_lines=<factory>, embeddings=<factory>, depth_maps=<factory>, raw_outputs=None, infer_time_us=0, queue_time_us=0, hw_infer_time_us=0, status_message='', skew_us=0)

BatchInferItem

class neoruntime_ipc_sdk.BatchInferItem(image, model_id, timeout_ms=5000, priority=4)[source]

A single inference request within a batch.

image: ndarray

model_id: str

timeout_ms: int = 5000

priority: int = 4

__init__(image, model_id, timeout_ms=5000, priority=4)

DetectedObject

class neoruntime_ipc_sdk.DetectedObject(label: 'str', score: 'float', bbox: 'BoundingBox', class_id: 'int' = 0, track_id: 'int | None' = None)[source]

label: str

score: float

bbox: BoundingBox

class_id: int = 0

track_id: int | None = None

__init__(label, score, bbox, class_id=0, track_id=None)

BoundingBox

class neoruntime_ipc_sdk.BoundingBox(x: 'float', y: 'float', width: 'float', height: 'float')[source]

x: float

y: float

width: float

height: float

to_xyxy()[source]

to_xywh()[source]

__init__(x, y, width, height)

LandmarkPoint

class neoruntime_ipc_sdk.LandmarkPoint(x: 'float', y: 'float', confidence: 'float' = 1.0)[source]

x: float

y: float

confidence: float = 1.0

__init__(x, y, confidence=1.0)

LandmarkSet

class neoruntime_ipc_sdk.LandmarkSet(type: 'str', points: 'list[LandmarkPoint]' = <factory>)[source]

type: str

points: list[LandmarkPoint]

__init__(type, points=<factory>)

SegmentationMask

class neoruntime_ipc_sdk.SegmentationMask(class_id: 'int', label: 'str', confidence: 'float', bbox: 'BoundingBox', mask_rle: 'bytes', mask_width: 'int', mask_height: 'int')[source]

class_id: int

label: str

confidence: float

bbox: BoundingBox

mask_rle: bytes

mask_width: int

mask_height: int

to_numpy_mask()[source]

Decode RLE to HxW bool numpy array.

__init__(class_id, label, confidence, bbox, mask_rle, mask_width, mask_height)

OcrLine

class neoruntime_ipc_sdk.OcrLine(text: 'str', confidence: 'float', bbox: 'BoundingBox')[source]

text: str

confidence: float

bbox: BoundingBox

__init__(text, confidence, bbox)

Embedding

class neoruntime_ipc_sdk.Embedding(dim: 'int', data: 'list[float]')[source]

dim: int

data: list[float]

__init__(dim, data)

DepthMap

class neoruntime_ipc_sdk.DepthMap(width: 'int', height: 'int', data: 'np.ndarray')[source]

width: int

height: int

data: ndarray

__init__(width, height, data)

Classification

class neoruntime_ipc_sdk.Classification(type: 'str', class_id: 'int', label: 'str', confidence: 'float')[source]

type: str

class_id: int

label: str

confidence: float

__init__(type, class_id, label, confidence)

ModelInfo

class neoruntime_ipc_sdk.ModelInfo(model_id: 'str', model_path: 'str', version: 'str' = '', inputs: 'list[dict]' = <factory>, outputs: 'list[dict]' = <factory>, estimated_tops: 'float' = 0.0, estimated_memory: 'int' = 0, load_timestamp: 'int' = 0, batch_size: 'int' = 1)[source]

model_id: str

model_path: str

version: str = ''

inputs: list[dict]

outputs: list[dict]

estimated_tops: float = 0.0

estimated_memory: int = 0

load_timestamp: int = 0

batch_size: int = 1

__init__(model_id, model_path, version='', inputs=<factory>, outputs=<factory>, estimated_tops=0.0, estimated_memory=0, load_timestamp=0, batch_size=1)

使用示例

单次推理

from neoruntime_ipc_sdk import InferenceClient
import numpy as np

inf = InferenceClient()

# 准备图像 (numpy array)
image = np.zeros((1080, 1920, 3), dtype=np.uint8)

# 执行推理
result = inf.infer(image, model_id="person_v1")

# 处理结果
for obj in result.objects:
    print(f"{obj.label}: {obj.score:.2f}")
    print(f"  位置: ({obj.bbox.x}, {obj.bbox.y})")
    print(f"  大小: {obj.bbox.width}x{obj.bbox.height}")

# 便捷方法
if result.has_person():
    print("检测到人员")

person_count = result.count_by_label("person")
print(f"人员数量: {person_count}")

persons = result.get_objects_by_label("person")

流式推理

# 订阅视频流推理结果
for frame_seq, result in inf.subscribe(
    stream="main",
    model="person_v1",
    fps=15
):
    print(f"帧 {frame_seq}: 检测到 {len(result.objects)} 个对象")

    for obj in result.objects:
        if obj.score > 0.8:
            print(f"  高置信度: {obj.label} ({obj.score:.2f})")

张量推理

import numpy as np

inf = InferenceClient()

# 准备输入张量
input1 = np.random.randn(1, 3, 224, 224).astype(np.float32)
input2 = np.random.randn(1, 3, 112, 112).astype(np.float32)

# 执行推理
outputs = inf.infer_with_tensors(
    model_id="custom_model",
    inputs=[input1, input2],
    input_names=["input_main", "input_sub"]
)

# 处理输出张量
for i, output in enumerate(outputs):
    print(f"输出 {i}: shape={output.shape}")

模型管理

# 列出所有模型
models = inf.list_models()
for model in models:
    print(f"模型ID: {model.model_id}")
    print(f"路径: {model.model_path}")
    print(f"版本: {model.version}")
    print(f"输入: {model.inputs}")
    print(f"输出: {model.outputs}")
    print(f"估算 TOPS: {model.estimated_tops}")
    print(f"估算内存: {model.estimated_memory} bytes")

# 获取模型详情
info = inf.get_model_info("person_v1")
if info:
    print(f"模型ID: {info.model_id}")
    print(f"路径: {info.model_path}")
    print(f"版本: {info.version}")

# 注册新模型
model_id = inf.register_model(
    model_path="/opt/models/custom.hef",
    model_id="custom_v1"
)
print(f"注册模型 ID: {model_id}")

# 注销模型
inf.unregister_model("custom_v1")

获取统计信息

stats = inf.get_stats()

print(f"设备利用率: {stats['device_utilization']}%")
print(f"设备温度: {stats['device_temperature']}°C")
print(f"总内存: {stats['total_memory_bytes']} bytes")
print(f"已用内存: {stats['used_memory_bytes']} bytes")

for model_stat in stats['model_stats']:
    print(f"模型: {model_stat['model_id']}")
    print(f"  总推理次数: {model_stat['total_inferences']}")
    print(f"  总错误数: {model_stat['total_errors']}")
    print(f"  平均延迟: {model_stat['avg_latency_us']}us")
    print(f"  当前 QPS: {model_stat['current_qps']}")
    print(f"  队列深度: {model_stat['queue_depth']}")

会话管理

# 创建会话
session_id = inf.create_session(
    session_id="my_session",
    app_id="my_app",
    allowed_models=["person_v1", "car_v1"],
    max_qps=10,
    max_concurrent=2,
    priority=4
)
print(f"会话ID: {session_id}")

# 使用会话进行推理
result = inf.infer(image, model_id="person_v1", session_id=session_id)

# 销毁会话
inf.destroy_session(session_id)

处理不同类型的结果

result = inf.infer(image, model_id="face_detection")

# 检测结果
for obj in result.objects:
    # 边界框
    x1, y1, x2, y2 = obj.bbox.to_xyxy()
    print(f"边界框: ({x1}, {y1}) - ({x2}, {y2})")

# 分类结果
for cls in result.classifications:
    print(f"分类: {cls.type} - {cls.label}: {cls.confidence:.2f}")

# 关键点
for lm_set in result.landmarks:
    print(f"关键点集类型: {lm_set.type}")
    for point in lm_set.points:
        print(f"  点: ({point.x}, {point.y}), 置信度: {point.confidence}")

# 原始输出
if result.raw_outputs:
    print(f"原始输出数量: {len(result.raw_outputs)}")

# 性能信息
print(f"推理时间: {result.infer_time_us}us")
print(f"排队时间: {result.queue_time_us}us")

上下文管理器

# 使用上下文管理器自动管理连接
with InferenceClient() as inf:
    result = inf.infer(image, model_id="person_v1")
    print(f"检测到 {len(result.objects)} 个对象")

错误处理

from grpc import RpcError

try:
    result = inf.infer(image, model_id="nonexistent_model")
except RpcError as e:
    print(f"推理失败: {e.details()}")
except RuntimeError as e:
    print(f"运行时错误: {e}")

分割结果

result = inf.infer(image, model_id="segmentation_v1")

for mask in result.masks:
    print(f"掩码: {mask.label} (置信度: {mask.confidence:.2f})")
    print(f"  边界框: ({mask.bbox.x}, {mask.bbox.y}, {mask.bbox.width}, {mask.bbox.height})")

    # 解码 RLE 掩码为 numpy bool 数组 (H x W)
    np_mask = mask.to_numpy_mask()
    print(f"  掩码尺寸: {np_mask.shape}, 像素数: {np_mask.sum()}")

OCR 结果

result = inf.infer(image, model_id="ocr_v1")

for line in result.ocr_lines:
    print(f"文本: '{line.text}' (置信度: {line.confidence:.2f})")
    print(f"  位置: ({line.bbox.x}, {line.bbox.y})")

嵌入 (CLIP 图像)

result = inf.infer(image, model_id="clip_vit_b32")

for emb in result.embeddings:
    print(f"嵌入维度: {emb.dim}")
    # 可用于相似度检索等场景
    import numpy as np
    vec = np.array(emb.data)
    print(f"  L2 范数: {np.linalg.norm(vec):.4f}")

CLIP 文本编码

# 通过 NPU 将文本编码为 CLIP 嵌入
embedding = inf.encode_text("a person walking in the park")
print(f"嵌入长度: {len(embedding)}")

# 编码多段文本并计算相似度
import numpy as np
emb1 = np.array(inf.encode_text("a cat"))
emb2 = np.array(inf.encode_text("a dog"))
similarity = np.dot(emb1, emb2) / (np.linalg.norm(emb1) * np.linalg.norm(emb2))
print(f"相似度: {similarity:.4f}")

深度估计

result = inf.infer(image, model_id="depth_v1")

for dm in result.depth_maps:
    print(f"深度图: {dm.width}x{dm.height}")
    print(f"  最小深度: {dm.data.min():.2f}, 最大深度: {dm.data.max():.2f}")

运行时更新后处理配置

import json

# 运行时更新 CLIP 文本提示词
config = json.dumps({
    "prompts": ["a person", "a car", "a bicycle"],
    "score_threshold": 0.3
})
inf.update_postprocess_config("clip_vit_b32", config)

# 运行时调整检测模型的分数门限（无需重新注册）
inf.update_postprocess_config("yolov8n", json.dumps({
    "detection_threshold": 0.5
}))

detection_threshold 仅在模型后处理解析为家族函数（hailo_yolov8n/s/m —— 检测模型注册时 variant JSON 不带 backend_function 的默认路径）时于运行时生效；通用插件导出会忽略 JSON 调参。iou_threshold 与 max_boxes 虽被链路接受但无行为效果：抑制与封顶发生在 HEF 编译期集成的硬件 NMS 里。未知键会使 RPC 抛出服务端的拒绝信息（-2801）。

GenAI (LLM/VLM)

import json

# 创建 GenAI 会话
session_id = inf.genai_create_session(
    hef_path="/opt/models/llm.hef",
    kind="llm"
)

# 流式生成 token
messages = [json.dumps({"role": "user", "content": "Hello, who are you?"})]
full_response = ""
for token in inf.genai_generate(
    session_id=session_id,
    messages=messages,
    max_tokens=256,
    temperature=0.7,
    do_sample=True
):
    print(token, end="", flush=True)
    full_response += token

# VLM: 携带图像输入生成
with open("image.jpg", "rb") as f:
    img_data = f.read()

vlm_session = inf.genai_create_session("/opt/models/vlm.hef", kind="vlm")
messages = [json.dumps({"role": "user", "content": "Describe this image."})]
for token in inf.genai_generate(
    session_id=vlm_session,
    messages=messages,
    images=[img_data]
):
    print(token, end="", flush=True)

# 中止进行中的生成
inf.genai_abort(session_id)

# 清理会话
inf.genai_destroy_session(session_id)
inf.genai_destroy_session(vlm_session)