对话式RAG与检索策略#

1
# pip install langchain langchain-openai langchain-community faiss-cpu
2

3
from langchain.chains import create_history_aware_retriever, create_retrieval_chain
4
from langchain.chains.combine_documents import create_stuff_documents_chain
5
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
6
from langchain_core.messages import HumanMessage, AIMessage
7
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
8
from langchain_community.vectorstores import FAISS
9

10
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
11
vectorstore = FAISS.from_texts(
12
    [
13
        "LangChain 由 Harrison Chase 于 2022 年 10 月开源。",
14
        "Harrison Chase 后来创立了 LangChain 相关公司，并推动 LangSmith、LangGraph 等产品发展。",
15
        "LangGraph 是 LangChain 生态中的有状态 Agent 编排框架。",
16
    ],
17
    embeddings,
18
)
19
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})
20

21
contextualize_q_prompt = ChatPromptTemplate.from_messages([
22
    ("system",
23
     "给定聊天历史和用户最新问题，如果问题引用了历史上下文，"
24
     "请将它改写成一个脱离聊天历史也能独立理解的问题。"
25
     "只做改写，不要回答问题。"),
26
    MessagesPlaceholder(variable_name="chat_history"),
27
    ("human", "{input}"),
28
])
29

30
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
31

32
history_aware_retriever = create_history_aware_retriever(
33
    llm, retriever, contextualize_q_prompt
34
)
35

36
qa_prompt = ChatPromptTemplate.from_messages([
37
    ("system",
38
     "你是一个知识库问答助手。请基于以下参考资料回答用户问题。"
39
     "如果资料中没有答案，请明确说明不知道。\n\n"
40
     "{context}"),
41
    MessagesPlaceholder(variable_name="chat_history"),
42
    ("human", "{input}"),
43
])
44

45
question_answer_chain = create_stuff_documents_chain(llm, qa_prompt)
46
rag_chain = create_retrieval_chain(history_aware_retriever, question_answer_chain)
47

48
chat_history = [
49
    HumanMessage(content="LangChain 是什么时候开源的？"),
50
    AIMessage(content="LangChain 由 Harrison Chase 于 2022 年 10 月开源。"),
51
]
52

53
result = rag_chain.invoke({
54
    "input": "它的作者后来做了什么？",
55
    "chat_history": chat_history,
56
})
57

58
print(result["answer"])

1
# pip install langchain langchain-openai langchain-chroma lark
2

3
from langchain_core.documents import Document
4
from langchain.chains.query_constructor.schema import AttributeInfo
5
from langchain.retrievers.self_query.base import SelfQueryRetriever
6
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
7
from langchain_chroma import Chroma
8

9
docs = [
10
    Document(
11
        page_content="一名宇航员穿越虫洞寻找新家园。",
12
        metadata={"genre": "science fiction", "year": 2014, "rating": 8.6},
13
    ),
14
    Document(
15
        page_content="玩具们在主人不在时会活过来。",
16
        metadata={"genre": "animated", "year": 1995, "rating": 8.3},
17
    ),
18
]
19

20
vectorstore = Chroma.from_documents(
21
    docs,
22
    embedding=OpenAIEmbeddings(model="text-embedding-3-small"),
23
    collection_name="self_query_demo",
24
)
25

26
metadata_field_info = [
27
    AttributeInfo(name="genre", description="电影类型", type="string"),
28
    AttributeInfo(name="year", description="电影上映年份", type="integer"),
29
    AttributeInfo(name="rating", description="电影评分，范围 1-10", type="float"),
30
]
31

32
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
33

34
retriever = SelfQueryRetriever.from_llm(
35
    llm,
36
    vectorstore,
37
    "电影简介",
38
    metadata_field_info,
39
    enable_limit=True,
40
    verbose=True,
41
)
42

43
results = retriever.invoke("找两部 2010 年之后评分高于 8 分的科幻电影")
44
for doc in results:
45
    print(doc.page_content, doc.metadata)

1
# pip install langchain langchain-openai langchain-community faiss-cpu rank-bm25
2

3
from langchain.retrievers import EnsembleRetriever
4
from langchain_community.retrievers import BM25Retriever
5
from langchain_community.vectorstores import FAISS
6
from langchain_openai import OpenAIEmbeddings
7

8
texts = [
9
    "PEP 703 提出了让 CPython 支持可选的无 GIL 模式。",
10
    "Python 3.12 改进了错误提示信息。",
11
    "LangChain 使用 Runnable 作为统一抽象。",
12
    "GIL 是 Python 多线程性能讨论中的核心概念。",
13
]
14

15
bm25_retriever = BM25Retriever.from_texts(texts)
16
bm25_retriever.k = 3
17

18
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
19
vectorstore = FAISS.from_texts(texts, embeddings)
20
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
21

22
ensemble_retriever = EnsembleRetriever(
23
    retrievers=[bm25_retriever, vector_retriever],
24
    weights=[0.4, 0.6],
25
)
26

27
docs = ensemble_retriever.invoke("PEP 703 和禁用 GIL 的实现思路是什么？")
28
for doc in docs:
29
    print(doc.page_content)