LLM应用开发——多模态LLM应用

from openai import OpenAI
import base64

client = OpenAI()

def encode_image(image_path: str) -> str:
    """将图像编码为 base64"""
    with open(image_path, "rb") as f:
        return base64.b64encode(f.read()).decode("utf-8")

def analyze_image(image_path: str, prompt: str = "描述这张图片") -> str:
    """分析图像内容"""
    
    base64_image = encode_image(image_path)
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": prompt},
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:image/jpeg;base64,{base64_image}",
                            "detail": "high"  # low, high, auto
                        }
                    }
                ]
            }
        ],
        max_tokens=1000
    )
    
    return response.choices[0].message.content

# 使用
result = analyze_image("photo.jpg", "这张图片中有什么？请详细描述。")
print(result)

# 使用 Sentence-Transformers 加载 CLIP 模型进行检索
from sentence_transformers import SentenceTransformer, util
from PIL import Image

# 加载模型
model = SentenceTransformer('clip-ViT-B-32')

# 编码图像
img_emb = model.encode(Image.open('car.jpg'))

# 编码文本
text_emb = model.encode(['a red sports car', 'a blue truck', 'a cat'])

# 计算相似度
cos_scores = util.cos_sim(img_emb, text_emb)
print(f"相似度分数: {cos_scores}")

import cv2
import base64

def process_video(video_path: str, seconds_per_frame: int = 2):
    video = cv2.VideoCapture(video_path)
    base64_frames = []
    
    while video.isOpened():
        success, frame = video.read()
        if not success:
            break
        
        # 按频率采样
        fps = video.get(cv2.CAP_PROP_FPS)
        frame_id = int(video.get(cv2.CAP_PROP_POS_FRAMES))
        
        if frame_id % int(fps * seconds_per_frame) == 0:
            _, buffer = cv2.imencode(".jpg", frame)
            base64_frames.append(base64.b64encode(buffer).decode("utf-8"))
            
    video.release()
    return base64_frames

# 将 base64_frames 放入 GPT-4o 的 messages 列表中发送

def analyze_image_url(image_url: str, prompt: str) -> str:
    """分析 URL 图像"""
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": prompt},
                    {
                        "type": "image_url",
                        "image_url": {"url": image_url}
                    }
                ]
            }
        ]
    )
    
    return response.choices[0].message.content

# 使用
url = "https://example.com/image.jpg"
result = analyze_image_url(url, "分析这张图片的主要内容")

def compare_images(image_paths: list, prompt: str) -> str:
    """对比多张图像"""
    
    content = [{"type": "text", "text": prompt}]
    
    for path in image_paths:
        base64_image = encode_image(path)
        content.append({
            "type": "image_url",
            "image_url": {
                "url": f"data:image/jpeg;base64,{base64_image}"
            }
        })
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": content}],
        max_tokens=1500
    )
    
    return response.choices[0].message.content

# 使用
result = compare_images(
    ["before.jpg", "after.jpg"],
    "对比这两张图片，描述它们的主要区别"
)

from pydantic import BaseModel
from typing import List, Optional
import instructor

client = instructor.from_openai(OpenAI())

class Invoice(BaseModel):
    """发票信息"""
    invoice_number: str
    date: str
    vendor_name: str
    total_amount: float
    items: List[dict]
    tax: Optional[float] = None

def parse_invoice(image_path: str) -> Invoice:
    """解析发票图像"""
    
    base64_image = encode_image(image_path)
    
    return client.chat.completions.create(
        model="gpt-4o",
        response_model=Invoice,
        messages=[
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": "请从这张发票图片中提取所有信息"},
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:image/jpeg;base64,{base64_image}",
                            "detail": "high"
                        }
                    }
                ]
            }
        ]
    )

# 使用
invoice = parse_invoice("invoice.jpg")
print(f"发票号: {invoice.invoice_number}")
print(f"金额: {invoice.total_amount}")

class ChartData(BaseModel):
    """图表数据"""
    chart_type: str  # bar, line, pie, etc.
    title: Optional[str] = None
    x_axis_label: Optional[str] = None
    y_axis_label: Optional[str] = None
    data_points: List[dict]
    insights: List[str]

def analyze_chart(image_path: str) -> ChartData:
    """分析图表并提取数据"""
    
    base64_image = encode_image(image_path)
    
    return client.chat.completions.create(
        model="gpt-4o",
        response_model=ChartData,
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": """分析这张图表：
1. 识别图表类型
2. 提取所有数据点
3. 总结主要洞察"""
                    },
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:image/jpeg;base64,{base64_image}",
                            "detail": "high"
                        }
                    }
                ]
            }
        ]
    )

class Product(BaseModel):
    """产品信息"""
    name: str
    brand: Optional[str] = None
    category: str
    estimated_price_range: Optional[str] = None
    key_features: List[str]
    condition: Optional[str] = None  # new, used, damaged

def identify_product(image_path: str) -> Product:
    """识别图片中的产品"""
    
    base64_image = encode_image(image_path)
    
    response = client.chat.completions.create(
        model="gpt-4o",
        response_model=Product,
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": "识别图片中的产品，提取详细信息"
                    },
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:image/jpeg;base64,{base64_image}"
                        }
                    }
                ]
            }
        ]
    )
    
    return response

from openai import OpenAI

client = OpenAI()

def generate_image(prompt: str, size: str = "1024x1024") -> str:
    """生成图像"""
    
    response = client.images.generate(
        model="dall-e-3",
        prompt=prompt,
        size=size,  # 1024x1024, 1792x1024, 1024x1792
        quality="hd",  # standard, hd
        n=1
    )
    
    return response.data[0].url

# 使用
image_url = generate_image(
    "一只穿着宇航服的柯基犬在月球上行走，背景是地球，科幻风格，高清细节"
)
print(f"生成的图像: {image_url}")

def create_variation(image_path: str) -> str:
    """创建图像变体"""
    
    with open(image_path, "rb") as f:
        response = client.images.create_variation(
            model="dall-e-2",
            image=f,
            n=1,
            size="1024x1024"
        )
    
    return response.data[0].url

def edit_image(
    image_path: str,
    mask_path: str,
    prompt: str
) -> str:
    """编辑图像（需要蒙版）"""
    
    with open(image_path, "rb") as image_file:
        with open(mask_path, "rb") as mask_file:
            response = client.images.edit(
                model="dall-e-2",
                image=image_file,
                mask=mask_file,
                prompt=prompt,
                n=1,
                size="1024x1024"
            )
    
    return response.data[0].url

from openai import OpenAI

client = OpenAI()

def transcribe_audio(audio_path: str, language: str = None) -> str:
    """将音频转为文字"""
    
    with open(audio_path, "rb") as audio_file:
        transcript = client.audio.transcriptions.create(
            model="whisper-1",
            file=audio_file,
            language=language,  # 可选，如 "zh", "en"
            response_format="text"
        )
    
    return transcript

# 使用
text = transcribe_audio("meeting.mp3", language="zh")
print(text)

def transcribe_with_timestamps(audio_path: str) -> dict:
    """带时间戳的音频转录"""
    
    with open(audio_path, "rb") as audio_file:
        transcript = client.audio.transcriptions.create(
            model="whisper-1",
            file=audio_file,
            response_format="verbose_json",
            timestamp_granularities=["word", "segment"]
        )
    
    return {
        "text": transcript.text,
        "segments": [
            {
                "start": seg.start,
                "end": seg.end,
                "text": seg.text
            }
            for seg in transcript.segments
        ],
        "words": [
            {
                "word": word.word,
                "start": word.start,
                "end": word.end
            }
            for word in transcript.words
        ] if hasattr(transcript, 'words') else []
    }

# 使用
result = transcribe_with_timestamps("speech.mp3")
for segment in result["segments"]:
    print(f"[{segment['start']:.2f}s - {segment['end']:.2f}s]: {segment['text']}")

def translate_audio(audio_path: str) -> str:
    """将音频翻译为英文"""
    
    with open(audio_path, "rb") as audio_file:
        translation = client.audio.translations.create(
            model="whisper-1",
            file=audio_file,
            response_format="text"
        )
    
    return translation

# 将中文音频翻译为英文
english_text = translate_audio("chinese_speech.mp3")

from pathlib import Path

def text_to_speech(
    text: str,
    output_path: str,
    voice: str = "alloy"
) -> str:
    """将文字转为语音"""
    
    # 可用声音: alloy, echo, fable, onyx, nova, shimmer
    response = client.audio.speech.create(
        model="tts-1-hd",  # tts-1 或 tts-1-hd
        voice=voice,
        input=text
    )
    
    response.stream_to_file(output_path)
    return output_path

# 使用
text_to_speech(
    "欢迎使用我们的语音合成服务！这是一个测试。",
    "output.mp3",
    voice="nova"
)

def stream_speech(text: str):
    """流式生成语音"""
    
    response = client.audio.speech.create(
        model="tts-1",
        voice="alloy",
        input=text
    )
    
    # 流式写入
    with open("stream_output.mp3", "wb") as f:
        for chunk in response.iter_bytes():
            f.write(chunk)

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.multi_modal_llms.openai import OpenAIMultiModal
from llama_index.core.schema import ImageNode, TextNode
import base64

class MultiModalRAG:
    """多模态 RAG 系统"""
    
    def __init__(self):
        self.text_index = None
        self.image_index = None
        self.mm_llm = OpenAIMultiModal(model="gpt-4o")
    
    def load_documents(self, directory: str):
        """加载文档和图像"""
        from llama_index.readers.file import ImageReader
        
        # 加载文本文档
        text_docs = SimpleDirectoryReader(
            directory,
            required_exts=[".txt", ".md", ".pdf"]
        ).load_data()
        
        # 加载图像
        image_reader = ImageReader()
        image_docs = []
        for img_path in Path(directory).glob("*.{jpg,png,jpeg}"):
            image_docs.extend(image_reader.load_data(img_path))
        
        # 创建索引
        self.text_index = VectorStoreIndex.from_documents(text_docs)
        self.image_index = VectorStoreIndex.from_documents(image_docs)
    
    def query(self, question: str) -> str:
        """多模态查询"""
        # 检索相关文本
        text_retriever = self.text_index.as_retriever(similarity_top_k=3)
        text_nodes = text_retriever.retrieve(question)
        
        # 检索相关图像
        image_retriever = self.image_index.as_retriever(similarity_top_k=2)
        image_nodes = image_retriever.retrieve(question)
        
        # 构建多模态上下文
        context_parts = []
        
        for node in text_nodes:
            context_parts.append(f"文本: {node.text}")
        
        for node in image_nodes:
            if hasattr(node, 'image'):
                context_parts.append(f"[图像: {node.metadata.get('file_name', 'image')}]")
        
        context = "\n\n".join(context_parts)
        
        # 生成回答
        prompt = f"""基于以下上下文回答问题。

上下文:
{context}

问题: {question}

回答:"""
        
        response = self.mm_llm.complete(prompt)
        return str(response)

from openai import OpenAI
import numpy as np
from typing import List
import faiss

client = OpenAI()

class ImageSearchEngine:
    """图像搜索引擎"""
    
    def __init__(self, dimension: int = 1536):
        self.dimension = dimension
        self.index = faiss.IndexFlatIP(dimension)  # 内积相似度
        self.images = []  # 存储图像路径
        self.descriptions = []  # 存储图像描述
    
    def describe_image(self, image_path: str) -> str:
        """生成图像描述"""
        base64_image = encode_image(image_path)
        
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {
                    "role": "user",
                    "content": [
                        {"type": "text", "text": "详细描述这张图片，包括主要内容、颜色、风格等"},
                        {
                            "type": "image_url",
                            "image_url": {"url": f"data:image/jpeg;base64,{base64_image}"}
                        }
                    ]
                }
            ]
        )
        
        return response.choices[0].message.content
    
    def get_embedding(self, text: str) -> List[float]:
        """获取文本嵌入"""
        response = client.embeddings.create(
            model="text-embedding-3-small",
            input=text
        )
        return response.data[0].embedding
    
    def add_image(self, image_path: str):
        """添加图像到索引"""
        # 生成描述
        description = self.describe_image(image_path)
        
        # 获取嵌入
        embedding = self.get_embedding(description)
        embedding = np.array([embedding]).astype('float32')
        
        # 添加到索引
        self.index.add(embedding)
        self.images.append(image_path)
        self.descriptions.append(description)
    
    def search(self, query: str, top_k: int = 5) -> List[dict]:
        """搜索相似图像"""
        query_embedding = self.get_embedding(query)
        query_embedding = np.array([query_embedding]).astype('float32')
        
        distances, indices = self.index.search(query_embedding, top_k)
        
        results = []
        for i, idx in enumerate(indices[0]):
            if idx != -1:
                results.append({
                    "image_path": self.images[idx],
                    "description": self.descriptions[idx],
                    "score": float(distances[0][i])
                })
        
        return results

# 使用
engine = ImageSearchEngine()
engine.add_image("photo1.jpg")
engine.add_image("photo2.jpg")

results = engine.search("一只在草地上奔跑的狗")
for r in results:
    print(f"图像: {r['image_path']}, 相似度: {r['score']:.3f}")

import cv2
from typing import List
import tempfile
import os

def extract_frames(
    video_path: str,
    num_frames: int = 10
) -> List[str]:
    """从视频中提取帧"""
    
    cap = cv2.VideoCapture(video_path)
    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    
    # 计算采样间隔
    interval = total_frames // num_frames
    
    frames = []
    frame_paths = []
    
    for i in range(num_frames):
        cap.set(cv2.CAP_PROP_POS_FRAMES, i * interval)
        ret, frame = cap.read()
        
        if ret:
            # 保存帧
            temp_path = tempfile.mktemp(suffix=".jpg")
            cv2.imwrite(temp_path, frame)
            frame_paths.append(temp_path)
    
    cap.release()
    return frame_paths

def analyze_video(video_path: str, prompt: str) -> str:
    """分析视频内容"""
    
    # 提取帧
    frame_paths = extract_frames(video_path, num_frames=8)
    
    # 构建消息
    content = [{"type": "text", "text": prompt}]
    
    for path in frame_paths:
        base64_image = encode_image(path)
        content.append({
            "type": "image_url",
            "image_url": {"url": f"data:image/jpeg;base64,{base64_image}"}
        })
    
    # 调用 GPT-4V
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": content}],
        max_tokens=2000
    )
    
    # 清理临时文件
    for path in frame_paths:
        os.remove(path)
    
    return response.choices[0].message.content

# 使用
analysis = analyze_video(
    "demo.mp4",
    "这个视频展示了什么内容？请按时间顺序描述主要事件。"
)
print(analysis)

class VideoSummarizer:
    """视频摘要生成器"""
    
    def __init__(self):
        self.client = OpenAI()
    
    def summarize(
        self,
        video_path: str,
        audio_language: str = "zh"
    ) -> dict:
        """生成视频摘要"""
        
        # 1. 提取音频转文字
        audio_path = self._extract_audio(video_path)
        transcript = self._transcribe_audio(audio_path, audio_language)
        
        # 2. 提取关键帧
        frame_paths = extract_frames(video_path, num_frames=10)
        
        # 3. 分析视觉内容
        visual_description = self._analyze_frames(frame_paths)
        
        # 4. 综合生成摘要
        summary = self._generate_summary(transcript, visual_description)
        
        # 清理
        os.remove(audio_path)
        for path in frame_paths:
            os.remove(path)
        
        return {
            "transcript": transcript,
            "visual_description": visual_description,
            "summary": summary
        }
    
    def _extract_audio(self, video_path: str) -> str:
        """从视频提取音频"""
        import moviepy.editor as mp
        
        video = mp.VideoFileClip(video_path)
        audio_path = tempfile.mktemp(suffix=".mp3")
        video.audio.write_audiofile(audio_path, logger=None)
        video.close()
        
        return audio_path
    
    def _transcribe_audio(self, audio_path: str, language: str) -> str:
        """转录音频"""
        with open(audio_path, "rb") as f:
            transcript = self.client.audio.transcriptions.create(
                model="whisper-1",
                file=f,
                language=language
            )
        return transcript.text
    
    def _analyze_frames(self, frame_paths: List[str]) -> str:
        """分析帧"""
        content = [
            {"type": "text", "text": "描述这些视频帧展示的主要视觉内容和场景变化"}
        ]
        
        for path in frame_paths:
            content.append({
                "type": "image_url",
                "image_url": {"url": f"data:image/jpeg;base64,{encode_image(path)}"}
            })
        
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": content}]
        )
        
        return response.choices[0].message.content
    
    def _generate_summary(self, transcript: str, visual: str) -> str:
        """生成综合摘要"""
        prompt = f"""基于以下视频的音频转录和视觉描述，生成一个全面的摘要。

音频转录：
{transcript}

视觉内容：
{visual}

请生成：
1. 视频主题
2. 主要内容概述
3. 关键要点（列表形式）
4. 时长估计和视频类型"""

        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}]
        )
        
        return response.choices[0].message.content

from openai import OpenAI
from typing import Optional
import base64

client = OpenAI()

class MultiModalCustomerService:
    """多模态智能客服"""
    
    def __init__(self, system_prompt: str = None):
        self.system_prompt = system_prompt or """你是一个专业的客服助手，可以：
1. 回答用户的文字问题
2. 分析用户上传的产品图片
3. 识别发票、收据等单据
4. 处理语音消息

请始终保持专业、友好的态度。"""
        self.conversation = []
    
    def chat(
        self,
        text: Optional[str] = None,
        image_path: Optional[str] = None,
        audio_path: Optional[str] = None
    ) -> str:
        """多模态对话"""
        
        user_content = []
        
        # 处理文字
        if text:
            user_content.append({"type": "text", "text": text})
        
        # 处理图像
        if image_path:
            base64_image = encode_image(image_path)
            user_content.append({
                "type": "image_url",
                "image_url": {"url": f"data:image/jpeg;base64,{base64_image}"}
            })
            if not text:
                user_content.insert(0, {"type": "text", "text": "请分析这张图片"})
        
        # 处理音频
        if audio_path:
            # 先转文字
            with open(audio_path, "rb") as f:
                transcript = client.audio.transcriptions.create(
                    model="whisper-1",
                    file=f
                )
            user_content.append({
                "type": "text",
                "text": f"[语音消息转文字]: {transcript.text}"
            })
        
        # 添加用户消息
        self.conversation.append({
            "role": "user",
            "content": user_content
        })
        
        # 调用 API
        messages = [
            {"role": "system", "content": self.system_prompt}
        ] + self.conversation
        
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            max_tokens=1000
        )
        
        assistant_message = response.choices[0].message.content
        
        # 保存助手回复
        self.conversation.append({
            "role": "assistant",
            "content": assistant_message
        })
        
        return assistant_message

# 使用示例
cs = MultiModalCustomerService()

# 文字咨询
response = cs.chat(text="这款产品的保修期是多久？")

# 图片咨询
response = cs.chat(
    text="这个产品有质量问题，请看图片",
    image_path="damage.jpg"
)

# 语音咨询
response = cs.chat(audio_path="question.mp3")

# 1. 图像压缩
from PIL import Image
import io

def compress_image(image_path: str, max_size: int = 1024) -> str:
    """压缩图像以减少 token 消耗"""
    img = Image.open(image_path)
    
    # 调整尺寸
    ratio = min(max_size / img.width, max_size / img.height)
    if ratio < 1:
        new_size = (int(img.width * ratio), int(img.height * ratio))
        img = img.resize(new_size, Image.LANCZOS)
    
    # 压缩
    buffer = io.BytesIO()
    img.save(buffer, format="JPEG", quality=85)
    
    return base64.b64encode(buffer.getvalue()).decode()

# 2. 选择合适的 detail 参数
# - low: 快速，低成本，适合简单任务
# - high: 详细分析，适合 OCR、细节识别
# - auto: 自动选择