实战:多模态 AI 应用开发
图像输入 + 文本 Prompt + LLM 推理 + 结果渲染前端 全流程实战
项目概述
目标:构建一个能"看图回答问题"的多模态应用。用户上传一张图片并输入自然语言问题,系统调用多模态大模型理解图像内容,最终在前端返回推理结果。
技术栈:
- 模型与推理:选用 Qwen-VL / LLaVA 等开源多模态模型,也可通过 API 调用 GPT-4V。
- 后端框架:FastAPI,提供 RESTful 接口与 SSE 流式推送。
- 前端界面:Gradio,快速搭建可交互的 Web UI,支持图像上传、文本输入和结果展示。
- 部署:Docker + GPU 环境,生产级推理服务。
Step 1:后端模型选择
选择多模态模型
目前主流的多模态模型方案包括:
| 模型 | 特点 | 适用场景 |
|---|---|---|
| Qwen-VL | 阿里通义千问多模态版本,中英文能力强,支持多轮对话 | 通用图文理解、文档问答 |
| LLaVA | 开源社区广泛使用,基于 LLaMA/Vicuna + CLIP 视觉编码器 | 学术研究、私有化部署 |
| GPT-4V | OpenAI 闭源模型,多模态理解能力最强 | 快速原型验证、高精度场景 |
选择依据:如果需要私有化部署且预算受限,推荐 Qwen-VL 或 LLaVA;如果追求开箱即用的最佳效果,可通过 API 接入 GPT-4V。
模型加载与推理准备
以 Qwen-VL-Chat 为例,使用 transformers 库加载模型:
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
model_name = "Qwen/Qwen-VL-Chat"
tokenizer = AutoTokenizer.from_pretrained(
model_name,
trust_remote_code=True,
)
model = AutoModelForCausalLM.from_pretrained(
model_name,
device_map="auto",
torch_dtype=torch.bfloat16,
trust_remote_code=True,
).eval()
print("模型加载完成,设备:", model.device)若使用 LLaVA,加载方式类似,只需替换 model_name 为 "liuhaotian/llava-v1.6-vicuna-7b" 并调整对应的 processor。
Step 2:API 接口开发
使用 FastAPI 定义推理端点,接收图像文件与文本问题,返回模型回答。同时加入图像预处理逻辑,确保输入符合模型要求。
图像预处理
多模态模型通常要求输入图像尺寸归一化到固定大小(如 224x224 或 336x336),并执行归一化(mean/std)。transforms 工具链如下:
from torchvision import transforms
from PIL import Image
def preprocess_image(image: Image.Image, target_size: int = 224) -> Image.Image:
transform = transforms.Compose([
transforms.Resize((target_size, target_size)),
transforms.ToTensor(),
transforms.Normalize(
mean=[0.48145466, 0.4578275, 0.40821073],
std=[0.26862954, 0.26130258, 0.27577711],
),
])
return transform(image)Qwen-VL 的 tokenizer 内置了图像处理能力,因此无需手动预处理,直接传入 PIL Image 即可。
FastAPI 端点
from fastapi import FastAPI, File, UploadFile, Form
from fastapi.responses import StreamingResponse
import io
from PIL import Image
app = FastAPI(title="多模态推理 API")
@app.post("/v1/chat")
async def chat(
image: UploadFile = File(...),
query: str = Form(...),
):
# 读取上传的图片
contents = await image.read()
pil_image = Image.open(io.BytesIO(contents)).convert("RGB")
# 构造多模态输入
messages = [
{
"role": "user",
"content": [
{"type": "image", "image": pil_image},
{"type": "text", "text": query},
],
}
]
# 模型推理
text, _ = model.chat(
tokenizer,
query=None,
history=None,
messages=messages,
)
return {"answer": text}Step 3:前端界面构建
Gradio 提供了一套声明式 API,几行代码即可构建带图像上传和文本交互的 Web 界面。
import gradio as gr
def predict(image, query):
if image is None:
return "请先上传一张图片"
# 将 numpy 数组转为 PIL Image
pil_img = Image.fromarray(image).convert("RGB")
messages = [
{
"role": "user",
"content": [
{"type": "image", "image": pil_img},
{"type": "text", "text": query},
],
}
]
response, _ = model.chat(tokenizer, query=None, history=None, messages=messages)
return response
# 构建界面
with gr.Blocks(title="多模态问答系统", theme="soft") as demo:
gr.Markdown("# 多模态 AI 问答")
with gr.Row():
with gr.Column():
image_input = gr.Image(label="上传图片", type="numpy")
text_input = gr.Textbox(label="输入问题", placeholder="请描述这张图片...")
submit_btn = gr.Button("提交")
with gr.Column():
output = gr.Textbox(label="模型回答", lines=10)
submit_btn.click(
fn=predict,
inputs=[image_input, text_input],
outputs=output,
)通过 gr.Blocks 可以实现灵活的布局:左侧放置上传区域和输入框,右侧展示模型回答。Gradio 会自动处理前端渲染、文件上传和 CORS 等细节。
Step 4:流式输出
大模型推理通常需要数秒甚至更长时间,流式输出可以显著提升用户体验。后端通过 SSE(Server-Sent Events)逐 token 推送结果。
后端流式端点
from fastapi.responses import StreamingResponse
import asyncio
import json
async def stream_chat(messages):
response = model.chat_stream(
tokenizer,
query=None,
history=None,
messages=messages,
)
for chunk in response:
yield json.dumps({"token": chunk}) + "\n"
await asyncio.sleep(0.01)
@app.post("/v1/chat/stream")
async def chat_stream(
image: UploadFile = File(...),
query: str = Form(...),
):
contents = await image.read()
pil_image = Image.open(io.BytesIO(contents)).convert("RGB")
messages = [
{
"role": "user",
"content": [
{"type": "image", "image": pil_image},
{"type": "text", "text": query},
],
}
]
return StreamingResponse(
stream_chat(messages),
media_type="text/event-stream",
)Gradio 流式渲染
Gradio 原生支持流式输出,只需将 fn 改为生成器函数即可自动逐 token 渲染:
def predict_stream(image, query):
if image is None:
yield "请先上传一张图片"
return
pil_img = Image.fromarray(image).convert("RGB")
messages = [
{
"role": "user",
"content": [
{"type": "image", "image": pil_img},
{"type": "text", "text": query},
],
}
]
for chunk in model.chat_stream(tokenizer, query=None, history=None, messages=messages):
yield chunk将 Gradio 的 submit 按钮绑定到 predict_stream 即可实现实时逐字输出效果。
Step 5:结果渲染
图文混排显示
当模型返回包含结构化信息的结果时(如检测到的物体列表、坐标等),可以在前端做可视化渲染。Gradio 支持 gr.HTML 组件,可以灵活绘制图文混排内容:
import matplotlib.pyplot as plt
import matplotlib.patches as patches
def render_result(image, detections):
"""在图像上绘制检测框并返回 HTML 排版结果"""
fig, ax = plt.subplots(1, figsize=(8, 8))
ax.imshow(image)
for det in detections:
x, y, w, h = det["bbox"]
rect = patches.Rectangle(
(x, y), w, h,
linewidth=2, edgecolor="red", facecolor="none",
)
ax.add_patch(rect)
ax.text(
x, y - 5, det["label"],
color="white", fontsize=10,
bbox=dict(facecolor="red", alpha=0.6),
)
plt.axis("off")
# 保存为 base64 嵌入 HTML
buf = io.BytesIO()
plt.savefig(buf, format="png", bbox_inches="tight")
buf.seek(0)
img_b64 = base64.b64encode(buf.read()).decode()
plt.close()
html = f"""
<div style="display:flex; gap:20px; align-items:flex-start;">
<img src="data:image/png;base64,{img_b64}" style="max-width:500px;">
<div style="max-width:400px;">
<h4>检测结果</h4>
<ul>{"".join(f"<li>{d['label']} (置信度: {d['score']:.2f})</li>" for d in detections)}</ul>
</div>
</div>
"""
return html该函数将检测框绘制在原始图像上,并生成一个图文混排的 HTML 片段,供 Gradio 的 gr.HTML 组件渲染。
完整代码
以下整合了上述所有步骤,形成可运行的 app.py:
import io
import base64
from PIL import Image
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
import gradio as gr
import matplotlib.pyplot as plt
import matplotlib.patches as patches
# ---------- 模型加载 ----------
model_name = "Qwen/Qwen-VL-Chat"
tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)
model = AutoModelForCausalLM.from_pretrained(
model_name,
device_map="auto",
torch_dtype=torch.bfloat16,
trust_remote_code=True,
).eval()
# ---------- 推理函数 ----------
def predict(image, query):
if image is None:
return "请先上传一张图片"
pil_img = Image.fromarray(image).convert("RGB")
messages = [
{
"role": "user",
"content": [
{"type": "image", "image": pil_img},
{"type": "text", "text": query},
],
}
]
response, _ = model.chat(
tokenizer, query=None, history=None, messages=messages
)
return response
# ---------- 流式推理 ----------
def predict_stream(image, query):
if image is None:
yield "请先上传一张图片"
return
pil_img = Image.fromarray(image).convert("RGB")
messages = [
{
"role": "user",
"content": [
{"type": "image", "image": pil_img},
{"type": "text", "text": query},
],
}
]
for chunk in model.chat_stream(
tokenizer, query=None, history=None, messages=messages
):
yield chunk
# ---------- 结果渲染 ----------
def predict_with_rendering(image, query):
if image is None:
yield "请先上传一张图片"
return
pil_img = Image.fromarray(image).convert("RGB")
messages = [
{
"role": "user",
"content": [
{"type": "image", "image": pil_img},
{"type": "text", "text": query},
],
}
]
response, history = model.chat(
tokenizer, query=None, history=None, messages=messages
)
# 如果返回包含检测信息,进行可视化
# 此处为简化示例,实际需根据模型输出格式解析
html = f"""
<div style="display:flex; gap:20px; align-items:flex-start;">
<img src="data:image/png;base64,{pil_to_b64(pil_img)}" style="max-width:400px;">
<div>
<h4>模型回答</h4>
<p>{response}</p>
</div>
</div>
"""
yield html
def pil_to_b64(pil_img):
buf = io.BytesIO()
pil_img.save(buf, format="PNG")
return base64.b64encode(buf.getvalue()).decode()
# ---------- Gradio 界面 ----------
with gr.Blocks(title="多模态问答系统", theme="soft") as demo:
gr.Markdown("# 多模态 AI 问答")
with gr.Tab("普通问答"):
with gr.Row():
with gr.Column():
img_in = gr.Image(label="上传图片", type="numpy")
txt_in = gr.Textbox(label="输入问题", placeholder="请描述这张图片...")
btn = gr.Button("提交")
with gr.Column():
txt_out = gr.Textbox(label="模型回答", lines=10)
btn.click(fn=predict, inputs=[img_in, txt_in], outputs=txt_out)
with gr.Tab("流式问答"):
with gr.Row():
with gr.Column():
img_s = gr.Image(label="上传图片", type="numpy")
txt_s = gr.Textbox(label="输入问题", placeholder="请描述这张图片...")
btn_s = gr.Button("提交")
with gr.Column():
txt_so = gr.Textbox(label="模型回答", lines=10)
btn_s.click(
fn=predict_stream,
inputs=[img_s, txt_s],
outputs=txt_so,
)
with gr.Tab("图文渲染"):
with gr.Row():
with gr.Column():
img_r = gr.Image(label="上传图片", type="numpy")
txt_r = gr.Textbox(label="输入问题", placeholder="请描述这张图片...")
btn_r = gr.Button("提交")
with gr.Column():
html_out = gr.HTML(label="渲染结果")
btn_r.click(
fn=predict_with_rendering,
inputs=[img_r, txt_r],
outputs=html_out,
)
if __name__ == "__main__":
demo.launch(server_name="0.0.0.0", server_port=7860)运行与部署
启动应用:
pip install gradio fastapi uvicorn pillow transformers torch torchvision matplotlib
python app.py生产环境建议使用 Docker + NVIDIA Container Toolkit 进行 GPU 加速部署,并配合 Nginx 反向代理提供 HTTPS 服务。
总结
本文从模型选择、后端接口开发、前端界面构建、流式输出到结果渲染,完整地呈现了一个多模态 AI 应用的开发流程。核心架构可概括为:
图像输入 -> 前端上传 (Gradio) -> 后端推理 (FastAPI + Qwen-VL) -> 流式返回 (SSE) -> 结果渲染 (HTML/图文混排)
该架构具备良好的可扩展性:模型可替换为 LLaVA 或 GPT-4V,前端也可替换为 React/Vue 等框架,通过 REST API 对接即可。