1 背景

我们在处理word，pdf文档过程中，经常需要提取关键的信息，比如我们在阅读一个招标文件时，就需要精准的提取标段的行业类型、项目类型和服务业务类型。而这些类型是我们系统内已经定义好了的。也就是说，将招标文件里的业务类型、行业类型、项目类型映射到我们系统的类型。
由于招标文件是用自然语言描述的，它不是一个结构化的数据。如果我们使用传统的方法比如正则表达式匹配检索的方法来提取类型，那么效果很差，基本上是匹配不到的。本文描述了一种“双路找回与加权打分”算法，利用本地嵌入模型，结合LLM大模型兜底的方法，精确提取分类并映射到系统类型。

2 数据准备

假设我们系统内已有的分类：

• 业务分类：biz_types_tree.json
  示例数据：

[
  {
    "id": 1001,
    "parent_id": 0,
    "type_name": "工程類",
    "children": [
      {
        "id": 2001,
        "parent_id": 1001,
        "type_name": "EPC總承包",
        "parent_name": "工程類",
        "full_path_name": "工程類-EPC總承包"
      },
      {
        "id": 2002,
        "parent_id": 1001,
        "type_name": "施工",
        "parent_name": "工程類",
        "full_path_name": "工程類-施工"
      }
    ]
  }
]

• 行业及项目分类
  示例数据

[
  {
    "id": 1,
    "parent_id": 0,
    "type_name": "房屋建築工程",
    "full_path_name": "房屋建築工程",
    "industry_id": 1,
    "industry_name": "房建",
    "parent_name": "",
    "children": [
      {
        "id": 100,
        "parent_id": 1,
        "type_name": "居住建築",
        "full_path_name": "房屋建築工程-居住建築",
        "industry_id": 1,
        "industry_name": "房建",
        "parent_name": "房屋建築工程",
        "children": [
          {
            "id": 101,
            "parent_id": 100,
            "type_name": "住宅",
            "full_path_name": "房屋建築工程-居住建築-住宅",
            "industry_id": 1,
            "industry_name": "房建",
            "parent_name": "居住建築",
            "children": [
              {
                "id": 10101,
                "parent_id": 101,
                "type_name": "普通住宅",
                "full_path_name": "房屋建築工程-居住建築-住宅-普通住宅",
                "industry_id": 1,
                "industry_name": "房建",
                "parent_name": "住宅"
              }
            ]
          }
        ]
      },
      {
        "id": 200,
        "parent_id": 1,
        "type_name": "公共建築",
        "full_path_name": "房屋建築工程-公共建築",
        "industry_id": 1,
        "industry_name": "房建",
        "parent_name": "房屋建築工程",
        "children": [
          {
            "id": 201,
            "parent_id": 200,
            "type_name": "醫療建築",
            "full_path_name": "房屋建築工程-公共建築-醫療建築",
            "industry_id": 1,
            "industry_name": "房建",
            "parent_name": "公共建築",
            "children": [
              {
                "id": 20101,
                "parent_id": 201,
                "type_name": "醫院",
                "full_path_name": "房屋建築工程-公共建築-醫療建築-醫院",
                "industry_id": 1,
                "industry_name": "房建",
                "parent_name": "醫療建築"
              }
            ]
          }
        ]
      }
    ]
  }
]

3 依赖库安装

在项目根目录下创建requirements.txt，内容如下

flask
sentence-transformers
scikit-learn
jieba
numpy
requests
PyJWT

4 准备key

4.1 准备大模型api key

准备一个大模型平台的apikey，如千问、deepseek

4.1 生成secret

secret用于flask服务接口的认证

openssl rand -hex 32

将生成的字符串保存起来，备用

5 服务代码

5.1 环境变量

在项目根目录下创建.env文件，内容如下

LLM_API_KEY=sk-42*******0a7****
LLM_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions
LLM_MODEL_NAME=qwen3-max
SERVER_PORT=50**
JWT_SECRET_KEY=11822d3******************aca4b3b5e4****

将4.1获取的大模型api key填入LLM_*, 这里以千问为例。
将4.1生成的secret填入JWT_SECRET_KEY。

5.2 app服务代码

app.py

from flask import Flask, request, jsonify
from sentence_transformers import SentenceTransformer, util
from functools import wraps
import numpy as np
import jieba
import json
import os
import requests
import jwt

app = Flask(__name__)

# ==================== 配置區域 ====================
# JWT 配置
# ⚠️ 生產環境下請通過環境變量注入此密鑰，確保 app.py 與 generate_token.py 使用相同密鑰！
JWT_SECRET_KEY = os.getenv("JWT_SECRET_KEY", "your_super_secret_key_change_me")

# 本地模型配置
EMBEDDING_MODEL_NAME = 'shibing624/text2vec-base-chinese'
CONFIDENCE_THRESHOLD = 0.65  # 置信度閾值

# LLM API 配置
LLM_API_KEY = os.getenv("LLM_API_KEY", "your_llm_provider_api_key")
LLM_BASE_URL = os.getenv("LLM_BASE_URL", "your_llm_provider_url")
LLM_MODEL_NAME = os.getenv("LLM_MODEL_NAME", "your_llm_provider_model_name")
# ==================================================
# 服務端口
SERVER_PORT = int(os.getenv("SERVER_PORT", "5080"))

# ==================== 1. 服務初始化（一次性加載） ====================
print("正在載入本地中文語意模型...")
MODEL = SentenceTransformer(EMBEDDING_MODEL_NAME)

def flatten_nodes(nodes):
    flat_list = []
    for node in nodes:
        if "full_path_name" in node and node["full_path_name"]:
            flat_list.append(node)
        if "children" in node and isinstance(node["children"], list):
            flat_list.extend(flatten_nodes(node["children"]))
    return flat_list

print("正在加載並扁平化本地 JSON 分類標準...")
with open("biz_types_tree.json", "r", encoding="utf-8") as f:
    FLAT_BIZ = flatten_nodes(json.load(f))
with open("project_types_tree.json", "r", encoding="utf-8") as f:
    FLAT_IND = flatten_nodes(json.load(f))

print("正在預計算分類路徑的 Embedding 向量...")
BIZ_PATHS = [item["full_path_name"] for item in FLAT_BIZ]
IND_PATHS = [item["full_path_name"] for item in FLAT_IND]

BIZ_EMBEDDINGS = MODEL.encode(BIZ_PATHS, convert_to_tensor=True)
IND_EMBEDDINGS = MODEL.encode(IND_PATHS, convert_to_tensor=True)
print("👉 系統初始化完畢！隨時準備響應請求。")

# ==================== 2. 本地規則引擎干預表 ====================
BIZ_PRIORITY_WORDS = {
    "施工": {"target_ids": [2002], "weight": 0.20},
    "EPC": {"target_ids": [2001], "weight": 0.25},
    "總承包": {"target_ids": [2001], "weight": 0.25},
    "新建": {"target_ids": [2002], "weight": 0.15},
    "擴建": {"target_ids": [2002], "weight": 0.15},
    "監理": {"target_ids": [2003], "weight": 0.20},
}

# ==================== 3. JWT 身份驗證裝飾器 ====================
def token_required(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        token = None
        auth_header = request.headers.get('Authorization')
        if auth_header:
            parts = auth_header.split()
            if len(parts) == 2 and parts[0].lower() == 'bearer':
                token = parts[1]

        if not token:
            return jsonify({"error": "Missing token. Authorization header must be 'Bearer <TOKEN>'"}), 401

        try:
            # 驗證由本地腳本生成的永久 Token
            data = jwt.decode(token, JWT_SECRET_KEY, algorithms=["HS256"])
            current_user = data.get('user')
        except jwt.InvalidTokenError:
            return jsonify({"error": "Invalid token"}), 401

        return f(current_user, *args, **kwargs)
    return decorated

# ==================== 4. 大模型 Fallback 決策函數 ====================
def call_llm_decision(project_desc, candidate_biz, candidate_ind):
    if not LLM_API_KEY or LLM_API_KEY == "your_deepseek_api_key_here":
        print("⚠️ 未檢測到有效的 LLM_API_KEY，跳過大模型決策，直接返回本地最高分結果。")
        return None

    prompt = f"""
你是一個極其專業的工程項目分類標籤專家。請閱讀下方的【項目描述】，並從【候選業務類型】和【候選行業及項目類型】中，挑選出唯一最脗合的分類。

【項目描述】
{project_desc}

【候選業務類型】
{json.dumps(candidate_biz, ensure_ascii=False, indent=2)}

【候選行業及項目類型】
{json.dumps(candidate_ind, ensure_ascii=False, indent=2)}

【任務要求】
1. 仔細分析項目描述，判斷其實際的工程性質（如：包含大面積新建大樓，應歸為“施工”而非單純“修繕”）。
2. 從給出的候選列表中，選擇一組最合適的分類，並提取其對應的 ID 和名稱。
3. 必須嚴格按照以下 JSON 格式輸出，不要包含任何 Markdown 標記（如 ```json）、不要有任何額外的解釋文字：
{{
  "biz_id": 選擇的業務類型id,
  "biz_name": "選擇的業務類型名稱",
  "industry_id": 選擇的行業類型id,
  "industry_name": "選擇的行業類型名稱",
  "project_id": 選擇的項目類型id,
  "project_name": "選擇的項目類型名稱"
}}
"""

    headers = {
        "Authorization": f"Bearer {LLM_API_KEY}",
        "Content-Type": "application/json"
    }
    payload = {
        "model": LLM_MODEL_NAME,
        "messages": [{"role": "user", "content": prompt}],
        "temperature": 0.1,
        "response_format": {"type": "json_object"}
    }

    try:
        response = requests.post(LLM_BASE_URL, headers=headers, json=payload, timeout=15)
        response_json = response.json()
        llm_output = response_json['choices'][0]['message']['content'].strip()
        if llm_output.startswith("```"):
            llm_output = llm_output.split("\n", 1)[1].rsplit("\n", 1)[0].strip()
        return json.loads(llm_output)
    except Exception as e:
        print(f"❌ 調用大模型決策失敗: {e}")
        return None

# ==================== 5. 受保護的分類路由接口 ====================
@app.route('/api/classify', methods=['POST'])
@token_required  # 認證裝飾器
def classify_project(current_user):
    data = request.json or {}
    project_desc = data.get("description", "")
    project_name = data.get("name", "")
    force_llm = data.get("force_llm", False)

    if not project_desc:
        return jsonify({"error": "description is required"}), 400

    combined_text = f"{project_name} {project_name} {project_desc}"
    segmented_text = " ".join(jieba.cut(combined_text))
    
    desc_embedding = MODEL.encode(segmented_text, convert_to_tensor=True)
    
    biz_scores = util.cos_sim(desc_embedding, BIZ_EMBEDDINGS)[0].cpu().numpy()
    ind_scores = util.cos_sim(desc_embedding, IND_EMBEDDINGS)[0].cpu().numpy()
    
    for word, config in BIZ_PRIORITY_WORDS.items():
        if word in combined_text:
            for idx, biz_item in enumerate(FLAT_BIZ):
                if biz_item["id"] in config["target_ids"]:
                    biz_scores[idx] += config["weight"]

    best_biz_idx = np.argmax(biz_scores)
    best_ind_idx = np.argmax(ind_scores)
    
    biz_confidence = float(biz_scores[best_biz_idx])
    ind_confidence = float(ind_scores[best_ind_idx])

    top5_biz_indices = np.argsort(biz_scores)[-5:][::-1]
    top5_ind_indices = np.argsort(ind_scores)[-5:][::-1]
    
    candidate_biz = [FLAT_BIZ[idx] for idx in top5_biz_indices]
    candidate_ind = [FLAT_IND[idx] for idx in top5_ind_indices]

    matched_biz = FLAT_BIZ[best_biz_idx]
    matched_ind = FLAT_IND[best_ind_idx]

    is_low_confidence = (biz_confidence < CONFIDENCE_THRESHOLD) or (ind_confidence < CONFIDENCE_THRESHOLD)
    
    if force_llm or is_low_confidence:
        path_used = "L2_LLM_Fallback"
        print(f"⚠️ 正在啟動大模型精準決策 (調用用戶: {current_user})...")
        
        for item in candidate_biz: item.pop("children", None)
        for item in candidate_ind: item.pop("children", None)
        
        llm_result = call_llm_decision(combined_text, candidate_biz, candidate_ind)
        if llm_result:
            llm_result["path_used"] = path_used
            llm_result["local_confidence"] = {"biz": biz_confidence, "industry": ind_confidence}
            return jsonify(llm_result)
        else:
            print("LLM 決策失敗，自動降級採用本地最高分結果。")
            path_used = "L1_Local_Fallback_Due_To_LLM_Error"
    else:
        path_used = "L1_Local_FastPath"

    response_data = {
        "biz_id": matched_biz["id"],
        "biz_name": matched_biz["type_name"],
        "industry_id": matched_ind.get("industry_id", None),
        "industry_name": matched_ind.get("industry_name", ""),
        "project_id": matched_ind["id"],
        "project_name": matched_ind["type_name"],
        "path_used": path_used,
        "local_confidence": {
            "biz": biz_confidence,
            "industry": ind_confidence
        }
    }
    
    return jsonify(response_data)

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=SERVER_PORT, debug=False)

6 运行

pip install -r requirements.txt
python app.py

第一次运行，将下载模型到本地，需要一定的时间

但很多时候模型下载失败，解决本办法是:

export HF_ENDPOINT=https://hf-mirror.com
python app.py

如果还是失败，那么可以离线下载嵌入式模型，地址https://huggingface.co/shibing624/text2vec-base-chinese/tree/main, 下载这个目录下的所有文件和子文件夹的文件，命名文件夹为text2vec-base-chinese。打包后上传到服务器，解压到app.py所在目录，并修改代码：

# 本地模型配置
EMBEDDING_MODEL_NAME = './text2vec-base-chinese'

这样修改后运行，app就不会从网上下载模型了，直接加载本地下载好的文件。

7 测试

curl --location --request POST 'http://172.16.*.*:5080/api/classify' \
--header 'Authorization: Bearer eyJhbGciOiJIUzI1*********************t5UTZ-YVxl2X-gw' \
--header 'Content-Type: application/json' \
--data-raw '{
    "name": "警察学院长湖校区3号学员公寓楼改造工程（重） 施工招标公告",
    "description": "警察学院长湖校区3号学员公寓楼改造工程（重） 施工招标公告"
}'

结果：

{
    "biz_id": 2002,
    "biz_name": "施工",
    "industry_id": 1,
    "industry_name": "房建",
    "local_confidence": {
        "biz": 0.6451624631881714,
        "industry": 0.5711911916732788
    },
    "path_used": "L2_LLM_Fallback",
    "project_id": 20203,
    "project_name": "高等院校建筑"
}