多模态大模型应用开发：从GPT-4V到LLaVA的实战指南

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

# 多模态模型架构对比
"""
1. Encoder-Decoder架构（如BLIP）
   - Image Encoder: ViT/Llama
   - Text Decoder: GPT-2
   - Cross-attention连接

2. Encoder-only架构（如CLIP）
   - Image Encoder: ViT
   - Text Encoder: BERT
   - 对比学习

3. Decoder-only架构（如GPT-4V）
   - 统一的Transformer解码器
   - 多模态输入投影
   - 端到端生成

4. 项目架构（如LLaVA）
   - 预训练的视觉编码器
   - 预训练的语言模型
   - 简单的连接层训练
"""

from transformers import AutoProcessor, LlavaForConditionalGeneration

# 加载LLaVA模型
model = LlavaForConditionalGeneration.from_pretrained(
    "llava-hf/llava-1.5-7b-hf"
)
processor = AutoProcessor.from_pretrained(
    "llava-hf/llava-1.5-7b-hf"
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81

import torch
from PIL import Image
from transformers import AutoModelForCausalLM, AutoProcessor

class ImageUnderstanding:
    """图像理解助手"""

    def __init__(self, model_name="llava-hf/llava-1.5-7b-hf"):
        self.model = AutoModelForCausalLM.from_pretrained(
            model_name,
            torch_dtype=torch.float16,
            device_map="auto"
        )
        self.processor = AutoProcessor.from_pretrained(model_name)

    def analyze_image(
        self,
        image_path: str,
        question: str
    ) -> str:
        """分析图像内容"""

        # 加载图像
        image = Image.open(image_path).convert("RGB")

        # 准备输入
        prompt = f"USER: <image>\n{question}\nASSISTANT:"

        inputs = self.processor(
            text=prompt,
            images=image,
            return_tensors="pt"
        ).to(self.model.device)

        # 生成响应
        with torch.no_grad():
            outputs = self.model.generate(
                **inputs,
                max_new_tokens=500,
                do_sample=False,
            )

        # 解码响应
        response = self.processor.decode(
            outputs[0],
            skip_special_tokens=True
        )

        return response.split("ASSISTANT:")[-1].strip()

    def describe_scene(self, image_path: str) -> str:
        """描述场景"""
        return self.analyze_image(
            image_path,
            "Please describe this image in detail, including objects, people, activities, and the environment."
        )

    def extract_text(self, image_path: str) -> str:
        """提取文字"""
        return self.analyze_image(
            image_path,
            "Extract all text visible in this image. Organize it logically."
        )

    def detect_objects(self, image_path: str) -> str:
        """检测物体"""
        return self.analyze_image(
            image_path,
            "List all objects visible in this image with their locations and relationships."
        )

# 使用示例
assistant = ImageUnderstanding()

# 分析图片
description = assistant.describe_scene("path/to/image.jpg")
print(description)

# 提取文字
text = assistant.extract_text("document.jpg")
print(text)

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146

from typing import List
import clip
import torch
from PIL import Image

class MultimodalRAG:
    """多模态RAG系统"""

    def __init__(self):
        # 加载CLIP模型
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        self.model, self.preprocess = clip.load("ViT-B/32", device=self.device)

        # 初始化向量存储
        self.image_store = []
        self.text_store = []

    def add_documents(
        self,
        images: List[str] = None,
        texts: List[str] = None
    ):
        """添加多模态文档"""

        # 处理图像
        if images:
            for img_path in images:
                image = Image.open(img_path)
                image_input = self.preprocess(image).unsqueeze(0).to(self.device)

                with torch.no_grad():
                    image_features = self.model.encode_image(image_input)

                self.image_store.append({
                    "path": img_path,
                    "features": image_features.cpu(),
                    "metadata": {"type": "image"}
                })

        # 处理文本
        if texts:
            for text in texts:
                text_input = clip.tokenize([text]).to(self.device)

                with torch.no_grad():
                    text_features = self.model.encode_text(text_input)

                self.text_store.append({
                    "content": text,
                    "features": text_features.cpu(),
                    "metadata": {"type": "text"}
                })

    def retrieve(
        self,
        query_image: str = None,
        query_text: str = None,
        top_k: int = 5
    ) -> List[dict]:
        """跨模态检索"""

        query_features = None

        # 图像查询
        if query_image:
            image = Image.open(query_image)
            image_input = self.preprocess(image).unsqueeze(0).to(self.device)

            with torch.no_grad():
                query_features = self.model.encode_image(image_input)

        # 文本查询
        elif query_text:
            text_input = clip.tokenize([query_text]).to(self.device)

            with torch.no_grad():
                query_features = self.model.encode_text(text_input)

        # 计算相似度
        results = []

        # 检索图像
        for doc in self.image_store:
            similarity = torch.cosine_similarity(
                query_features,
                doc["features"].to(self.device)
            ).item()

            results.append({
                "content": doc["path"],
                "score": similarity,
                "type": "image"
            })

        # 检索文本
        for doc in self.text_store:
            similarity = torch.cosine_similarity(
                query_features,
                doc["features"].to(self.device)
            ).item()

            results.append({
                "content": doc["content"],
                "score": similarity,
                "type": "text"
            })

        # 排序并返回top-k
        results.sort(key=lambda x: x["score"], reverse=True)

        return results[:top_k]

    def multimodal_rag(
        self,
        query_image: str,
        query_text: str
    ) -> str:
        """多模态RAG问答"""

        # 检索相关内容
        image_results = self.retrieve(query_image=query_image)
        text_results = self.retrieve(query_text=query_text)

        # 构建prompt
        context = "Retrieved Information:\n"

        context += "\nRelevant Images:\n"
        for result in image_results[:3]:
            context += f"- {result['content']}\n"

        context += "\nRelevant Texts:\n"
        for result in text_results[:3]:
            context += f"- {result['content']}\n"

        prompt = f"""
        {context}

        Based on the above retrieved information and the provided image,
        answer the following question:
        {query_text}
        """

        # 使用多模态LLM生成答案
        response = self.analyze_image(query_image, prompt)

        return response

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94

import cv2
import numpy as np
from typing import List

class VideoAnalyzer:
    """视频分析器"""

    def __init__(self, mlm_model):
        self.mlm_model = mlm_model

    def extract_key_frames(
        self,
        video_path: str,
        num_frames: int = 10
    ) -> List[str]:
        """提取关键帧"""

        cap = cv2.VideoCapture(video_path)
        total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
        frame_indices = np.linspace(0, total_frames - 1, num_frames, dtype=int)

        key_frames = []
        for idx in frame_indices:
            cap.set(cv2.CAP_PROP_POS_FRAMES, idx)
            ret, frame = cap.read()

            if ret:
                frame_path = f"frame_{idx}.jpg"
                cv2.imwrite(frame_path, frame)
                key_frames.append(frame_path)

        cap.release()
        return key_frames

    def analyze_video(
        self,
        video_path: str,
        question: str
    ) -> str:
        """分析视频内容"""

        # 提取关键帧
        key_frames = self.extract_key_frames(video_path)

        # 分析关键帧
        frame_descriptions = []
        for frame_path in key_frames:
            description = self.mlm_model.describe_scene(frame_path)
            frame_descriptions.append(description)

        # 综合分析
        prompt = f"""
        Here are descriptions of key frames from a video:
        {chr(10).join([f'Frame {i+1}: {desc}' for i, desc in enumerate(frame_descriptions)])}

        Based on these frame descriptions, answer:
        {question}
        """

        # 使用文本LLM生成答案
        response = self.text_llm.generate(prompt)

        return response

    def detect_actions(self, video_path: str) -> List[str]:
        """检测动作"""

        key_frames = self.extract_key_frames(video_path, num_frames=20)

        actions = []
        for i, frame_path in enumerate(key_frames):
            action = self.mlm_model.analyze_image(
                frame_path,
                "What action is being performed in this frame? Be concise."
            )
            actions.append(f"Frame {i+1}: {action}")

        return actions

    def summarize_video(self, video_path: str) -> str:
        """视频摘要"""

        # 提取关键帧
        key_frames = self.extract_key_frames(video_path)

        # 生成摘要
        prompt = "Create a concise summary of this video based on these key frames:"
        for i, frame_path in enumerate(key_frames[:5]):
            frame_desc = self.mlm_model.describe_scene(frame_path)
            prompt += f"\nFrame {i+1}: {frame_desc}"

        summary = self.mlm_model.text_llm.generate(prompt)

        return summary

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90

import whisper
import torch
from typing import Dict

class MultimodalAudioAssistant:
    """多模态音频助手"""

    def __init__(self):
        # 加载Whisper模型
        self.audio_model = whisper.load_model("base")

        # 加载多模态LLM
        self.mlm_model = AutoModelForCausalLM.from_pretrained(
            "llava-hf/llava-1.5-7b-hf"
        )

    def transcribe_audio(
        self,
        audio_path: str
    ) -> Dict[str, any]:
        """转录音频"""

        # 转录
        result = self.audio_model.transcribe(
            audio_path,
            language="zh",
            task="transcribe"
        )

        return {
            "text": result["text"],
            "language": result["language"],
            "segments": result["segments"]
        }

    def audio_question_answering(
        self,
        audio_path: str,
        question: str
    ) -> str:
        """音频问答"""

        # 转录音频
        transcription = self.transcribe_audio(audio_path)
        audio_text = transcription["text"]

        # 使用LLM回答问题
        prompt = f"""
        Audio Transcript:
        {audio_text}

        Question: {question}

        Based on the audio transcript, provide a detailed answer.
        """

        response = self.mlm_model.generate(prompt)

        return response

    def audio_visual_sync(
        self,
        audio_path: str,
        video_path: str
    ) -> Dict[str, any]:
        """音视频同步分析"""

        # 转录音频
        audio_text = self.transcribe_audio(audio_path)

        # 提取视频关键帧
        key_frames = self.extract_key_frames(video_path)

        # 分析音视频关联
        prompt = f"""
        Audio: {audio_text['text']}

        Visual Content:
        {[self.mlm_model.describe_scene(frame) for frame in key_frames[:3]]}

        Analyze the relationship between the audio and visual content.
        Are they consistent? What is the overall message?
        """

        analysis = self.mlm_model.generate(prompt)

        return {
            "audio_text": audio_text,
            "visual_summary": analysis
        }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

from transformers import DonutProcessor, VisionEncoderDecoderModel

class DocumentAnalyzer:
    """文档分析器"""

    def __init__(self):
        # 加载Donut模型
        self.processor = DonutProcessor.from_pretrained(
            "naver-clova-ix/donut-base-finetuned-docvqa"
        )
        self.model = VisionEncoderDecoderModel.from_pretrained(
            "naver-clova-ix/donut-base-finetuned-docvqa"
        )

    def analyze_document(
        self,
        image_path: str,
        question: str
    ) -> str:
        """分析文档图像"""

        # 加载文档图像
        image = Image.open(image_path).convert("RGB")

        # 准备输入
        prompt = f"<s>{question}</s>"
        task_prompt = f"<s_docvqa><s_question>{question}</s_answer><s>"

        inputs = self.processor(
            image,
            task_prompt,
            return_tensors="pt"
        )

        # 生成答案
        with torch.no_grad():
            outputs = self.model.generate(
                inputs.pixel_values,
                inputs.input_ids[0:1],
            )

        # 解码
        generated_text = self.processor.batch_decode(outputs)[0]

        # 提取答案
        answer = generated_text.split("</s_answer>")[-1].strip()

        return answer

    def extract_table(self, image_path: str) -> str:
        """提取表格"""

        return self.analyze_document(
            image_path,
            "Extract all tables from this document in Markdown format."
        )

    def extract_form_fields(self, image_path: str) -> Dict[str, str]:
        """提取表单字段"""

        result = self.analyze_document(
            image_path,
            "List all form fields in this document with their labels and values."
        )

        # 解析结果为结构化数据
        fields = {}
        for line in result.split("\n"):
            if ":" in line:
                label, value = line.split(":", 1)
                fields[label.strip()] = value.strip()

        return fields

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98

from fastapi import FastAPI, UploadFile, File
from pydantic import BaseModel
from typing import Optional

app = FastAPI()

class MultimodalChatBot:
    """多模态聊天机器人"""

    def __init__(self):
        self.vlm_model = LlavaForConditionalGeneration.from_pretrained(
            "llava-hf/llava-1.5-7b-hf",
            torch_dtype=torch.float16,
            device_map="auto"
        )
        self.processor = AutoProcessor.from_pretrained(
            "llava-hf/llava-1.5-7b-hf"
        )

        # 对话历史
        self.conversation_history = {}

    async def chat(
        self,
        user_id: str,
        message: str,
        image: Optional[UploadFile] = None
    ) -> str:
        """多模态对话"""

        # 获取历史
        history = self.conversation_history.get(user_id, [])

        # 准备输入
        if image:
            # 有图像
            image_bytes = await image.read()
            image_pil = Image.open(io.BytesIO(image_bytes)).convert("RGB")

            prompt = self._build_prompt_with_image(history, message)

            inputs = self.processor(
                text=prompt,
                images=image_pil,
                return_tensors="pt"
            ).to(self.vlm_model.device)

            # 生成
            with torch.no_grad():
                outputs = self.vlm_model.generate(
                    **inputs,
                    max_new_tokens=500,
                    do_sample=True,
                    temperature=0.7,
                )

            response = self.processor.decode(outputs[0], skip_special_tokens=True)

        else:
            # 纯文本
            prompt = self._build_prompt(history, message)
            response = self.text_llm.generate(prompt)

        # 更新历史
        history.append({"role": "user", "content": message})
        history.append({"role": "assistant", "content": response})
        self.conversation_history[user_id] = history[-10:]  # 保留最近10轮

        return response

    def _build_prompt_with_image(self, history, message):
        prompt = "USER: <image>\n"

        for h in history:
            prompt += f"{h['role'].upper()}: {h['content']}\n"

        prompt += f"USER: {message}\nASSISTANT:"
        return prompt

    def _build_prompt(self, history, message):
        prompt = ""
        for h in history:
            prompt += f"{h['role'].upper()}: {h['content']}\n"

        prompt += f"USER: {message}\nASSISTANT:"
        return prompt

chatbot = MultimodalChatBot()

@app.post("/chat/{user_id}")
async def chat_endpoint(
    user_id: str,
    message: str = Form(...),
    image: UploadFile = File(None)
):
    response = await chatbot.chat(user_id, message, image)

    return {"response": response}