粤ICP备17114055号
微信扫码
添加专属顾问
支持私有云部署
我要投稿
利用AI智能体增强RAG系统，提升生成式AI助手性能 核心内容： 1. RAG系统在企业中的应用及其优势 2. 传统RAG系统架构与局限性分析 3. 纠正性RAG系统的设计与实现
在当今的人工智能领域，生成式人工智能（Generative AI）正迅速发展，越来越多的企业希望利用定制数据构建自己的生成式 AI 助手。检索增强生成系统（Retrieval Augmented Generation systems，简称 RAG 系统）因其能够避免对大语言模型（Large Language Models，LLMs）进行昂贵的微调而迅速流行起来。RAG 系统的一个关键优势在于，它可以轻松集成企业自身的数据，增强大语言模型的智能，并为用户的问题提供更具上下文的答案。
然而，传统的 RAG 系统也面临着一系列问题，这些问题可能导致其性能不佳，甚至给出错误的答案。那么，有没有一种方法可以改进这些问题呢？答案是肯定的。在本文中，我们将探讨如何利用 AI 智能体（AI Agents）来增强传统 RAG 系统的能力，并解决其一些局限性。
传统的 RAG 系统架构通常由两个主要步骤组成：数据处理和索引、检索和响应生成。
这种两步式工作流程在行业中被广泛用于构建传统的 RAG 系统，但它也存在一些局限性。
传统 RAG 系统存在几个明显的局限性：
在本文中，我们将特别关注 RAG 系统无法访问实时数据的局限性，并确保检索到的文档块确实与回答问题相关。这样可以使 RAG 系统能够回答关于最近事件和实时数据的问题，并减少出现幻觉的可能性。
纠正性 RAG 系统的灵感来自于论文《Corrective Retrieval Augmented Generation》（作者：Yan 等人）中提出的解决方案。其关键思想是像往常一样从向量数据库中检索文档块，然后使用 LLM 检查每个检索到的文档块是否与输入问题相关。
如果所有检索到的文档块都相关，那么就像标准的 RAG 管道一样，将其发送到 LLM 进行正常的响应生成。然而，如果某些检索到的文档与输入问题不相关，我们会重新表述输入查询，在网络上搜索与输入问题相关的新信息，然后将其发送到 LLM 以生成响应。
这种方法的关键创新点包括在网络上搜索，用实时数据扩充向量数据库中的静态信息，以及检查检索到的文档是否与输入问题相关，这是简单的嵌入余弦相似度无法捕捉到的。
AI 智能体或智能 AI 系统在 2024 年有了显著的发展，它使我们能够构建能够推理、分析、交互并自动采取行动的生成式 AI 系统。智能 AI 的核心思想是构建完全自主的系统，能够在最少的人工干预下理解和管理复杂的工作流程和任务。智能系统可以掌握细微的概念，设定并追求目标，通过任务进行推理，并根据不断变化的条件调整其行动。这些系统可以由单个智能体或多个智能体组成。
我们可以使用各种框架来构建智能 AI 系统，包括 CrewAI、LangChain、LangGraph、AutoGen 等等。使用这些框架可以让我们轻松开发复杂的工作流程。需要记住的是，智能体基本上是一个或多个大语言模型，它们可以访问一组工具，并根据特定的基于提示的指令来回答用户问题。
在我们的实际实现中，我们将使用 LangGraph 来构建我们的智能 RAG 系统。LangGraph 建立在 LangChain 之上，有助于创建循环图，这对于开发由 LLM 驱动的 AI 智能体至关重要。它的接口受到广泛使用的 NetworkX 库的启发，能够通过循环计算步骤协调和检查点多个链（或参与者）。LangGraph 将智能体工作流程视为循环图结构。
LangGraph 智能体的主要组件包括：
LangGraph 利用这些来促进具有状态持久性的循环 LLM 调用执行，这是 AI 智能体经常需要的。
智能纠正性 RAG 系统的工作流程包含了多个关键组件和执行流程。该系统中有两个主要的流程：
我们从用户查询开始，该查询将发送到向量数据库（我们将使用 Chroma），并检索一些上下文文档。如果用户查询基于最近的事件或超出我们向量数据库中初始数据范围的主题，那么有可能无法检索到任何文档。
接下来，我们将用户查询和上下文文档发送给 LLM，并使其充当文档评分器。它将根据每个上下文文档在含义和上下文方面是否与用户查询相关，将其评为“是”或“否”。
然后是决策节点，这里有两种可能的路径：
下一步是调用网络搜索工具，在我们的实现中，我们将使用 Tavily Web Search API 工具在网络上搜索并获取相关信息作为上下文文档，然后将它们添加到从向量数据库中检索到的任何相关上下文文档列表中。
最后一步是通过相同的 RAG 流程，使用查询和上下文文档（包括从网络检索到的实时信息）生成响应。
我们将使用 LangGraph 实现我们之前讨论的智能 RAG 系统。我们将从 Wikipedia 加载一些文档到我们的向量数据库（Chroma 数据库）中，并使用 Tavily Search 工具进行网络搜索。与 LLM 的连接和提示将使用 LangChain 进行，并且智能体将使用 LangGraph 构建。对于我们的 LLM，我们将使用 ChatGPT GPT-4o，这是一个功能强大的 LLM，对工具调用具有原生支持。不过，你也可以自由使用任何其他 LLM，包括开源 LLM，建议使用经过微调以进行工具调用的强大 LLM 以获得最佳性能。
我们首先安装必要的依赖项，这些是我们构建系统将使用的库：
!pip install langchain==0.2.0
!pip install langchain-openai==0.1.7
!pip install langchain-community==0.2.0
!pip install langgraph==0.1.1
!pip install langchain-chroma==0.1.1
我们使用 getpass() 函数输入我们的 Open AI 密钥，这样我们就不会意外地在代码中暴露我们的密钥。
from getpass import getpass
OPENAI_KEY = getpass('Enter Open AI API Key: ')
同样，我们使用 getpass() 函数输入我们的 Tavily Search API 密钥，并从指定位置获取免费的 API 密钥。
TAVILY_API_KEY = getpass('Enter Tavily Search API Key: ')
接下来，我们设置一些系统环境变量，这些变量将在稍后对 LLM 进行身份验证和搜索 API 时使用。
import os
os.environ['OPENAI_API_KEY'] = OPENAI_KEY
os.environ['TAVILY_API_KEY'] = TAVILY_API_KEY
我们从 Wikipedia 中选取一部分文档来构建用于检索和搜索的向量数据库。我们已经从 Wikipedia 中提取了这些文档并将它们存储在一个存档文件中。
from langchain_openai import OpenAIEmbeddings
openai_embed_model = OpenAIEmbeddings(model='text-embedding-3-small')
# 使用 Google Colab 下载
!gdown 1oWBnoxBZ1Mpeond8XDUSO6J9oAjcRDyW
import gzip
import json
from langchain.docstore.document import Document
from langchain.text_splitter import RecursiveCharacterTextSplitter
wikipedia_filepath ='simplewiki-2020-11-01.jsonl.gz'
docs = []
with gzip.open(wikipedia_filepath, 'rt', encoding='utf8') as fIn:
for line in fIn:
data = json.loads(line.strip())
docs.append({
'metadata': {
'title': data.get('title'),
'article_id': data.get('id')
},
'data':''.join(data.get('paragraphs')[0:3])
})
docs = [doc for doc in docs for x in ['india'] if x in doc['data'].lower().split()]
docs = [Document(page_content=doc['data'], metadata=doc['metadata']) for doc in docs]
splitter = RecursiveCharacterTextSplitter(chunk_size=2000, chunk_overlap=300)
chunked_docs = splitter.split_documents(docs)
from langchain_chroma import Chroma
chroma_db = Chroma.from_documents(documents=chunked_docs,
collection_name='rag_wikipedia_db',
embedding=openai_embed_model,
collection_metadata={"hnsw:space": "cosine"},
persist_directory="./wikipedia_db")
我们使用“带阈值的相似性检索”策略，该策略使用余弦相似性，并根据用户输入查询检索前 3 个最相似的文档，同时引入一个截止值，以不返回任何低于特定相似性阈值（在这种情况下为 0.3）的文档。
similarity_threshold_retriever = chroma_db.as_retriever(search_type="similarity_score_threshold",
search_kwargs={"k": 3,
"score_threshold": 0.3})
我们可以通过一些示例查询来测试我们的检索器是否正常工作。
我们使用 LLM（在我们的例子中是 GPT-4o）来判断任何检索到的文档是否与给定问题相关，答案将是“是”或“否”。
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_openai import ChatOpenAI
class GradeDocuments(BaseModel):
"""Binary score for relevance check on retrieved documents."""
binary_score: str = Field(
description="Documents are relevant to the question, 'yes' or 'no'"
)
llm = ChatOpenAI(model="gpt-4o", temperature=0)
structured_llm_grader = llm.with_structured_output(GradeDocuments)
SYS_PROMPT = """You are an expert grader assessing relevance of a retrieved document to a user question.
Follow these instructions for grading:
- If the document contains keyword(s) or semantic meaning related to the question, grade it as relevant.
- Your grade should be either 'yes' or 'no' to indicate whether the document is relevant to the question or not."""
grade_prompt = ChatPromptTemplate.from_messages(
[
("system", SYS_PROMPT),
("human", """Retrieved document:
{document}
User question:
{question}
"""),
]
)
doc_grader = (grade_prompt | structured_llm_grader)
我们可以通过一些示例用户查询来测试这个评分器，看看从向量数据库中检索到的上下文文档的相关性如何。
我们将检索器连接到 LLM（在我们的例子中是 GPT-4o），并构建我们的问答 RAG 链。这将是我们的传统 RAG 系统，我们稍后将其与 AI 智能体集成。
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_core.output_parsers import StrOutputParser
from operator import itemgetter
prompt = """You are an assistant for question-answering tasks.
Use the following pieces of retrieved context to answer the question.
If no context is present or if you don't know the answer, just say that you don't know the answer.
Do not make up the answer unless it is there in the provided context.
Give a detailed answer and to the point answer with regard to the question.
Question:
{question}
Context:
{context}
Answer:
"""
prompt_template = ChatPromptTemplate.from_template(prompt)
chatgpt = ChatOpenAI(model_name='gpt-4o', temperature=0)
def format_docs(docs):
return"\n\n".join(doc.page_content for doc in docs)
qa_rag_chain = (
{
"context": (itemgetter('context') | RunnableLambda(format_docs)),
"question": itemgetter('question')
}
|
prompt_template
|
chatgpt
|
StrOutputParser()
)
我们可以通过一些查询来测试我们的传统 RAG 系统。
我们构建一个查询重写器，它将使用 LLM（在我们的例子中是 GPT-4o）将输入的用户查询重写为更适合网络搜索的更好版本。这将帮助我们从网络上为我们的查询获取更好的上下文信息。
llm = ChatOpenAI(model="gpt-4o", temperature=0)
SYS_PROMPT = """Act as a question re-writer and perform the following task:
- Convert the following input question to a better version that is optimized for web search.
- When re-writing, look at the input question and try to reason about the underlying semantic intent / meaning.
"""
re_write_prompt = ChatPromptTemplate.from_messages(
[
("system", SYS_PROMPT),
("human", """Here is the initial question:
{question}
Formulate an improved question.
""",
),
]
)
question_rewriter = (re_write_prompt
|
llm
|
StrOutputParser())
我们可以用一个示例问题来测试查询重写器，看看它是如何工作的：
query = "who won the champions league in 2024?"
question_rewriter.invoke({"question": query})
输出结果可能类似：
Who was the winner of the 2024 UEFA Champions League?
可以看到，查询重写器将原始问题优化为更适合网络搜索的形式。
我们将使用 Tavily API 进行网络搜索，因此要建立与该 API 的连接。在搜索时，我们将使用前 3 个搜索结果作为额外的上下文信息，当然你也可以自由加载更多的搜索结果。
from langchain_community.tools.tavily_search import TavilySearchResults
tv_search = TavilySearchResults(max_results=3, search_depth='advanced',max_tokens=10000)
from typing import List
from typing_extensions import TypedDict
class GraphState(TypedDict):
"""
Represents the state of our graph.
Attributes:
question: question
generation: LLM response generation
web_search_needed: flag of whether to add web search - yes or no
documents: list of context documents
"""
question: str
generation: str
web_search_needed: str
documents: List[str]
def retrieve(state):
"""
Retrieve documents
Args:
state (dict): The current graph state
Returns:
state (dict): New key added to state, documents - that contains retrieved context documents
"""
print("---RETRIEVAL FROM VECTOR DB---")
question = state["question"]
# Retrieval
documents = similarity_threshold_retriever.invoke(question)
return {"documents": documents, "question": question}
web_search_needed 标志设置为 “Yes”。否则，如果所有文档在上下文上都与给定的用户查询相关，它会将该标志设置为 “No”。它还会更新状态图，确保上下文文档仅包含相关文档。def grade_documents(state):
"""
Determines whether the retrieved documents are relevant to the question
by using an LLM Grader.
If any document are not relevant to question or documents are empty - Web Search needs to be done
If all documents are relevant to question - Web Search is not needed
Helps filtering out irrelevant documents
Args:
state (dict): The current graph state
Returns:
state (dict): Updates documents key with only filtered relevant documents
"""
print("---CHECK DOCUMENT RELEVANCE TO QUESTION---")
question = state["question"]
documents = state["documents"]
# Score each doc
filtered_docs = []
web_search_needed = "No"
if documents:
for d in documents:
score = doc_grader.invoke(
{"question": question, "document": d.page_content}
)
grade = score.binary_score
if grade == "yes":
print("---GRADE: DOCUMENT RELEVANT---")
filtered_docs.append(d)
else:
print("---GRADE: DOCUMENT NOT RELEVANT---")
web_search_needed = "Yes"
continue
else:
print("---NO DOCUMENTS RETRIEVED---")
web_search_needed = "Yes"
return {"documents": filtered_docs, "question": question,
"web_search_needed": web_search_needed}
def rewrite_query(state):
"""
Rewrite the query to produce a better question.
Args:
state (dict): The current graph state
Returns:
state (dict): Updates question key with a re-phrased or re-written question
"""
print("---REWRITE QUERY---")
question = state["question"]
documents = state["documents"]
# Re-write question
better_question = question_rewriter.invoke({"question": question})
return {"documents": documents, "question": better_question}
from langchain.schema import Document
def web_search(state):
"""
Web search based on the re-written question.
Args:
state (dict): The current graph state
Returns:
state (dict): Updates documents key with appended web results
"""
print("---WEB SEARCH---")
question = state["question"]
documents = state["documents"]
# Web search
docs = tv_search.invoke(question)
web_results = "\n\n".join([d["content"] for d in docs])
web_results = Document(page_content=web_results)
documents.append(web_results)
return {"documents": documents, "question": question}
generation 字段，以便在智能体图中的任何时候都可以访问它，并根据需要将响应输出给用户。def generate_answer(state):
"""
Generate answer from context document using LLM
Args:
state (dict): The current graph state
Returns:
state (dict): New key added to state, generation, that contains LLM generation
"""
print("---GENERATE ANSWER---")
question = state["question"]
documents = state["documents"]
# RAG generation
generation = qa_rag_chain.invoke({"context": documents, "question": question})
return {"documents": documents, "question": question,
"generation": generation}
web_search_needed 标志，并决定是进行网络搜索还是生成响应。它将返回要调用的函数名称。如果需要进行网络搜索，系统将返回 rewrite_query 字符串，促使我们的智能 RAG 系统遵循查询重写、搜索和响应生成的流程。如果不需要进行网络搜索，该函数将返回 generate_answer 字符串，使我们的 RAG 系统进入从给定上下文文档和查询生成响应的常规流程。你将在我们智能体图的条件节点中使用这个函数，根据两种可能的路径将流程路由到正确的函数。def decide_to_generate(state):
"""
Determines whether to generate an answer, or re-generate a question.
Args:
state (dict): The current graph state
Returns:
str: Binary decision for next node to call
"""
print("---ASSESS GRADED DOCUMENTS---")
web_search_needed = state["web_search_needed"]
if web_search_needed == "Yes":
# All documents have been filtered check_relevance
# We will re-generate a new query
print("---DECISION: SOME or ALL DOCUMENTS ARE NOT RELEVANT TO QUESTION, REWRITE QUERY---")
return "rewrite_query"
else:
# We have relevant documents, so generate answer
print("---DECISION: GENERATE RESPONSE---")
return "generate_answer"
我们使用 LangGraph 并根据上一节实现的函数将智能体构建为一个图，按照我们的智能 RAG 系统架构将它们放入相关节点，并根据定义的工作流程用相关边连接它们。
from langgraph.graph import END, StateGraph
agentic_rag = StateGraph(GraphState)
# Define the nodes
agentic_rag.add_node("retrieve", retrieve) # retrieve
agentic_rag.add_node("grade_documents", grade_documents) # grade documents
agentic_rag.add_node("rewrite_query", rewrite_query) # transform_query
agentic_rag.add_node("web_search", web_search) # web search
agentic_rag.add_node("generate_answer", generate_answer) # generate answer
# Build graph
agentic_rag.set_entry_point("retrieve")
agentic_rag.add_edge("retrieve", "grade_documents")
agentic_rag.add_conditional_edges(
"grade_documents",
decide_to_generate,
{"rewrite_query": "rewrite_query", "generate_answer": "generate_answer"},
)
agentic_rag.add_edge("rewrite_query", "web_search")
agentic_rag.add_edge("web_search", "generate_answer")
agentic_rag.add_edge("generate_answer", END)
# Compile
agentic_rag = agentic_rag.compile()
我们可以使用以下代码可视化我们的智能 RAG 系统工作流程：
from IPython.display import Image, display, Markdown
display(Image(agentic_rag.get_graph().draw_mermaid_png()))
最后，我们准备在一些用户查询上实时测试我们的智能 RAG 系统！由于我们在图节点的相关函数中放置了打印语句，因此在图执行时我们可以看到它们的打印输出。
query = "what is the capital of India?"
response = agentic_rag.invoke({"question": query})
输出如下：
---RETRIEVAL FROM VECTOR DB---
---CHECK DOCUMENT RELEVANCE TO QUESTION---
---GRADE: DOCUMENT RELEVANT---
---GRADE: DOCUMENT NOT RELEVANT---
---GRADE: DOCUMENT NOT RELEVANT---
---ASSESS GRADED DOCUMENTS---
---DECISION: SOME or ALL DOCUMENTS ARE NOT RELEVANT TO QUESTION, REWRITE QUERY---
---REWRITE QUERY---
---WEB SEARCH---
---GENERATE ANSWER—
可以看到，从向量数据库中检索到的一些文档不相关，因此它成功地从网络上检索了上下文信息并生成了响应。我们现在可以查看生成的响应：
display(Markdown(response['generation']))
输出：
The capital city of India is New Delhi. It is a union territory within the
larger metropolitan area of Delhi and is situated in the north-central part
of the country on the west bank of the Yamuna River. New Delhi was formally
dedicated as the capital in 1931 and has a population of about 9.4 million
people.
query = "who won the champions league in 2024?"
response = agentic_rag.invoke({"question": query})
输出：
---RETRIEVAL FROM VECTOR DB---
---CHECK DOCUMENT RELEVANCE TO QUESTION---
---GRADE: DOCUMENT NOT RELEVANT---
---ASSESS GRADED DOCUMENTS---
---DECISION: SOME or ALL DOCUMENTS ARE NOT RELEVANT TO QUESTION, REWRITE QUERY---
---REWRITE QUERY---
---WEB SEARCH---
---GENERATE ANSWER---
系统似乎按预期工作，由于没有上下文文档，它使用网络搜索工具从网络上检索新信息来生成对我们查询的响应。我们现在可以查看响应：
display(Markdown(response['generation']))
输出：
The winner of the 2024 UEFA Champions League was Real Madrid. They secured
victory in the final against Borussia Dortmund with goals from Dani Carvajal
and Vinicius Junior.
query = "Tell me about India"
response = agentic_rag.invoke({"question": query})
输出：
---RETRIEVAL FROM VECTOR DB---
---CHECK DOCUMENT RELEVANCE TO QUESTION---
---GRADE: DOCUMENT RELEVANT---
---GRADE: DOCUMENT RELEVANT---
---GRADE: DOCUMENT RELEVANT---
---ASSESS GRADED DOCUMENTS---
---DECISION: GENERATE RESPONSE---
---GENERATE ANSWER—
我们的智能 RAG 系统运行良好，在这种情况下，由于所有检索到的文档都与回答用户问题相关，所以它没有进行网络搜索。我们现在可以查看响应：
display(Markdown(response['generation']))
输出：
India is a country located in Asia, specifically at the center of South Asia.
It is the seventh largest country in the world by area and the largest in
South Asia. . . . . . .
India has a rich and diverse history that spans thousands of years,
encompassing various languages, cultures, periods, and dynasties. The
civilization began in the Indus Valley, . . . . . .
在本指南中，我们深入了解了传统 RAG 系统当前面临的挑战、AI 智能体的作用和重要性，以及智能 RAG 系统如何应对其中的一些挑战。我们详细讨论了受《Corrective Retrieval Augmented Generation》论文启发的智能纠正性 RAG 系统的详细系统架构和工作流程。最后但同样重要的是，我们使用 LangGraph 实现了这个智能 RAG 系统，并在各种场景下进行了测试。你可以通过这个 Colab 笔记本轻松访问代码，并尝试通过添加更多功能（如额外的幻觉检查等）来改进这个系统。
如果你想深入学习如何构建这样的智能 AI 系统，可以参加 GenAI Pinnacle Program。在这里，你将通过与生成式 AI 专家的一对一指导、超过 200 小时的强化学习的高级课程以及对 26 多种 GenAI 工具和库的掌握，彻底革新你的 AI 学习和开发之旅。提升你的技能，成为 AI 领域的领导者。
希望本文能为你在 RAG 系统和 AI 智能体的探索中提供有价值的参考，让我们一起期待在人工智能领域不断创新和突破，构建更智能、更高效的系统，为各个行业带来变革性的影响 。无论是企业应用场景，还是学术研究方向，对这些技术的深入理解和应用都将为我们打开新的大门，创造更多可能。
53AI，企业落地大模型首选服务商
产品：场景落地咨询+大模型应用平台+行业解决方案
承诺：免费场景POC验证，效果验证后签署服务协议。零风险落地应用大模型，已交付160+中大型企业
2024-10-27
2024-09-04
2024-07-18
2024-05-05
2024-06-20
2024-06-13
2024-07-09
2024-07-09
2024-05-19
2024-07-07
回到顶部