大模型Rag-指令調度

本文主要記錄根據用戶問題指令，基于大模型做Rag，匹配最相關描述集進行指令調度，可用于匹配后端接口以及展示答案及圖表等。

1.指令查詢處理邏輯

1.實現思路

指令識別：主要根據用戶的問題q計算與指令描述集is = [i0, ... , im]和指令關鍵詞集ks = [k0, ... , km]之間的匹配距離sim，選取最匹配的指令,top1/top5
用戶指令與指令描述集embedding向量匹配相似度
關鍵詞處理：提取指令中的關鍵參數據，用戶的問題q計算與指令提示詞集ps = [p0, ... , pm]，從用戶問題中抽取指令執行需要的參數集

2.指令查詢實現參數介紹

參數示例

{ attribute_ids: "[{"paramName":"項目名稱","paramCode":"prejectname","paramPrompt":"請提取出項目名稱","paramWeight":"0.4"}]", uni_key: "001", candidate_value: "科技園工人數量", name: "查詢科技園工人數量", description: "查詢科技園工人數量", tip_word: "", id: 2, keyword: "科技園，工人，數量", instruction_id: 2 }

參數代表含義

"attribute_ids": 屬性值

"uni_key": 參數編碼示例:projectname

"candidate_value": 候選值

"name": "查詢科技園工人數量"

"description": "查詢科技園工人數量"

"tip_word": ""，參數不全提示詞

"id": 2,

"keyword": 關鍵詞,

"instruction_id": 2?

3.指令查詢實現細節

這一部分主要對指令描述集和指令關鍵詞集的處理，針對參數為id,description,keyword

指令識別，主要根據用戶的問題q計算與指令描述集is = [i0, ... , im]和指令關鍵詞集ks = [k0, ... , km]之間的匹配距離sim

存儲在在self.INSTTS中，記錄id，指令描述集description,指令關鍵詞集keyword,示例如下

# 提取內容
ists = jsonobj["data"]
for ist in ists:cid = ist["id"]cdes = ist["description"]ckws = ist["keyword"]self.INSTTS.append((cid, cdes, ckws))# self.INSTTS值
self.INSTTS = [(2, '查詢科技園工人數量', '科技園,工人，數量'), (4, '查詢3月份營業額', 營業額'), (5, '打開百度首頁', '打開,百度'), (6, '查詢某項目某地區某段時間的營業額', '查詢，項目，地區，營業額'),  (7, '請幫我打開智慧工地指揮中心', '打開，智慧工地，指揮中心'), (8, '打開數據標注', '打開,數據標注'), (9, '測試表格指令', '輸出,測試表格'), (10, '查詢七日氣溫', '查詢，氣溫'), (12, '測試折線圖', '測試折線圖'), (13, '測試餅圖', '測試餅圖'), (14, '測試柱狀圖', '測試柱狀圖')]

指令描述集向量化,記錄id2des,id2kws

self.id2key：記錄編號和id，用于后續查詢

self.id2des: 記錄id和des指令描述，用于向量化

self.id2kws: 記錄id和kws關鍵詞，用于計算用戶query和關鍵詞相似度

num0 = 0
for record in self.INSTTS:_id = record[0]_des = record[1]_kws = record[2]d_embeddings = self.model.encode([self.instruction + _des], normalize_embeddings=True)self.index.add(d_embeddings)self.i2key[num0] = _idself.id2des[_id] = _desself.id2kws[_id] = _kwsnum0 += 1

4.屬性查詢實現細節

這一部分主要針對指令，對應參數為tip_word:參數不全提示詞,attribute_ids：屬性值的處理，其中attribute_ids的參數示例如下

attribute_ids: "[{"paramName":"項目名稱","paramCode":"prejectname","paramPrompt":"請提取出項目名稱","paramWeight":"0.4"}]",

參數代表意思解析

attribute_ids: "[{ "paramName":"項目名稱", "paramCode":"prejectname", "paramPrompt":"請提取出項目名稱", # 需要提取的內容 "paramWeight":"0.4"}]" # 參數權重

參數的存儲，記錄TIPSINFOS，cvalue，PARAMINFOS

self.TIPSINFOS：記錄id和tip_word，后續用于根據id提取對應參數

cvalue: 記錄paramName, paramCode, paramPrompt, paramWeight

self.PARAMINFOS：記錄id和cvalue

jsonobj = json.loads(response.text)
params = jsonobj["data"]
for param in params:print(param)cid = param["id"]if "tip_word" in param:tip_word = param["tip_word"]self.TIPSINFOS[cid] = tip_wordcvalue = []if "attribute_ids" in param and len(param["attribute_ids"])>0:attrs = json.loads(param["attribute_ids"])for attr in attrs:paramName = attr["paramName"]paramCode = attr["paramCode"]paramPrompt = attr["paramPrompt"]paramWeight = attr["paramWeight"]if len(paramName)>0 and len(paramCode)>0 and len(paramPrompt)>0 and len(paramWeight)>0 :cvalue.append((paramName, paramCode, paramPrompt, paramWeight))print(cid, paramName, paramCode, paramPrompt, paramWeight)self.PARAMINFOS[cid] = cvalue

2.用戶指令與指令描述集處理

1.計算用戶指令與指令描述集相似度，選取topk

整體代碼

question:用戶指令問題

k:topk選取

theata:相似度閾值

def find_vec2(self, question, k, theata):print(question)q_embeddings = self.model.encode([self.instruction + question], normalize_embeddings=True)# print(q_embeddings)D, I = self.index.search(q_embeddings, k)  # v為待檢索向量，返回的I為每個待檢索query最相似TopK的索引list，D為其對應的距離result = []for i in range(k):sort_k = i + 1sim_i = I[0][i]uuid = self.i2key.get(sim_i, "none")sim_v = D[0][i]midkws = self.id2kws.get(uuid, "none")score = self.calculate_score(sim_v, question, midkws)  # 計算得分if score < theata:doc = {}doc["score"] = scoredoc["action_id"] = uuidresult.append(doc)return result

uuid = self.i2key.get(sim_i, "none")：根據編號查詢id

midkws = self.id2kws.get(uuid, "none"): 根據id查詢kws關鍵詞

進一步計算相似度

score = self.calculate_score(sim_v, question, midkws) ：根據sim_v向量得分,question用戶指令問題,midkws，kws關鍵詞進一步計算匹配度得分

def calculate_score(self, score, question, kws):if kws == "none" or kws == "":return int(score * 1000)else:ckws = kws.split(",")chit = 0for kw in ckws:if question.find(kw) > -1:      # 檢測問題中是否包含關鍵詞chit += 1# print(score, chit, question, kws)return int(score * 1000) - 100 * chit

2.加載指令候選參數

如果命中指令，提取指令候選參數

self.TIPSINFOS：參數不全提示詞

self.PARAMINFOS：參數屬性信息

optstr = self.TIPSINFOS.get(action_id, "指令參數不全。")   # 通過action_id獲取參數不全提示詞
cur_parms = self.PARAMINFOS.get(action_id, [])  # 通過action_id獲取屬性信息

3.依據大模型判斷是否參數不全

沒有參數不全提取任務

如果cur_parms的長度為0，證明沒有判斷參數不全的任務，直接返回結果

self.SUCCESS_SIGN：參數提前標志，默認為1，代表成功(沒有參數提取任務也設置為1)

self.EMPTY_PARMS: 缺失參數字符，默認為空字符串

self.MISS_VALUE：缺失字符，默認為空字符串

if len(cur_parms) == 0:return self.SUCCESS_SIGN, self.EMPTY_PARMS, self.MISS_VALUE

有參數不全提取任務

datestr = time.strftime("%Y%m%d", time.localtime())
parmnames = []
for cp in cur_parms:parmnames.append(cp[0])
parmnames_str = "、".join(parmnames)
newquestion1 = "問題：'" + "我想查看" + question + "'\n" + \"假設今天是" + datestr + "，請根據以上問題，提取"+parmnames_str+"，要求：\n"
for n1 in range(len(cur_parms)):record = str(n1+1) + "、" + cur_parms[n1][2] + "；\n"newquestion1 = newquestion1 + recordnewquestion1 = newquestion1 + "請以JSON格式輸出，要求：\n"
for n1 in range(len(cur_parms)):record = str(n1+1) + "、" + cur_parms[n1][0] + "用" + cur_parms[n1][1] + "表示；\n"newquestion1 = newquestion1 + recordprint(newquestion1)rststr = self.excute_prompt(str(newquestion1)) # 訪問星火大模型

封裝提示詞給大模型

示例 newquestion1值

這個示例中，問題為我想查看打開百度，經過屬性信息將要提取的參數封裝到prompt提示詞中

"問題：'我想查看打開百度' 假設今天是20250417，請根據以上問題，提取項目名稱，要求： 1、請提取出項目名稱；請以JSON格式輸出，要求： 1、項目名稱用projectname表示； "

訪問大模型之后的rststr值

'```json

{

"projectname": "查看打開百度"

}

```'

參數不全處理邏輯

默認設置參數缺失

這里cur_parms=[('項目名稱', 'projectname', '請提取出項目名稱', '0.4')]，

默認設置miss_value中每一個參數都是缺失的

keywords_value = []
for cp in cur_parms:keywords_value.append(cp[1])
miss_value = {key: '1' for key in keywords_value}

4.參數不全處理

這里使用的星火大模型，模型返回的結果中，如果參數不全，會出現"由于問題中"的提示字符，這里通過它進行判斷，如果要提取的參數key在大模型返回的結果中，則設置miss_value[key] = '0'，表示該參數key不缺失，例如miss_value={'projectname': '0'}代表projectname不缺失。

if "由于問題中" in rststr:# rst = {key: '' for key in keywords_value}rststr = self.excute_prompt(str(newquestion1))if "由于問題中" in rststr:rststr = self.excute_prompt(str(newquestion1))if "由于問題中" in rststr:return self.PARAMTER_PARTIAL, optstr, miss_valueelse:for key in keywords_value:if key in rststr:miss_value[key] = '0'else:for key in keywords_value:if key in rststr:miss_value[key] = '0'
else:for key in keywords_value:if key in rststr:miss_value[key] = '0'
# 打印結果
print(miss_value)

5.解析大模型返回的參數

sindex = rststr.find("{")
eindex = rststr.find("}")
print(sindex, eindex)
if sindex == -1 and eindex == -1:return self.PARAMTER_PARTIAL, optstr, miss_valuerststrjson = rststr[sindex:eindex + 1]
print(rststrjson)
rststrjson = rststrjson.replace("'", "\"")
print(rststrjson)rst = {}
rstjson = json.loads(rststrjson)
n = len(cur_parms)
# rst = json.dumps(rst)

6.計算相似度

根據大模型返回參數和配置的參數計算相似度

for n1 in range(len(cur_parms)):midcode = cur_parms[n1][1]midtheta = 0.4try:midtheta = float(cur_parms[n1][3])except:passprint(midcode, midtheta)if midcode in rstjson:midvalue = rstjson[midcode]if midtheta > 0.0:msim, mval = self.search.match_parameter(action_id, midcode, midvalue, theata=midtheta)if msim > 0:rst[midcode] = mvalelse:  # 查詢的項目名稱與參數完全不同print("msim is 0.0.")rst[midcode] = ''miss_value[midcode] = '1'n-=1continue# return self.FAILED_SIGN, json.dumps(rst), miss_valueelse:print("midtheta is 0.0.")rst[midcode] = midvalueelse:print(midcode, " not in ", rstjson)rst["optstr"] = optstrreturn self.FAILED_SIGN, json.dumps(rst), miss_value
print(rst)
if n != len(cur_parms):rst["optstr"] = optstrreturn self.FAILED_SIGN, json.dumps(rst), miss_value
return self.SUCCESS_SIGN, json.dumps(rst), miss_value

參數相似度計算，找到最相似的參數

def match_parameter(self, action_id, uni_key, parm, theata=0.4):midkey = str(action_id) + "_" + uni_key# 存儲參數parm中的元素cwds = {}for i in range(len(parm)):cwds[parm[i]] = ""rstsim = 0.0    # 初始化相似度結果為0.0rstsss = ""     # 初始化最匹配的候選值為空字符串"""# 遍歷 self.PARAMS 字典中以 midkey 為鍵的所有候選值及其拆分后的字典表示# sss 表示候選值，wds 表示候選值拆分后的字典"""for sss, wds in self.PARAMS[midkey].items():# print(sss, wds)csim = self.sim_parameter(cwds, wds)    # 計算輸入的cwds 與 庫里的 wds相似度# 如果相似度csim大于預先設定的theata閾值，并且相似度csim大于之前設定的最大相似度rstsimif csim >= theata and csim > rstsim:# 更新最大相似度為當前相似度rstsim = csim# 更新最匹配的候選值為當前候選值rstsss = sss# 打印匹配結果，包括輸入參數、相似度（保留三位小數）、最匹配的候選值和閾值print("match_parameter --- ", parm, round(rstsim, 3), rstsss, theata)# 返回最大相似度和最匹配的候選值return rstsim, rstsss

計算交并比

def sim_parameter(self, parm1, parm2):union_ws = {}join_ws = {}for w1 in parm1.keys():union_ws[w1] = ""if w1 in parm2:join_ws[w1] = ""for w2 in parm2.keys():union_ws[w2] = ""return 1.0 * len(join_ws.keys()) / len(union_ws.keys())