当前位置： 首页 > news >正文

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

news 来源：原创 2024/7/2 3:45:01

这示例它实现了一个基于LangGraph的系统，用于处理文档检索和生成答案的过程。
好的，我会按照Markdown格式完整翻译并保留文件结构和格式：

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

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 OpenAIEmbeddingsurls = ["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()

状态

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

from typing import Dict, TypedDictfrom langchain_core.messages import BaseMessageclass 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 structurethat supports addition with `operator.add`. This could be used, for instance, to accumulate messagesor other pieces of data throughout the graph."""keys: Dict[str, any]

节点和边

每个 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_scoreif grade == "yes":print("---GRADE: DOCUMENT RELEVANT---")filtered_docs.append(d)else:print("---GRADE: DOCUMENT NOT RELEVANT---")search = "Yes"  # 进行网络搜索continuereturn {"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, StateGraphworkflow = 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. LangChain：是一个用于构建语言模型应用的Python库，提供了文本分割、文档加载、向量存储、嵌入和检索等功能。
  2. Tavily Search：是一个网络搜索引擎，可以用于补充检索过程中的数据源。
  3. RecursiveCharacterTextSplitter：用于将长文本分割成更小的块，以便更好地处理和索引。
  4. Chroma：是一个向量数据库，可以存储和检索文档的嵌入表示。
  5. OpenAIEmbeddings：使用OpenAI的模型来生成文档的嵌入表示。
  6. StateGraph：是一个用于构建和执行状态图的类，状态图是一种用于控制流程的有向图。

总结：

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

相关文章：

  • 在VS Code中快速生成Vue模板的技巧
  • 查看 RK3568 Android SDK 版本的详细指南
  • 猫头虎分享已解决Bug || 前端领域技术问题解析
  • 常见的网络设备
  • Java算法常用技巧
  • web前端之vue一键部署的shell脚本和它的点.bat文件、海螺AI、ChatGPT
  • 基于PHP的草莓种植信息管理系统
  • SpringCloud学习笔记 - 1、Boot和Cloud版本选型
  • 代码规范性思考
  • 硬件开发笔记（十八）：核心板与底板之间的连接方式介绍说明：板对板连接器
  • 微服务feign组件学习
  • 嵌入式技术学习——c51单片机——蜂鸣器
  • Android开发系列（三）Jetpack Compose 之TextField
  • 泛微开发修炼之旅--20关于Ecology中如何查询正文文件的物理文件，并修改正文中的内容的解决方案
  • Chromium 开发指南2024 Mac篇-Xcode安装（二）
  • @angular/forms 源码解析之双向绑定
  • 【Leetcode】104. 二叉树的最大深度
  • 2017 年终总结 —— 在路上
  • axios请求、和返回数据拦截，统一请求报错提示_012
  • CentOS 7 防火墙操作
  • css布局，左右固定中间自适应实现
  • CSS实用技巧干货
  • css系列之关于字体的事
  • IIS 10 PHP CGI 设置 PHP_INI_SCAN_DIR
  • Java基本数据类型之Number
  • magento2项目上线注意事项
  • Making An Indicator With Pure CSS
  • Promise面试题，控制异步流程
  • 持续集成与持续部署宝典Part 2：创建持续集成流水线
  • 翻译 | 老司机带你秒懂内存管理 - 第一部（共三部）
  • 关于使用markdown的方法（引自CSDN教程）
  • 规范化安全开发 KOA 手脚架
  • 理解IaaS, PaaS, SaaS等云模型 (Cloud Models)
  • 设计模式（12）迭代器模式（讲解+应用）
  • 推荐一个React的管理后台框架
  • 微信端页面使用-webkit-box和绝对定位时，元素上移的问题
  • 问题之ssh中Host key verification failed的解决
  • 系统认识JavaScript正则表达式
  • 数据可视化之下发图实践
  • ​​​​​​​​​​​​​​汽车网络信息安全分析方法论
  • ​一帧图像的Android之旅 :应用的首个绘制请求
  • # Swust 12th acm 邀请赛# [ E ] 01 String [题解]
  • #APPINVENTOR学习记录
  • （14）目标检测_SSD训练代码基于pytorch搭建代码
  • （BFS）hdoj2377-Bus Pass
  • （补）B+树一些思想
  • （七）Java对象在Hibernate持久化层的状态
  • （推荐）叮当——中文语音对话机器人
  • .NET 3.0 Framework已经被添加到WindowUpdate
  • .net core 控制台应用程序读取配置文件app.config
  • .NET 自定义中间件 判断是否存在 AllowAnonymousAttribute 特性 来判断是否需要身份验证
  • .pop ----remove 删除
  • .stream().map与.stream().flatMap的使用
  • .vimrc php,修改home目录下的.vimrc文件，vim配置php高亮显示
  • /usr/lib/mysql/plugin权限_给数据库增加密码策略遇到的权限问题