构建LangChain应用程序的示例代码：37、基于LangGraph的文档检索与答案生成系统教程_langgraph文档

这示例它实现了一个基于LangGraph的系统，用于处理文档检索和生成答案的过程。

---

! pip install langchain_community tiktoken langchain-openai langchainhub chromadb langchain langgraph tavily-python
1

CRAG

Corrective-RAG 是一篇最新的论文，介绍了一种有趣的主动 RAG 方法。

该框架根据问题对检索到的文档进行评分：

1. 正确的文档 -

   • 如果至少有一个文档超过了相关性的阈值，则继续生成。
   • 在生成之前，它会进行知识细化。
   • 这会将文档分成“知识条带”。
   • 它对每个条带进行评分，并过滤掉无关的条带。

2. 含糊或错误的文档 -

   • 如果所有文档都低于相关性阈值或评分器不确定，则框架会寻找额外的数据源。
   • 它会使用网络搜索来补充检索。
   • 论文中的图表还表明，这里使用了查询重写。

论文链接：https://arxiv.org/pdf/2401.15884.pdf

---

让我们使用 LangGraph 从头开始实现这一点。

我们可以做一些简化：

• 作为初步尝试，让我们跳过知识细化阶段。如果需要，可以将其添加回节点中。
• 如果任何文档不相关，我们选择使用网络搜索来补充检索。
• 我们将使用 Tavily Search 进行网络搜索。
• 我们将使用查询重写来优化网络搜索查询。

设置 TAVILY_API_KEY。

检索器

让我们索引3篇博客文章。

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import WebBaseLoader
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

urls = [
    "https://lilianweng.github.io/posts/2023-06-23-agent/",
    "https://lilianweng.github.io/posts/2023-03-15-prompt-engineering/",
    "https://lilianweng.github.io/posts/2023-10-25-adv-attack-llm/",
]

docs = [WebBaseLoader(url).load() for url in urls]
docs_list = [item for sublist in docs for item in sublist]

text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=250, chunk_overlap=0
)
doc_splits = text_splitter.split_documents(docs_list)

# Add to vectorDB
vectorstore = Chroma.from_documents(
    documents=doc_splits,
    collection_name="rag-chroma",
    embedding=OpenAIEmbeddings(),
)
retriever = vectorstore.as_retriever()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

状态

我们将定义一个图。
我们的状态将是 dict 。
我们可以从任何图形节点 state[‘keys’] 访问它。

from typing import Dict, TypedDict

from langchain_core.messages import BaseMessage


class GraphState(TypedDict):
    """
    Represents the state of an agent in the conversation.

    Attributes:
        keys: A dictionary where each key is a string and the value is expected to be a list or another structure
              that supports addition with `operator.add`. This could be used, for instance, to accumulate messages
              or other pieces of data throughout the graph.
    """

    keys: Dict[str, any]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

节点和边

每个 node 将简单地修改 state 。
每个 edge 将选择接下来调用哪个 node 。
它将遵循上面显示的图表。

import json
import operator
from typing import Annotated, Sequence, TypedDict

# 导入langchain相关模块
from langchain import hub
from langchain.output_parsers import PydanticOutputParser
from langchain.output_parsers.openai_tools import PydanticToolsParser
from langchain.prompts import PromptTemplate
from langchain.schema import Document
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_community.vectorstores import Chroma
from langchain_core.messages import BaseMessage, FunctionMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_core.runnables import RunnablePassthrough
from langchain_core.utils.function_calling import convert_to_openai_tool
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langgraph.prebuilt import ToolInvocation

### 节点函数 ###

def retrieve(state):
    """
    检索文档

    参数:
        state (dict): 代理当前状态，包括所有键。

    返回:
        dict: 在状态中添加新的键'documents'，包含检索到的文档。
    """
    print("---RETRIEVE---")
    state_dict = state["keys"]
    question = state_dict["question"]
    documents = retriever.invoke(question)
    return {"keys": {"documents": documents, "question": question}}

def generate(state):
    """
    生成回答

    参数:
        state (dict): 代理当前状态，包括所有键。

    返回:
        dict: 在状态中添加新的键'generation'，包含生成的回答。
    """
    print("---GENERATE---")
    state_dict = state["keys"]
    question = state_dict["question"]
    documents = state_dict["documents"]

    # 提示模板
    prompt = hub.pull("rlm/rag-prompt")

    # 大语言模型
    llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0, streaming=True)

    # 后处理函数
    def format_docs(docs):
        return "\n\n".join(doc.page_content for doc in docs)

    # 链
    rag_chain = prompt | llm | StrOutputParser()

    # 运行
    generation = rag_chain.invoke({"context": documents, "question": question})
    return {
        "keys": {"documents": documents, "question": question, "generation": generation}
    }

def grade_documents(state):
    """
    判断检索到的文档是否与问题相关。

    参数:
        state (dict): 代理当前状态，包括所有键。

    返回:
        dict: 在状态中添加新的键'filtered_documents'，包含相关的文档。
    """
    print("---CHECK RELEVANCE---")
    state_dict = state["keys"]
    question = state_dict["question"]
    documents = state_dict["documents"]

    # 数据模型
    class grade(BaseModel):
        """相关性检查的二进制评分。"""

        binary_score: str = Field(description="相关性评分 'yes' 或 'no'")

    # 大语言模型
    model = ChatOpenAI(temperature=0, model="gpt-4-0125-preview", streaming=True)

    # 工具
    grade_tool_oai = convert_to_openai_tool(grade)

    # 绑定工具和强制调用的语言模型
    llm_with_tool = model.bind(
        tools=[convert_to_openai_tool(grade_tool_oai)],
        tool_choice={"type": "function", "function": {"name": "grade"}},
    )

    # 解析器
    parser_tool = PydanticToolsParser(tools=[grade])

    # 提示模板
    prompt = PromptTemplate(
        template="""你是一个评分员，评估检索到的文档与用户问题的相关性。\n 
        这是检索到的文档：\n\n {context} \n\n
        这是用户的问题：{question} \n
        如果文档包含与用户问题相关的关键词或语义，请评为相关。\n
        给出一个 'yes' 或 'no' 的二进制评分，表示文档是否与问题相关。""",
        input_variables=["context", "question"],
    )

    # 链
    chain = prompt | llm_with_tool | parser_tool

    # 评分
    filtered_docs = []
    search = "No"  # 默认不进行网络搜索来补充检索
    for d in documents:
        score = chain.invoke({"question": question, "context": d.page_content})
        grade = score[0].binary_score
        if grade == "yes":
            print("---GRADE: DOCUMENT RELEVANT---")
            filtered_docs.append(d)
        else:
            print("---GRADE: DOCUMENT NOT RELEVANT---")
            search = "Yes"  # 进行网络搜索
            continue

    return {
        "keys": {
            "documents": filtered_docs,
            "question": question,
            "run_web_search": search,
        }
    }

def transform_query(state):
    """
    转换查询以生成更好的问题。

    参数:
        state (dict): 代理当前状态，包括所有键。

    返回:
        dict: 保存新的问题。
    """
    print("---TRANSFORM QUERY---")
    state_dict = state["keys"]
    question = state_dict["question"]
    documents = state_dict["documents"]

    # 创建一个提示模板，包含格式指令和查询
    prompt = PromptTemplate(
        template="""你正在生成一个优化检索的问题。\n 
        查看输入并试图推理其潜在的语义意图。\n 
        这是初始问题：
        \n ------- \n
        {question} 
        \n ------- \n
        生成一个改进的问题：""",
        input_variables=["question"],
    )

    # 评分员
    model = ChatOpenAI(temperature=0, model="gpt-4-0125-preview", streaming=True)

    # 链
    chain = prompt | model | StrOutputParser()
    better_question = chain.invoke({"question": question})

    return {"keys": {"documents": documents, "question": better_question}}

def web_search(state):
    """
    使用Tavily进行网络搜索。

    参数:
        state (dict): 代理当前状态，包括所有键。

    返回:
        state (dict): 将网络搜索结果附加到文档中。
    """
    print("---WEB SEARCH---")
    state_dict = state["keys"]
    question = state_dict["question"]
    documents = state_dict["documents"]

    tool = TavilySearchResults()
    docs = tool.invoke({"query": question})
    web_results = "\n".join([d["content"] for d in docs])
    web_results = Document(page_content=web_results)
    documents.append(web_results)

    return {"keys": {"documents": documents, "question": question}}

### 边函数 ###

def decide_to_generate(state):
    """
    决定是生成回答还是重新生成问题。

    参数:
        state (dict): 代理当前状态，包括所有键。

    返回:
        dict: 在状态中添加新的键'filtered_documents'，包含相关的文档。
    """
    print("---DECIDE TO GENERATE---")
    state_dict = state["keys"]
    question = state_dict["question"]
    filtered_documents = state_dict["documents"]
    search = state_dict["run_web_search"]

    if search == "Yes":
        # 所有文档已被过滤
        # 我们将重新生成一个新的查询
        print("---DECISION: TRANSFORM QUERY and RUN WEB SEARCH---")
        return "transform_query"
    else:
        # 我们有相关文档，所以生成回答
        print("---DECISION: GENERATE---")
        return "generate"

import pprint
from langgraph.graph import END, StateGraph

workflow = StateGraph(GraphState)

# 定义节点
workflow.add_node("retrieve", retrieve)  # 检索
workflow.add_node("grade_documents", grade_documents)  # 评分文档
workflow.add_node("generate", generate)  # 生成
workflow.add_node("transform_query", transform_query)  # 转换查询
workflow.add_node("web_search", web_search)  # 网络搜索

# 构建图
workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "grade_documents")
workflow.add_conditional_edges(
    "grade_documents",
    decide_to_generate,
    {
        "transform_query": "transform_query",
        "generate": "generate",
    },
)
workflow.add_edge("transform_query", "web_search")
workflow.add_edge("web_search", "generate")
workflow.add_edge("generate", END)

# 编译
app = workflow.compile()

# 运行
inputs = {"keys": {"question": "Explain how the different types of agent memory work?"}}
for output in app.stream(inputs):
    for key, value in output.items():
        pprint.pprint(f"Output from node '{key}':")
        pprint.pprint("---")
        pprint.pprint(value["keys"], indent=2, width=80, depth=None)
    pprint.pprint("\n---\n")

# 对不在上下文中的问题进行修正
inputs = {"keys": {"question": "What is the approach taken in the AlphaCodium paper?"}}
for output in app.stream(inputs):
    for key, value in output.items():
        pprint.pprint(f"Output from node '{key}':")
        pprint.pprint("---")
        pprint.pprint(value["keys"], indent=2, width=80, depth=None)
    pprint.pprint("\n---\n")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278

---

扩展知识点：

1. LangChain：是一个用于构建语言模型应用的Python库，提供了文本分割、文档加载、向量存储、嵌入和检索等功能。
2. Tavily Search：是一个网络搜索引擎，可以用于补充检索过程中的数据源。
3. RecursiveCharacterTextSplitter：用于将长文本分割成更小的块，以便更好地处理和索引。
4. Chroma：是一个向量数据库，可以存储和检索文档的嵌入表示。
5. OpenAIEmbeddings：使用OpenAI的模型来生成文档的嵌入表示。
6. StateGraph：是一个用于构建和执行状态图的类，状态图是一种用于控制流程的有向图。

总结：

本文介绍了一个使用LangGraph实现的系统，该系统通过文档检索、文档评估、问题转换和网络搜索等步骤，来生成针对特定问题的答案。系统的核心是一个状态图，它定义了各个节点和边，通过这些节点和边来控制整个检索和生成流程。代码中使用了多个库，包括langchain、langchain_community、langchain_openai等，这些库为系统提供了文本分割、文档加载、向量存储、嵌入和检索等功能。