Claude4.8的多模态扩展：整合华为云图像识别实现“看图说话”

[用户上传图片] → [华为云图像识别服务] → [结构化标签文本] → [Claude4.8自然语言生成] → [最终描述文本]

# 华为云SDK
pip install huaweicloudsdkcore huaweicloudsdkimage huaweicloudsdkobs

# Web框架与工具
pip install flask flask-cors pillow

# Claude API
pip install anthropic

# 华为云凭证
HUAWEI_AK = "your-access-key"
HUAWEI_SK = "your-secret-key"
HUAWEI_PROJECT_ID = "your-project-id"
HUAWEI_REGION = "cn-east-3"

# OBS配置
OBS_BUCKET_NAME = "image-to-text-bucket"

# Claude API
CLAUDE_API_KEY = "your-claude-api-key"

from huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkimage.v2 import ImageClient, RunImageTaggingRequest
from huaweicloudsdkimage.v2.region.image_region import ImageRegion
import base64
import json

class ImageAnalyzer:
    """华为云图像分析器，负责提取图片的结构化标签"""
    
    def __init__(self, ak, sk, region, project_id):
        credentials = BasicCredentials(ak, sk)
        self.client = ImageClient.new_builder() \
            .with_credentials(credentials) \
            .with_region(ImageRegion.value_of(region)) \
            .build()
        self.project_id = project_id
    
    def analyze(self, image_path):
        """
        分析图片，返回结构化标签
        包含：物体标签、置信度、场景类别等
        """
        # 读取图片并转base64
        with open(image_path, "rb") as f:
            image_base64 = base64.b64encode(f.read()).decode("utf-8")
        
        # 构建请求
        request = RunImageTaggingRequest()
        request.body = {
            "image": image_base64,
            "language": "zh",
            "threshold": 30,  # 置信度阈值，低于此值的标签会被过滤
            "limit": 20       # 最多返回20个标签
        }
        
        # 调用华为云图像识别API
        response = self.client.run_image_tagging(request)
        
        # 解析返回结果
        tags = []
        for tag in response.result.tags:
            tags.append({
                "tag": tag.tag,           # 标签名，如"建筑"、"天空"
                "type": tag.type,         # 标签类型，如"object"、"scene"
                "confidence": tag.confidence,  # 置信度，如98.5
                "i18n_tag": tag.i18n_tag  # 国际化标签
            })
        
        # 按置信度降序排列
        tags.sort(key=lambda x: x["confidence"], reverse=True)
        
        return {
            "tags": tags,
            "dominant_objects": [t for t in tags if t["confidence"] > 80][:5],
            "scene_hint": self._infer_scene(tags)
        }
    
    def _infer_scene(self, tags):
        """根据标签推断场景类型"""
        scene_keywords = {
            "城市街景": ["建筑", "汽车", "街道", "行人", "交通灯"],
            "自然风光": ["山", "水", "树", "天空", "花"],
            "室内场景": ["桌子", "椅子", "窗户", "灯光", "墙"],
            "美食": ["食物", "盘子", "餐具", "餐厅"],
            "人物肖像": ["人脸", "眼睛", "头发", "衣服"]
        }
        
        tag_names = [t["tag"] for t in tags]
        best_scene = "通用场景"
        best_score = 0
        
        for scene, keywords in scene_keywords.items():
            score = sum(1 for k in keywords if k in tag_names)
            if score > best_score:
                best_score = score
                best_scene = scene
        
        return best_scene

import anthropic

class ImageDescriber:
    """基于Claude4.8，将图像标签转化为自然语言描述"""
    
    def __init__(self, api_key):
        self.client = anthropic.Anthropic(api_key=api_key)
    
    def generate_description(self, image_analysis_result, style="vivid"):
        """
        根据图像分析结果生成描述文本
        
        参数：
            image_analysis_result: ImageAnalyzer.analyze()的返回结果
            style: 描述风格，"vivid"(生动) / "concise"(简洁) / "technical"(技术向)
        """
        # 构建详细的Prompt
        tags_info = json.dumps(image_analysis_result["tags"], ensure_ascii=False, indent=2)
        scene = image_analysis_result["scene_hint"]
        dominant = image_analysis_result["dominant_objects"]
        
        style_guide = {
            "vivid": "请用生动、富有文学色彩的语言描述这个画面，让读者仿佛身临其境。",
            "concise": "请用简洁明了的语言，一句话概括这个画面的核心内容。",
            "technical": "请用技术化的语言描述这个画面，适合用于图像检索的文本索引。"
        }
        
        prompt = f"""你是一位专业的图片解说员。根据以下计算机视觉系统从图片中识别出的结构化标签，生成一段自然的看图说话描述。

【识别到的场景类型】{scene}

【置信度最高的主体物体】
{json.dumps(dominant, ensure_ascii=False, indent=2)}

【完整标签列表（含置信度）】
{tags_info}

【描述要求】
{style_guide.get(style, style_guide["vivid"])}
- 不要罗列标签，要自然连贯地组织语言
- 合理推断物体之间的空间关系和场景氛围
- 如果标签置信度较低，使用“可能”、“似乎”等词语
- 描述长度控制在100-200字

请直接输出描述文本："""
        
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=300,
            temperature=0.7,  # 增加一点创造性
            messages=[{"role": "user", "content": prompt}]
        )
        
        return response.content[0].text.strip()

from flask import Flask, request, jsonify, render_template
from werkzeug.utils import secure_filename
import os
import time

app = Flask(__name__)
app.config["UPLOAD_FOLDER"] = "./uploads"
os.makedirs(app.config["UPLOAD_FOLDER"], exist_ok=True)

# 初始化分析器
analyzer = ImageAnalyzer(HUAWEI_AK, HUAWEI_SK, HUAWEI_REGION, HUAWEI_PROJECT_ID)
describer = ImageDescriber(CLAUDE_API_KEY)

@app.route("/")
def index():
    return render_template("index.html")

@app.route("/analyze", methods=["POST"])
def analyze_image():
    """上传图片并获取描述"""
    if "image" not in request.files:
        return jsonify({"error": "未上传图片"}), 400
    
    file = request.files["image"]
    style = request.form.get("style", "vivid")
    
    # 保存上传的图片
    filename = secure_filename(file.filename)
    filepath = os.path.join(app.config["UPLOAD_FOLDER"], filename)
    file.save(filepath)
    
    try:
        # 步骤1：华为云图像识别
        start_time = time.time()
        analysis_result = analyzer.analyze(filepath)
        tag_time = time.time() - start_time
        
        # 步骤2：Claude4.8生成描述
        start_time = time.time()
        description = describer.generate_description(analysis_result, style=style)
        desc_time = time.time() - start_time
        
        return jsonify({
            "success": True,
            "description": description,
            "tags": analysis_result["tags"],
            "scene": analysis_result["scene_hint"],
            "timing": {
                "image_analysis": f"{tag_time:.2f}s",
                "text_generation": f"{desc_time:.2f}s",
                "total": f"{tag_time + desc_time:.2f}s"
            }
        })
    
    finally:
        # 清理临时文件
        if os.path.exists(filepath):
            os.remove(filepath)

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080)