---
title: qwen3-embedding-0.6b
canonical_url: "https://www.modelscope.cn/models/xiaowangge/qwen3-embedding-0.6b"
md_url: "https://www.modelscope.cn/models/xiaowangge/qwen3-embedding-0.6b.md"
repository: xiaowangge/qwen3-embedding-0.6b
chinese_name: qwen3-embedding-0.6b
last_updated: 2025-08-09
license: "Apache License 2.0"
pipeline_tag: feature-extraction
tasks:
  - feature-extraction
  - sentence-similarity
  - text-generation
  - sentence-similarity
model_type:
  - qwen3
architectures:
  - Qwen3ForCausalLM
base_model:
  - Qwen/Qwen3-Embedding-0.6B
base_model_relation: quantized
library_name:
  - transformer
  - onnx
  - pytorch
frameworks:
  - Pytorch
language:
  - zh
  - en
inference_backends:
  - "deploy_task text/emb"
  - "lmdeploy_turbomind 0.9.1"
  - "sglang 0.5.2"
  - "vllm 0.9.2"
downloads: 176181
stars: 11
---

# qwen3-embedding-0.6b

> qwen3-embedding-0.6b - xiaowangge 在 ModelScope 开源的模型。转换说明 从 modelscope 网站 Qwen/Qwen3-Embedding-0.6B 模型转换而来。 提交编号：729a03da5b9dcd17805d30b64c87f75d31ce40f8。 分别使用 transformers 和 onnxruntime 进行 cpu 推理（池化方式：last token），精度评测如下：

xiaowangge/qwen3-embedding-0.6b 是 ModelScope 魔搭社区上的feature-extraction、sentence-similarity、text-generation模型，采用 Apache License 2.0 许可，基于 Qwen/Qwen3-Embedding-0.6B 构建，可用 deploy_task text/emb、lmdeploy_turbomind 0.9.1、sglang 0.5.2 部署。

- **Repository**: xiaowangge/qwen3-embedding-0.6b
- **License**: Apache License 2.0
- **Tasks**: feature-extraction, sentence-similarity, text-generation, sentence-similarity
- **Base model**: Qwen/Qwen3-Embedding-0.6B
- **Inference backends**: deploy_task text/emb, lmdeploy_turbomind 0.9.1, sglang 0.5.2, vllm 0.9.2
- **Downloads**: 176181
- **Stars**: 11
- **Last updated**: 2025-08-09

Source: https://www.modelscope.cn/models/xiaowangge/qwen3-embedding-0.6b

---

### 转换说明
- 从 modelscope 网站 `Qwen/Qwen3-Embedding-0.6B` 模型转换而来。
- 提交编号：729a03da5b9dcd17805d30b64c87f75d31ce40f8。
- 分别使用 transformers 和 onnxruntime 进行 cpu 推理（池化方式：last token），精度评测如下：

```
最大差异： 5.1409006e-07
最小差异： 0.0
平均差异： 7.7263245e-08
```

### 下载模型

#### SDK 下载

```bash
# 安装 ModelScope
pip install modelscope
```

```python
# SDK 模型下载
from modelscope import snapshot_download
model_dir = snapshot_download('xiaowangge/qwen3-embedding-0.6b')
```

#### Git 下载
```
# Git模型下载
git clone https://www.modelscope.cn/xiaowangge/qwen3-embedding-0.6b.git
```

### 本地应用

> 默认为 `cpu` 推理，`gpu` 加速推理需要安装 `onnxruntime_gpu`。

#### 环境

> 推荐使用虚拟环境进行包管理，基础环境要求： python >= 3.11。

```
coloredlogs==15.0.1
fastapi==0.115.13
numpy==2.3.0
onnxruntime_gpu==1.22.0
pydantic==2.11.7
transformers==4.51.3
uvicorn==0.34.3
```

#### 程序

> 示例接口服务基本支持个人仓库下的所有文本向量模型，可以用于 `bce` 或 `bge` 系列模型推理。

```python
import asyncio
import json
import logging
import os
import threading
import time
import uuid
from concurrent.futures import ThreadPoolExecutor
from contextlib import asynccontextmanager
from pathlib import Path
from typing import Any, Dict, List

import coloredlogs
import numpy as np
import onnxruntime as ort
import uvicorn
from fastapi import FastAPI, Header
from pydantic import BaseModel
from transformers import AutoTokenizer, PreTrainedTokenizerFast

# 获取日志对象
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)

# 创建日志文件
log_file_path = "app.log"
file_formatter = logging.Formatter(
    fmt="%(levelname)s %(asctime)s - [t-%(threadName)s] (%(filename)s:%(lineno)d) %(funcName)s : %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
)
file_handler = logging.FileHandler(log_file_path, encoding="utf-8")
file_handler.setFormatter(file_formatter)
logger.addHandler(file_handler)

# 创建日志服务
coloredlogs.install(
    level="DEBUG",
    datefmt="%Y-%m-%d %H:%M:%S",
    milliseconds=True,
    fmt="%(levelname)s %(asctime)s - [t-%(threadName)s] (%(filename)s:%(lineno)d) %(funcName)s : %(message)s",
    field_styles={
        "levelname": {"color": "green"},
        "asctime": {"color": "magenta"},
        "name": {"color": "blue", "bold": True},
        "thread": {"color": "yellow"},
        "funcName": {"color": "green"},
    },
)


class EmbeddingRequest(BaseModel):
    """
    嵌入请求对象
    """
    model: str
    input: List[str]
    normalize: bool = True


class OnnxModelManager:
    """
    ONNX 模型管理器
    """

    def __init__(self, base_path: Path, max_executors: int = 5, max_active_models: int = 5,
                 max_idle_time: int = 5 * 60):
        self.__base_path: Path = base_path
        self.__max_active_models: int = max_active_models
        self.__max_idle_time: int = max_idle_time
        self.__available_models: List[str] = []
        self.__loaded_models: Dict[str, OnnxModel] = {}
        self.__embedding_executor: ThreadPoolExecutor = ThreadPoolExecutor(max_workers=max_executors,
                                                                           thread_name_prefix="embedding")
        self.__lock: threading.RLock = threading.RLock()
        self.__stop_event: threading.Event = threading.Event()

    def init_available_model(self):
        """
        初始化可用模型
        """
        logger.info("Find available models...")
        if self.__base_path is None:
            raise ValueError("Please set model path before run app")
        if not self.__base_path.exists():
            raise ValueError(f"Base model path({self.__base_path}) does not exist")
        if not self.__base_path.is_dir():
            raise ValueError(f"Base model path({self.__base_path}) is not a directory")
        # 扫描模型目录
        for model_path in self.__base_path.iterdir():
            if model_path.is_dir():
                self.__available_models.append(model_path.name)

        logger.info(f"Available models: {self.__available_models}")

    def load_model(self, model_name: str):
        """
        加载模型
        :param model_name: 模型名称
        """
        with self.__lock:
            # 检查是否超过最大激活模型数
            if len(self.__loaded_models) >= self.__max_active_models:
                raise ValueError(
                    f"Too many active models: {len(self.__loaded_models)}, max: {self.__max_active_models}")

            if model_name not in self.__available_models:
                raise ValueError(f"Model {model_name} is not available")

            base_model_path = self.__base_path.joinpath(model_name)
            if not base_model_path.exists():
                raise ValueError(f"Model path({base_model_path}) does not exist")

            # 加载模型配置
            model_config_path = base_model_path.joinpath("onnx_config.json")
            model_config = {}
            if model_config_path.exists():
                model_config_str = model_config_path.read_text(encoding="utf-8")
                try:
                    model_config = json.loads(model_config_str)
                except Exception as e:
                    logger.warning(f"Failed to load model config: {e}")
            logger.debug(f"Model config: {model_config}")

            # 加载分词器
            tokenizer_path = base_model_path.joinpath("tokenizer.json")
            if not tokenizer_path.exists():
                raise ValueError(f"Tokenizer path({tokenizer_path}) does not exist")
            # padding_side 默认设为 left，推理其它模型时，需要考虑 padding_side 是否设为 right
            padding_side = model_config.get("padding_side", "left")
            tokenizer = AutoTokenizer.from_pretrained(base_model_path, trust_remote_code=True,
                                                      padding_side=padding_side)
            logger.info(f"Loaded tokenizer for model: {model_name}")

            # 加载模型
            model_path = self.__base_path.joinpath(model_name).joinpath("model.onnx")
            if not model_path.exists():
                raise ValueError(f"Model path({model_path}) does not exist")
            logger.info(f"Loaded model for name: {model_name}")

            # 缓存模型会话
            ort_session = ort.InferenceSession(model_path, providers=["CUDAExecutionProvider", "CPUExecutionProvider"])
            self.__loaded_models[model_name] = OnnxModel(base_model_path, tokenizer, ort_session, model_config)
            logger.info(f"Loaded model success, providers: {ort_session.get_providers()}")

    def unload_models(self):
        """
        卸载模型
        """
        while not self.__stop_event.wait(timeout=10):
            with self.__lock:
                logger.info("Checking idle models...")
                # 遍历模型列表，判断是否超过最大空闲时间
                for model_name, onnx_model in list(self.__loaded_models.items()):
                    last_access_time = onnx_model.get_last_access_time()
                    if int(time.time() - last_access_time) > self.__max_idle_time:
                        logger.info(f"Unload model {model_name}")
                        logger.info(f"Unloading idle model: {model_name}")
                        model = self.__loaded_models.pop(model_name)
                        model.release()
        logger.info("Unload models thread stopped.")

    def inference(self, model_name: str, inputs: List[str], normalize: bool = True):
        """
        推理
        :param model_name: 模型名称
        :param inputs: 输入数据列表
        :param normalize: 是否归一化
        :return: (标记数量，特征向量)
        """
        # 先尝试无锁读取
        with self.__lock:
            onnx_model = self.__loaded_models.get(model_name)

        if onnx_model is None:
            self.load_model(model_name)
            with self.__lock:
                onnx_model = self.__loaded_models.get(model_name)
                if onnx_model is None:
                    raise ValueError(f"Failed to load model {model_name}")

        return onnx_model.inference(inputs, normalize)

    async def inference_async(self, model_name: str, inputs: List[str],
                              normalize: bool = True):
        """
        异步推理，耗时操作运行在线程池，避免阻塞主协程
        :param self: 当前类对象
        :param model_name: 模型名称
        :param inputs: 输入数据列表
        :param normalize: 是否归一化
        :return: (标记数量，特征向量)
        """
        event_loop = asyncio.get_event_loop()
        return await event_loop.run_in_executor(self.__embedding_executor, self.inference, model_name,
                                                inputs, normalize)

    def stop(self):
        """
        停止
        """
        logger.info("Stopping model manager...")
        self.__stop_event.set()
        self.__embedding_executor.shutdown(wait=True)
        logger.info("Model manager stopped")


class OnnxModel:
    """
    ONNX 模型
    """

    def __init__(self, base_model_path: Path, tokenizer: PreTrainedTokenizerFast, ort_session: ort.InferenceSession,
                 model_config: Dict[str, Any]):
        self.__base_model_path: Path = base_model_path
        self.__tokenizer: PreTrainedTokenizerFast = tokenizer
        self.__ort_session: ort.InferenceSession = ort_session
        self.__model_config: Dict[str, Any] = model_config
        self.__last_access_time: float = time.time()

    def update_last_access_time(self):
        """
        更新最后访问时间
        """
        self.__last_access_time = time.time()

    def get_last_access_time(self):
        """
        获取最后访问时间
        :return: 最后访问时间（单位：秒）
        """
        return self.__last_access_time

    def release(self):
        """
        释放资源
        """
        try:
            if self.__ort_session:
                self.__ort_session.end_profiling()
        except Exception:
            pass
        self.__tokenizer = None
        self.__ort_session = None

    def inference(self, inputs: List[str], normalize: bool = True):
        """
        推理
        :param inputs: 输入数据列表
        :param normalize: 是否归一化
        :return: (标记数量，特征向量)
        """
        # 校验模型是否可用
        if self.__tokenizer is None or self.__ort_session is None:
            raise ValueError(f"Model has been released： {self.__base_model_path.name}")
        # 检查输入长度
        if len(inputs) > 16:
            raise ValueError("Input length must less than 16")

        # 更新最后访问时间
        self.__last_access_time = time.time()
        logger.debug(f"Inference model: {self.__base_model_path.name}, inputs: {inputs}, normalize: {normalize}")

        # 获取编码配置
        max_length = self.__model_config.get("max_length", 1024)
        logger.info(f"Model config, max_length: {max_length}")
        # 输入编码
        encode_inputs = self.__tokenizer(inputs, return_tensors="np", padding="longest", truncation=True,
                                         max_length=max_length)

        # 计算标记数量
        tokens = np.sum(encode_inputs["attention_mask"], axis=1).tolist()
        total_tokens = sum(tokens)
        inputs = {
            "input_ids": encode_inputs["input_ids"].astype(np.int64),
            "attention_mask": encode_inputs["attention_mask"].astype(np.int64),
        }

        # 动态处理嵌入名称
        input_names = [input.name for input in self.__ort_session.get_inputs()]
        if "token_type_ids" in input_names and "token_type_ids" in encode_inputs:
            inputs["token_type_ids"] = encode_inputs["token_type_ids"].astype(np.int64)

        outputs = self.__ort_session.run(None, inputs)
        logger.debug(f"Inference model: {self.__base_model_path.name}, outputs: {outputs}")
        embeddings = outputs[0]

        # 读取池化配置
        pooling_type = self.__model_config.get("pooling_type", "cls")
        logger.info(f"Pooling type: {pooling_type}")
        # 根据池化类型获取池化向量，默认池化类型为 cls
        if pooling_type == "mean":
            embeddings = OnnxModel.get_pooling_mean(encode_inputs["attention_mask"], embeddings)
        elif pooling_type == "last_token":
            embeddings = OnnxModel.get_pooling_last_token(encode_inputs["attention_mask"], embeddings)
        else:
            if pooling_type != "cls":
                logger.warning(f"Unknown pooling type: {pooling_type}, use cls pooling instead")
            embeddings = OnnxModel.get_pooling_cls(embeddings)

        # 归一化嵌入向量
        if normalize:
            embeddings = OnnxModel.normalize_embeddings(embeddings)
        return total_tokens, embeddings

    @staticmethod
    def get_pooling_cls(token_embeddings: np.ndarray[np.float32]):
        """
        获取 CLS 池化方式的嵌入向量
        :param token_embeddings: 标记嵌入向量
        :return: 池化向量
        """
        return token_embeddings[:, 0, :]

    @staticmethod
    def get_pooling_mean(attention_mask: np.ndarray[np.float32], token_embeddings: np.ndarray[np.float32]):
        """
        获取 MEAN 池化方式的嵌入向量
        :param attention_mask: 注意力向量
        :param token_embeddings: 标记嵌入向量
        :return: 池化向量
        """
        extend_mask = attention_mask.reshape(token_embeddings.shape[0], token_embeddings.shape[1], 1)
        # 将 hidden_states 中填充位置设为 0
        mask_hidden_state = token_embeddings * extend_mask
        # 计算有效位置的隐藏状态之和
        sum_hidden = np.sum(mask_hidden_state, axis=1)
        # 计算有效位置的数量
        sum_mask = np.sum(attention_mask, axis=1)
        # 处理 NaN 和 0
        sum_mask = np.maximum(sum_mask, 1e-9)
        return sum_hidden / sum_mask

    @staticmethod
    def get_pooling_last_token(attention_mask: np.ndarray[np.float32], token_embeddings: np.ndarray[np.float32]):
        """
        获取 LAST TOKEN 池化方式的嵌入向量
        :param attention_mask: 注意力向量
        :param token_embeddings: 标记嵌入向量
        :return: 池化向量
        """
        left_padding = (attention_mask[:, -1] == 1).sum() == attention_mask.shape[0]
        if left_padding:
            return token_embeddings[:, -1, :]
        else:
            valid_index = attention_mask.sum(axis=1) - 1
            batch_size = token_embeddings.shape[0]
            return token_embeddings[np.arange(batch_size), valid_index, :]

    @staticmethod
    def normalize_embeddings(embedding: np.ndarray[np.float32]):
        """
        归一化嵌入向量（欧氏距离，L2 范数）
        :param embedding: 嵌入向量
        :return: 归一化向量
        """
        return embedding / np.linalg.norm(embedding, ord=2, axis=1, keepdims=True)


# onnx 模型管理器
onnx_model_manager: OnnxModelManager | None = None
# 默认密钥
random_api_key = f"Bearer {str(uuid.uuid4())}"
API_KEY = os.getenv("API_KEY", random_api_key)
logger.info(f"API_KEY: {API_KEY}")


def init_app():
    """
    初始化应用
    :return:
    """
    # 输出可用的推理组件
    available_providers = ort.get_available_providers()
    logger.info(f"ONNX Runtime available providers: {available_providers}")

    # 模型存放路径
    base_path = Path(os.getenv("MODEL_PATH", "/ai-models"))
    if not base_path.exists():
        raise ValueError(f"MODEL_PATH({base_path}) dose not exist")

    # 嵌入模型最大线程数， 默认 2
    max_executors = int(os.getenv("MAX_EXECUTORS", "2"))
    if max_executors <= 1:
        raise ValueError(f"MAX_EXECUTORS({max_executors}) must greater than 1")

    # 最大闲置时间，默认 300 秒，超过该时间后，模型将被卸载
    max_idle_time = int(os.getenv("MAX_IDLE_TIME", "300"))
    if max_idle_time <= 60:
        raise ValueError(f"MAX_IDLE_TIME({max_idle_time}) must greater than 0")

    # 最大运行模型数量，默认 3
    max_active_models = int(os.getenv("MAX_ACTIVE_MODELS", "3"))
    if max_active_models <= 1:
        raise ValueError(f"MAX_ACTIVE_MODELS({max_idle_time}) must greater than 1")

    logger.info(
        f"Initializing app, base_path: {base_path}, max_executors: {max_executors}, max_idle_time: {max_idle_time}s, "
        f"max_active_models: {max_active_models}")
    # 初始化模型管理器
    global onnx_model_manager
    onnx_model_manager = OnnxModelManager(base_path, max_executors, max_active_models, max_idle_time)
    onnx_model_manager.init_available_model()
    # 启动卸载线程
    thread = threading.Thread(target=onnx_model_manager.unload_models, name="unload-thread", daemon=True)
    thread.start()


@asynccontextmanager
async def lifespan(app: FastAPI):
    init_app()
    yield
    if onnx_model_manager:
        onnx_model_manager.stop()


app = FastAPI(lifespan=lifespan)


@app.post("/v1/embeddings")
async def embedding_data(request: EmbeddingRequest, authorization: str = Header(None)):
    """
    嵌入接口
    :param request: 请求对象
    :param authorization 认证信息
    :return: 嵌入结果
    """

    logger.debug(f"Get text embeddings, authorization: {authorization}")

    # 鉴权
    if API_KEY != authorization:
        return {
            "code": 401,
            "msg": "Request not authorized"
        }

    # 获取参数
    model = request.model
    inputs = request.input
    normalize = request.normalize

    # 校验参数
    if len(model) == 0:
        return {
            "code": 500,
            "msg": "Field model is blank"
        }
    if len(inputs) == 0:
        return {
            "code": 500,
            "msg": "Field inputs is empty"
        }

    # 异步推理，异常处理
    try:
        total_tokens, embeddings = await onnx_model_manager.inference_async(model, inputs, normalize)
    except Exception as e:
        logger.error(f"Embedding error, model: {model}, inputs: {inputs}, error: {e}")
        return {
            "code": 500,
            "msg": f"{e}"
        }
    data = [{"object": "embedding", "embedding": embedding.tolist(), "index": i} for i, embedding in
            enumerate(embeddings)]
    id = str(uuid.uuid4())
    return {
        "id": id,
        "model": model,
        "object": "list",
        "data": data,
        "usage": {
            "prompt_tokens": total_tokens,
            "total_tokens": total_tokens
        }
    }


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=7999)
```

#### 发送请求

> 支持传输 `normalize` 参数决定是否归一化。

```bash
curl --request POST \
  --url http://localhost:7999/v1/embeddings \
  --header 'Authorization: Bearer 11abe39f-afcd-43e6-8f88-1827d3959a94' \
  --header 'Content-Type: application/json' \
  --data '{
    "model": "qwen3-embedding-0.6b",
    "input": [
        "What is the capital of China?",
        "Explain gravity"
    ]
}'
```

#### 接收响应

> `embedding` 对象代表特征向量，由于输出太多，这里省略。

```json
{
	"id": "b14a42d7-8bcb-4099-b56c-793d3a67f97e",
	"model": "qwen3-embedding-0.6b-q8",
	"object": "list",
	"data": [
		{
			"object": "embedding",
			"embedding": [],
			"index": 0
		},
        {
			"object": "embedding",
			"embedding": [],
			"index": 1
		}
    ],
	"usage": {
		"prompt_tokens": 85,
		"total_tokens": 85
	}
```

#### 特性

 - 查看源码进行相关参数的配置，支持工作线程，模型自动卸载等参数。
 - 支持自动扫描模型目录，加载可用模型。
 - 支持模型长时间不用时自动卸载，支持通过模型目录下的 `onnx_config.json` 文件配置模型**池化方式**和**最大序列长度**。


#### 示例模型配置

> 默认填充方向为 `left`，序列长度为 `1024`，池化方式为 `cls`，这里可以通过自定义的 `onnx_config.json` 配置文件覆盖默认值。

```json
{
	"padding_side": "left",
	"pooling_type": "last_token",
	"max_length": 1024
}
```

#### 模型目录结构

```
|__ ai-models
|  |__ qwen3-embedding-0.6b
|     |__ onnx_config.json（可选）
|     |__ tokenizer.json
|     |__ model.onnx
|     |__ model.onnx_data（可选）
```

### 常见问题

 1. 接口应用只是一个简单的示例服务，需要自行测试。
 2. 如有相关疑问，请在讨论区提出。
 3. 默认 `API KEY` 会在应用启动时输出至日志文件或控制台。