2.6 结构整理、增加了cache命中缓存
This commit is contained in:
parent
19f7d7b38b
commit
d4a3446236
@ -8,6 +8,7 @@ import time
|
||||
from dashscope import Assistants, Messages, Runs, Threads
|
||||
from llama_index.indices.managed.dashscope import DashScopeCloudRetriever
|
||||
|
||||
from flask_app.general.llm.doubao import read_txt_to_string
|
||||
from flask_app.general.llm.通义千问long_plus import qianwen_long, upload_file, qianwen_plus
|
||||
|
||||
prompt = """
|
||||
@ -157,7 +158,7 @@ def llm_call(question, knowledge_name,file_id, result_queue, ans_index, llm_type
|
||||
else:
|
||||
result_queue.put((ans_index, (question, qianwen_res)))
|
||||
elif llm_type==3:
|
||||
# print(f"doubao! question:{question}")
|
||||
# print(f"doubao! question:{question}") #暂时废弃doubao！
|
||||
# doubao_res=doubao_model(question,need_extra)
|
||||
qianwen_plus_res=qianwen_plus(question,need_extra)
|
||||
if not qianwen_plus_res:
|
||||
@ -178,56 +179,62 @@ def llm_call(question, knowledge_name,file_id, result_queue, ans_index, llm_type
|
||||
print(f"LLM 调用失败，查询索引 {ans_index}，错误：{e}")
|
||||
result_queue.put((ans_index, None)) # 使用 None 作为失败的占位符
|
||||
|
||||
def multi_threading(queries, knowledge_name="", file_id="", llm_type=1,need_extra=False):
|
||||
|
||||
def multi_threading(queries, knowledge_name="", file_id="", llm_type=1, need_extra=False):
|
||||
if not queries:
|
||||
return []
|
||||
|
||||
print("多线程提问：starting multi_threading...")
|
||||
result_queue = queue.Queue()
|
||||
|
||||
def submit_task(executor, query, idx):
|
||||
# 提交任务，并返回对应的 future
|
||||
return executor.submit(llm_call, query, knowledge_name, file_id, result_queue, idx, llm_type, need_extra)
|
||||
|
||||
future_to_index = {}
|
||||
with concurrent.futures.ThreadPoolExecutor(max_workers=40) as executor:
|
||||
future_to_index = {
|
||||
executor.submit(llm_call, query, knowledge_name, file_id, result_queue, index, llm_type,need_extra): index
|
||||
for index, query in enumerate(queries)
|
||||
}
|
||||
if llm_type == 4:
|
||||
# 每组10个任务，组与组之间间隔5秒,提高缓存命中率
|
||||
group_size = 5
|
||||
for i in range(0, len(queries), group_size):
|
||||
group = queries[i:i + group_size]
|
||||
for idx, query in enumerate(group, start=i):
|
||||
future = submit_task(executor, query, idx)
|
||||
future_to_index[future] = idx
|
||||
# 若还有后续组，则等待5秒
|
||||
if i + group_size < len(queries):
|
||||
time.sleep(5)
|
||||
else:
|
||||
# 直接一次性提交所有任务
|
||||
for index, query in enumerate(queries):
|
||||
future = submit_task(executor, query, index)
|
||||
future_to_index[future] = index
|
||||
|
||||
# 收集所有任务的异常
|
||||
for future in concurrent.futures.as_completed(future_to_index):
|
||||
index = future_to_index[future]
|
||||
idx = future_to_index[future]
|
||||
try:
|
||||
future.result() # 确保任务完成，如果有未处理的异常会在这里抛出
|
||||
future.result()
|
||||
except Exception as exc:
|
||||
print(f"查询索引 {index} 生成了一个异常：{exc}")
|
||||
result_queue.put((index, None)) # 使用 None 作为失败的占位符
|
||||
print(f"查询索引 {idx} 生成了一个异常：{exc}")
|
||||
result_queue.put((idx, None))
|
||||
|
||||
# 初始化结果列表，确保按查询的索引顺序排列
|
||||
# 根据任务索引初始化结果列表
|
||||
results = [None] * len(queries)
|
||||
while not result_queue.empty():
|
||||
index, result = result_queue.get()
|
||||
results[index] = result
|
||||
# 可选：过滤掉所有结果为 None 的项
|
||||
# 如果希望保留 None 以表示失败的查询，可以注释掉以下代码
|
||||
|
||||
# 如果所有结果都为 None，则返回空列表；否则过滤掉 None 项
|
||||
if all(result is None for result in results):
|
||||
return []
|
||||
results = [r for r in results if r is not None]
|
||||
return results
|
||||
return [r for r in results if r is not None]
|
||||
|
||||
if __name__ == "__main__":
|
||||
start_time=time.time()
|
||||
# # # 读取问题列表
|
||||
# baseinfo_file_path = '/flask_app/static/提示词/基本信息工程标.txt'
|
||||
# questions =read_questions_from_file(baseinfo_file_path)
|
||||
# knowledge_name = "招标解析5word"
|
||||
# llm_type=1
|
||||
# results = multi_threading(questions, knowledge_name)
|
||||
# end_time = time.time()
|
||||
# if not results:
|
||||
# print("errror!")
|
||||
# else:
|
||||
# print("elapsed time:"+str(end_time-start_time))
|
||||
# # 打印结果
|
||||
# for question, response in results:
|
||||
# print(f"Response: {response}")
|
||||
|
||||
file_path = r"C:\Users\Administrator\Desktop\fsdownload\39b0c3b4-1807-456c-8330-c5c7d1b7a2ca\ztbfile_procurement\ztbfile_procurement_1.pdf"
|
||||
file_id = upload_file(file_path)
|
||||
# file_path = r"C:\Users\Administrator\Desktop\fsdownload\39b0c3b4-1807-456c-8330-c5c7d1b7a2ca\ztbfile_procurement\ztbfile_procurement_1.pdf"
|
||||
# file_id = upload_file(file_path)
|
||||
# questions=["该招标文件的项目名称是？项目编号（或招标编号）是？采购人（或招标人）是？采购代理机构（或招标代理机构）是？请按json格式给我提供信息，键名分别是'项目名称','项目编号','采购人','采购代理机构'，若存在未知信息，在对应的键值中填'未知'。","该招标文件的项目概况是？项目基本情况是？请按json格式给我提供信息，键名分别为'项目概况','项目基本情况'，若存在嵌套信息，嵌套内容键名以文件中对应字段命名，而嵌套键值必须与原文保持一致，若存在未知信息，在对应的键值中填'未知'。"]
|
||||
# results=multi_threading(questions,"",file_id,2) #1代表使用百炼rag 2代表使用qianwen-long
|
||||
# if not results:
|
||||
@ -237,20 +244,26 @@ if __name__ == "__main__":
|
||||
# for question, response in results:
|
||||
# print(f"Question: {question}")
|
||||
# print(f"Response: {response}")
|
||||
|
||||
# ques=["关于'资格要求',本采购文件第一章第二款要求的内容是怎样的？请按json格式给我提供信息，键名为'资格要求'，而键值需要完全与原文保持一致，不要擅自总结、删减，如果存在未知信息，请在对应键值处填'未知'。"]
|
||||
# # ques=["该招标文件的工程名称（项目名称）是？招标编号是？招标人是？招标代理机构是？请按json格式给我提供信息，键名分别是'工程名称','招标编号','招标人','招标代理机构'，若存在未知信息，在对应的键值中填'未知'。","该招标文件的工程概况（或项目概况）是？招标范围是？请按json格式给我提供信息，键名分别为'工程概况','招标范围'，若存在嵌套信息，嵌套内容键名以文件中对应字段命名，若存在未知信息，在对应的键值中填'未知'。"]
|
||||
# results = multi_threading(ques, "6.2视频会议docx")
|
||||
# if not results:
|
||||
# print("errror!")
|
||||
# else:
|
||||
# # 打印结果
|
||||
# for question, response in results:
|
||||
# print(f"Question: {question}")
|
||||
# print(f"Response: {response}")
|
||||
#
|
||||
query=[]
|
||||
for i in range(1,50):
|
||||
query.append("请返回这个数字："+str(i))
|
||||
res=multi_threading(query,"",file_id,2)
|
||||
processed_filepath=r"C:\Users\Administrator\Desktop\货物标\extract_files\107国道.txt"
|
||||
full_text=read_txt_to_string(processed_filepath)
|
||||
temp="请告诉我LED 全彩显示屏的功能是怎样的，请以JSON格式返回，键名为'LED 全彩显示屏'，键值为字符串。"
|
||||
user_query = f"文本内容：{full_text}\n" + temp
|
||||
# user_query='''
|
||||
# </tr><tr>
|
||||
# <td colspan="1" rowspan="1">3</td>
|
||||
# <td colspan="1" rowspan="1">大屏播控系统</td>
|
||||
# <td colspan="1" rowspan="1">1、具有多用户多权限管理功能,支持多用户同时登录客户端,每个用户根据自身不同权限管理显示屏;2、系统对输入信号源进行预监视,实现在播控前预先查看的功能。</td>
|
||||
# <td colspan="1" rowspan="1">中国</td>
|
||||
# <td colspan="1" rowspan="1">1</td>
|
||||
# <td colspan="1" rowspan="1">套</td>
|
||||
# </tr><tr>
|
||||
# 请告诉我大屏播控系统的功能是怎样的，请以JSON格式返回，键名为'大屏播控系统'，键值为字符串。
|
||||
# '''
|
||||
for i in range(1,15):
|
||||
query.append(user_query)
|
||||
res=multi_threading(query,"","",4)
|
||||
for _,response in res:
|
||||
print(response)
|
||||
# end_time = time.time()
|
||||
|
||||
@ -279,6 +279,9 @@ def qianwen_plus(user_query, need_extra=False):
|
||||
|
||||
"""
|
||||
使用之前上传的文件，根据用户查询生成响应，并实时显示流式输出。
|
||||
目前命中缓存局限：1.不足 256 Token 的内容不会被缓存。
|
||||
2.上下文缓存的命中概率并不是100%，即使是上下文完全一致的请求，也存在无法命中的概率，命中概率依据系统判断而定。
|
||||
3.若多线程同时请求，缓存无法及时更新！
|
||||
参数：
|
||||
- user_query: 用户查询
|
||||
- need_extra: 是否需要返回额外数据（默认 False）
|
||||
@ -326,7 +329,9 @@ def qianwen_plus(user_query, need_extra=False):
|
||||
usage = chunk_data.get('usage')
|
||||
if usage is not None:
|
||||
completion_tokens = usage.get('completion_tokens', 0)
|
||||
|
||||
# prompt_tokens_details = usage.get('prompt_tokens_details', {}) #命中tokens ，取消注释可以print
|
||||
# cache_hit = prompt_tokens_details.get('cached_tokens', 0)
|
||||
# print("命中："+str(cache_hit))
|
||||
# 处理 choices 信息
|
||||
choices = chunk_data.get('choices', [])
|
||||
if choices:
|
||||
|
||||
@ -30,7 +30,7 @@
|
||||
}
|
||||
}
|
||||
|
||||
6.请从提供的招标文件中提取与“信息公示媒介”相关的信息（如补充说明、文件澄清、评标结果等）公示媒介（如网址、官网）。按以下 JSON 格式输出信息，其中键名为“信息公示媒介”，键值为一个字符串列表。如果存在多个信息公示媒介，请分别将原文中相关表述逐字添加至字符串中。注意，若只有一个信息公示媒介，字符串列表中只包含一个字符串。示例输出格式如下，仅供参考：
|
||||
6.请从提供的招标文件中提取与“信息公示媒介”相关的信息，即补充说明、文件澄清、评标结果等信息的公示媒介（如网址、官网）。按以下 JSON 格式输出信息，其中键名为“信息公示媒介”，键值为一个字符串列表。如果存在多个信息公示媒介，请分别将原文中相关表述逐字添加至字符串中。注意，若只有一个信息公示媒介，字符串列表中只包含一个字符串。示例输出格式如下，仅供参考：
|
||||
{
|
||||
"信息公示媒介":["招标公告在政府采购网（www.test.gov.cn）发布。","中标结果将在采购网(www.test.bid.cn)予以公告。"]
|
||||
}
|
||||
|
||||
@ -30,7 +30,7 @@
|
||||
}
|
||||
}
|
||||
|
||||
6.请从提供的招标文件中提取与“信息公示媒介”相关的信息（如补充说明、文件澄清、评标结果等）公示媒介（如网址、官网）。按以下 JSON 格式输出信息，其中键名为“信息公示媒介”，键值为一个字符串列表。如果存在多个信息公示媒介，请分别将原文中相关表述逐字添加至字符串中。注意，若只有一个信息公示媒介，字符串列表中只包含一个字符串。示例输出格式如下，仅供参考：
|
||||
6.请从提供的招标文件中提取与“信息公示媒介”相关的信息，即补充说明、文件澄清、评标结果等信息的公示媒介（如网址、官网）。按以下 JSON 格式输出信息，其中键名为“信息公示媒介”，键值为一个字符串列表。如果存在多个信息公示媒介，请分别将原文中相关表述逐字添加至字符串中。注意，若只有一个信息公示媒介，字符串列表中只包含一个字符串。示例输出格式如下，仅供参考：
|
||||
{
|
||||
"信息公示媒介":["招标公告在政府采购网（www.test.gov.cn）发布。","中标结果将在采购网(www.test.bid.cn)予以公告。"]
|
||||
}
|
||||
|
||||
@ -646,7 +646,8 @@ def get_technical_requirements(invalid_path, processed_filepath, model_type=1):
|
||||
new_query=f"文件内容：{full_text}\n" + new_query
|
||||
queries.append(new_query)
|
||||
if model_type == 1:
|
||||
results = multi_threading(queries, "", "", 3, True) # 豆包
|
||||
# results = multi_threading(queries, "", "", 3, True) # 豆包
|
||||
results = multi_threading(queries, "", "", 4, True) # plus
|
||||
else:
|
||||
results = multi_threading(queries, "", file_id, 2, True) # qianwen-long
|
||||
temp_final = {}
|
||||
|
||||
@ -35,11 +35,11 @@ def fetch_procurement_reqs(procurement_path, invalid_path):
|
||||
busi_model_type=3 #long-stream
|
||||
processed_filepath = ""
|
||||
else:
|
||||
tech_model_type= 1 #doubao
|
||||
tech_model_type= 1 #doubao或者qianwen-plus
|
||||
busi_model_type =4 #qianwen-plus
|
||||
processed_filepath = convert_file_to_markdown(procurement_path,"extract3.txt") # invalid_path->markdown格式
|
||||
else:
|
||||
tech_model_type = 1 #doubao
|
||||
tech_model_type = 1 #doubao或者qianwen-plus
|
||||
busi_model_type = 4
|
||||
processed_filepath = convert_file_to_markdown(procurement_path) # 正常情况：procurement_path->markdown格式
|
||||
# processed_filepath = pdf2txt(procurement_path) # 纯文本提取
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user