LangChain部署RAG part2.搭建多模态RAG引擎（赋范大模型社区公开课听课笔记）

从零到一搭建多模态RAG引擎：代码解析与实战指南

一、多模态RAG系统核心技术栈

在开始编码前，需先明确整个系统的技术选型。本项目围绕“PDF解析→结构化转换→向量检索→智能问答”全流程设计，核心工具链如下：

技术模块	核心工具	作用
文档解析与预处理	Unstructured + PaddleOCR	提取PDF中的标题、段落、表格、图片等元素，支持中英文OCR
PDF渲染与图片处理	PyMuPDF（fitz）、Pillow	读取PDF页面、提取图片并转换色彩空间（如CMYK转RGB）
结构化转换	html2text、Markdown处理	将表格HTML转为Markdown，生成结构化中间文档
向量数据库	FAISS + OpenAI Embeddings	存储文本向量，实现高效相似性检索
智能体开发	LangChain + LangGraph	构建多节点工作流，实现“检索-评估-回答”自动化逻辑
前端交互	Agent Chat UI	提供可视化对话界面，支持多模态结果展示

二、环境搭建：从依赖安装到配置

环境准备是项目开发的基础，需确保Python版本兼容性与依赖包正确性，以下是详细步骤：

1. 创建虚拟环境

推荐使用Python 3.9~3.11版本（本文以3.10为例），通过conda或venv创建独立虚拟环境，避免依赖冲突：

# 方式1：使用conda创建
conda create -n pdf_rag python=3.10 -y
conda activate pdf_rag# 方式2：使用venv创建
python -m venv pdf_rag
pdf_rag\Scripts\activate  # Windows系统
# source pdf_rag/bin/activate  # Linux/Mac系统

2. 安装核心依赖

通过pip安装文档解析、OCR、向量处理等所需依赖，可指定华为/清华镜像源加速下载：

# 1. 文档解析核心库（支持PDF/Word/PPT等）
pip install "unstructured[all-docs]" --index-url https://mirrors.huaweicloud.com/repository/pypi/simple# 2. OCR引擎（PaddleOCR比Tesseract更优，支持中英文）
pip install paddlenlp paddleocr# 3. PDF与图片处理库
pip install PyMuPDF pillow matplotlib# 4. 结构化转换与向量数据库
pip install html2text faiss-cpu langchain-text-splitters# 5. LangChain生态（智能体开发）
pip install langchain-core langchain-community langchain-openai langgraph langsmith# 6. 环境变量与工具调用
pip install python-dotenv langchain-tavily

3. 配置环境变量

创建.env文件，存储API密钥（如OpenAI、DeepSeek）与LangSmith追踪配置，保障敏感信息安全：

# .env文件内容
DEEPSEEK_API_KEY=sk-c1a253xxxxxx  # 替换为你的DeepSeek API密钥
OPENAI_API_KEY=sk-proj-gExxxxxx   # 替换为你的OpenAI API密钥
LANGSMITH_TRACING=true            # 开启LangSmith流程追踪
LANGSMITH_API_KEY=lsv2_pt_b44xxxx # 替换为你的LangSmith API密钥
LANGSMITH_PROJECT=langraph_studio_chatbot  # LangSmith项目名
TAVILY_API_KEY=tvly-27xxxxxx      # 可选，用于Web搜索工具

三、PDF文档解析：从非结构化到结构化

多模态RAG的核心第一步是将PDF中的多类型元素（文本、表格、图片）提取并结构化。本项目采用“Unstructured + PaddleOCR”方案，实现高精度解析与重建，具体代码如下：

1. 提取PDF元素（文本、表格、图片元数据）

使用UnstructuredLoader加载PDF，通过hi_res模式（高分辨率OCR）处理复杂排版，同时开启表格结构检测：

from langchain_unstructured import UnstructuredLoader# 1. 配置PDF路径与解析参数
file_path = "0.LangChain技术生态介绍.pdf"  # 你的PDF文件路径
loader_local = UnstructuredLoader(file_path=file_path,strategy="hi_res",               # 高分辨率模式，适合复杂文档infer_table_structure=True,      # 自动解析表格结构ocr_languages="chi_sim+eng",     # 支持中英文OCRocr_engine="paddleocr"           # 指定PaddleOCR引擎
)# 2. 逐页加载并提取元素（返回LangChain Document对象列表）
docs_local = []
for doc in loader_local.lazy_load():docs_local.append(doc)# 3. 查看解析结果（每个doc包含文本内容与元数据）
print("解析元素类型：", [doc.metadata["category"] for doc in docs_local[:5]])
print("第1个元素内容：", docs_local[0].page_content)
print("第1个元素元数据（页码、坐标）：", docs_local[0].metadata)

• 关键说明：docs_local中的每个Document对象包含page_content（文本内容）与metadata（页码、元素类型、坐标、置信度等），为后续结构化提供基础。

2. 可视化解析结果（可选）

为验证解析准确性，可通过PyMuPDF渲染PDF页面，并绘制元素边界框（标题用紫色、表格用红色、图片用绿色）：

import fitz
import matplotlib.patches as patches
import matplotlib.pyplot as plt
from PIL import Imagedef plot_pdf_with_boxes(pdf_page, segments):"""绘制PDF页面与元素边界框"""# 1. 将PDF页面转为PIL图片pix = pdf_page.get_pixmap()pil_image = Image.frombytes("RGB", [pix.width, pix.height], pix.samples)# 2. 初始化绘图fig, ax = plt.subplots(1, figsize=(10, 10))ax.imshow(pil_image)# 3. 定义元素颜色映射category_to_color = {"Title": "orchid",       # 标题：紫色"Image": "forestgreen",  # 图片：绿色"Table": "tomato"        # 表格：红色}categories = set()# 4. 遍历元素，绘制边界框for segment in segments:# 坐标缩放：将PDF逻辑坐标转为图片像素坐标points = segment["coordinates"]["points"]layout_width = segment["coordinates"]["layout_width"]layout_height = segment["coordinates"]["layout_height"]scaled_points = [(x * pix.width / layout_width, y * pix.height / layout_height)for x, y in points]# 确定颜色（未定义类型默认蓝色）box_color = category_to_color.get(segment["category"], "deepskyblue")categories.add(segment["category"])# 绘制多边形框（通常为矩形）rect = patches.Polygon(scaled_points, linewidth=1, edgecolor=box_color, facecolor="none")ax.add_patch(rect)# 5. 添加图例与隐藏坐标轴legend_handles = [patches.Patch(color="deepskyblue", label="Text")]for category in ["Title", "Image", "Table"]:if category in categories:legend_handles.append(patches.Patch(color=category_to_color[category], label=category))ax.axis("off")ax.legend(handles=legend_handles, loc="upper right")plt.tight_layout()plt.show()def render_page(doc_list: list, page_number: int, print_text=True):"""渲染指定页码的PDF与元素"""# 1. 打开PDF并加载指定页面pdf_page = fitz.open(file_path).load_page(page_number - 1)  # 页码从0开始# 2. 筛选该页面的所有元素page_docs = [doc for doc in doc_list if doc.metadata.get("page_number") == page_number]segments = [doc.metadata for doc in page_docs]# 3. 绘制与打印文本plot_pdf_with_boxes(pdf_page, segments)if print_text:for doc in page_docs:print(f"【{doc.metadata['category']}】{doc.page_content}\n")# 调用函数，渲染第1页
render_page(docs_local, page_number=1)

3. PDF逆向转化为Markdown

将解析后的元素组装为Markdown文档（保留标题层级、表格结构、图片引用），生成可直接用于RAG的结构化数据：

import os
import fitz
from unstructured.partition.pdf import partition_pdf
from html2text import html2text# 1. 配置路径
pdf_path = "0.LangChain技术生态介绍.pdf"
output_dir = "pdf_images"  # 存储提取的图片
os.makedirs(output_dir, exist_ok=True)  # 不存在则创建文件夹# 2. 提取PDF元素（文本、表格、图片元数据）
elements = partition_pdf(filename=pdf_path,infer_table_structure=True,strategy="hi_res",ocr_languages="chi_sim+eng",ocr_engine="paddleocr"
)# 3. 提取图片并保存（处理CMYK转RGB）
doc = fitz.open(pdf_path)
image_map = {}  # 映射：页码 -> 图片路径列表
for page_num, page in enumerate(doc, start=1):image_map[page_num] = []# 遍历页面中的所有图片for img_index, img in enumerate(page.get_images(full=True), start=1):xref = img[0]  # 图片引用IDpix = fitz.Pixmap(doc, xref)# 图片保存路径（按页码+索引命名）img_path = os.path.join(output_dir, f"page{page_num}_img{img_index}.png")# 处理色彩空间：CMYK转RGBif pix.n < 5:  # RGB/Gray模式pix.save(img_path)else:  # CMYK模式pix = fitz.Pixmap(fitz.csRGB, pix)pix.save(img_path)image_map[page_num].append(img_path)# 4. 组装Markdown内容
md_lines = []
inserted_images = set()  # 避免重复插入图片for el in elements:cat = el.category  # 元素类型text = el.text     # 元素文本page_num = el.metadata.page_number  # 元素所在页码# 处理标题（一级标题#，二级标题##）if cat == "Title" and text.strip().startswith("- "):md_lines.append(text + "\n")  # 列表项标题，保持原样elif cat == "Title":md_lines.append(f"# {text}\n")elif cat in ["Header", "Subheader"]:md_lines.append(f"## {text}\n")# 处理表格（HTML转Markdown）elif cat == "Table":if hasattr(el.metadata, "text_as_html") and el.metadata.text_as_html:md_lines.append(html2text(el.metadata.text_as_html) + "\n")else:md_lines.append(el.text + "\n")# 处理图片（引用本地保存的图片）elif cat == "Image":for img_path in image_map.get(page_num, []):if img_path not in inserted_images:md_lines.append(f"![Image](./{img_path})\n")inserted_images.add(img_path)# 处理普通文本（段落、列表等）else:md_lines.append(text + "\n")# 5. 写入Markdown文件
output_md = "LangChain技术生态介绍【逆向转化版】.md"
with open(output_md, "w", encoding="utf-8") as f:f.write("\n".join(md_lines))print(f"✅ 转换完成！生成文件：{output_md}，图片保存至：{output_dir}")

四、向量数据库构建：文本切片与嵌入

结构化的Markdown文档需进一步切片为“语义片段”，并通过嵌入模型转为向量，存储到FAISS数据库中，为后续检索提供支持：

1. 文本切片（Markdown标题分层）

使用MarkdownHeaderTextSplitter按标题层级切片（如一级标题#、二级标题##），避免切断语义逻辑：

from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import MarkdownHeaderTextSplitter
from langchain_community.vectorstores import FAISS
import os# 1. 加载环境变量与嵌入模型
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
embed = OpenAIEmbeddings(api_key=OPENAI_API_KEY,base_url="https://ai.devtool.tech/proxy/v1",  # 可选，代理地址model="text-embedding-3-small"  # 轻量型嵌入模型，性价比高
)# 2. 读取Markdown文件
md_path = "LangChain技术生态介绍【逆向转化版】.md"
with open(md_path, "r", encoding="utf-8") as f:md_content = f.read()# 3. 按标题分层切片
headers_to_split_on = [("#", "Header 1"),   # 一级标题：#("##", "Header 2")   # 二级标题：##
]
markdown_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)
md_splits = markdown_splitter.split_text(md_content)  # 切片后的Document列表# 4. 查看切片结果
print(f"切片总数：{len(md_splits)}")
print(f"前3个切片标题：")
for i, split in enumerate(md_splits[:3]):print(f"  {i+1}. {split.metadata.get('Header 1', '')} - {split.metadata.get('Header 2', '')}")

2. 向量存储与本地保存

将切片后的文本转为向量，存储到FAISS数据库，并保存到本地（便于后续加载复用）：

# 1. 构建FAISS向量库
vector_store = FAISS.from_documents(md_splits, embedding=embed)# 2. 本地保存向量库（文件夹形式）
vs_save_path = "langchain_course_db"
vector_store.save_local(vs_save_path)
print(f"✅ 向量库已保存至：{vs_save_path}")# 3. 加载向量库（后续使用时）
loaded_vector_store = FAISS.load_local(folder_path=vs_save_path,embeddings=embed,allow_dangerous_deserialization=True  # 允许反序列化（本地文件安全时使用）
)
print("✅ 向量库加载成功，可用于检索！")

五、Agentic RAG系统开发：基于LangGraph的智能问答

传统RAG仅能“检索→回答”，而Agentic RAG通过多节点工作流（如“检索评估→问题改写→重新检索”）提升问答准确性。本项目基于LangGraph构建智能体，实现端到端多模态问答：

1. 核心代码：LangGraph工作流定义

from __future__ import annotations
import os
import asyncio
from typing import Literal
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain.tools.retriever import create_retriever_tool
from langgraph.graph import MessagesState, StateGraph, START, END
from langgraph.prebuilt import ToolNode
from pydantic import BaseModel, Field# 1. 加载环境变量
load_dotenv(override=True)# 2. 初始化LLM与嵌入模型
MODEL_NAME = "deepseek-chat"  # 主模型（DeepSeek对话模型）
# 主模型：用于对话与工具调用决策
model = init_chat_model(model=MODEL_NAME,model_provider="deepseek",temperature=0  # 0表示确定性输出，适合问答
)
# 评估模型：判断检索结果是否相关
grader_model = init_chat_model(model=MODEL_NAME,model_provider="deepseek",temperature=0
)
# 嵌入模型：与向量库一致
embed = OpenAIEmbeddings(api_key=os.getenv("OPENAI_API_KEY"),base_url="https://ai.devtool.tech/proxy/v1",model="text-embedding-3-small"
)# 3. 加载向量库并创建检索工具
VS_PATH = "langchain_course_db"  # 向量库路径
vector_store = FAISS.load_local(folder_path=VS_PATH,embeddings=embed,allow_dangerous_deserialization=True
)
# 创建检索工具（每次返回3条相关结果）
retriever_tool = create_retriever_tool(vector_store.as_retriever(search_kwargs={"k": 3}),name="retrieve_langchain_course",  # 工具名（需唯一）description="检索LangChain技术生态课程的相关内容，包括工具链、Agent开发、RAG集成等。"  # 工具描述
)# 4. 定义Prompt（指令设计是智能体核心）
# 系统指令：限定回答范围与工具调用逻辑
SYSTEM_INSTRUCTION = ("你是LangChain技术生态课程的助教，仅回答与LangChain相关的问题（如工具链、Agent开发、RAG集成）。\n""如果问题与课程无关，直接回复：'我不能回答与LangChain技术生态课程无关的问题。'\n""当现有上下文不足时，可调用工具`retrieve_langchain_course`获取相关资料。"
)
# 评估Prompt：判断检索结果是否与问题相关
GRADE_PROMPT = ("你是检索结果评估师，需判断文档是否与用户问题相关。\n""检索文档：\n{context}\n\n""用户问题：{question}\n""仅返回'yes'（相关）或'no'（不相关），无需额外内容。"
)
# 问题改写Prompt：优化不相关/模糊的问题
REWRITE_PROMPT = ("你是问题优化师，需将用户问题改写为与LangChain技术生态相关的清晰问题。\n""例如：'如何开发智能体？' → '如何基于LangChain开发Agent智能体？'\n""原始问题：\n{question}\n""优化后问题："
)
# 回答Prompt：基于上下文生成结构化回答
ANSWER_PROMPT = ("你是LangChain课程助教，需基于提供的上下文回答问题，要求：\n""1. 用Markdown格式，代码块用```包裹，图片引用保留路径；\n""2. 若上下文无相关信息，直接回复'我不知道。'\n""3. 优先引用课程中的代码示例与概念。\n\n""问题：{question}\n""上下文：{context}"
)# 5. 定义LangGraph节点（每个节点对应一个功能）
class GradeDoc(BaseModel):"""结构化输出：评估结果（yes/no）"""binary_score: str = Field(description="检索结果相关性，'yes'表示相关，'no'表示不相关")async def generate_query_or_respond(state: MessagesState):"""节点1：判断是否调用检索工具（或直接回答）"""response = await model.bind_tools([retriever_tool]).ainvoke([{"role": "system", "content": SYSTEM_INSTRUCTION},*state["messages"]  # 历史消息（含用户问题）])return {"messages": [response]}async def grade_documents(state: MessagesState) -> Literal["generate_answer", "rewrite_question"]:"""节点2：评估检索结果，决定生成回答或改写问题"""# 提取用户问题与检索结果question = state["messages"][0].content  # 第1条消息是原始问题context = state["messages"][-1].content  # 最后1条消息是检索结果# 调用评估模型prompt = GRADE_PROMPT.format(question=question, context=context)result = await grader_model.with_structured_output(GradeDoc).ainvoke([{"role": "user", "content": prompt}])# 返回下一个节点（相关则生成回答，否则改写问题）return "generate_answer" if result.binary_score.lower() == "yes" else "rewrite_question"async def rewrite_question(state: MessagesState):"""节点3：改写问题（使其更贴合课程内容）"""question = state["messages"][0].contentprompt = REWRITE_PROMPT.format(question=question)resp = await model.ainvoke([{"role": "user", "content": prompt}])# 将改写后的问题作为新的用户消息，回到工具调用决策节点return {"messages": [{"role": "user", "content": resp.content}]}async def generate_answer(state: MessagesState):"""节点4：基于上下文生成最终回答"""question = state["messages"][0].contentcontext = state["messages"][-1].contentprompt = ANSWER_PROMPT.format(question=question, context=context)resp = await model.ainvoke([{"role": "user", "content": prompt}])return {"messages": [resp]}# 6. 构建LangGraph工作流（图结构定义流程逻辑）
workflow = StateGraph(MessagesState)  # 状态类型：消息列表# 添加节点
workflow.add_node("generate_query_or_respond", generate_query_or_respond)  # 工具调用决策
workflow.add_node("retrieve", ToolNode([retriever_tool]))  # 检索工具节点
workflow.add_node("rewrite_question", rewrite_question)  # 问题改写
workflow.add_node("generate_answer", generate_answer)  # 生成回答# 定义边（流程逻辑）
workflow.add_edge(START, "generate_query_or_respond")  # 起点→决策节点
workflow.add_edge("generate_query_or_respond", "retrieve")  # 决策→检索
workflow.add_conditional_edges(  # 检索后→评估，分支到回答或改写"retrieve",grade_documents  # 条件函数：返回下一个节点名
)
workflow.add_edge("generate_answer", END)  # 回答→终点
workflow.add_edge("rewrite_question", "generate_query_or_respond")  # 改写→重新决策# 7. 编译智能体（生成可调用对象）
rag_agent = workflow.compile(name="langchain_rag_agent")
print("✅ Agentic RAG智能体编译完成！")

2. 智能体调用与测试

通过ainvoke异步调用智能体，支持多轮对话与工具自动调用：

# 测试函数：调用智能体并打印结果
async def test_rag_agent(question: str):print(f"用户问题：{question}")print("-" * 50)# 调用智能体（输入：消息列表，输出：更新后的消息列表）result = await rag_agent.ainvoke({"messages": [{"role": "user", "content": question}]})# 提取最终回答（最后一条消息）answer = result["messages"][-1].contentprint(f"智能体回答：\n{answer}")print("=" * 50 + "\n")# 测试案例1：相关问题（需检索）
await test_rag_agent("如何基于LangChain构建RAG系统？")# 测试案例2：不相关问题（直接拒绝）
await test_rag_agent("如何学习Python基础？")# 测试案例3：模糊问题（需改写）
await test_rag_agent("如何开发智能体？")

六、前端部署：Agent Chat UI可视化

为让系统更易用，可通过LangChain官方的agent-chat-ui搭建前端界面，支持可视化对话与工具调用记录查看：

# 1. 克隆前端仓库
git clone https://github.com/langchain-ai/agent-chat-ui.git# 2. 进入目录并安装依赖（需Node.js 16+）
cd agent-chat-ui
pnpm install  # 或 npm install# 3. 配置后端地址（需将Python智能体部署为API）
# 修改agent-chat-ui/src/lib/agent.ts中的API地址，指向你的后端接口# 4. 启动前端
pnpm dev
# 访问 http://localhost:3000 即可使用对话界面

七、总结与扩展

本文通过完整代码实现了多模态RAG引擎的核心流程，从PDF解析到智能问答，覆盖“数据处理→向量存储→智能体开发”全链路。开发者可基于此扩展：

1. 多模态支持增强：集成InternVL等模型，实现图片内容理解（如识别图表中的数据）；
2. 企业级部署：将向量库替换为Milvus/Weaviate，支持更大数据量；使用Docker容器化前后端，实现高可用；
3. 功能优化：添加对话记忆（Memory）、多工具调用（如Web搜索、SQL查询）、结果溯源（引用原始文档页码）。

如需深入学习大模型Agent开发，可参考课程《2025大模型Agent智能体开发实战》，获取更多企业级案例与进阶技术。