LangChain1.0实战之多模态RAG系统（二）——多模态RAG系统图片分析与语音转写功能实现

本期分享通过集成全模态大模型的能力，成功构建了支持图片和音频分析的多模态 RAG 系统。这种端到端的解决方案大大简化了传统多模态处理的复杂性，展现了未来大模型技术的发展趋势。下期分享笔者将和大家一起处理多模态PDF文档，也是我们多模态RAG系统的核心功能，大家敬请期待~

朝阳区靓仔_James

420人浏览 · 2025-12-28 11:15:00

朝阳区靓仔_James · 2025-12-28 11:15:00 发布

前言

上篇分享[《LangChain1.0实战之多模态RAG系统（一）——多模态RAG系统核心架构及智能问答功能开发》]中笔者系统介绍了基于 LangChain 1.0 构建多模态 RAG 系统的整体架构设计，并完成了智能问答基础模块的开发，实现了对话历史管理、流式响应等核心功能。

本篇笔者将在此基础上，进一步实现多模态 RAG 系统的两个重要扩展功能：图片内容分析与语音信息转写。通过引入对图像和语音数据的处理能力，让 RAG 系统真正具备理解和响应多种模态信息的能力。

学习前置要求：本文是系列的第二篇，强烈建议大家先掌握第一篇中介绍的项目架构、环境配置和基础代码实现，这将有助于大家更好地理解本文的内容。

本系列内容适合所有对 LangChain 感兴趣的学习者，无论之前是否接触过 LangChain。当然，如果大家已经学习过我的专栏[《深入浅出LangChain&LangGraph AI Agent 智能体开发》]，相信可以更快上手。该专栏基于笔者在实际项目中的深度使用经验，系统讲解了使用LangChain/LangGraph如何开发智能体，目前已更新 27讲，并持续补充实战与拓展内容。欢迎感兴趣的同学关注笔者微信公众号 大模型真好玩，每期分享涉及的代码均可在公众号私信: LangChain智能体开发免费获取。

PS:鉴于后台私信越来越多，我建了一些大模型交流群，大家在日常学习生活工作中遇到的大模型知识和问题都可以在群中分享出来大家一起解决！如果大家想交流大模型知识，可以关注我并回复加群。

一、图片分析功能实现

1.1 传统图片分析系统建设思路

在构建图片分析系统时，传统技术方案通常采用多阶段处理流程：

核心技术路径：

图像预处理：使用 OpenCV、PIL 等 Python 库对原始图像进行尺寸调整、色彩校正、噪声过滤等预处理操作
文字信息提取：通过 OCR 模型（如 MonkeyOCR、DeepSeek-OCR）识别并提取图像中的文本内容
语义理解分析：借助多模态大模型（如 Qwen-VL 系列）对图像进行深度语义解析，理解图像场景、对象关系等高层语义信息

这种分层处理架构虽然技术成熟，但存在流程复杂、误差累积等问题。随着多模态大模型能力的提升，笔者这里采用更简洁高效的端到端解决方案。

1.2 多模态 RAG 图片分析实现

本系统得益于Qwen3-Omni在文本、图像、音频全模态任务中保持的顶尖性能，实现一体化的图片分析功能。

1.2.1 数据结构回顾与设计

首先回顾笔者在[LangChain1.0实战之多模态RAG系统（一）——多模态RAG系统核心架构及智能问答功能开发]中定义的核心数据结构，多模态RAG系统会将图片经过base64编码转化为字符串中存储在ContentBlock中，与文字内容同步送入Qwen3-Omni大模型处理，这里不需要变动数据结构：

class ContentBlock(BaseModel):
type: str = Field(description="内容类型: text, image, audio")
content: Optional[str] = Field(description="内容数据")
class MessageRequest(BaseModel):
content_blocks: List[ContentBlock] = Field(default=[], description="内容块")
history: List[Dict[str, Any]] = Field(default=[], description="对话历史")
class MessageResponse(BaseModel):
content: str
timestamp: str
role: str

1.2.2 图像编码处理工具

在项目根目录创建 utils.py 文件，实现图像编码工具类，图像编码工具类负责识别用户上传图片格式并将其编码为base64字符串:

import base64
from fastapi import UploadFile, HTTPException
class ImageProcessor:
"""图像处理工具类"""
@staticmethod
def image_to_base64(image_file: UploadFile) -> str:
try:
# 读取文件内容
contents = image_file.file.read()
# 进行base64编码
base64_encoded = base64.b64encode(contents).decode('utf-8')
return base64_encoded
except Exception as e:
raise HTTPException(status_code=500, detail=f"图像编码失败: {str(e)}")
@staticmethod
def get_image_mime_type(filename: str) -> str:
extension = filename.split('.')[-1].lower()
mime_types = {
'jpg': 'image/jpeg',
'jpeg': 'image/jpeg',
'png': 'image/png',
'gif': 'image/gif',
'bmp': 'image/bmp',
'webp': 'image/webp'
}
return mime_types.get(extension, 'image/jpeg')

1.2.3 多模态消息构建

修改多模态消息构建逻辑，支持图像数据的处理，识别图片后缀格式并将其处理为base64字符串，构造格式为{"type":"image_url", "image_url":{"url": 图片base64格式}}的多模态大模型访问请求

def create_multimodal_message(request: MessageRequest, image_file: UploadFile) -> HumanMessage:
"""创建多模态消息"""
message_content = []
# 如果有图片
if image_file:
processor = ImageProcessor()
mime_type = processor.get_image_mime_type(image_file.filename)
base64_image = processor.image_to_base64(image_file)
message_content.append({
"type": "image_url",
"image_url": {
"url": f"data:{mime_type};base64,{base64_image}"
},
})
# 处理内容块
for i, block in enumerate(request.content_blocks):
if block.type == "text":
message_content.append({
"type": "text",
"text": block.content
})
elif block.type == "image":
# 只有base64格式的消息才会被接入
if block.content.startswith("data:image"):
message_content.append({
"type": "image_url",
"image_url": {
"url": block.content
},
})
return HumanMessage(content=message_content)

1.2.4 历史记录多模态支持

增强历史记录处理函数，添加图像分析的系统提示和多模态内容支持：

def convert_history_to_messages(history: List[Dict[str, Any]]) -> List[BaseMessage]:
"""将历史记录转换为 LangChain 消息格式，支持多模态内容"""
messages = []
# 添加系统消息
system_prompt = """
你是一个专业的多模态 RAG 助手，具备如下能：
1. 与用户对话的能力。
2. 图像内容识别和分析能力(OCR, 对象检测， 场景理解)
重要指导原则：
- 当用户上传图片并提出问题时，请结合图片内容和用户的具体问题来回答
- 仔细分析图片中的文字、图表、对象、场景等所有可见信息
- 根据用户的问题重点，有针对性地分析图片相关部分
- 如果图片包含文字，请准确识别并在回答中引用
- 如果用户只上传图片没有问题，则提供图片的全面分析
请以专业、准确、友好的方式回答
"""
messages.append(SystemMessage(content=system_prompt))
# 转换历史消息
for i, msg in enumerate(history):
content = msg.get("content", "")
content_blocks = msg.get("content_blocks", [])
message_content = []
if msg["role"] == "user":
for block in content_blocks:
if block.get("type") == "text":
message_content.append({
"type": "text",
"text": block.get("content", "")
})
elif block.get("type") == "image":
image_data = block.get("content", "")
if image_data.startswith("data:image"):
message_content.append({
"type": "image_url",
"image_url" : {
"url": image_data
}
})
messages.append(HumanMessage(content=message_content))
elif msg["role"] == "assistant":
messages.append(AIMessage(content=content))
return messages

1.2.5 接口格式调整

由于现在需要同时支持 JSON 数据和文件上传，将接口调整为 multipart/form-data 格式，修改chat_stream接口如下：

@app.post("/api/chat/stream")
async def chat_stream(
image_file: UploadFile = File(...),
content_blocks: str = Form(default="[]"),
history: str = Form(default="[]")
):
"""流式聊天接口（支持多模态）"""
try:
# 解析 JSON 字符串
try:
content_blocks_data = json.loads(content_blocks)
history_data = json.loads(history)
except json.JSONDecodeError as e:
raise HTTPException(status_code=400, detail=f"JSON 解析错误: {str(e)}")
# 创建请求对象（用于传递给其他函数）
request_data = MessageRequest(content_blocks=content_blocks_data, history=history_data)
# 转换消息历史
messages = convert_history_to_messages(request_data.history)
# 添加当前用户消息（支持多模态）
current_message = create_multimodal_message(request_data, image_file)
messages.append(current_message)
# 返回流式响应
return StreamingResponse(
generate_streaming_response(messages),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"Content-Type": "text/event-stream",
}
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))

完成以上五个核心模块的修改后，系统就具备了完整的图片分析能力，可以处理用户上传的图像并进行智能问答~

1.3 图片分析功能测试

完成代码开发后，我们通过 Postman 对图片分析功能进行完整测试。

测试配置步骤

1. 请求头设置
在 Postman 的 Headers 选项卡中配置：

Content-Type: multipart/form-data

2. 请求体配置
在 Body 选项卡中选择 form-data 格式，添加以下参数：

字段名	类型	值
`image_file`	File	选择 Gemini 3.0 Logo 图片文件
`content_blocks`	Text	`[{"type": "text", "content": "请分析这张图片"}]`
`history`	Text	`[]`

3. 测试执行
向 /api/chat/stream 端点发送 POST 请求，系统返回流式响应。

测试结果验证

从响应结果可见，系统成功识别并分析了测试图片：

准确识别出图片包含 Gemini 3.0 的 Logo
详细描述了 Logo 的设计元素和视觉特征
提供了完整的图片内容分析

测试结果表明，图片分析功能已正常集成到多模态 RAG 系统中，能够正确处理用户上传的图片并结合问题进行智能分析。

二、音频分析功能实现

2.1 多模态RAG音频处理的实现

音频分析功能的实现思路与图片分析类似，主要通过 Base64 编码和格式转换实现音频数据的处理与传输。

2.1.1 数据结构设计

音频数据采用与图片相同的存储方式，通过 Base64 编码存储在 content 字段中：

class ContentBlock(BaseModel):
type: str = Field(description="内容类型: text, image, audio")
content: Optional[str] = Field(description="内容数据")
class MessageRequest(BaseModel):
content_blocks: List[ContentBlock] = Field(default=[], description="内容块")
history: List[Dict[str, Any]] = Field(default=[], description="对话历史")
class MessageResponse(BaseModel):
content: str
timestamp: str
role: str

2.1.2 音频处理工具类

在 utils.py 中添加音频处理工具类，支持多种音频格式，针对音频格式的处理笔者额外添加了上传文件大小的相关限制：

class AudioProcessor:
"""音频处理工具类"""
@staticmethod
def audio_to_base64(audio_file: UploadFile) -> str:
try:
# 验证文件类型
if not AudioProcessor.is_valid_audio_type(audio_file.content_type, audio_file.filename):
raise HTTPException(
status_code=400,
detail="不支持的音频格式，支持的格式有: MP3, WAV, OGG, M4A, FLAC"
)
# 读取文件内容
contents = audio_file.file.read()
# 验证文件大小（可选：限制为10MB）
max_size = 10 * 1024 * 1024  # 10MB
if len(contents) > max_size:
raise HTTPException(
status_code=400,
detail=f"音频文件过大，最大支持 {max_size // 1024 // 1024}MB"
)
base64_encoded = base64.b64encode(contents).decode('utf-8')
return base64_encoded
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=f"音频编码失败: {str(e)}")
@staticmethod
def get_audio_mime_type(filename: str) -> str:
extension = filename.split('.')[-1].lower()
mime_types = {
'mp3': 'audio/mpeg',
'wav': 'audio/wav',
'm4a': 'audio/mp4',
}
return mime_types.get(extension, 'audio/mpeg')  # 默认为MP3
@staticmethod
def is_valid_audio_type(content_type: str, filename: str) -> bool:
# 获取支持的MIME类型列表
supported_mimes = {
'audio/mpeg', 'audio/wav', 'audio/mp4'
}
# 检查content_type
if content_type and content_type in supported_mimes:
return True
# 检查文件扩展名
file_extension = filename.split('.')[-1].lower()
supported_extensions = {'mp3', 'wav', 'm4a'}
return file_extension in supported_extensions

2.1.3 多模态消息构建增强

扩展 create_multimodal_message 函数，支持音频数据的处理, 首先提取音频文件后缀并将其处理为base64格式，同时构造格式为{"type":"audio_url", "audio_url":{"url": 音频base64格式}}多模态大模型访问请求：

def create_multimodal_message(request: MessageRequest, image_file: UploadFile | None, audio_file:UploadFile | None) -> HumanMessage:
"""创建多模态消息"""
message_content = []
# 如果有图片
if image_file:
processor = ImageProcessor()
mime_type = processor.get_image_mime_type(image_file.filename)
base64_image = processor.image_to_base64(image_file)
message_content.append({
"type": "image_url",
"image_url": {
"url": f"data:{mime_type};base64,{base64_image}"
},
})
if audio_file:
processor = AudioProcessor()
mime_type = processor.get_audio_mime_type(audio_file.filename)
base64_audio = processor.audio_to_base64(audio_file)
message_content.append({
"type": "audio_url",
"audio_url": {
"url": f"data:{mime_type};base64,{base64_audio}"
},
})
# 处理内容块
for i, block in enumerate(request.content_blocks):
if block.type == "text":
message_content.append({
"type": "text",
"text": block.content
})
elif block.type == "image":
# 只有base64格式的消息才会被接入
if block.content.startswith("data:image"):
message_content.append({
"type": "image_url",
"image_url": {
"url": block.content
},
})
elif block.type == "audio":
if block.content.startswith("data:audio"):
message_content.append({
"type": "audio_url",
"audio_url": {
"url": block.content
},
})
return HumanMessage(content=message_content)

2.1.4 历史记录处理增强

更新系统提示词，添加音频处理能力，并增强历史记录处理逻辑：

def convert_history_to_messages(history: List[Dict[str, Any]]) -> List[BaseMessage]:
"""将历史记录转换为 LangChain 消息格式，支持多模态内容"""
messages = []
# 添加系统消息
system_prompt = """
你是一个专业的多模态 RAG 助手，具备如下能：
1. 与用户对话的能力。
2. 图像内容识别和分析能力(OCR, 对象检测， 场景理解)
3. 音频转写与分析
重要指导原则：
- 当用户上传图片并提出问题时，请结合图片内容和用户的具体问题来回答
- 仔细分析图片中的文字、图表、对象、场景等所有可见信息
- 根据用户的问题重点，有针对性地分析图片相关部分
- 如果图片包含文字，请准确识别并在回答中引用
- 如果用户只上传图片没有问题，则提供图片的全面分析
请以专业、准确、友好的方式回答
"""
messages.append(SystemMessage(content=system_prompt))
# 转换历史消息
for i, msg in enumerate(history):
content = msg.get("content", "")
content_blocks = msg.get("content_blocks", [])
message_content = []
if msg["role"] == "user":
for block in content_blocks:
if block.get("type") == "text":
message_content.append({
"type": "text",
"text": block.get("content", "")
})
elif block.get("type") == "image":
image_data = block.get("content", "")
if image_data.startswith("data:image"):
message_content.append({
"type": "image_url",
"image_url" : {
"url": image_data
}
})
elif block.get("type") == "audio":
audio_data = block.get("content", "")
if audio_data.startswith("data:audio"):
message_content.append({
"type": "audio_url",
"image_url": {
"url": audio_data
}
})
messages.append(HumanMessage(content=message_content))
elif msg["role"] == "assistant":
messages.append(AIMessage(content=content))
return messages

2.1.5 接口完整实现

更新聊天接口，同时支持图片和音频文件上传：

@app.post("/api/chat/stream")
async def chat_stream(
image_file: UploadFile | None = File(None),
content_blocks: str = Form(default="[]"),
history: str = Form(default="[]"),
audio_file: UploadFile | None = File(None)
):
"""流式聊天接口（支持多模态）"""
try:
# 解析 JSON 字符串
try:
content_blocks_data = json.loads(content_blocks)
history_data = json.loads(history)
except json.JSONDecodeError as e:
raise HTTPException(status_code=400, detail=f"JSON 解析错误: {str(e)}")
# 创建请求对象（用于传递给其他函数）
request_data = MessageRequest(content_blocks=content_blocks_data, history=history_data)
# 转换消息历史
messages = convert_history_to_messages(request_data.history)
# 添加当前用户消息（支持多模态）
current_message = create_multimodal_message(request_data, image_file=image_file, audio_file=audio_file)
messages.append(current_message)
# 返回流式响应
return StreamingResponse(
generate_streaming_response(messages),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"Content-Type": "text/event-stream",
}
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))

2.2 功能测试

完成代码开发后，通过 Postman 对音频分析功能进行测试。

测试配置：

请求头：Content-Type: multipart/form-data
请求体（form-data格式）

字段名	类型	值
`audio_file`	File	选择测试音频文件（包含"你好"的录音）
`content_blocks`	Text	`[{"type": "text", "content": "请解析音频内容"}]`
`history`	Text	`[]`

测试结果：
系统成功识别并转写了音频内容，准确输出三个"你好"，验证了音频分析功能的正确性。

三、总结

[《深入浅出LangChain&LangGraph AI Agent 智能体开发》]专栏内容源自笔者在实际学习和工作中对 LangChain 与 LangGraph 的深度使用经验，旨在帮助大家系统性地、高效地掌握 AI Agent 的开发方法，在各大技术平台获得了不少关注与支持。目前已更新28讲，正在更新LangChain1.0实战篇，并随时补充笔者在实际工作中总结的拓展知识点。如果大家感兴趣，欢迎关注笔者的微信公众号 大模型真好玩，每期分享涉及的代码均可在公众号私信: LangChain智能体开发免费获取。

如何学习大模型 AI ？

由于新岗位的生产效率，要优于被取代岗位的生产效率，所以实际上整个社会的生产效率是提升的。

但是具体到个人，只能说是：

“最先掌握AI的人，将会比较晚掌握AI的人有竞争优势”。

这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。

我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。

我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑，所以在工作繁忙的情况下还是坚持各种整理和分享。但苦于知识传播途径有限，很多互联网行业朋友无法获得正确的资料得到学习提升，故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。

在这里插入图片描述