当前位置： 首页 > news >正文

【大模型实战篇】基于开源视觉大模型封装多模态信息提取工具

news 2025/8/27 8:05:01

1. 功能模块

封装一个统一的视觉模型调用工具，支持多种大模型（Qwen2.5-VL、GLM4V、GPT-4V）来分析图片内容。

主要模块有：

  1. 配置管理（DEFAULT_CONFIG）

    • 存储不同模型的调用参数（API URL、模型名、API Key、最大 Token、温度等）。

    • 每个模型都有一个配置字典。

  2. 视觉模型服务类（VisionModelService）
    封装了所有核心功能，包括：

    • _init_clients()：初始化各个模型的客户端实例（依赖 openai.OpenAI）。

    • encode_image_to_base64()：将本地图片转成 base64 字符串（供模型 API 使用）。

    • analyze_image_content()：构建请求，调用模型 API 分析图片，返回结果（尽量解析成 JSON）。

    • _get_default_prompt()：提供一个默认的中文提示词，让模型以 JSON 格式返回图片信息。

    • get_supported_models()：返回支持的模型列表。

    • test_connection()：测试指定模型是否能正常调用。

  3. 全局单例服务实例（_vision_service + 工厂方法）

    • get_vision_service()：保证全局只初始化一个 VisionModelService，避免重复创建客户端。

  4. 便捷函数（对外接口）

    • analyze_image_with_vision_model()：对外封装的“分析图片”函数。

    • test_vision_model_connection()：对外封装的“测试连接”函数。

  5. 主函数入口（if __name__ == "__main__":）

    • 先测试 qwen2.5_vl 的连接。

    • 如果 data/test.jpg 存在，则调用模型分析图片，否则提示图片不存在。

2. 调用流程

代码运行时的主要流程是：

  1. 初始化

    • 执行 get_vision_service() → 创建 VisionModelService 实例。

    • _init_clients() 根据配置初始化各个模型的 OpenAI 客户端。

  2. 测试模型连接（可选）

    • 调用 test_vision_model_connection("qwen2.5_vl")。

    • 发送一条简单消息 "请回复'连接测试成功'"。

    • 检查返回值，判断是否连接成功。

  3. 分析图片（核心逻辑）

    • 调用 analyze_image_with_vision_model(image_path, model_type, prompt)。

    • encode_image_to_base64() 把图片转成 base64。

    • 构造 messages 请求（包括文字 prompt 和图片 base64 URL）。

    • 调用 client.chat.completions.create() 生成结果。

    • 优先尝试把结果解析成 JSON，如果失败，就直接用原始文本包装成 JSON 格式返回。

3. 关键逻辑细节

  1. 多模型统一调用
    不同模型（Qwen、GLM、OpenAI）都统一用 VisionModelService 来调用，减少了代码重复。

  2. 图片处理

    • 图片被读取为二进制 → base64 编码 → data:image/jpeg;base64,xxx 的 URL → 传给模型。

    • 避免了直接上传本地路径的兼容性问题。

  3. 健壮性设计

    • API 调用失败会捕获异常，并返回 status: error。

    • 模型响应为空/不是 JSON 格式，也能兼容返回结果。

    • 默认超时时间设置为 60 秒，避免请求卡死。

  4. 默认提示词（Prompt Engineering）

    • 强制模型输出 JSON 格式结果。

    • 要求所有 key 必须是中文（比如 "产品名称":"xxx"）。

    • 特别强调提取 包装上的文字、成分表 等。

"""
视觉模型调用工具程序
支持多种视觉模型，包括Qwen2.5-VL等
"""
import json
import os
import sys
from typing import Dict, Any, Optional, List
from PIL import Image
from io import BytesIO
import base64
import timetry:from openai import OpenAI
except ImportError:print("警告：未安装openai库，请运行: pip install openai")OpenAI = Nonetry:from PIL import Image
except ImportError:print("警告：未安装PIL库，请运行: pip install pillow")Image = None# 默认配置
DEFAULT_CONFIG = {"qwen2.5_vl": {"api_base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1","model_name": "model","api_key": os.getenv("QWEN_API_KEY"),"max_tokens": 4096,"temperature": 0.1},"glm4v": {"api_base_url": "https://open.bigmodel.cn/api/paas/v4","model_name": "model","api_key": os.getenv("GLM_API_KEY"),"max_tokens": 4096,"temperature": 0.1},"gpt4v": {"api_base_url": "https://api.openai.com/v1","model_name": "model","api_key": os.getenv("OPENAI_API_KEY"),"max_tokens": 4096,"temperature": 0.1}
}class VisionModelService:"""视觉模型服务类"""def __init__(self, config: Dict[str, Any] = None):"""初始化视觉模型服务Args:config: 配置字典，包含各种模型的配置信息"""self.config = config or DEFAULT_CONFIGself.clients = {}self._init_clients()def _init_clients(self):"""初始化各种模型的客户端"""if OpenAI is None:print("警告：OpenAI库未安装，无法初始化客户端")returnfor model_type, model_config in self.config.items():try:client = OpenAI(api_key=model_config.get("api_key", "dummy-key"),base_url=model_config.get("api_base_url"),)self.clients[model_type] = {"client": client,"config": model_config}print(f"✅ {model_type} 客户端初始化成功")except Exception as e:print(f"❌ {model_type} 客户端初始化失败: {e}")def encode_image_to_base64(self, image_path: str) -> str:"""将图片文件编码为base64字符串Args:image_path: 图片文件路径Returns:base64编码的图片字符串"""try:with open(image_path, "rb") as image_file:return base64.b64encode(image_file.read()).decode('utf-8')except Exception as e:print(f"编码图片失败: {e}")return ""def analyze_image_content(self, image_path: str, model_type: str = "qwen2.5_vl",prompt: str = None,image_info: Dict[str, Any] = None) -> Dict[str, Any]:"""分析图片内容Args:image_path: 图片文件路径model_type: 模型类型 (qwen2.5_vl, glm4v, gpt4v)prompt: 自定义提示词image_info: 图片基本信息Returns:图片分析结果"""if model_type not in self.clients:return {"status": "error","error": f"不支持的模型类型: {model_type}","supported_models": list(self.clients.keys())}client_info = self.clients[model_type]client = client_info["client"]config = client_info["config"]try:# 将图片编码为base64base64_image = self.encode_image_to_base64(image_path)if not base64_image:return {"status": "error","error": "图片编码失败","image_path": image_path}# 使用默认提示词或自定义提示词if prompt is None:prompt = self._get_default_prompt()# 构建请求消息messages = [{"role": "user","content": [{"type": "text","text": prompt},{"type": "image_url","image_url": {"url": f"data:image/jpeg;base64,{base64_image}"}}]}]# 调用模型API（增加超时时间）response = client.chat.completions.create(model=config["model_name"],messages=messages,max_tokens=config.get("max_tokens", 4096),temperature=config.get("temperature", 0.1),timeout=60  # 增加到60秒超时)# 解析响应if response.choices and len(response.choices) > 0:content = response.choices[0].message.content# 尝试解析JSON格式的响应try:parsed_content = json.loads(content)return {"status": "success","model_type": model_type,"content": parsed_content,"raw_content": content,"image_info": image_info or {}}except json.JSONDecodeError:# 如果不是JSON格式，返回原始文本return {"status": "success","model_type": model_type,"content": {"分析结果": content},"raw_content": content,"image_info": image_info or {}}else:return {"status": "error","error": "模型响应为空","model_type": model_type}except Exception as e:return {"status": "error","error": f"模型调用失败: {str(e)}","model_type": model_type,"image_path": image_path}def _get_default_prompt(self) -> str:"""获取默认的图片分析提示词"""return """你是一个专业的图片内容分析AI。请分析这张图片并提取其中的所有信息，以标准的JSON键值对格式返回，并且所有键名必须使用中文。要求：
1. 提取图片中的所有可见文字信息，并尽可能组织成有意义的键值对
2. 识别产品信息：名称、规格、价格、功效、成分、使用方法等
3. 如果能识别出明确的标题和对应内容，请组织成"标题":"内容"的格式
4. 对于无法形成键值对的信息，可以用描述性的键名
5. 特别注意包装上的文字信息，包括品牌、产品名称、成分表、营养信息等不要输出解释性文字，直接返回标准的JSON键值对格式。"""def get_supported_models(self) -> List[str]:"""获取支持的模型列表"""return list(self.clients.keys())def test_connection(self, model_type: str = "qwen2.5_vl") -> Dict[str, Any]:"""测试模型连接Args:model_type: 模型类型Returns:连接测试结果"""if model_type not in self.clients:return {"status": "error","error": f"不支持的模型类型: {model_type}"}try:client_info = self.clients[model_type]client = client_info["client"]config = client_info["config"]# 发送一个简单的测试请求messages = [{"role": "user","content": "请回复'连接测试成功'"}]response = client.chat.completions.create(model=config["model_name"],messages=messages,max_tokens=10,temperature=0)if response.choices and len(response.choices) > 0:return {"status": "success","model_type": model_type,"message": "连接测试成功","response": response.choices[0].message.content}else:return {"status": "error","error": "模型响应为空"}except Exception as e:return {"status": "error","error": f"连接测试失败: {str(e)}","model_type": model_type}# 全局实例
_vision_service = Nonedef get_vision_service(config: Dict[str, Any] = None) -> VisionModelService:"""获取视觉模型服务实例Args:config: 配置字典Returns:VisionModelService实例"""global _vision_serviceif _vision_service is None:_vision_service = VisionModelService(config)return _vision_servicedef analyze_image_with_vision_model(image_path: str, model_type: str = "qwen2.5_vl",prompt: str = None,config: Dict[str, Any] = None) -> Dict[str, Any]:"""使用视觉模型分析图片的便捷函数Args:image_path: 图片文件路径model_type: 模型类型prompt: 自定义提示词config: 配置字典Returns:分析结果"""service = get_vision_service(config)return service.analyze_image_content(image_path, model_type, prompt)def test_vision_model_connection(model_type: str = "qwen2.5_vl", config: Dict[str, Any] = None) -> Dict[str, Any]:"""测试视觉模型连接的便捷函数Args:model_type: 模型类型config: 配置字典Returns:测试结果"""service = get_vision_service(config)return service.test_connection(model_type)# 示例用法
if __name__ == "__main__":# 测试连接print("测试视觉模型连接...")result = test_vision_model_connection("qwen2.5_vl")print(f"连接测试结果: {result}")# 如果有测试图片，可以进行分析test_image = "data/test.jpg"if os.path.exists(test_image):print(f"\n分析图片: {test_image}")analysis_result = analyze_image_with_vision_model(test_image, "qwen2.5_vl")print(f"分析结果: {json.dumps(analysis_result, ensure_ascii=False, indent=2)}")else:print(f"\n测试图片不存在: {test_image}")

http://www.dtcms.com/a/351833.html

相关文章：

  • 第1节: 微服务架构设计篇
  • WhisperLiveKit：实时语音转文本
  • 面试总结（1）
  • 2. LangChain4J 中的 Hello World
  • 未来几年哪些行业有潜力
  • ICBC_TDR_UShield2_Install.exe [ICBC UKEY]
  • ARP地址解析协议
  • 【AI算力平台】算力高效调度策略——GPU调度
  • mysql 执行sql流程概述
  • FreeRTOS学习笔记（四）：任务执行与切换
  • ProfiNet 转 Ethernet/IP基于西门子 S7 - 1500 与罗克韦尔 PLC 的汽车零部件加工线协同案例
  • 基于微服务的水果分销系统-项目分享
  • LeetCode 3000.对角线最长的矩形的面积：一次遍历
  • 【golang长途旅行第32站】反射
  • 【机器学习深度学习】连续微调与权重合并的陷阱与最佳实践
  • 修改C盘缓存文件路径
  • MongoDB /redis/mysql 界面化的数据查看页面App
  • UCIE Specification详解（八）
  • 在MiniOB源码中学习使用Flex与Bison解析SQL语句-第一节
  • Rust 环境搭建与 SeekStorm 项目编译部署（支持中文）
  • Robrain V2.0正式登场：落地人形机器人，引爆智能进化革命
  • Ubuntu操作系统下使用mysql、mongodb、redis
  • [特殊字符] CentOS 7 升级 OpenSSH 10.0p2 完整教程（含 Telnet 备份）
  • 如果 我退休了
  • 汽车域控中Hypervisor方案极致安全原理与弊端
  • APP UI自动化测试的思路总结
  • 破解豆瓣Ajax动态加载：Python爬取完整长评论和短评
  • Java面试实战系列【JVM篇】- JVM内存结构与运行时数据区详解（私有区域）
  • 数据结构：链式队列尝试；0826
  • poi生成word固定表格列宽