在快速发展的人工智能领域，Agentic RAG 已成为一种改变游戏规则的信息检索和生成方法。这种先进的技术结合了检索增强生成（RAG）与自主代理的力量，提供了一种更动态、更能感知上下文的信息处理和生成方法。随着企业和研究人员寻求增强人工智能能力，了解和实施代理 RAG 已成为在竞争中保持领先的关键。

本文利用两个强大的工具深入探讨了掌握 Agentic RAG 的复杂性： LangChain 和 CrewAI。文章探讨了从传统 RAG 到代理 RAG 的演变过程，强调了其中的主要区别和优势。文章还探讨了 LangChain 如何作为实施代理 RAG 的基础，并演示了如何利用 CrewAI 创建更复杂、更高效的人工智能系统。

RAG 的演变：从传统到代理

传统 RAG 的局限性

传统的检索增强生成（RAG）系统通过将大型语言模型（LLM）与向量数据库相结合，克服了现成 LLM 的局限性，从而彻底改变了人工智能。然而，这些系统在多任务处理时面临挑战，不适合复杂的使用案例。在构建简单的问答聊天机器人、支持机器人等之前，这种方法还可以，但一旦事情变得有点复杂，传统的 RAG 方法就会失效。它们往往难以将检索到的数据与上下文联系起来，导致肤浅的回复可能无法完全解决查询的细微差别。

Agentic RAG 简介

代理 RAG 是传统 RAG 的一种演变，它整合了人工智能代理来增强 RAG 方法。这种方法采用自主代理来分析初步发现，并战略性地选择有效的数据检索工具。这些人工智能代理有能力将复杂的任务分解成几个子任务，使其变得易于处理。它们还拥有记忆（如聊天记录），因此知道发生了什么以及需要进一步采取什么步骤。此外，这些人工智能代理非常聪明，只要有解决任务的需求，它们就能调用任何应用程序接口或工具。代理可以提出逻辑、推理并采取相应的行动。这就是代理 RAG 方法的突出之处。该系统将复杂的查询分解为易于管理的部分，为每个部分分配特定的代理，同时保持无缝协调。

代理 RAG 的主要优势和使用案例

与传统系统相比，Agentic RAG 具有众多优势。它的自主代理可以独立工作，并行高效地处理复杂的查询。系统的适应性使其能够根据新信息或不断变化的用户需求动态调整策略。在市场营销方面，Agentic RAG 可以分析客户数据，生成个性化通信，并提供实时竞争情报。它还能加强营销活动管理中的决策制定，改进搜索引擎优化策略。

LangChain： Agentic RAG 的支柱

LangChain 概览

LangChain 已成为构建大型语言模型（LLM）应用程序的强大框架，其功能呈指数级增长。与其他框架相比，它是一种多功能工具，与各种平台的兼容性更强。LangChain 的核心是集成尖端技术，通过每次交互来提高模型性能。该框架采用模块化原则，在处理自然语言交互时具有灵活性和适应性。

Agentic RAG 的基本组件

LangChain 的架构支持短期和长期记忆功能，这对 Agentic RAG 系统至关重要。短期记忆利用上下文中的学习，而长期记忆则利用外部矢量存储来实现无限的信息保留和快速检索。这些组件使 LangChain 能够出色地理解对话中的上下文、语气和细微差别，从而实现更像人的交互。

将 LangChain 与外部工具集成

为了实现Agentic RAG，LangChain 可以与各种外部工具集成。这种集成引入了智能代理，它们可以随着时间的推移进行规划、推理和学习。系统通常包括负责问题解答和总结的文档代理，以及负责监督和协调其工作的元代理。这种分层结构增强了需要战略规划和细微决策的任务的能力，将代理的性能提升到新的高度。

利用 CrewAI 实现高级Agentic RAG

CrewAI 简介

CrewAI 是一个开源框架，旨在创建和管理智能代理团队。与传统的聊天机器人不同，这些代理可以协作和共享信息，共同完成复杂的任务。CrewAI 是一个复杂的平台，可帮助企业有效地组织人工智能运营，模拟软件开发团队的角色和职责。

实施多代理工作流程

CrewAI 允许用户为代理定义任务、角色、目标和背景故事，从而促进多代理工作流程。这种方法可提高组织内的生产力、决策流程和产品设计。该框架支持各种协作模式，包括顺序、分层和异步工作流。利用 CrewAI，团队可以简化操作，并通过协调努力实现效率最大化。

优化代理互动和决策

CrewAI 通过角色扮演、重点维护和工具利用等功能优化代理互动。该平台纳入了安全措施和协议的防护栏，确保可靠和合乎道德的操作。记忆功能使代理能够存储和回忆过去的互动，从而增强决策过程。通过将 CrewAI 与 Groq 的 Llama3-70B 等高级语言模型集成，企业可以进一步改进内容生成和任务执行。

Agentic RAG 工作流程教程

我们将了解代理如何参与 RAG 系统，通过调用工具检索最相关的信息。

我将使用 SingleStore Notebooks（就像 Google colab 或 Jupyter Notebooks，但具有更多功能）来运行我的代码。你也可以使用同样的方法。SingleStore 有一个免费共享层，你可以免费注册并开始使用服务。

创建好自己的 SingleStore 笔记本后，让我们继续添加下面的代码并逐步运行它。

安装所需的程序库

!pip install crewai==0.28.8 crewai_tools==0.1.6 langchain_community==0.0.29 sentence-transformers langchain-groq --quiet

from langchain_openai import ChatOpenAI
import os
from crewai_tools import PDFSearchTool
from langchain_community.tools.tavily_search import TavilySearchResults
from crewai_tools  import tool
from crewai import Crew
from crewai import Task
from crewai import Agent

提及 Groq API 密钥

import os
# Set the API key
os.environ['GROQ_API_KEY'] = 'Add Your Groq API Key'

提及正在使用的 LLM

llm = ChatOpenAI(
    openai_api_base="https://api.groq.com/openai/v1",
    openai_api_key=os.environ['GROQ_API_KEY'],
    model_name="llama3-8b-8192",
    temperature=0.1,
    max_tokens=1000,
)

加载你想使用的数据/自定义数据。我使用的是一份关于 “关注就是你所需要的一切 ”的公开 PDF 文件。

import requests
pdf_url = 'https://proceedings.neurips.cc/paper_files/paper/2017/file/3f5ee243547dee91fbd053c1c4a845aa-Paper.pdf'
response = requests.get(pdf_url)
with open('attenstion_is_all_you_need.pdf', 'wb') as file:
    file.write(response.content)

创建一个 RAG 工具变量来传递我们的 PDF 文件

rag_tool = PDFSearchTool(pdf='attenstion_is_all_you_need.pdf',
    config=dict(
        llm=dict(
            provider="groq", # or google, openai, anthropic, llama2, ...
            config=dict(
                model="llama3-8b-8192",
                # temperature=0.5,
                # top_p=1,
                # stream=true,
            ),
        ),
        embedder=dict(
            provider="huggingface", # or openai, ollama, ...
            config=dict(
                model="BAAI/bge-small-en-v1.5",
                #task_type="retrieval_document",
                # title="Embeddings",
            ),
        ),
    )
)

rag_tool.run("How did self-attention mechanism evolve in large language models?")

我们将使用 Tavily 作为我们的代理使用的工具。

Tavily 搜索 API 是专为 LLM 和 RAG 优化的搜索引擎，旨在提供高效、快速和持久的搜索结果。因此，让我们提及并设置 Tavily API 密钥。

import os
# Set the Tavily API key
os.environ['TAVILY_API_KEY'] = 'Add Your Tavily API Key'

web_search_tool = TavilySearchResults(k=3)

web_search_tool.run("What is self-attention mechansim in large language models?")

让我们定义一个工具

@tool
def router_tool(question):
  """Router Function"""
  if 'self-attention' in question:
    return 'vectorstore'
  else:
    return 'web_search'

创建合作代理

Router_Agent = Agent(
  role='Router',
  goal='Route user question to a vectorstore or web search',
  backstory=(
    "You are an expert at routing a user question to a vectorstore or web search."
    "Use the vectorstore for questions on concept related to Retrieval-Augmented Generation."
    "You do not need to be stringent with the keywords in the question related to these topics. Otherwise, use web-search."
  ),
  verbose=True,
  allow_delegation=False,
  llm=llm,
)

Retriever_Agent = Agent(
role="Retriever",
goal="Use the information retrieved from the vectorstore to answer the question",
backstory=(
    "You are an assistant for question-answering tasks."
    "Use the information present in the retrieved context to answer the question."
    "You have to provide a clear concise answer."
),
verbose=True,
allow_delegation=False,
llm=llm,
)

Grader_agent =  Agent(
  role='Answer Grader',
  goal='Filter out erroneous retrievals',
  backstory=(
    "You are a grader assessing relevance of a retrieved document to a user question."
    "If the document contains keywords related to the user question, grade it as relevant."
    "It does not need to be a stringent test.You have to make sure that the answer is relevant to the question."
  ),
  verbose=True,
  allow_delegation=False,
  llm=llm,
)

hallucination_grader = Agent(
    role="Hallucination Grader",
    goal="Filter out hallucination",
    backstory=(
        "You are a hallucination grader assessing whether an answer is grounded in / supported by a set of facts."
        "Make sure you meticulously review the answer and check if the response provided is in alignmnet with the question asked"
    ),
    verbose=True,
    allow_delegation=False,
    llm=llm,
)

answer_grader = Agent(
    role="Answer Grader",
    goal="Filter out hallucination from the answer.",
    backstory=(
        "You are a grader assessing whether an answer is useful to resolve a question."
        "Make sure you meticulously review the answer and check if it makes sense for the question asked"
        "If the answer is relevant generate a clear and concise response."
        "If the answer gnerated is not relevant then perform a websearch using 'web_search_tool'"
    ),
    verbose=True,
    allow_delegation=False,
    llm=llm,
)

为这些代理定义任务

router_task = Task(
    description=("Analyse the keywords in the question {question}"
    "Based on the keywords decide whether it is eligible for a vectorstore search or a web search."
    "Return a single word 'vectorstore' if it is eligible for vectorstore search."
    "Return a single word 'websearch' if it is eligible for web search."
    "Do not provide any other premable or explaination."
    ),
    expected_output=("Give a binary choice 'websearch' or 'vectorstore' based on the question"
    "Do not provide any other premable or explaination."),
    agent=Router_Agent,
    tools=[router_tool],
)

retriever_task = Task(
    description=("Based on the response from the router task extract information for the question {question} with the help of the respective tool."
    "Use the web_serach_tool to retrieve information from the web in case the router task output is 'websearch'."
    "Use the rag_tool to retrieve information from the vectorstore in case the router task output is 'vectorstore'."
    ),
    expected_output=("You should analyse the output of the 'router_task'"
    "If the response is 'websearch' then use the web_search_tool to retrieve information from the web."
    "If the response is 'vectorstore' then use the rag_tool to retrieve information from the vectorstore."
    "Return a claer and consise text as response."),
    agent=Retriever_Agent,
    context=[router_task],
   #tools=[retriever_tool],
)

grader_task = Task(
    description=("Based on the response from the retriever task for the quetion {question} evaluate whether the retrieved content is relevant to the question."
    ),
    expected_output=("Binary score 'yes' or 'no' score to indicate whether the document is relevant to the question"
    "You must answer 'yes' if the response from the 'retriever_task' is in alignment with the question asked."
    "You must answer 'no' if the response from the 'retriever_task' is not in alignment with the question asked."
    "Do not provide any preamble or explanations except for 'yes' or 'no'."),
    agent=Grader_agent,
    context=[retriever_task],
)

hallucination_task = Task(
    description=("Based on the response from the grader task for the quetion {question} evaluate whether the answer is grounded in / supported by a set of facts."),
    expected_output=("Binary score 'yes' or 'no' score to indicate whether the answer is sync with the question asked"
    "Respond 'yes' if the answer is in useful and contains fact about the question asked."
    "Respond 'no' if the answer is not useful and does not contains fact about the question asked."
    "Do not provide any preamble or explanations except for 'yes' or 'no'."),
    agent=hallucination_grader,
    context=[grader_task],
)
answer_task = Task(
    description=("Based on the response from the hallucination task for the quetion {question} evaluate whether the answer is useful to resolve the question."
    "If the answer is 'yes' return a clear and concise answer."
    "If the answer is 'no' then perform a 'websearch' and return the response"),
    expected_output=("Return a clear and concise response if the response from 'hallucination_task' is 'yes'."
    "Perform a web search using 'web_search_tool' and return ta clear and concise response only if the response from 'hallucination_task' is 'no'."
    "Otherwise respond as 'Sorry! unable to find a valid response'."),
    context=[hallucination_task],
    agent=answer_grader,
    #tools=[answer_grader_tool],
)

定义用例流程

rag_crew = Crew(
    agents=[Router_Agent, Retriever_Agent, Grader_agent, hallucination_grader, answer_grader],
    tasks=[router_task, retriever_task, grader_task, hallucination_task, answer_task],
    verbose=True,
)

提问

inputs ={"question":"How does self-attention mechanism help large language models?"}

启动管道

result = rag_crew.kickoff(inputs=inputs)

你应该看到这些代理的神奇作用，从而获得最终的输出/响应。

你可以通过运行来查看最终的正确响应

print(result)

Agentic RAG、LangChain 和 CrewAI 之旅展示了人工智能驱动的信息处理和生成方面的突破性进展。这些工具对企业和研究人员如何处理复杂查询和做出数据驱动型决策产生了重大影响。通过结合自主代理、灵活框架和协作式人工智能团队的优势，企业现在可以更高效、更准确地处理复杂的 RAG 相关任务。