# 核心依赖pip install openai          # 兼容大多数模型的 API 接口pip install pillow          # 图片处理pip install httpx           # 异步 HTTP 请求pip install python-dotenvpip install rich            # 终端美化输出pip install base64          # 图片编码（标准库）# 可选：本地模型pip install transformers torch  # 本地 LLaVApip install gradio              # WebUI

# config.pyimport osfrom dotenv import load_dotenvload_dotenv()# === OpenAI / GPT-4o ===OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")OPENAI_BASE_URL = "https://api.openai.com/v1"# === 通义千问 VL（推荐）===QWEN_API_KEY = os.getenv("DASHSCOPE_API_KEY")QWEN_BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1"QWEN_MODEL = "qwen-vl-max"# === 默认使用 Qwen-VL ===DEFAULT_API_KEY = QWEN_API_KEYDEFAULT_BASE_URL = QWEN_BASE_URLDEFAULT_MODEL = QWEN_MODEL

# .envOPENAI_API_KEY=sk-xxxxDASHSCOPE_API_KEY=sk-xxxx   # 阿里云灵积 API Key

# utils/image_utils.pyimport base64import httpxfrom pathlib import Pathfrom PIL import Imageimport iodef image_to_base64(image_path: str) -> str:    """本地图片转 Base64 编码"""    with open(image_path, "rb") as f:        data = f.read()    return base64.b64encode(data).decode("utf-8")def url_to_base64(image_url: str) -> str:    """网络图片转 Base64"""    response = httpx.get(image_url, timeout=30)    response.raise_for_status()    return base64.b64encode(response.content).decode("utf-8")def resize_image(image_path: str, max_size: int = 1024) -> str:    """    压缩图片（节省 Token）    API 通常对图片分辨率有优化建议    """    img = Image.open(image_path)    # 保持比例缩放    width, height = img.size    if max(width, height) > max_size:        ratio = max_size / max(width, height)        new_size = (int(width * ratio), int(height * ratio))        img = img.resize(new_size, Image.LANCZOS)    # 保存到内存    buffer = io.BytesIO()    img.save(buffer, format="JPEG", quality=85)    buffer.seek(0)    return base64.b64encode(buffer.read()).decode("utf-8")def get_image_info(image_path: str) -> dict:    """获取图片基本信息"""    img = Image.open(image_path)    file_size = Path(image_path).stat().st_size    return {        "width": img.width,        "height": img.height,        "mode": img.mode,        "format": img.format,        "size_kb": file_size // 1024,    }

# vision/image_analyzer.pyfrom openai import OpenAIfrom utils.image_utils import image_to_base64, resize_imagefrom config import DEFAULT_API_KEY, DEFAULT_BASE_URL, DEFAULT_MODELfrom pathlib import Pathclass ImageAnalyzer:    """多模态图片分析器"""    def __init__(self, api_key=None, base_url=None, model=None):        self.client = OpenAI(            api_key=api_key or DEFAULT_API_KEY,            base_url=base_url or DEFAULT_BASE_URL,        )        self.model = model or DEFAULT_MODEL    def _build_image_content(self, image_source: str) -> dict:        """构建图片消息内容（支持本地路径或 URL）"""        if image_source.startswith(("http://", "https://")):            # 直接使用 URL            return {                "type": "image_url",                "image_url": {"url": image_source, "detail": "high"}            }        else:            # 本地文件转 Base64            suffix = Path(image_source).suffix.lower()            mime_map = {".jpg": "jpeg", ".jpeg": "jpeg", ".png": "png", ".gif": "gif", ".webp": "webp"}            mime = mime_map.get(suffix, "jpeg")            b64 = resize_image(image_source)   # 自动压缩            return {                "type": "image_url",                "image_url": {"url": f"data:image/{mime};base64,{b64}", "detail": "high"}            }    def describe(self, image_source: str, prompt: str = "请详细描述这张图片的内容") -> str:        """图片内容描述"""        response = self.client.chat.completions.create(            model=self.model,            messages=[                {                    "role": "user",                    "content": [                        self._build_image_content(image_source),                        {"type": "text", "text": prompt}                    ]                }            ],            max_tokens=1024,        )        return response.choices[0].message.content    def ask(self, image_source: str, question: str) -> str:        """视觉问答"""        return self.describe(image_source, question)    def analyze_multiple(self, image_sources: list, question: str) -> str:        """多图片联合分析"""        content = []        for img in image_sources:            content.append(self._build_image_content(img))        content.append({"type": "text", "text": question})        response = self.client.chat.completions.create(            model=self.model,            messages=[{"role": "user", "content": content}],            max_tokens=2048,        )        return response.choices[0].message.content# 快速测试if __name__ == "__main__":    analyzer = ImageAnalyzer()    # 描述网络图片    result = analyzer.describe(        "https://upload.wikimedia.org/wikipedia/commons/thumb/3/3a/Cat03.jpg/1200px-Cat03.jpg",        "用中文详细描述这张图片，包括颜色、构图和细节"    )    print(result)

# vision/vqa.pyfrom image_analyzer import ImageAnalyzerfrom typing import List, Dictimport jsonclass VisualQA:    """视觉问答系统"""    def __init__(self):        self.analyzer = ImageAnalyzer()    def single_qa(self, image: str, question: str) -> str:        """单图问答"""        return self.analyzer.ask(image, question)    def batch_qa(self, image: str, questions: List[str]) -> Dict[str, str]:        """批量问答（一张图，多个问题）"""        # 将所有问题合并为一次请求，节省 API 调用        combined_question = "请回答以下所有问题，用 JSON 格式返回，key 为问题编号：\n"        for i, q in enumerate(questions, 1):            combined_question += f"Q{i}: {q}\n"        combined_question += "\n返回格式：{\"Q1\": \"答案1\", \"Q2\": \"答案2\", ...}"        raw = self.analyzer.ask(image, combined_question)        # 解析 JSON 结果        try:            # 提取 JSON 部分            start = raw.find("{")            end = raw.rfind("}") + 1            answers = json.loads(raw[start:end])            return {questions[i]: answers.get(f"Q{i+1}", "") for i in range(len(questions))}        except Exception:            return {"raw_answer": raw}    def structured_extraction(self, image: str, schema: dict) -> dict:        """        结构化信息提取        从图片中按指定字段提取结构化数据        schema 示例：        {            "商品名称": "字符串",            "价格": "数字",            "规格": "字符串列表",            "是否有优惠": "布尔值"        }        """        prompt = f"""请从图片中提取以下信息，以 JSON 格式返回：需要提取的字段：{json.dumps(schema, ensure_ascii=False, indent=2)}注意：- 如果图片中没有某个字段的信息，返回 null- 严格按照指定的数据类型返回- 只返回 JSON，不要其他文字JSON 结果："""        raw = self.analyzer.ask(image, prompt)        try:            start = raw.find("{")            end = raw.rfind("}") + 1            return json.loads(raw[start:end])        except Exception:            return {"error": "解析失败", "raw": raw}# 使用示例if __name__ == "__main__":    vqa = VisualQA()    # 从商品图片提取结构化数据    product_schema = {        "商品名称": "字符串",        "品牌": "字符串",        "价格": "数字（元）",        "主要成分": "字符串列表",        "净含量": "字符串"    }    result = vqa.structured_extraction("product.jpg", product_schema)    print(json.dumps(result, ensure_ascii=False, indent=2))

# vision/chart_extractor.pyimport jsonimport pandas as pdfrom image_analyzer import ImageAnalyzerclass ChartExtractor:    """图表数据提取器"""    def __init__(self):        self.analyzer = ImageAnalyzer()    def extract_table(self, image: str) -> pd.DataFrame:        """        从图片中提取表格数据，返回 DataFrame        支持：截图表格、扫描件表格、Excel 截图等        """        prompt = """请识别图片中的表格，将其转换为 JSON 格式。要求：1. 返回格式为 {"headers": ["列1", "列2", ...], "rows": [["值1", "值2", ...], ...]}2. 保持原始数据，不要修改数值3. 如果单元格为空，填写 null4. 只返回 JSON，不要其他内容JSON："""        raw = self.analyzer.ask(image, prompt)        try:            start = raw.find("{")            end = raw.rfind("}") + 1            data = json.loads(raw[start:end])            df = pd.DataFrame(data["rows"], columns=data["headers"])            return df        except Exception as e:            print(f"解析失败：{e}")            return pd.DataFrame()    def extract_chart_data(self, image: str) -> dict:        """        从折线图/柱状图/饼图中提取数据        """        prompt = """请分析这张图表，提取其中的数据。请返回以下 JSON 格式：{    "chart_type": "图表类型（折线图/柱状图/饼图/散点图等）",    "title": "图表标题",    "x_axis": "X轴标签",    "y_axis": "Y轴标签",    "data_series": [        {            "name": "数据系列名称",            "data": [{"x": "x值", "y": "y值"}, ...]        }    ],    "summary": "一句话总结图表趋势"}只返回 JSON："""        raw = self.analyzer.ask(image, prompt)        try:            start = raw.find("{")            end = raw.rfind("}") + 1            return json.loads(raw[start:end])        except Exception:            return {"raw": raw}    def compare_charts(self, images: list, question: str = "对比这些图表，总结主要差异") -> str:        """多图表对比分析"""        return self.analyzer.analyze_multiple(images, question)# 使用示例if __name__ == "__main__":    extractor = ChartExtractor()    # 提取表格数据并保存为 CSV    df = extractor.extract_table("sales_table.png")    if not df.empty:        df.to_csv("extracted_data.csv", index=False, encoding="utf-8-sig")        print("✅ 表格数据已提取：")        print(df.to_string())    # 分析折线图    chart_data = extractor.extract_chart_data("revenue_chart.png")    print("\n📊 图表数据：")    print(json.dumps(chart_data, ensure_ascii=False, indent=2))

# vision/ocr_engine.pyimport reimport jsonfrom image_analyzer import ImageAnalyzerclass SmartOCR:    """    智能 OCR 引擎    比传统 OCR 更强：理解上下文，输出结构化结果    """    def __init__(self):        self.analyzer = ImageAnalyzer()    def extract_text(self, image: str, preserve_layout: bool = True) -> str:        """        提取图片中的文字        Args:            preserve_layout: 是否保持原始排版布局        """        if preserve_layout:            prompt = "请识别图片中的所有文字，尽量保持原始排版和换行格式，直接输出文字内容。"        else:            prompt = "请识别并提取图片中的所有文字，按阅读顺序输出。"        return self.analyzer.ask(image, prompt)    def extract_invoice(self, image: str) -> dict:        """发票信息提取"""        schema = {            "发票类型": "字符串",            "发票号码": "字符串",            "开票日期": "字符串（YYYY-MM-DD）",            "购买方名称": "字符串",            "购买方税号": "字符串",            "销售方名称": "字符串",            "销售方税号": "字符串",            "商品明细": "列表，每项包含名称、数量、单价、金额",            "不含税金额": "数字",            "税率": "字符串",            "税额": "数字",            "价税合计": "数字"        }        prompt = f"""请从这张发票图片中提取以下信息，返回 JSON 格式：{json.dumps(schema, ensure_ascii=False, indent=2)}只返回 JSON："""        raw = self.analyzer.ask(image, prompt)        try:            start = raw.find("{")            end = raw.rfind("}") + 1            return json.loads(raw[start:end])        except Exception:            return {"error": "解析失败", "raw_text": raw}    def extract_id_card(self, image: str) -> dict:        """        身份证信息提取        ⚠️ 注意：生产环境请遵守数据合规要求        """        prompt = """请提取身份证上的信息，返回 JSON：{"姓名": "", "性别": "", "民族": "", "出生日期": "", "住址": "", "身份证号": ""}只返回 JSON："""        raw = self.analyzer.ask(image, prompt)        try:            start = raw.find("{")            end = raw.rfind("}") + 1            return json.loads(raw[start:end])        except Exception:            return {"raw": raw}    def extract_business_card(self, image: str) -> dict:        """名片信息提取"""        prompt = """请提取名片上的所有信息，返回 JSON：{    "姓名": "",    "职位": "",    "公司": "",    "手机": [],    "邮箱": [],    "微信": "",    "地址": "",    "网站": ""}只返回 JSON："""        raw = self.analyzer.ask(image, prompt)        try:            start = raw.find("{")            end = raw.rfind("}") + 1            return json.loads(raw[start:end])        except Exception:            return {"raw": raw}

# vision/batch_processor.pyimport asyncioimport aiohttpfrom pathlib import Pathfrom typing import List, Callablefrom openai import AsyncOpenAIfrom utils.image_utils import resize_imagefrom config import DEFAULT_API_KEY, DEFAULT_BASE_URL, DEFAULT_MODELfrom rich.progress import Progress, SpinnerColumn, TextColumn, BarColumnfrom rich.console import Consoleconsole = Console()class BatchImageProcessor:    """异步批量图片处理器"""    def __init__(self, max_concurrent: int = 5):        """        Args:            max_concurrent: 最大并发请求数（避免超出 API 速率限制）        """        self.client = AsyncOpenAI(            api_key=DEFAULT_API_KEY,            base_url=DEFAULT_BASE_URL,        )        self.semaphore = asyncio.Semaphore(max_concurrent)        self.model = DEFAULT_MODEL    async def _process_single(self, image_path: str, prompt: str) -> dict:        """处理单张图片（异步）"""        async with self.semaphore:            try:                b64 = resize_image(image_path)                response = await self.client.chat.completions.create(                    model=self.model,                    messages=[{                        "role": "user",                        "content": [                            {                                "type": "image_url",                                "image_url": {"url": f"data:image/jpeg;base64,{b64}"}                            },                            {"type": "text", "text": prompt}                        ]                    }],                    max_tokens=512,                )                return {                    "file": image_path,                    "result": response.choices[0].message.content,                    "success": True,                }            except Exception as e:                return {"file": image_path, "result": str(e), "success": False}    async def process_all(self, image_paths: List[str], prompt: str) -> List[dict]:        """批量处理所有图片"""        tasks = [self._process_single(img, prompt) for img in image_paths]        results = []        with Progress(            SpinnerColumn(),            TextColumn("[progress.description]{task.description}"),            BarColumn(),            TextColumn("{task.completed}/{task.total}"),        ) as progress:            task = progress.add_task("处理图片中...", total=len(tasks))            for coro in asyncio.as_completed(tasks):                result = await coro                results.append(result)                status = "✅" if result["success"] else "❌"                progress.update(task, advance=1, description=f"{status}{Path(result['file']).name}")        return results    def run(self, image_dir: str, prompt: str, output_file: str = "results.json") -> List[dict]:        """同步入口"""        import json        # 扫描图片文件        path = Path(image_dir)        images = [str(f) for f in path.rglob("*")                   if f.suffix.lower() in [".jpg", ".jpeg", ".png", ".webp"]]        console.print(f"📂 找到 {len(images)} 张图片，开始批量处理...")        results = asyncio.run(self.process_all(images, prompt))        # 统计        success = sum(1 for r in results if r["success"])        console.print(f"\n✅ 成功：{success} / {len(results)}")        # 保存结果        with open(output_file, "w", encoding="utf-8") as f:            json.dump(results, f, ensure_ascii=False, indent=2)        console.print(f"💾 结果已保存：{output_file}")        return results# 使用示例if __name__ == "__main__":    processor = BatchImageProcessor(max_concurrent=3)    # 批量分析产品图片    results = processor.run(        image_dir="./product_images",        prompt="请识别图中的产品名称、品牌和主要特征，返回 JSON 格式",        output_file="product_analysis.json"    )

# multimodal_rag.pyfrom langchain_community.vectorstores import Chromafrom langchain_community.embeddings import HuggingFaceEmbeddingsfrom langchain.schema import Documentfrom vision.image_analyzer import ImageAnalyzerfrom pathlib import Pathfrom typing import Listimport hashlibimport jsonclass MultimodalRAG:    """图文混合知识库"""    def __init__(self, persist_dir: str = "./mm_vectorstore"):        self.analyzer = ImageAnalyzer()        self.embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-m3")        self.persist_dir = persist_dir        self.vectorstore = None        self._load_or_create()    def _load_or_create(self):        """加载或创建向量库"""        self.vectorstore = Chroma(            persist_directory=self.persist_dir,            embedding_function=self.embeddings,        )    def add_image(self, image_path: str, extra_context: str = "") -> str:        """        将图片添加到知识库        1. 用多模态模型生成图片描述        2. 将描述文本向量化存入向量库        3. 元数据中保存原始图片路径        """        print(f"🔍 分析图片：{Path(image_path).name}")        # 生成图片的详细描述        description = self.analyzer.ask(            image_path,            "请详细描述这张图片的所有内容，包括文字、数据、图形元素，尽可能详尽"        )        # 合并描述和额外上下文        full_content = description        if extra_context:            full_content = f"{extra_context}\n\n图片描述：{description}"        # 为图片生成唯一ID        img_id = hashlib.md5(image_path.encode()).hexdigest()[:8]        # 存入向量库        doc = Document(            page_content=full_content,            metadata={                "source": image_path,                "type": "image",                "image_id": img_id,            }        )        self.vectorstore.add_documents([doc])        print(f"  ✅ 已索引（ID: {img_id}）")        return img_id    def query(self, question: str, top_k: int = 3) -> dict:        """        检索 + 问答（图文统一检索）        """        # 检索相关内容        docs = self.vectorstore.similarity_search(question, k=top_k)        if not docs:            return {"answer": "未找到相关内容", "sources": []}        # 分类文档和图片        text_contexts = []        image_sources = []        for doc in docs:            if doc.metadata.get("type") == "image":                image_sources.append(doc.metadata["source"])                text_contexts.append(f"[图片内容]\n{doc.page_content}")            else:                text_contexts.append(doc.page_content)        # 构建回答        from openai import OpenAI        from config import DEFAULT_API_KEY, DEFAULT_BASE_URL, DEFAULT_MODEL        client = OpenAI(api_key=DEFAULT_API_KEY, base_url=DEFAULT_BASE_URL)        context = "\n\n---\n\n".join(text_contexts)        prompt = f"""根据以下检索到的内容回答问题：{context}问题：{question}请基于上述内容给出准确的回答："""        response = client.chat.completions.create(            model=DEFAULT_MODEL,            messages=[{"role": "user", "content": prompt}],            max_tokens=1024,        )        return {            "answer": response.choices[0].message.content,            "sources": [doc.metadata.get("source", "") for doc in docs],            "doc_count": len(docs),        }

# app.py - 完整的多模态分析工具import gradio as grimport jsonfrom vision.image_analyzer import ImageAnalyzerfrom vision.chart_extractor import ChartExtractorfrom vision.ocr_engine import SmartOCRanalyzer = ImageAnalyzer()extractor = ChartExtractor()ocr = SmartOCR()def analyze_image(image, task, custom_question):    """图片分析主函数"""    if image is None:        return "请先上传图片"    # 保存临时文件    import tempfile    from PIL import Image as PILImage    img = PILImage.fromarray(image)    with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as tmp:        img.save(tmp.name)        temp_path = tmp.name    try:        if task == "📝 内容描述":            return analyzer.describe(temp_path)        elif task == "📊 提取表格":            df = extractor.extract_table(temp_path)            if df.empty:                return "未检测到表格数据"            return df.to_markdown()        elif task == "📈 分析图表":            data = extractor.extract_chart_data(temp_path)            return json.dumps(data, ensure_ascii=False, indent=2)        elif task == "🔤 文字识别（OCR）":            return ocr.extract_text(temp_path)        elif task == "🧾 发票识别":            result = ocr.extract_invoice(temp_path)            return json.dumps(result, ensure_ascii=False, indent=2)        elif task == "💼 名片识别":            result = ocr.extract_business_card(temp_path)            return json.dumps(result, ensure_ascii=False, indent=2)        elif task == "❓ 自定义问题":            if not custom_question.strip():                return "请输入你的问题"            return analyzer.ask(temp_path, custom_question)        else:            return "请选择分析任务"    finally:        import os        os.unlink(temp_path)  # 清理临时文件with gr.Blocks(title="🎨 智能图片分析工具", theme=gr.themes.Soft()) as demo:    gr.Markdown("# 🎨 智能图片分析工具\n> 基于多模态大模型 · 支持图表/OCR/VQA")    with gr.Row():        with gr.Column():            image_input = gr.Image(label="上传图片", type="numpy", height=400)            task_selector = gr.Radio(                choices=[                    "📝 内容描述",                    "📊 提取表格",                    "📈 分析图表",                    "🔤 文字识别（OCR）",                    "🧾 发票识别",                    "💼 名片识别",                    "❓ 自定义问题",                ],                label="选择分析任务",                value="📝 内容描述"            )            custom_q = gr.Textbox(                label="自定义问题（选择「自定义问题」时生效）",                placeholder="请输入你的问题..."            )            analyze_btn = gr.Button("🚀 开始分析", variant="primary", size="lg")        with gr.Column():            output = gr.Textbox(label="分析结果", lines=20, show_copy_button=True)    analyze_btn.click(        fn=analyze_image,        inputs=[image_input, task_selector, custom_q],        outputs=output    )    gr.Examples(        examples=[            [None, "📝 内容描述", ""],            [None, "🔤 文字识别（OCR）", ""],            [None, "❓ 自定义问题", "图中有哪些重要数据？"],        ],        inputs=[image_input, task_selector, custom_q]    )if __name__ == "__main__":    demo.launch(server_port=7861, inbrowser=True)

📈 进阶路径初级：单图分析 API 调用  ↓中级：批量处理 + 结构化提取 + OCR  ↓高级：图文混合 RAG + 多轮对话 + 本地部署（LLaVA）  ↓专家：视频理解 + 实时流处理 + 微调专属模型