当前位置：首页 > news >正文

基于 RAG 的三级工单智能分类系统：从自然语言到工单分类的完整落地方案

news 2026/6/2 4:19:56

在政务热线（如“12345”）和企业客服系统中，用户描述往往是自然语言的 —— 模糊、不规范、信息冗余。如何把用户一句“帮我查下未成年人相关的问题”，精准匹配到结构化的工单分类体系？

传统做法通常依赖关键词规则，但是当用户输入和数据库里面设定的关键词不一致时候，检索结果即为0。所以要利用大模型的理解能力，将用户的问题类型关键字尽可能映射到已有的类型字段里面。

本文构建一个完整可运行的RAG（Retrieval-Augmented Generation）工单智能分类系统。
系统基于：

SiliconFlow：BAAI/bge-large-zh-v1.5 向量模型
DeepSeek Chat：关键词抽取
PostgreSQL 存储工单分类主数据
FastAPI 提供服务接口
本地 classify_embeddings.json 作为向量缓存

一、系统整体架构

┌──────────────┐ │ 用户输入内容 │ └───────┬──────┘ ▼ ┌──────────────┐ │ 关键词抽取API │（deepseek-v3-250324） └───────┬──────┘ ▼ ┌──────────────┐ │ 构造查询向量 │ BGE 中文大模型 └───────┬──────┘ ▼ ┌──────────────┐ │ 分类向量缓存 │ classify_embeddings.json └───────┬──────┘ ▼ ┌──────────────┐ │ 余弦相似度匹配│ └───────┬──────┘ ▼ ┌──────────────┐ │ 返回匹配结果 │ └──────────────┘

二、分类向量库构建（预处理）

要想让模型理解层级语义，我们将其组合成：

民生保障 -> 基本社会服务 -> 留守儿童关爱保护

并用bge-large-zh-v1.5生成向量

缓存到：classify_embeddings.json

""" embedding_service.py ------------------------------------- 用于从 PostgreSQL 中加载工单分类字典表， 并调用 Silicon Flow BAAI Embedding 模型生成每个分类的向量， 最终生成一个 classify_embeddings.json 缓存文件。 """ import os import json import numpy as np import pandas as pd import requests from db import get_connection # 数据库连接 # ========================== # 配置 # ========================== CACHE_FILE = "classify_embeddings.json" EMBEDDING_MODEL = "BAAI/bge-large-zh-v1.5" API_KEY = "sk- wemlrqfnhhuvcjjmbpeuagzseatlitnrzdzmvcruvty" API_URL = "https://api.siliconflow.cn/v1/embeddings" EMBEDDING_DIM = 1024 # ========================== # 生成 embedding # ========================== def get_embedding(text: str) -> list[float]: """调用 Silicon Flow BAAI Embedding API 获取文本向量""" headers = { "Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json" } data = { "model": EMBEDDING_MODEL, "input": text } response = requests.post(API_URL, json=data, headers=headers) response.raise_for_status() # 返回 embedding return response.json()["data"][0]["embedding"] # ========================== # 构造层级路径 # ========================== def build_hierarchy(df: pd.DataFrame, type_code: str) -> str: """从当前分类向上递归找到完整路径，如 '公共安全 -> 社会治安 -> 黄赌毒黑'""" path = [] current = df.loc[df["typeCode"] == type_code] while not current.empty: type_name = current.iloc[0]["typeName"] path.insert(0, type_name) parent_code = current.iloc[0]["parentTypeCode"] # 没有父类则停止 if not parent_code or parent_code == "0": break current = df.loc[df["typeCode"] == parent_code] return " -> ".join(path) # ========================== # 生成并缓存分类向量 # ========================== def generate_classify_embeddings(): """读取分类表并生成每条分类的 embedding""" print("✅ 开始读取数据库分类表...") conn = get_connection() sql = ''' SELECT "typeCode", "typeName", "parentTypeCode", "level" FROM system.dic_work_order_classify; ''' df = pd.read_sql(sql, conn) conn.close() print(f"✅ 成功读取 {len(df)} 条分类数据") embeddings = [] for _, row in df.iterrows(): path = build_hierarchy(df, row["typeCode"]) print(f"🔹 生成向量: {path}") try: vec = get_embedding(path) except Exception as e: print(f"❌ 生成向量失败: {path}, 错误: {e}") continue embeddings.append({ "typeCode": row["typeCode"], "typeName": row["typeName"], "hierarchy": path, "embedding": vec }) with open(CACHE_FILE, "w", encoding="utf-8") as f: json.dump(embeddings, f, ensure_ascii=False, indent=2) print(f"✅ 向量生成完成，共 {len(embeddings)} 条，缓存文件已保存：{CACHE_FILE}") return embeddings # ========================== # 加载或生成缓存 # ========================== def load_or_create_embeddings(): """加载缓存文件，如果不存在则重新生成""" if os.path.exists(CACHE_FILE): print("⚙️ 检测到缓存文件，正在加载分类向量...") with open(CACHE_FILE, "r", encoding="utf-8") as f: return json.load(f) else: print("⚙️ 未检测到缓存文件，开始生成分类向量...") return generate_classify_embeddings() # ========================== # 相似度计算 # ========================== def cosine_similarity(vec1, vec2): """计算余弦相似度""" v1 = np.array(vec1) v2 = np.array(vec2) return np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2))

三、关键词提取模块（DeepSeek）

""" keyword_extractor_api.py ========================= 关键词提取模块 功能： 1. 调用固定的 Apifox 接口（deepseek-v3-250324 模型）提取中文关键词 2. 输入：自然语言文本 3. 输出：关键词列表 4. 可选终端测试（仅在本地调试使用） 注意： - URL、API Key 和模型写死在代码中 - FastAPI 调用时直接调用 extract_keywords_api(text) """ import requests # ========================= # 固定 API 配置 # ========================= API_URL = "https://ark.cn-beijing.volces.com/api/v3/chat/completions" API_KEY = "9b6b8bd0- -4e53-ae63-91f89dbbddb8" MODEL = "deepseek-v3-250324" def extract_keywords_api(text: str) -> list[str]: """ 调用 Apifox 接口提取关键词 输入： text: str, 自然语言文本 输出： keywords: list[str], 提取的中文关键词列表 """ headers = { "Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json" } payload = { "model": MODEL, "messages": [ { "role": "user", "content": f"请从以下文本中提取关键词，返回中文关键字列表，逗号分隔：{text}" } ], "max_tokens": 200 } response = requests.post(API_URL, headers=headers, json=payload) response.raise_for_status() content = response.json()['choices'][0]['message']['content'] keywords = [k.strip() for k in content.replace("，", ",").split(",") if k.strip()] return keywords # 可选终端测试 if __name__ == "__main__": text = input("请输入需要提取关键词的文本：").strip() keywords = extract_keywords_api(text) print("原始输入：", text) print("提取关键词：", keywords)

四、查询向量构造与匹配

""" router.py ========================= FastAPI 路由模块 功能： 1. 接收用户自然语言工单描述 2. 调用 keyword_extractor_api 提取关键词 3. 将原始文本 + 关键词生成查询向量 4. 与缓存的分类向量进行相似度匹配 5. 调用大模型进行语义检测 6. 返回 JSON，包括： - 原始输入 - 提取的关键词 - 匹配结果（包含 typeCode、typeName、hierarchy、similarity） - 大模型语义检测结果（predicted_category, reason） """ from fastapi import APIRouter from pydantic import BaseModel import numpy as np from embedding_service import load_or_create_embeddings, get_embedding from keyword_extractor_api import extract_keywords_api # 写死 URL/Key #from semantic_detector import detect_semantic # 新增大模型检测模块 # ========================= # 定义路由 # ========================= router = APIRouter(prefix="/api", tags=["工单分类"]) # ========================= # 请求体模型 # ========================= class WorkOrderRequest(BaseModel): description: str # 用户自然语言工单描述 top_n: int = 5 # 返回前 N 条匹配 threshold: float = 0.5 # 相似度阈值 # ========================= # 余弦相似度计算 # ========================= def cosine_similarity(a: np.ndarray, b: np.ndarray) -> float: """ 计算余弦相似度 """ return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))) # ========================= # 主接口：匹配多条工单分类 # ========================= @router.post("/match") def match_workorder(req: WorkOrderRequest): """ 输入自然语言描述： 1. 调用 keyword_extractor_api 提取关键词 2. 构造查询向量（原始描述 + 关键词） 3. 加载分类向量并计算相似度 4. 调用大模型进行语义检测 5. 返回 JSON，包括 input, keywords, matches, semantic_detection """ # 1️⃣ 提取关键词 keywords = extract_keywords_api(req.description) if not keywords: return { "input": req.description, "keywords": [], "matches": [], "semantic_detection": {}, "message": "未提取到关键词" } # 2️⃣ 构造查询文本并生成查询向量 query_text = f"{req.description}，关键词：{','.join(keywords)}" query_embedding = get_embedding(query_text) # 使用 BAAI/bge-large-zh-v1.5 # 3️⃣ 加载分类向量 classify_embeddings = load_or_create_embeddings() # 4️⃣ 计算相似度并筛选 results = [] for item in classify_embeddings: score = cosine_similarity(np.array(item["embedding"]), query_embedding) if score >= req.threshold: results.append({ "typeCode": item["typeCode"], "typeName": item["typeName"], "hierarchy": item["hierarchy"], "similarity": round(score, 4) }) # 按相似度排序 results.sort(key=lambda x: x["similarity"], reverse=True) # 5️⃣ 调用大模型进行语义检测 #semantic_result = detect_semantic(req.description) # 6️⃣ 返回给 Apifox return { "input": req.description, "keywords": keywords, "matches": results[:req.top_n], #"semantic_detection": semantic_result }

五、FastAPI 完整服务

""" run.py ========================= 主启动文件 功能： 1. 启动 FastAPI 服务 2. 系统启动时预加载分类向量缓存 3. 注册工单分类路由 """ from fastapi import FastAPI import uvicorn from router import router from embedding_service import load_or_create_embeddings # 创建 FastAPI 实例 app = FastAPI(title="RAG 三级工单分类系统（rag_workorder_hierarchy）") # =================================================== # 启动事件：加载向量缓存 # =================================================== @app.on_event("startup") def startup_event(): """ 系统启动时预加载分类向量缓存 """ load_or_create_embeddings() # 注册路由 app.include_router(router) # =================================================== # 程序入口 # =================================================== if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000)

核心接口

/api/match：