LLM 大语言模型实战 (十)-LangChain RAG 语义检索及 BM25Retriever 精确筛选

2024-09-28 23:15:22 ⋅ 920 ⋅ 1 ⋅ 0

LangChain 六大模块

LLM
prompt
chains
retrieval
memory
agents

chain 有两种输入： input (prompt) 或者 dict；

dict 更好用，因为可以有名字

安装依赖

!pip install openai
!pip install langchain_community

basic RAG

from operator import itemgetter
from dotenv import dotenv_values
from langchain.prompts import ChatPromptTemplate
from langchain_community.chat_models import ChatOpenAI
# from langchain_community.embeddings import OpenAIEmbeddings
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain.retrievers import BM25Retriever, EnsembleRetriever
from langchain.document_loaders import TextLoader
ENV_CONFIG = dotenv_values("../.env")
# ENV_CONFIG

vectorstore = FAISS.load_local("../data/db_index", 
    OpenAIEmbeddings(api_key=ENV_CONFIG.get("API_KEY"), 
                               base_url=ENV_CONFIG.get("BASE_URL")),
                               allow_dangerous_deserialization=True
)
retriever = vectorstore.as_retriever()
vectorstore.similarity_search("rag的本质是什么", k=2)

BM25Retriever

使用BM25Retriever的主要优势在于它可以在初步筛选阶段提供更高的精确度，尤其是在处理短文本或查询时。BM25算法基于词频和逆文档频率（TF-IDF），适合处理信息检索中的经典查询需求，可以有效捕捉到文档中的重要关键词。而Embedding方法则通过语义相似性进行匹配，适合处理更复杂的上下文。

结合这两种方法，可以利用BM25的精确性和Embedding的语义理解能力，形成一个强大的检索系统。通过EnsembleRetriever，可以更好地平衡不同类型查询的表现，最终提升检索效果。这样，用户在检索时能获得更全面、更相关的结果。

# BM25 在业界用的非常多，个人认为没有办法被向量召回完全替代的，至少可以作为补充

text_loader = TextLoader("../data/text.txt")
text_docs = text_loader.load()

bm25_retriever = BM25Retriever.from_documents(documents=text_docs, k=1)
bm25_retriever
BM25Retriever(vectorizer=<rank_bm25.BM25Okapi object at 0x7fd140fc7ee0>, k=1)
emb_retriever = vectorstore.as_retriever(search_kwargs={"k": 1})
emb_retriever
VectorStoreRetriever(tags=['FAISS', 'OpenAIEmbeddings'], vectorstore=<langchain_community.vectorstores.faiss.FAISS object at 0x7fd14107edf0>, search_kwargs={'k': 1})
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, emb_retriever], weights=[0.5, 0.5]
)

# {"question"： "rag的本质是什么？", "context": "doc1, doc2"}
template = """Answer the question based only on the following context:
{context}

Question: {question}
""" 
prompt = ChatPromptTemplate.from_template(template)

# template = """Answer the question based only on the following context:
# {context}

# Question: {question}
# """
# prompt = ChatPromptTemplate.from_template(template)

model = ChatOpenAI(
    api_key=ENV_CONFIG.get("API_KEY"), base_url=ENV_CONFIG.get("BASE_URL")
)
/var/folders/nn/zmjw6gy57jg40k_tswcx0mpc0000gp/T/ipykernel_22920/3683481803.py:8: LangChainDeprecationWarning: The class `ChatOpenAI` was deprecated in LangChain 0.0.10 and will be removed in 1.0. An updated version of the class exists in the :class:`~langchain-openai package and should be used instead. To use it run `pip install -U :class:`~langchain-openai` and import as `from :class:`~langchain_openai import ChatOpenAI``.
  model = ChatOpenAI(
#  chain也是一个函数， 函数就是一个 DAG
# 解读 Chain； 
chain = (
    {"context": ensemble_retriever, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)
print(({"context": ensemble_retriever, "question": RunnablePassthrough()}
    | prompt).invoke("Rag的本质是什么").messages[0].content)
Answer the question based only on the following context:
[Document(metadata={'source': '../data/text.txt'}, page_content='RAG的本质是让模型获取正确的Context(上下文)，利用ICL (In Context Learning)的能力，输出正确的响应。它综合利用了固化在模型权重中的参数化知识和存在外部存储中的非参数化知识(知识库、数据库等)。RAG分为两阶段：使用编码模型（如 BM25、DPR、ColBERT 等）根据问题找到相关的文档。生成阶段：以找到的上下文作为基础，系统生成文本。\n\n\n通用语言模型通过微调就可以完成几类常见任务，比如分析情绪和识别命名实体。这些任务不需要额外的背景知识就可以完成。\n\n\n要完成更复杂和知识密集型的任务，可以基于语言模型构建一个系统，访问外部知识源来做到。这样的实现与事实更加一性，生成的答案更可靠，还有助于缓解“幻觉”问题。\n\n\nRAG 会接受输入并检索出一组相关/支撑的文档，并给出文档的来源（例如维基百科）。这些文档作为上下文和输入的原始提示词组合，送给文本生成器得到最终的输出。这样 RAG 更加适应事实会随时间变化的情况。这非常有用，因为 LLM 的参数化知识是静态的。RAG 让语言模型不用重新训练就能够获取最新的信息，基于检索生成产生可靠的输出。\n\nLewis 等人（2021）提出一个通用的 RAG 微调方法。这种方法使用预训练的 seq2seq 作为参数记忆，用维基百科的密集向量索引作为非参数记忆（使通过神经网络预训练的检索器访问）。\n\n最近，基于检索器的方法越来越流行，经常与 ChatGPT 等流行 LLM 结合使用来提高其能力和事实一致性。'), Document(page_content='RAG的本质是让模型获取正确的Context(上下文)，利用ICL (In Context Learning)的能力，输出正确的响应。它综合利用了固化在模型权重中的参数化知识和存在外部存储中的非参数化知识(知识库、数据库等)。RAG分为两阶段：使用编码模型（如 BM25、DPR、ColBERT 等）根据问题找到相关的文档。生成阶段：以找到的上下文作为基础，系统生成文本。\n\n\n通用语言模型通过微调就可以完成几类常见任务，比如分析情绪和识别命名实体。这些任务不需要额外的背景知识就可以完成。\n\n\n要完成更复杂和知识密集型的任务，可以基于语言模型构建一个系统，访问外部知识源来做到。这样的实现与事实更加一性，生成的答案更可靠，还有助于缓解“幻觉”问题。', metadata={'source': PosixPath('../data/text.txt')})]

Question: Rag的本质是什么
# type(chain)
model.invoke("rag的本质是什么")

AIMessage(content='rag的本质是一种粗糙、粗糙或不整洁的布料，通常用于清洁、擦拭或擦洗。它可以由各种不同类型的材料制成，如棉、亚麻、纤维等。由于其耐用性和吸水性，rag通常用于家庭清洁、车辆清洁和其他清洁工作中。', additional_kwargs={}, response_metadata={'token_usage': {'completion_tokens': 121, 'prompt_tokens': 16, 'total_tokens': 137, 'completion_tokens_details': {'reasoning_tokens': 0}}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None}, id='run-63001da4-5071-47c1-bf58-a0026bcb1771-0')
chain.invoke("rag的本质是什么？")

'RAG的本质是让模型获取正确的Context(上下文)，利用ICL (In Context Learning)的能力，输出正确的响应。'
type(chain)
langchain_core.runnables.base.RunnableSequence
chain.invoke("RAG的本质是什么？")
'RAG的本质是让模型获取正确的Context(上下文)，利用ICL (In Context Learning)的能力，输出正确的响应。'

上面是一个简单的 RAG

langChain 自带的 RAG （工作用，学习不用）

# ## 用 写好的 chain

# from langchain.chains import RetrievalQA

# qa = RetrievalQA.from_chain_type(llm=model, 
#                                  chain_type="stuff", 
#                                  retriever=retriever

#                                  )
# qa.invoke('rag的本质是什么')

RAG 复杂化一下

希望传入一点别的东西；

  • 希望规定语言是什么
  • chain的另外一种写法，（通过 itemgetter 获取 input key value 内容）
template = """Answer the question based only on the following context:
{context}

Question: {question}

Answer in the following language: {language}
"""
prompt = ChatPromptTemplate.from_template(template)

chain = (
    {
        "context": itemgetter("question") | ensemble_retriever,
        "question": itemgetter("question"),
        "language": itemgetter("language"),
    }
    | prompt
    | model
    | StrOutputParser()
)
chain.invoke({"question": "RAG的本质是什么？", "language": "English"})
'The essence of RAG is to allow models to obtain the correct context, utilize the ability of In Context Learning (ICL), and output the correct response.'

format document

先看一下 document 是什么

docs_test = ensemble_retriever.get_relevant_documents("rag的本质是什么")
/var/folders/nn/zmjw6gy57jg40k_tswcx0mpc0000gp/T/ipykernel_22920/3374453711.py:1: LangChainDeprecationWarning: The method `BaseRetriever.get_relevant_documents` was deprecated in langchain-core 0.1.46 and will be removed in 1.0. Use :meth:`~invoke` instead.
  docs_test = ensemble_retriever.get_relevant_documents("rag的本质是什么")
len(docs_test)
2
print(docs_test[0].page_content)
RAG的本质是让模型获取正确的Context(上下文)，利用ICL (In Context Learning)的能力，输出正确的响应。它综合利用了固化在模型权重中的参数化知识和存在外部存储中的非参数化知识(知识库、数据库等)。RAG分为两阶段：使用编码模型（如 BM25、DPR、ColBERT 等）根据问题找到相关的文档。生成阶段：以找到的上下文作为基础，系统生成文本。

通用语言模型通过微调就可以完成几类常见任务，比如分析情绪和识别命名实体。这些任务不需要额外的背景知识就可以完成。

要完成更复杂和知识密集型的任务，可以基于语言模型构建一个系统，访问外部知识源来做到。这样的实现与事实更加一性，生成的答案更可靠，还有助于缓解“幻觉”问题。

RAG 会接受输入并检索出一组相关/支撑的文档，并给出文档的来源（例如维基百科）。这些文档作为上下文和输入的原始提示词组合，送给文本生成器得到最终的输出。这样 RAG 更加适应事实会随时间变化的情况。这非常有用，因为 LLM 的参数化知识是静态的。RAG 让语言模型不用重新训练就能够获取最新的信息，基于检索生成产生可靠的输出。

Lewis 等人（2021）提出一个通用的 RAG 微调方法。这种方法使用预训练的 seq2seq 作为参数记忆，用维基百科的密集向量索引作为非参数记忆（使通过神经网络预训练的检索器访问）。

最近，基于检索器的方法越来越流行，经常与 ChatGPT 等流行 LLM 结合使用来提高其能力和事实一致性。
print(docs_test[1].page_content)
RAG的本质是让模型获取正确的Context(上下文)，利用ICL (In Context Learning)的能力，输出正确的响应。它综合利用了固化在模型权重中的参数化知识和存在外部存储中的非参数化知识(知识库、数据库等)。RAG分为两阶段：使用编码模型（如 BM25、DPR、ColBERT 等）根据问题找到相关的文档。生成阶段：以找到的上下文作为基础，系统生成文本。

通用语言模型通过微调就可以完成几类常见任务，比如分析情绪和识别命名实体。这些任务不需要额外的背景知识就可以完成。

要完成更复杂和知识密集型的任务，可以基于语言模型构建一个系统，访问外部知识源来做到。这样的实现与事实更加一性，生成的答案更可靠，还有助于缓解“幻觉”问题。

format document

  1. 用户可能有多轮
  2. 太长了 -> 历史对话总结，
  3. 用户说法不规范，-> 格式化/标准化用户文法
  4. format document

rag的本质是富尔玛 -> 请问RAG的本质是什么？

from langchain.schema import format_document
from langchain_core.messages import AIMessage, HumanMessage, get_buffer_string
from langchain_core.runnables import RunnableParallel
from langchain.prompts.prompt import PromptTemplate
# 这里 chat history 是用来 【标准化用户的输入的问题】

_template = """Given the following conversation and a follow up question, 
rephrase the follow up question to be a standalone question, in its original language.

Chat History:  
{chat_history}
Follow Up Input: {question}
Standalone question:"""
CONDENSE_QUESTION_PROMPT = PromptTemplate.from_template(_template)
template = """Answer the question based only on the following context:
{context}

Question: {question}
"""
ANSWER_PROMPT = ChatPromptTemplate.from_template(template)
template = """Answer the question based only on the following context:
{context}

Question: {question}
"""
ANSWER_PROMPT = ChatPromptTemplate.from_template(template)

DEFAULT_DOCUMENT_PROMPT = PromptTemplate.from_template(template="{page_content}")

def _combine_documents(
    docs, document_prompt=DEFAULT_DOCUMENT_PROMPT, document_separator="\n\n"
):
    # 重新排序； 把第二个放到 最末尾
    final_docs = [docs[0]]
    for i in range(2, len(docs)):
        final_docs.append(docs[i])
    final_docs.append(docs[1])

    doc_strings = [format_document(doc, document_prompt) for doc in final_docs]
    return document_separator.join(doc_strings)
# print('\n'.join([x.page_content for x in docs_test]))
_inputs = RunnableParallel(
    standalone_question=RunnablePassthrough.assign(
        chat_history=lambda x: get_buffer_string(x["chat_history"])
    )
    | CONDENSE_QUESTION_PROMPT
    | ChatOpenAI(temperature=0, api_key=ENV_CONFIG.get("API_KEY"), base_url=ENV_CONFIG.get("BASE_URL"))
    | StrOutputParser(),
)

_context = {
    "context": itemgetter("standalone_question") | ensemble_retriever | _combine_documents,
    "question": lambda x: x["standalone_question"],
}
conversational_qa_chain = _inputs | _context | ANSWER_PROMPT | ChatOpenAI(api_key=ENV_CONFIG.get("API_KEY"), base_url=ENV_CONFIG.get("BASE_URL")) | StrOutputParser()
# vector store 里面；
# 可以读 mysql 的数据  -> Chroma, Faiss
# conversational_qa_chain = _inputs | _context | ANSWER_PROMPT | ChatOpenAI(api_key=ENV_CONFIG.get("API_KEY"), base_url=ENV_CONFIG.get("BASE_URL")) | StrOutputParser()

# # {"context": xxx, "question"： xxxx}
# # 假设要同时返回 retriever 检索回的 document
# conversational_qa_chain2 = _inputs | _context | {
#     "ans": ANSWER_PROMPT | ChatOpenAI(api_key=ENV_CONFIG.get("API_KEY"), base_url=ENV_CONFIG.get("BASE_URL")) | StrOutputParser(),
#     "doc": itemgetter("context")
# }
# conversational_qa_chain2.invoke({"question": "RAG 的本质是什么？",
#         "chat_history": [],})
# 这里是为了让大家知道 langchain 的每一步都输入什么？（可以自己采用类似的调试方式）
# conversational_qa_chain

# (_inputs).invoke({"question": "RAG 的本质是什么？",
#         "chat_history": [],})

print((_inputs | _context | ANSWER_PROMPT).invoke({"question": "RAG 的本质是什么？",
        "chat_history": [],}))
messages=[HumanMessage(content='Answer the question based only on the following context:\nRAG的本质是让模型获取正确的Context(上下文)，利用ICL (In Context Learning)的能力，输出正确的响应。它综合利用了固化在模型权重中的参数化知识和存在外部存储中的非参数化知识(知识库、数据库等)。RAG分为两阶段：使用编码模型（如 BM25、DPR、ColBERT 等）根据问题找到相关的文档。生成阶段：以找到的上下文作为基础，系统生成文本。\n\n\n通用语言模型通过微调就可以完成几类常见任务，比如分析情绪和识别命名实体。这些任务不需要额外的背景知识就可以完成。\n\n\n要完成更复杂和知识密集型的任务，可以基于语言模型构建一个系统，访问外部知识源来做到。这样的实现与事实更加一性，生成的答案更可靠，还有助于缓解“幻觉”问题。\n\n\nRAG 会接受输入并检索出一组相关/支撑的文档，并给出文档的来源（例如维基百科）。这些文档作为上下文和输入的原始提示词组合，送给文本生成器得到最终的输出。这样 RAG 更加适应事实会随时间变化的情况。这非常有用，因为 LLM 的参数化知识是静态的。RAG 让语言模型不用重新训练就能够获取最新的信息，基于检索生成产生可靠的输出。\n\nLewis 等人（2021）提出一个通用的 RAG 微调方法。这种方法使用预训练的 seq2seq 作为参数记忆，用维基百科的密集向量索引作为非参数记忆（使通过神经网络预训练的检索器访问）。\n\n最近，基于检索器的方法越来越流行，经常与 ChatGPT 等流行 LLM 结合使用来提高其能力和事实一致性。\n\nRAG的本质是让模型获取正确的Context(上下文)，利用ICL (In Context Learning)的能力，输出正确的响应。它综合利用了固化在模型权重中的参数化知识和存在外部存储中的非参数化知识(知识库、数据库等)。RAG分为两阶段：使用编码模型（如 BM25、DPR、ColBERT 等）根据问题找到相关的文档。生成阶段：以找到的上下文作为基础，系统生成文本。\n\n\n通用语言模型通过微调就可以完成几类常见任务，比如分析情绪和识别命名实体。这些任务不需要额外的背景知识就可以完成。\n\n\n要完成更复杂和知识密集型的任务，可以基于语言模型构建一个系统，访问外部知识源来做到。这样的实现与事实更加一性，生成的答案更可靠，还有助于缓解“幻觉”问题。\n\nQuestion: 你能解释一下 RAG 的本质是什么吗？\n', additional_kwargs={}, response_metadata={})]
(_inputs | _context).invoke({
        "question": "RAG 的本质是什么？",
        "chat_history": [],
    })
{'context': 'RAG的本质是让模型获取正确的Context(上下文)，利用ICL (In Context Learning)的能力，输出正确的响应。它综合利用了固化在模型权重中的参数化知识和存在外部存储中的非参数化知识(知识库、数据库等)。RAG分为两阶段：使用编码模型（如 BM25、DPR、ColBERT 等）根据问题找到相关的文档。生成阶段：以找到的上下文作为基础，系统生成文本。\n\n\n通用语言模型通过微调就可以完成几类常见任务，比如分析情绪和识别命名实体。这些任务不需要额外的背景知识就可以完成。\n\n\n要完成更复杂和知识密集型的任务，可以基于语言模型构建一个系统，访问外部知识源来做到。这样的实现与事实更加一性，生成的答案更可靠，还有助于缓解“幻觉”问题。\n\n\nRAG 会接受输入并检索出一组相关/支撑的文档，并给出文档的来源（例如维基百科）。这些文档作为上下文和输入的原始提示词组合，送给文本生成器得到最终的输出。这样 RAG 更加适应事实会随时间变化的情况。这非常有用，因为 LLM 的参数化知识是静态的。RAG 让语言模型不用重新训练就能够获取最新的信息，基于检索生成产生可靠的输出。\n\nLewis 等人（2021）提出一个通用的 RAG 微调方法。这种方法使用预训练的 seq2seq 作为参数记忆，用维基百科的密集向量索引作为非参数记忆（使通过神经网络预训练的检索器访问）。\n\n最近，基于检索器的方法越来越流行，经常与 ChatGPT 等流行 LLM 结合使用来提高其能力和事实一致性。\n\nRAG的本质是让模型获取正确的Context(上下文)，利用ICL (In Context Learning)的能力，输出正确的响应。它综合利用了固化在模型权重中的参数化知识和存在外部存储中的非参数化知识(知识库、数据库等)。RAG分为两阶段：使用编码模型（如 BM25、DPR、ColBERT 等）根据问题找到相关的文档。生成阶段：以找到的上下文作为基础，系统生成文本。\n\n\n通用语言模型通过微调就可以完成几类常见任务，比如分析情绪和识别命名实体。这些任务不需要额外的背景知识就可以完成。\n\n\n要完成更复杂和知识密集型的任务，可以基于语言模型构建一个系统，访问外部知识源来做到。这样的实现与事实更加一性，生成的答案更可靠，还有助于缓解“幻觉”问题。',
 'question': 'RAG 的本质是什么？'}
conversational_qa_chain.invoke(
    {
        "question": "RAG 的本质是什么？",
        "chat_history": [],
    }
)
'RAG的本质是让模型获取正确的Context(上下文)，利用ICL (In Context Learning)的能力，输出正确的响应。'

更复杂的 RAG，把 检索的 document 也返回

from operator import itemgetter

from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory(
    return_messages=True, output_key="answer", input_key="question"
)
/var/folders/nn/zmjw6gy57jg40k_tswcx0mpc0000gp/T/ipykernel_22920/1732780203.py:5: LangChainDeprecationWarning: Please see the migration guide at: https://python.langchain.com/docs/versions/migrating_memory/
  memory = ConversationBufferMemory(
# 设置一个 memory; 自动维护，而不是类似于上面的【用户传入 chat_history 、 注意这里的区别】
# 上一个 demo 是用户要输入 history, 这里是通过 memory 维护，而不用每一次 invoke的时候都加上
loaded_memory = RunnablePassthrough.assign(
    chat_history=RunnableLambda(memory.load_memory_variables) | itemgetter("history"),
)
# 标准化用户的输入问题
standalone_question = {
    "standalone_question": {
        "question": lambda x: x["question"],
        "chat_history": lambda x: get_buffer_string(x["chat_history"]),
    }
    | CONDENSE_QUESTION_PROMPT
    | ChatOpenAI(temperature=0, api_key=ENV_CONFIG.get("API_KEY"), base_url=ENV_CONFIG.get("BASE_URL"))
    | StrOutputParser(),
}
# 检索 Document 
retrieved_documents = {
    "docs": itemgetter("standalone_question") | retriever,
    "question": lambda x: x["standalone_question"],
}
# 通过检索回来的 doc 和标准化后的 question 构建 prompt
final_inputs = {
    "context": lambda x: _combine_documents(x["docs"]),
    "question": itemgetter("question"),
}
# 返回答案，两条路
# 一条直接 获得检索内容
# 一条 构建 prompt -> chatmodel -> output parser
answer = {
    "answer": final_inputs | ANSWER_PROMPT | ChatOpenAI(api_key=ENV_CONFIG.get("API_KEY"), base_url=ENV_CONFIG.get("BASE_URL")),
    "docs": itemgetter("docs"),
}
# 构建完整链
final_chain = loaded_memory | standalone_question | retrieved_documents | answer
inputs = {"question": "Lewis提出了什么方法？"}
result = final_chain.invoke(inputs)
result
# 用户的 memory 需要自己存一下；
# 但是相对于手动传 history 好很多了；
memory.save_context(inputs, {"answer": result["answer"].content})
memory.load_memory_variables({})
# 这里看到不要穿 history, 因为通过 memory 完成了；
inputs = {"question": "how did Rag really work?"}
result = final_chain.invoke(inputs)
result

为者常成，行者常至

自由转载-非商用-非衍生-保持署名（创意共享3.0许可证）