当前样例实现基于已有知识库进行一次完整的多轮检索问答
完成“对接指南“页面的注册账号、实名认证、AK/SK 密钥获取和签名获取，完成知识库的创建及文档上传后，可调用以下脚本实现知识库的一次完整检索问答
详细 API 接口说明参考: search_knowledge（新）chat_completions（新）
import json import requests from volcengine.auth.SignerV4 import SignerV4 from volcengine.base.Request import Request from volcengine.Credentials import Credentials collection_name = "Vehicle_Kashiwa_01" #知识库名称 project_name = "default" query = "这篇论文是关于什么的" #本轮问题 ak = "your ak" sk = "your sk" account_id = "your account_id" g_knowledge_base_domain = "api-knowledgebase.mlp.cn-beijing.volces.com" base_prompt = """# 任务 你是一位在线客服，你的首要任务是通过巧妙的话术回复用户的问题，你需要根据「参考资料」来回答接下来的「用户问题」，这些信息在 <context></context> XML tags 之内，你需要根据参考资料给出准确，简洁的回答。 你的回答要满足以下要求： 1. 回答内容必须在参考资料范围内，尽可能简洁地回答问题，不能做任何参考资料以外的扩展解释。 2. 回答中需要根据客户问题和参考资料保持与客户的友好沟通。 3. 如果参考资料不能帮助你回答用户问题，告知客户无法回答该问题，并引导客户提供更加详细的信息。 4. 为了保密需要，委婉地拒绝回答有关参考资料的文档名称或文档作者等问题。 # 任务执行 现在请你根据提供的参考资料，遵循限制来回答用户的问题，你的回答需要准确和完整。 # 参考资料 <context> {} </context> """ def prepare_request(method, path, params=None, data=None, doseq=0): if params: for key in params: if ( isinstance(params[key], int) or isinstance(params[key], float) or isinstance(params[key], bool) ): params[key] = str(params[key]) elif isinstance(params[key], list): if not doseq: params[key] = ",".join(params[key]) r = Request() r.set_shema("http") r.set_method(method) r.set_connection_timeout(10) r.set_socket_timeout(10) headers = { "Accept": "application/json", "Content-Type": "application/json; charset=utf-8", "Host": g_knowledge_base_domain, "V-Account-Id": account_id, } r.set_headers(headers) if params: r.set_query(params) r.set_host(g_knowledge_base_domain) r.set_path(path) if data is not None: r.set_body(json.dumps(data)) # 生成签名 credentials = Credentials(ak, sk, "air", "cn-north-1") SignerV4.sign(r, credentials) return r def search_knowledge(): method = "POST" path = "/api/knowledge/collection/search_knowledge" #知识库检索 request_params = { "project": "default", "name": "Vehicle_Kashiwa_01", "query": "这篇论文是关于什么的", "limit": 10, "pre_processing": { "need_instruction": True, #是否需要拼接 instruction，非对称问题建议开启 "rewrite": False, #多轮改写开关，仅根据 messages 信息对本轮问题进行改写 "return_token_usage": True, "messages": [ { "role": "system", "content": "" }, { "role": "user", "content": "这篇论文是关于什么的" } ] }, "dense_weight": 0.5, #调节语义检索权重，1 代表纯语义检索，仅在向量化模型选择语义+关键词模型时生效 "post_processing": { "get_attachment_link": True, #是否返回原始图片，仅当创建知识库开启 OCR 时生效，否则自动跳过图片 "chunk_group": True, #是否对召回切片按照文档进行聚合 "rerank_only_chunk": False, #是否对召回切片按照文档语序进行重排 "rerank_switch": False, #是否开启重排 "chunk_diffusion_count": 0 #是否召回切片的临近切片，如 1 代表额外召回当前切片的上下各一个切片 } } info_req = prepare_request(method=method, path=path, data=request_params) rsp = requests.request( method=info_req.method, url="http://{}{}".format(g_knowledge_base_domain, info_req.path), headers=info_req.headers, data=info_req.body ) # print("search res = {}".format(rsp.text)) return rsp.text def chat_completion(message, stream=False, return_token_usage=True, temperature=0.7, max_tokens=4096): method = "POST" path = "/api/knowledge/chat/completions" #大模型生成 request_params = { "messages": message, #通过 messages 字段传入 prompt，召回切片，历史多轮对话 "stream": True, #是否流式返回 "return_token_usage": True, "model": "Doubao-pro-32k", #默认走系统公共接入点，如需配置方舟上创建的私有接入点，可在此传入 ep_id，且额外指定 api_key 参数，详细说明见 chat_completions 接口 "max_tokens": 4096, "temperature": 0.7, "model_version": "240828" } info_req = prepare_request(method=method, path=path, data=request_params) rsp = requests.request( method=info_req.method, url="http://{}{}".format(g_knowledge_base_domain, info_req.path), headers=info_req.headers, data=info_req.body ) print("chat completion res = {}".format(rsp.text)) def generate_prompt(rsp_txt): # system prompt 拼接 rsp = json.loads(rsp_txt) if rsp["code"] != 0: return prompt = "" rsp_data = rsp["data"] points = rsp_data["result_list"] for point in points: # 先拼接系统字段 doc_info = point["doc_info"] for system_field in ["doc_name","title","chunk_title","content"] : if system_field == 'doc_name' or system_field == 'title': if system_field in doc_info: prompt += f"{system_field}: {doc_info[system_field]}\n" else: if system_field in point: if system_field == "content" and doc_info["doc_type"] == "faq.xlsx": question_field = "original_question" prompt += f"content: 当询问到相似问题时，请参考对应答案进行回答：问题:“{point[question_field]}”。答案:“{point[system_field]}”\n" #faq 最佳实践 else: prompt += f"{system_field}: {point[system_field]}\n" if "table_chunk_fields" in point: table_chunk_fields = point["table_chunk_fields"] for self_field in [] : # 使用 next() 从 table_chunk_fields 中找到第一个符合条件的项目 find_one = next((item for item in table_chunk_fields if item["field_name"] == self_field), None) if find_one: prompt += f"{self_field}: {find_one['field_value']}\n" prompt += "---\n" return base_prompt.format(prompt) def search_knowledge_and_chat_completion(): # 1.执行search_knowledge rsp_txt = search_knowledge() # 2.生成prompt prompt = generate_prompt(rsp_txt) # todo:用户需要本地缓存对话信息，并按照顺序依次加入到 messages 中 # 3.拼接 message 对话, 问题对应 role 为 user，系统对应 role 为 system, 答案对应 role 为 assistant, 内容对应 content messages =[ { "role": "system", "content": prompt #大模型指令 + 检索结果 }, { "role": "user", "content": "2024 年最新的 RAG 论文都有什么？" #多轮历史对话 }, { "role": "assistant", "content": "2024 年最新的 RAG 论文有：《XX》,《xx》" #多轮历史对话 }, { "role": "user", "content": query } ] # 4.调用chat_completion chat_completion(messages) if __name__ == "__main__": search_knowledge_and_chat_completion()