前言

笔者深入解析了 LangChain 1.0 的 create_agent API，重点介绍了 MCP 协议工具集成、结构化输出、记忆管理和中间步中间件机制四大核心能力。通过高德地图 MCP 接入、动态模型选择等实战案例，展示了如何构建具备外部工具调用、记忆保持与流程可控的智能体应用。

掌握了基础知识后，要想检验自己对 LangChain 1.0 的理解，并进一步提升开发能力，最好的方式莫过于通过实际项目来锤炼技能。从本期开始，笔者将推出一个完整的实战系列—— 《LangChain1.0 搭建多模态 RAG 知识库》 ，覆盖从前端到后端的全流程开发。在本系列中笔者将分享：

• LangChain1.0 多模态智能体的基本开发流程
• LangChain1.0 文档处理与向量库构建技巧
• 基于 FastAPI 的后端服务开发
• 使用 React 构建现代化前端界面

作为系列的第一篇，本文将从整体架构出发，系统介绍多模态 RAG 系统的核心组成与技术栈选择。后续文章将逐步拆解 LangChain + FastAPI 后端的关键模块，带领大家从零搭建完整的系统。

本系列内容适合所有对 LangChain 感兴趣的学习者，无论之前是否接触过 LangChain。

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

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

首先介绍本系列项目的核心架构，本项目构建的是一个多模态RAG智能对话系统，支持文本、图像、音频、PDF等多种格式的输入，具备智能问答、图片分析、音频转写、PDF解析四大核心功能。为体现企业级智能体系统的开发标准，本项目采用前后端分离的现代化架构，帮助大家理解真实场景中的系统设计与开发流程。整体架构如下所示：

在基于 LangChain 1.0 的后端开发中，笔者将实现四个功能独立的智能体模块，分别对应不同的多模态输入处理任务：

尽管本项目离真正的生产级别系统仍有差异，但通过实现多模态输入的基本处理流程、系统架构的设计规划、LangChain 1.0 核心特性的应用，以及项目的整体架构与部署实践，相信大家能够系统掌握构建现代多模态 RAG 系统的关键技能，为开发更复杂的应用打下基础。

二、核心功能讲解方式

由于篇幅限制，笔者无法逐行解析所有代码细节，但将通过以下方式确保大家的学习效果：

• 提供完整源码

  所有代码将完整开放，供大家随时查阅和运行

• 重点剖析架构

  着重讲解各智能体的功能设计与构建方法

• 掌握核心逻辑

  帮助大家深入理解项目架构与关键技术实现

具体分享安排如下：

1. 智能体开发循序渐进

   从最基础的智能问答智能体入手，逐步扩展到图片解析、音频处理、PDF解析等复杂场景，并同步讲解如何通过 FastAPI 对智能体功能进行服务化封装。

2. 前端开发与交互原理

   解析现代化前端组件的开发流程，重点介绍大模型智能问答场景下前端与后端的核心交互机制。

3. 项目部署实战

   带领大家完成前后端应用的本地部署，确保项目能够完整运行。

接下来，笔者就从最基础的智能问答智能体开始，深入讲解其实现原理与开发过程。

三、智能问答智能体搭建

作为多模态RAG系统的核心基础，智能问答功能是处理文本、图像、音频和PDF等各类输入的通用能力。本节将从零开始构建智能问答智能体。

3.1 智能问答智能体后端搭建

1. 环境配置与依赖引入

首先引入必要的依赖包，确保具备LangChain、FastAPI等核心库的支持。

import json
import uvicorn
from typing import List, Dict, Any, AsyncGenerator
from datetime import datetime
from pydantic import BaseModel, Field
from fastapi import HTTPException, FastAPI
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
from langchain.chat_models import init_chat_model
from langchain.messages import SystemMessage, HumanMessage, AIMessage
from langchain_core.messages import BaseMessage

2. 多模态模型初始化

为处理多模态数据，笔者选用阿里巴巴通义千问于2025年9月发布的全模态模型Qwen3-Omni-30B-A3B-Instruct，通过硅基流动提供的API接口进行接入。langchain接入硅基流动大模型的方法同样使用init_chat_model api

def get_chat_model():
try:
model = init_chat_model(
model="Qwen/Qwen3-Omni-30B-A3B-Instruct",
model_provider="openai",
base_url="https://api.siliconflow.cn/v1/",
api_key="你注册的硅基流动api key",
)
return model
except Exception as e:
raise HTTPException(status_code=500, detail=f"模型初始化失败: {str(e)}")

3. 数据结构定义

使用Pydantic严格定义请求和响应的数据格式，确保类型安全：对于pydantic使用不熟悉的读者可参考笔者的文章深入浅出LangGraph AI Agent智能体开发教程（六）—LangGraph 底层API入门。这里通过ContentBlock统一封装多模态数据，使用type字段区分数据类型，为后续扩展预留接口。

class ContentBlock(BaseModel):
type: str = Field(description="内容类型: text, image, audio")
content: 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

4. 多模态消息构建

将前端请求转换为LangChain可识别的消息格式：

def create_multimodal_message(request: MessageRequest) -> HumanMessage:
"""创建多模态消息"""
message_content = []
# 处理内容块
for i, block in enumerate(request.content_blocks):
if block.type == "text":
message_content.append({
"type": "text",
"text": block.content
})
return HumanMessage(content=message_content[0]["text"])

5. 对话历史管理

维护完整的对话上下文，确保智能体具备记忆能力：

def convert_history_to_messages(history: List[Dict[str, Any]]) -> List[BaseMessage]:
"""将历史记录转换为 LangChain 消息格式，支持多模态内容"""
messages = []
# 添加系统消息
system_prompt = """
你是一个专业的多模态 RAG 助手，具备与用户对话的能力， 请以专业、准确、友好的方式回答用户所提问题。
"""
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", "")
})
messages.append(HumanMessage(content=message_content))
elif msg["role"] == "assistant":
messages.append(AIMessage(content=content))
return messages

6. 流式响应生成

实现实时响应的流式输出机制，注意流式响应需要用到python的生成器yield机制，大家不熟悉可学习廖雪峰老师相关课程 https://liaoxuefeng.com/books/python/advanced/generator/index.html

async def generate_streaming_response(
messages: List[BaseMessage]
) -> AsyncGenerator[str, None]:
"""生成流式响应"""
try:
model = get_chat_model()
# 创建流式响应
full_response = ""
chunk_count = 0
async for chunk in model.astream(messages):
chunk_count += 1
if hasattr(chunk, 'content') and chunk.content:
content = chunk.content
full_response += content
# 直接发送每个chunk的内容，避免重复
data = {
"type": "content_delta",
"content": content,
"timestamp": datetime.now().isoformat()
}
yield f"data: {json.dumps(data, ensure_ascii=False)}\n\n"
# 发送完成信号
final_data = {
"type": "message_complete",
"full_content": full_response,
"timestamp": datetime.now().isoformat(),
}
yield f"data: {json.dumps(final_data, ensure_ascii=False)}\n\n"
except Exception as e:
error_data = {
"type": "error",
"error": str(e),
"timestamp": datetime.now().isoformat()
}
yield f"data: {json.dumps(error_data, ensure_ascii=False)}\n\n"

7. 流式聊天接口

提供实时交互的流式API端点：

async def chat_stream(request: MessageRequest):
"""流式聊天接口（支持多模态）"""
try:
# 转换消息历史
messages = convert_history_to_messages(request.history)
# 添加当前用户消息（支持多模态）
current_message = create_multimodal_message(request)
messages.append(current_message)
# 返回流式响应
return StreamingResponse(
generate_streaming_response(messages),
media_type="text/plain",
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))

8. 同步聊天接口

为简单场景提供一次性响应的同步接口:

async def chat_sync(request: MessageRequest):
"""同步聊天接口（支持多模态）"""
try:
# 转换消息历史
messages = convert_history_to_messages(request.history)
# 添加当前用户消息（支持多模态）
current_message = create_multimodal_message(request)
messages.append(current_message)
# 获取模型响应
model = get_chat_model()
response = await model.ainvoke(messages)
return MessageResponse(
content=response.content,
role="assistant",
timestamp=datetime.now().isoformat(),
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))

3.2 FastAPI后端服务封装

FastAPI是基于Python标准类型提示的现代Web框架，以其卓越的性能和开发效率著称。FastAPI底层基于 Starlette（用于 Web 微服务）和 Pydantic（用于数据验证），它的名字就揭示了其核心特点：Fast（快速）和 API。下面笔者将智能体功能封装为完整的API服务：

1. 应用初始化与CORS配置

app = FastAPI(
title="多模态 RAG 工作台 API",
description="基于 LangChain 1.0 的智能对话 API",
version="1.0.0"
)
# 配置跨域访问
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)

2. API路由注册

@app.get和@app.post是FastAPI的路径操作装饰器，它们告诉 FastAPI 下面的函数处理哪个路径和哪种 HTTP 方法。我们需要将chat_stream和chat_sync方法使用路径操作装饰器包装，使得可以访问api接口得到结果：

@app.post("/api/chat/stream")
async def chat_stream(request: MessageRequest):
@app.post("/api/chat")
async def chat_sync(request: MessageRequest):

3. 创建应用

if __name__ == "__main__":
uvicorn.run(
app,
host="localhost",
port=8000
)

至此，笔者完成了智能问答智能体的核心后端搭建。在后续章节中，将基于此基础逐步扩展图片解析、音频处理等更多模态的智能体能力。

四、使用 Postman 测试接口正确性

为确保智能问答智能体功能正常运行，笔者使用 Postman 对 API 接口进行完整测试。

4.1 测试单轮对话

1. 首先运行主程序启动服务端：

python main.py运行代码文件，服务正常启动后，显示如下结果

2. 配置 Postman 请求

1. 创建新请求：

• 方法：POST
• URL：http://localhost:8000/api/chat/stream

2. 设置请求头：

• 添加 Content-Type 为 application/json

3. 在 Body 中填入测试数据，验证基础问答功能，点击 Send 发送请求后，Postman 将展示流式响应效果：

{
"content_blocks": [
{
"type": "text",
"content": "什么是人工智能？"
}
],
"history": []
}

4. 响应说明

   流式接口会实时返回多个数据块（content_delta），最终会返回完整响应内容（message_complete），此结果证明智能问答接口运行正常

4.2 多轮对话测试

验证对话历史记忆功能，使用包含上下文的测试数据：

{
"content_blocks": [
{
"type": "text",
"content": "你好我叫什么名字？"
}
],
"history": [
{
"role": "user",
"content_blocks": [
{
"type": "text",
"content": "你好，我是苍进空"
}
]
},
{
"role": "assistant",
"content": "你好！我是多模态 RAG 助手，很高兴为您服务。"
}
]
}

测试结果如下，可以看到多模态大模型完全记着笔者刚才的提问，准确说出了名字：苍进空！

以上就是我们今天要分享的全部内容啦！

五、总结

本文系统介绍了基于LangChain 1.0开发多模态RAG系统的核心架构，通过前后端分离架构，构建支持文本、图像、音频和PDF处理的四大功能模块。同时编写完成智能问答基础模块的开发工作，实现了对话历史管理和流式响应，并使用Postman完成接口测试验证。本实战项目为后续扩展更复杂的多模态应用奠定了坚实基础。

如何学习大模型 AI ？

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

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

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

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

我在一线科技企业深耕十二载，见证过太多因技术卡位而跃迁的案例。那些率先拥抱 AI 的同事，早已在效率与薪资上形成代际优势，我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在大模型的学习中的很多困惑。我们整理出这套 AI 大模型突围资料包：

• ✅ 从零到一的 AI 学习路径图
• ✅ 大模型调优实战手册（附医疗/金融等大厂真实案例）
• ✅ 百度/阿里专家闭门录播课
• ✅ 大模型当下最新行业报告
• ✅ 真实大厂面试真题
• ✅ 2025 最新岗位需求图谱

所有资料 ⚡️ ，朋友们如果有需要 《AI大模型入门+进阶学习资源包》，下方扫码获取~

① 全套AI大模型应用开发视频教程

（包含提示工程、RAG、LangChain、Agent、模型微调与部署、DeepSeek等技术点）

② 大模型系统化学习路线

作为学习AI大模型技术的新手，方向至关重要。 正确的学习路线可以为你节省时间，少走弯路；方向不对，努力白费。这里我给大家准备了一份最科学最系统的学习成长路线图和学习规划，带你从零基础入门到精通！

③ 大模型学习书籍&文档

学习AI大模型离不开书籍文档，我精选了一系列大模型技术的书籍和学习文档（电子版），它们由领域内的顶尖专家撰写，内容全面、深入、详尽，为你学习大模型提供坚实的理论基础。

④ AI大模型最新行业报告

2025最新行业报告，针对不同行业的现状、趋势、问题、机会等进行系统地调研和评估，以了解哪些行业更适合引入大模型的技术和应用，以及在哪些方面可以发挥大模型的优势。

⑤ 大模型项目实战&配套源码

学以致用，在项目实战中检验和巩固你所学到的知识，同时为你找工作就业和职业发展打下坚实的基础。

⑥ 大模型大厂面试真题

面试不仅是技术的较量，更需要充分的准备。在你已经掌握了大模型技术之后，就需要开始准备面试，我精心整理了一份大模型面试题库，涵盖当前面试中可能遇到的各种技术问题，让你在面试中游刃有余。

以上资料如何领取？

为什么大家都在学大模型？

最近科技巨头英特尔宣布裁员2万人，传统岗位不断缩减，但AI相关技术岗疯狂扩招，有3-5年经验，大厂薪资就能给到50K*20薪！

不出1年，“有AI项目经验”将成为投递简历的门槛。

风口之下，与其像“温水煮青蛙”一样坐等被行业淘汰，不如先人一步，掌握AI大模型原理+应用技术+项目实操经验，“顺风”翻盘！

这些资料真的有用吗？

这份资料由我和鲁为民博士(北京清华大学学士和美国加州理工学院博士)共同整理，现任上海殷泊信息科技CEO，其创立的MoPaaS云平台获Forrester全球’强劲表现者’认证，服务航天科工、国家电网等1000+企业，以第一作者在IEEE Transactions发表论文50+篇，获NASA JPL火星探测系统强化学习专利等35项中美专利。本套AI大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。

资料内容涵盖了从入门到进阶的各类视频教程和实战项目，无论你是小白还是有些技术基础的技术人员，这份资料都绝对能帮助你提升薪资待遇，转行大模型岗位。

以上全套大模型资料如何领取？