基于高阶RAG的本地知识库问答机器人

from langchain_community.document_loaders.csv_loader import CSVLoader
from langchain_community.document_loaders import PyPDFLoader
# 读取csv，返回list
def load_csv(path):
    # 每条记录为一个元素
    loader = CSVLoader(
        file_path="project_1/data/clean.csv",
        encoding='utf-8' # 编码
    )
    data = loader.load()
    return data 
# 读取pdf，返回list
def load_pdf(path):
    # 是以每页为一个元素的
    loader = PyPDFLoader("project_1/data/1.pdf")
    pages = loader.load_and_split()
    return pages

from langchain_text_splitters import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(    chunk_size=200, # 指定每个文本块的目标大小，这里设置为200个字符。    chunk_overlap=50, # 指定文本块之间的重叠字符数，这里设置为50个字符。    length_function=len, # 用于测量文本长度的函数，这里使用Python内置的`len`函数。    is_separator_regex=False, # 指定`separators`中的分隔符是否应被视为正则表达式，这里设置为False，表示分隔符是字面字符。    separators=["\n\n",  "\n",   " ",    ".",    ",",     "，",  "。", ] # 定义用于分割文本的分隔符列表。)
pages = load_pdf("project_1/data/1.pdf")
texts = text_splitter.split_documents([pages[0].page_content])

from langchain.embeddings import HuggingFaceBgeEmbeddingsmodel_name = "project_2/bge-large-zh-v1.5"model_kwargs = {'device': 'cuda'}# # 当向量都被规范化（归一化）后，它们的范数都是1。# 余弦相似度的计算只需要向量的点积运算，从而减少了计算的复杂度，加快了处理速度。hf = HuggingFaceBgeEmbeddings(    encode_kwargs = {'normalize_embeddings': True}     model_name=model_name,    model_kwargs=model_kwargs,    encode_kwargs=encode_kwargs,    query_instruction="为这个句子生成表示以用于检索相关文章：")
embedding = hf.embed_query("你好！")len(embedding)

# 快速创建数据库from langchain_chroma import Chromadb = Chroma.from_documents(   documents = texts,    embedding = hf,   ids = None,   collection_name = 'test1',   collection_metadata = {"hnsw:space": "cosine"},   persist_directory = 'project_1/chroma_db'   )
# 相似度方法通过查询文本检索数据，会自动调用向量模型query = "网络首发是什么"docs = db.similarity_search(query)print(docs[0].page_content)

# 创建一个检索器retriever = vectorstore.as_retriever(search_kwargs={"k": 10})
query = "萧炎是谁?"
# Get relevant documents ordered by relevance scoredocs = retriever.invoke(query)

# 重排        tokenizer = AutoTokenizer.from_pretrained(self.config.rerank_model)        model = AutoModelForSequenceClassification.from_pretrained(self.config.rerank_model)        model.eval()        pairs = [[new_query, doc.page_content] for doc in docs]        with torch.no_grad():            inputs = tokenizer(pairs, padding=True, truncation=True, return_tensors='pt', max_length=512)            scores = model(**inputs, return_dict=True).logits.view(-1, ).float()

# 实例化自定义模型llm = Qwen(mode_name_or_path = "/root/autodl-tmp/Qwen1.5-7B-Chat")
prompt = ChatPromptTemplate.from_template(PROMPT_TEMPLATE)
output_parser = StrOutputParser()
# 构建 chainchain = prompt | llm | output_parser
res = chain.invoke(                    {                    "context": format_docs(reordered_docs),                    "question": "小医仙是谁？"                    }                    )
print(res)

prompt = ChatPromptTemplate.from_template(PROMPT_TEMPLATE)
        chat_llm_chain = LLMChain(            llm=self.llm,            prompt=prompt,            verbose=True,        )
        # 生成回复         res = chat_llm_chain.predict(                                # 格式化聊天记录为JSON字符串                                chat_history = "\n".join([f"{entry['role']}: {entry['content']}" for entry in self.memory]),                                # chat_history = "\n".join(self.memory),                                context=context,                                question=query                                )                # res = chat_llm_chain.invoke({"context": context,"question": query})        logger.info(f"模型回复：{res}")
        # 更新聊天历史        self.memory.append({'role': 'user', 'content': query})        self.memory.append({'role': 'assistant', 'content': res})        print(self.memory)
        return res