PythonAI算法
Agentic RAG 技术实践：基于 LangChain 与 OpenAI
本文介绍了 Agentic RAG 技术及其在复杂多文档问答中的应用。通过构建文档智能体与元智能体架构，结合 LangChain、OpenAI 及 FAISS 向量存储，实现了具备工具调用能力的检索增强生成系统。内容涵盖环境配置、RAG 链构建、LangGraph 工作流设计以及实际问答测试，展示了如何利用外部搜索与学术检索工具提升 LLM 回答的准确性与时效性。
本文介绍了 Agentic RAG 技术及其在复杂多文档问答中的应用。通过构建文档智能体与元智能体架构，结合 LangChain、OpenAI 及 FAISS 向量存储，实现了具备工具调用能力的检索增强生成系统。内容涵盖环境配置、RAG 链构建、LangGraph 工作流设计以及实际问答测试，展示了如何利用外部搜索与学术检索工具提升 LLM 回答的准确性与时效性。
Agentic RAG（智能体检索增强生成）是一种新兴的智能体驱动解决方案，能够应对复杂的多文档问答挑战。这项技术通过智能化手段，不仅提升了处理文本数据的效率，还极大地增强了问答系统的准确性和深度。与传统 RAG 相比，Agentic RAG 不完全依赖于大型语言模型直接生成答案，而是引入智能体来承担需要深思熟虑、逻辑推理、工具运用以及持续学习的任务。
Agentic RAG 是由智能体驱动的技术，能够灵活处理多文档问答任务。这项技术不仅能比较文档、总结内容，还能对多个摘要进行对比分析。通过智能化手段，提高了解决复杂问题的能力，让问答更加高效和精确。
基本架构是为每个文档设置一个文档智能体，每个文档智能体都能在其自己的文档内执行问答和摘要。然后，设置一个顶级智能体（元智能体）来管理所有这些较低级别的文档智能体。这种分层结构允许系统并行处理信息并进行综合决策。
实现 Agentic RAG 主要依赖以下核心组件：
安装所需的依赖项：
!pip install -qU langchain langchain_openai langgraph arxiv duckduckgo-search
!pip install -qU faiss-cpu pymupdf
设置环境变量（以 Google Colab 为例）：
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 并构建 LCEL RAG 链：
from operator import itemgetter
from langchain.schema.output_parser import StrOutputParser
from langchain.schema.runnable import RunnablePassthrough
from langchain_openai import ChatOpenAI
openai_chat_model = ChatOpenAI(model="gpt-3.5-turbo")
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")}
)
为了使智能体能够回答各种问题并引入外部知识，通常会给其配备一个工具箱。这里整合了 DuckDuckGo 网络搜索和 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 通过 StatefulGraph 来实现，这个结构使用 AgentState 对象在图的不同节点间传递信息。核心是 AgentState 对象，它被存储在一个 TypedDict 中，键为'messages'，值为 BaseMessage 的序列。
定义状态类：
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 负责调用工具。
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.set_entry_point("agent")
设置条件边进行路由：
def should_continue(state):
last_message = state["messages"][-1]
if "function_call" not in 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()
调用 LangGraph 进行提问，展示智能体的工具调用能力。
from langchain_core.messages import HumanMessage
inputs = {"messages": [HumanMessage(content="在大型语言模型的背景下，RAG 是什么？它是什么时候出现的？")}]
response = app.invoke(inputs)
print(response)
响应示例：
系统首先识别出需要搜索外部信息，调用 duckduckgo_search 工具获取关于 RAG 的定义和历史背景，随后结合搜索结果生成最终回答。例如，当询问论文作者时，智能体会自动调用 Arxiv 工具检索具体元数据。
Agentic RAG 通过将检索增强生成与智能体工作流相结合，显著提升了 LLM 在处理动态信息和复杂任务时的表现。利用 LangGraph 的状态管理机制，我们可以灵活地编排模型调用与工具执行，实现循环迭代和条件分支。本文展示了从基础 RAG 链到完整智能体工作流的构建过程，为开发具备自主规划能力的 AI 应用提供了参考方案。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online