句子推理模型使用

bge-small-zh-v1.5 是北京智源研究院（BAAI）推出的轻量级中文文本嵌入模型，属于 FlagEmbedding 系列 1.5 版本，以 BERT 为基础架构，主打轻量化、高性能与合理相似度分布，适合资源受限场景下的中文语义检索、文本分类等任务。

以下是一个简单的使用脚本。

主程序

import warnings
import torch
import os
import json
import numpy as np
import time
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import uvicorn
from sentence_transformers import SentenceTransformer



# ---------------------- 1. 全局配置 ----------------------
warnings.filterwarnings("ignore")
os.environ["CUDA_VISIBLE_DEVICES"] = ""
torch.cuda.is_available = lambda: False
torch.set_num_threads(4)

# 模型路径配置
MODEL_PATH = "local_model/bge-small-zh-v1.5"

# 静态JSON配置文件路径
CONFIG_PATH = "config/command_list.json"

# 全局缓存（配置+指令库向量）
CONFIG_CACHE = {}
CMD_EMBEDDINGS_CACHE = None


# ---------------------- 2. 工具函数：加载/更新配置 ----------------------
def load_config():
    """从静态JSON文件加载配置"""
    global CONFIG_CACHE, CMD_EMBEDDINGS_CACHE
    try:
        # 从静态JSON文件读取配置
        print("📄 从静态JSON文件加载配置...")
        
        # 读取配置文件
        with open(CONFIG_PATH, "r", encoding="utf-8") as f:
            CONFIG_CACHE = json.load(f)

        # 验证配置格式
        if not isinstance(CONFIG_CACHE.get("command_list"), list) or len(CONFIG_CACHE["command_list"]) == 0:
            raise ValueError("command_list必须是非空列表")
        if not isinstance(CONFIG_CACHE.get("threshold"), (int, float)):
            raise ValueError("threshold必须是数字")
        
        command_list = CONFIG_CACHE["command_list"]
        threshold = CONFIG_CACHE["threshold"]
        
        # 预编码指令库，更新向量缓存
        CMD_EMBEDDINGS_CACHE = model.encode(
            command_list,
            convert_to_numpy=True,
            normalize_embeddings=True
        )
        print(f"✅ 配置加载成功：指令库{len(command_list)}条，阈值{threshold}")
        return CONFIG_CACHE
    except Exception as e:
        raise RuntimeError(f"加载配置失败：{e}")


def save_config(new_config):
    """保存新配置到静态JSON文件"""
    global CONFIG_CACHE, CMD_EMBEDDINGS_CACHE
    
    try:
        # 保存到静态JSON文件
        print("📄 保存配置到静态JSON文件...")
        
        # 验证新配置格式
        if not isinstance(new_config.get("command_list"), list) or len(new_config["command_list"]) == 0:
            raise ValueError("command_list必须是非空列表")
        if not isinstance(new_config.get("threshold"), (int, float)) or new_config["threshold"] < 0 or new_config["threshold"] > 1:
            raise ValueError("threshold必须是0-1之间的数字")

        # 写入配置文件
        with open(CONFIG_PATH, "w", encoding="utf-8") as f:
            json.dump(new_config, f, ensure_ascii=False, indent=2)

        # 重新加载配置（更新缓存）
        load_config()
        return True
    except Exception as e:
        raise ValueError(f"保存配置失败：{e}")


# ---------------------- 3. 加载模型 + 初始化配置 ----------------------
print("🔍 加载模型...")
try:
    model = SentenceTransformer(MODEL_PATH, device="cpu")
    # 初始化加载配置
    load_config()
except Exception as e:
    raise RuntimeError(f"服务启动失败：{e}")

# ---------------------- 4. FastAPI初始化 ----------------------
app = FastAPI(title="船舶指令推理API", version="1.0")


# ---------------------- 5. 数据模型定义 ----------------------
# 更新配置请求体
class UpdateConfigRequest(BaseModel):
    command_list: list[str]  # 新的指令库
    threshold: float = 0.75  # 新的匹配阈值


# 推理请求体（仅需ASR文本）
class InferRequest(BaseModel):
    asr_text: str  # ASR语音转文字输入


# ---------------------- 6. 接口1：查询当前配置（新增） ----------------------
@app.get("/get_config", summary="查询当前配置（指令库/阈值）")
async def get_config():
    try:
        # 加载最新配置
        config = load_config()

        return {
            "code": 200,
            "msg": "查询成功",
            "data": {
                "command_list": config.get("command_list", []),
                "threshold": config.get("threshold", 0.75),
                "update_time": int(time.time())  # 使用当前时间作为更新时间
            }
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"查询配置失败：{e}")


# ---------------------- 7. 接口2：更新配置 ----------------------
@app.post("/update_config", summary="更新配置（静态JSON数据源）")
async def update_config(request: UpdateConfigRequest):
    try:
        # 构造新配置
        new_config = {
            "command_list": request.command_list,
            "threshold": request.threshold
        }
        # 保存配置到静态JSON文件
        save_config(new_config)
        return {
            "code": 200,
            "msg": "配置更新成功",
            "data": new_config
        }
    except ValueError as e:
        raise HTTPException(status_code=400, detail=f"参数错误：{e}")
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"配置更新失败：{e}")


# ---------------------- 8. 接口3：推理（仅需ASR文本） ----------------------
@app.post("/infer", summary="船舶指令推理（根据配置读取数据源）")
async def infer(request: InferRequest):
    try:
        start_time = time.time()

        # 检查配置缓存，如果为空则重新加载
        if not CONFIG_CACHE or CMD_EMBEDDINGS_CACHE is None:
            load_config()

        if not CONFIG_CACHE or CMD_EMBEDDINGS_CACHE is None:
            raise RuntimeError("配置加载失败，请检查数据库连接和配置")

        # 编码ASR输入
        asr_embedding = model.encode(
            [request.asr_text],
            convert_to_numpy=True,
            normalize_embeddings=True
        )

        # 计算相似度匹配
        similarity = np.dot(asr_embedding, CMD_EMBEDDINGS_CACHE.T)[0]
        top_idx = np.argmax(similarity)
        matched_command = CONFIG_CACHE["command_list"][top_idx]
        similarity_score = round(float(similarity[top_idx]), 3)
        is_execute = similarity_score >= CONFIG_CACHE["threshold"]

        # 计算耗时
        infer_time = round((time.time() - start_time) * 1000, 2)

        return {
            "code": 200,
            "msg": "推理成功",
            "data": {
                "asr_text": request.asr_text, # 传入的ASR 语音转文字输入文本
                "matched_command": matched_command, # 系统从指令库中匹配到的最相似船舶指令
                "similarity_score": similarity_score, # 输入文本与匹配指令的语义相似度得分
                "threshold": CONFIG_CACHE["threshold"], # 推理执行的相似度阈值（静态配置值）
                "is_execute": is_execute, # 是否执行匹配到的船舶指令（核心业务判断）
                "infer_time_ms": infer_time # 本次推理的耗时（毫秒）
            }
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"推理失败：{e}")


# ---------------------- 9. 健康检查接口 ----------------------
@app.get("/health", summary="服务健康检查")
async def health_check():
    return {
        "code": 200,
        "msg": "服务运行中",
        "data": {
            "model_loaded": True if 'model' in locals() else False,
            "config_loaded": True if CONFIG_CACHE else False,
            "command_count": len(CONFIG_CACHE.get("command_list", [])),
            "threshold": CONFIG_CACHE.get("threshold"),
            "cpu_only": True,
            "timestamp": int(time.time())
        }
    }


# ---------------------- 10. 启动服务 ----------------------
if __name__ == "__main__":
    uvicorn.run(
        app,
        host="0.0.0.0",
        port=9529,
        workers=1,
        log_level="error"
    )

配置文件

{
  "command_list": [
    "开启1号阀门",
    "开启2号阀门",
    "开启3号阀门",
    "关闭1号阀门",
    "关闭2号阀门",
    "关闭3号阀门"
  ],
  "threshold": 0.8 #权重
}

构建docker镜像

docker build ship-api:v1.0.0

运行docker镜像

docker run -d -p 9529:9529 -v /www/dk_project/dk_app/config:/app/config --name ship-inference-api ship-api:v1.0.0