GPT-4 Vision 图像识别完整指南

import openai
import base64
import requests
from PIL import Image
import io

# 配置 API
openai.api_key = "your-api-key"
openai.api_base = "https://api.n1n.ai/v1"

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

def analyze_image(image_path, prompt):
    """使用 GPT-4 Vision 分析图像"""
    # 编码图像
    base64_image = encode_image(image_path)
    
    response = openai.ChatCompletion.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)

# 分析网络图片
def analyze_url_image(image_url, prompt):
    """分析网络图片"""
    response = openai.ChatCompletion.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

# 多图像分析
def analyze_multiple_images(image_paths, prompt):
    """同时分析多张图像"""
    messages = [{"type": "text", "text": prompt}]
    
    # 添加所有图像
    for path in image_paths:
        base64_image = encode_image(path)
        messages.append({
            "type": "image_url",
            "image_url": {
                "url": f"data:image/jpeg;base64,{base64_image}",
                "detail": "low"  # 多图时建议用 low 节省 token
            }
        })
    
    response = openai.ChatCompletion.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": messages
        }],
        max_tokens=2000
    )
    
    return response.choices[0].message.content

# 使用示例：比较多张图片
images = ["product1.jpg", "product2.jpg", "product3.jpg"]
comparison = analyze_multiple_images(
    images,
    "请比较这三个产品的设计特点，指出它们的相似之处和差异"
)

# 图像序列分析（如连续帧）
def analyze_image_sequence(frames, task="describe_changes"):
    """分析图像序列，如视频帧"""
    
    if task == "describe_changes":
        prompt = "这是一个时间序列的图像。请描述从第一张到最后一张图像发生的变化。"
    elif task == "detect_motion":
        prompt = "分析这些连续帧中的运动和动作。"
    elif task == "summarize_story":
        prompt = "根据这些图像讲述一个连贯的故事。"
    
    return analyze_multiple_images(frames, prompt)

# OCR 文字识别应用
def extract_text_from_image(image_path, language="中英文"):
    """从图像中提取文字"""
    prompt = f"""请提取图像中的所有{language}文字内容。
    要求：
    1. 保持原始格式和排版
    2. 标注不确定的文字
    3. 如果有表格，请用 Markdown 格式呈现
    4. 识别所有可见的文字，包括水印、标签等
    """
    
    return analyze_image(image_path, prompt)

# 文档分析
def analyze_document(image_path):
    """分析文档图像"""
    prompt = """请分析这个文档图像：
    1. 文档类型（发票、合同、报告等）
    2. 关键信息提取（日期、金额、参与方等）
    3. 文档结构和布局
    4. 重要条款或要点
    请以结构化的方式输出结果。
    """
    
    result = analyze_image(image_path, prompt)
    
    # 结构化处理结果
    return {
        "raw_text": extract_text_from_image(image_path),
        "analysis": result
    }

# 手写识别
def recognize_handwriting(image_path):
    """识别手写内容"""
    prompt = """请识别图像中的手写内容：
    1. 转录所有手写文字
    2. 如果字迹不清，请标注 [?]
    3. 保持原始的段落结构
    4. 识别任何图表或符号
    """
    
    return analyze_image(image_path, prompt)

# 表格数据提取
def extract_table_data(image_path):
    """从图像中提取表格数据"""
    prompt = """请将图像中的表格转换为 CSV 格式：
    1. 第一行为表头
    2. 使用逗号分隔
    3. 处理合并单元格
    4. 保留所有数据
    
    输出格式示例：
    列1,列2,列3
    数据1,数据2,数据3
    """
    
    csv_data = analyze_image(image_path, prompt)
    
    # 可以进一步处理为 DataFrame
    import pandas as pd
    from io import StringIO
    
    try:
        df = pd.read_csv(StringIO(csv_data))
        return df
    except:
        return csv_data

# 高级应用场景

# 1. 医疗图像分析（注意：仅供参考，不能替代专业诊断）
def analyze_medical_image(image_path):
    """分析医疗图像"""
    prompt = """请分析这张医疗图像（仅供教育参考）：
    1. 描述可见的解剖结构
    2. 指出任何明显的异常
    3. 图像质量评估
    4. 建议可能需要的进一步检查
    
    声明：这只是AI分析，不能替代专业医疗诊断。
    """
    
    return analyze_image(image_path, prompt)

# 2. 产品质量检测
def quality_inspection(image_path, product_type="electronic"):
    """产品质量检测"""
    prompt = f"""请检查这个{product_type}产品的图像：
    1. 外观缺陷（划痕、凹陷、污渍等）
    2. 组装问题（对齐、间隙、松动等）
    3. 标签和标识是否正确
    4. 整体质量评分（1-10）
    
    请详细列出所有发现的问题。
    """
    
    return analyze_image(image_path, prompt)

# 3. 安全监控分析
def security_analysis(image_path):
    """安全监控图像分析"""
    prompt = """分析这张监控图像：
    1. 场景中的人数
    2. 人员活动和行为
    3. 可疑或异常情况
    4. 环境安全隐患
    5. 时间和光照条件
    """
    
    return analyze_image(image_path, prompt)

# 4. 艺术作品分析
def analyze_artwork(image_path):
    """艺术作品分析"""
    prompt = """请分析这件艺术作品：
    1. 艺术风格和流派
    2. 使用的技法和媒介
    3. 色彩运用和构图
    4. 主题和象征意义
    5. 可能的创作时期
    6. 艺术价值评估
    """
    
    return analyze_image(image_path, prompt)

# 5. 房地产评估
def analyze_property(image_paths):
    """房地产图像分析"""
    prompt = """请评估这些房产照片：
    1. 房间类型和功能
    2. 装修风格和质量
    3. 空间大小估算
    4. 设施和家具情况
    5. 维护状况
    6. 改进建议
    """
    
    return analyze_multiple_images(image_paths, prompt)

# 性能优化策略

class VisionOptimizer:
    """GPT-4 Vision 优化工具"""
    
    @staticmethod
    def resize_image(image_path, max_size=(2048, 2048)):
        """调整图像大小以优化性能"""
        from PIL import Image
        
        img = Image.open(image_path)
        img.thumbnail(max_size, Image.Resampling.LANCZOS)
        
        # 转换为 RGB（如果需要）
        if img.mode in ('RGBA', 'LA'):
            background = Image.new('RGB', img.size, (255, 255, 255))
            background.paste(img, mask=img.split()[-1])
            img = background
        
        # 保存优化后的图像
        optimized_path = f"optimized_{image_path}"
        img.save(optimized_path, "JPEG", quality=85)
        
        return optimized_path
    
    @staticmethod
    def calculate_token_cost(image_path, detail="high"):
        """估算图像 token 消耗"""
        from PIL import Image
        
        img = Image.open(image_path)
        width, height = img.size
        
        if detail == "low":
            # 低细节模式固定 85 tokens
            return 85
        
        # 高细节模式计算
        # 首先缩放到 2048x2048 内
        if width > 2048 or height > 2048:
            ratio = min(2048/width, 2048/height)
            width = int(width * ratio)
            height = int(height * ratio)
        
        # 然后缩放到 768px 的最短边
        if min(width, height) > 768:
            ratio = 768 / min(width, height)
            width = int(width * ratio)
            height = int(height * ratio)
        
        # 计算 512x512 的块数
        tiles_width = (width + 511) // 512
        tiles_height = (height + 511) // 512
        total_tiles = tiles_width * tiles_height
        
        # 每个块 170 tokens，加上基础 85 tokens
        return 170 * total_tiles + 85
    
    @staticmethod
    def batch_process_images(image_paths, prompt, batch_size=5):
        """批量处理图像"""
        results = []
        
        for i in range(0, len(image_paths), batch_size):
            batch = image_paths[i:i+batch_size]
            
            # 优化每张图像
            optimized_batch = []
            for path in batch:
                optimized_path = VisionOptimizer.resize_image(path)
                optimized_batch.append(optimized_path)
            
            # 批量分析
            result = analyze_multiple_images(optimized_batch, prompt)
            results.append(result)
            
            # 清理临时文件
            for path in optimized_batch:
                os.remove(path)
        
        return results

# 使用示例
optimizer = VisionOptimizer()

# 1. 计算 token 成本
image = "large_photo.jpg"
tokens = optimizer.calculate_token_cost(image, detail="high")
cost = tokens * 0.01 / 1000  # 假设价格
print(f"预计消耗 {tokens} tokens，成本约 ${cost:.4f}")

# 2. 优化大图像
optimized = optimizer.resize_image("huge_image.png")

# 3. 批量处理
images = ["img1.jpg", "img2.jpg", "img3.jpg", "img4.jpg", "img5.jpg"]
results = optimizer.batch_process_images(
    images,
    "描述每张图片的主要内容",
    batch_size=3
)

# 流式处理大量图像
import asyncio
import aiohttp
from typing import List, AsyncGenerator

class VisionStreamer:
    """流式处理图像分析"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.n1n.ai/v1"
    
    async def analyze_image_stream(
        self,
        image_path: str,
        prompt: str
    ) -> AsyncGenerator[str, None]:
        """流式返回图像分析结果"""
        
        base64_image = encode_image(image_path)
        
        async with aiohttp.ClientSession() as session:
            async with session.post(
                f"{self.base_url}/chat/completions",
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json"
                },
                json={
                    "model": "gpt-4o",
                    "messages": [{
                        "role": "user",
                        "content": [
                            {"type": "text", "text": prompt},
                            {
                                "type": "image_url",
                                "image_url": {
                                    "url": f"data:image/jpeg;base64,{base64_image}"
                                }
                            }
                        ]
                    }],
                    "stream": True
                }
            ) as response:
                async for line in response.content:
                    if line:
                        decoded = line.decode('utf-8').strip()
                        if decoded.startswith("data: "):
                            data = decoded[6:]
                            if data != "[DONE]":
                                try:
                                    import json
                                    chunk = json.loads(data)
                                    content = chunk['choices'][0]['delta'].get('content', '')
                                    if content:
                                        yield content
                                except:
                                    pass

# 使用流式处理
async def main():
    streamer = VisionStreamer("your-api-key")
    
    # 实时输出分析结果
    async for chunk in streamer.analyze_image_stream(
        "complex_image.jpg",
        "详细分析这张图片的所有元素"
    ):
        print(chunk, end='', flush=True)

# 运行
asyncio.run(main())