Agentic RAG技术是一种新兴的智能体驱动解决方案，能够应对复杂的多文档问答挑战。这项技术通过智能化手段，不仅提升了我们处理文本数据的效率，还极大地增强了问答系统的准确性和深度。本文带大家来了解这项技术，阐述其如何提升信息检索和分析的效率与准确性。

1 Agentic RAG简介

Agentic RAG是由智能体驱动的技术，能够灵活处理多文档问答任务。这项技术不仅能比较文档、总结内容，还能对多个摘要进行对比分析。

正因如此，我们不完全依赖于大型语言模型，而是通过智能体来承担那些需要深思熟虑、逻辑推理、工具运用以及持续学习的任务。Agentic RAG通过智能化手段，提高了解决复杂问题的能力，让问答更加高效和精确。

2 基本架构

基本架构是为每个文档设置一个文档智能体，每个文档智能体都能在其自己的文档内执行问答和摘要。

然后，设置一个顶级智能体（元智能体）来管理所有这些较低级别的文档智能体。

3 技术栈

技术栈包括以下几个核心组件：

1. Langchain：特别是LCEL，这是用于构建大型语言模型（LLM）应用的编排框架。
2. OpenAI：提供我们需要的大型语言模型（LLM）。
3. FAISS-cpu：用作向量存储解决方案。
4. 数据源：我们使用ArxivLoader来检索arXiv上发表文章的元数据。

4 代码实现

安装所需的依赖项：

!pip install -qU langchain langchain_openai langgraph arxiv duckduckgo-search
!pip install -qU faiss-cpu pymupdf

设置环境变量：

from google.colab import userdata
from uuid import uuid4
import os
#
os.environ['OPENAI_API_KEY'] = userdata.get('OPENAI_API_KEY')
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = f"AIE1 - LangGraph - {uuid4().hex[0:8]}"
os.environ["LANGCHAIN_API_KEY"] = userdata.get('LANGCHAIN_API_KEY')

使用LCEL实例化一个简单的检索链。

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import ArxivLoader
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

# 加载特定主题的文档
docs = ArxivLoader(query="Retrieval Augmented Generation", load_max_docs=5).load()

# 将文档分割成更小的块
text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=350, chunk_overlap=50
)

chunked_documents = text_splitter.split_documents(docs)
#
# 实例化嵌入模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-small",openai_api_key=os.environ['OPENAI_API_KEY'])
# 创建索引-将文档块加载到向量存储中
faiss_vectorstore = FAISS.from_documents(
    documents=chunked_documents,
    embedding=embeddings,
)
# 创建检索器
retriever = faiss_vectorstore.as_retriever()

生成RAG提示：

from langchain_core.prompts import ChatPromptTemplate

RAG_PROMPT = """\
使用以下上下文回答用户的查询。如果你不能回答问题，请回答“我不知道”。

问题：
{question}

上下文：
{context}
"""

rag_prompt = ChatPromptTemplate.from_template(RAG_PROMPT)

实例化LLM：

from langchain_openai import ChatOpenAI

openai_chat_model = ChatOpenAI(model="gpt-3.5-turbo")

构建LCEL RAG链：

from operator import itemgetter
from langchain.schema.output_parser import StrOutputParser
from langchain.schema.runnable import RunnablePassthrough

retrieval_augmented_generation_chain = (
       {"context": itemgetter("question") 
    | retriever, "question": itemgetter("question")}
    | RunnablePassthrough.assign(context=itemgetter("context"))
    | {"response": rag_prompt | openai_chat_model, "context": itemgetter("context")}
)
#
retrieval_augmented_generation_chain

响应：

{
  "response": "AIMessage(content='Retrieval Augmented Generation 是一种结合了深度学习技术和传统检索技术的文本生成范式。它通过显式获取知识以插件方式，引领了许多NLP任务的可扩展性和可能缓解文本生成的难度。它涉及从检索到的人类编写的参考资料中生成文本，而不是从头开始生成。', response_metadata={'token_usage': {'completion_tokens': 73, 'prompt_tokens': 2186, 'total_tokens': 2259}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': 'fp_b28b39ffa8', 'finish_reason': 'stop', 'logprobs': None})",
  "context": "[Document(page_content='...'), Document(page_content='...'), Document(page_content='...')]"
}

构建工具箱：

为了使智能体能够回答各种问题并引入外部知识，通常会给其配备一个工具箱。

LangChain社区的存储库（https://github.com/langchain-ai/langchain/tree/master/libs/community/langchain_community/tools）中提供了众多工具，这里选择了几种来使用，这样可以更直观地展示LangGraph的循环特性。

具体来说，将整合以下工具：

• Duck Duck Go网络搜索（https://github.com/langchain-ai/langchain/tree/master/libs/community/langchain_community/tools/ddg_search）
• Arxiv学术文献检索（https://github.com/langchain-ai/langchain/tree/master/libs/community/langchain_community/tools/arxiv)）

from langchain_community.tools.ddg_search import DuckDuckGoSearchRun
from langchain_community.tools.arxiv.tool import ArxivQueryRun
from langgraph.prebuilt import ToolExecutor
tool_belt = [
    DuckDuckGoSearchRun(),
    ArxivQueryRun()
]

tool_executor = ToolExecutor(tool_belt)

实例化OpenAI函数调用：

from langchain_core.utils.function_calling import convert_to_openai_function
#
model = ChatOpenAI(temperature=0)
#
functions = [convert_to_openai_function(t) for t in tool_belt]
model = model.bind_functions(functions)

使用LangGraph：

LangGraph通过StatefulGraph来实现，这个结构使用AgentState对象在图的不同节点间传递信息。

虽然有很多配置选项，但核心是AgentState对象，它被存储在一个TypedDict中。这个对象的键是“messages”，对应的值是一个BaseMessages的序列，每当状态发生变化时，新的消息就会被添加到这个序列中。

from typing import TypedDict, Annotated, Sequence
import operator
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
  messages: Annotated[Sequence[BaseMessage], operator.add]

构建节点

call_model是一个节点，将...嗯...调用模型

call_tool是一个节点，将调用工具

```python
from langgraph.prebuilt import ToolInvocation
import json
from langchain_core.messages import FunctionMessage

def call_model(state):
  messages = state["messages"]
  response = model.invoke(messages)
return {"messages" : [response]}

def call_tool(state):
  last_message = state["messages"][-1]

  action = ToolInvocation(
      tool=last_message.additional_kwargs["function_call"]["name"],
      tool_input=json.loads(
          last_message.additional_kwargs["function_call"]["arguments"]
      )
  )

  response = tool_executor.invoke(action)

  function_message = FunctionMessage(content=str(response), name=action.tool)

return {"messages" : [function_message]}

构建工作流：

from langgraph.graph import StateGraph, END

workflow = StateGraph(AgentState)

workflow.add_node("agent", call_model)
workflow.add_node("action", call_tool)
workflow.nodes

设置入口点：

workflow.set_entry_point("agent")

构建条件边进行路由：

def should_continue(state):
  last_message = state["messages"][-1]

if"function_call"notin last_message.additional_kwargs:
    return"end"

return"continue"

workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue" : "action",
        "end" : END
    }
)

最后将条件边连接到智能体节点和动作节点：

workflow.add_edge("action", "agent")

编译工作流：

app = workflow.compile()
#
app

调用LangGraph - 提问：

from langchain_core.messages import HumanMessage

inputs = {"messages" : [HumanMessage(content="在大型语言模型的背景下，RAG是什么？它是什么时候出现的？")]}
response = app.invoke(inputs)
print(response)

响应：

{
  "messages": [
    HumanMessage(content="在大型语言模型的背景下，RAG是什么？它是什么时候出现的？"),
    AIMessage(content='', additional_kwargs={'function_call': {'arguments': '{"query":"RAG in the context of Large Language Models"}', 'name': 'duckduckgo_search'}}, response_metadata={'token_usage': {'completion_tokens': 25, 'prompt_tokens': 171, 'total_tokens': 196}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': 'fp_b28b39ffa8', 'finish_reason': 'function_call', 'logprobs': None}),
    
FunctionMessage(content="大型语言模型（LLMs）是处理和生成文本的非常强大的工具。然而，它们天生难以理解更广泛的信息背景，特别是在处理长篇对话或复杂任务时。这就是大型上下文窗口和检索增强生成（RAG）发挥作用的地方。这些先进的、泛化的语言模型是在庞大的数据集上训练的，使它们能够理解和生成类似人类的文本。在RAG的背景下，LLMs被用来基于用户查询和从向量数据库检索的上下文信息生成完整的响应。在语言模型快速发展的背景下，检索增强生成（RAG）和长上下文大型语言模型（LLMs）之间的辩论引起了广泛关注。检索增强生成（RAG）是一个AI框架，通过将模型基于外部知识源来补充LLM的内部信息表示，从而提高LLM生成响应的质量。在基于LLM的问答系统中实施RAG有两个主要好处：它确保模型有... RAG代表检索增强生成。RAG使大型语言模型（LLM）能够访问和利用最新的信息。因此，它提高了LLM的响应质量和相关性。下面是一个简单的RAG实现图。", name='duckduckgo_search'),
    AIMessage(content="在大型语言模型（LLMs）的背景下，RAG代表检索增强生成。它是一个AI框架，通过将模型基于外部知识源来补充LLM的内部信息表示，从而提高LLM生成响应的质量。RAG使LLM能够访问和利用最新的信息，从而提高模型生成的响应的相关性和质量。RAG作为增强LLM在理解和生成类似人类文本的能力的方式，在语言模型快速发展的领域中出现。", response_metadata={'token_usage': {'completion_tokens': 117, 'prompt_tokens': 491, 'total_tokens': 608}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': 'fp_3bc1b5746c', 'finish_reason': 'stop', 'logprobs': None})
  ]
}

响应内容：

RAG代表检索增强生成，在大型语言模型（LLMs）的背景下。它是一个AI框架，通过将模型基于外部知识源来补充LLM的内部信息表示，从而提高LLM生成响应的质量。RAG使LLM能够访问和利用最新的信息，从而提高模型生成的响应的相关性和质量。RAG作为增强LLM在理解和生成类似人类文本的能力的方式，在语言模型快速发展的领域中出现。

提问：

问题 = "检索增强生成论文的主要作者是谁 - 他们参加了哪所大学？"

inputs = {"messages" : [HumanMessage(content=问题)]}

response = app.invoke(inputs) print(response['messages'][-1].content)

**响应：**

关于“检索增强生成”论文的主要作者是Huayang Li。不幸的是，提供的摘要中没有提到他们参加的大学。

**提问：**

问题 = "检索增强生成论文的主要作者是谁？"

inputs = {"messages" : [HumanMessage(content=问题)]}

response = app.invoke(inputs)
print(response['messages'][-1].content)

响应：论文“检索增强文本生成”的主要作者是Huayang Li、Yixuan Su、Deng Cai、Yan Wang和Lemao Liu