Claude与DeepSeek对接实战：如何实现图片传输功能

1次阅读

没有评论

共计 3195 个字符，预计需要花费 8 分钟才能阅读完成。

在 Claude 与 DeepSeek 的 API 对接过程中，图片传输是常见但容易踩坑的需求场景。比如智能客服系统需要发送产品示意图，内容审核平台要提交待检测图片，或者知识库需要嵌入带图说明。这些场景都面临几个技术挑战：

二进制数据处理：图片需要特殊编码才能在 JSON 中传输
尺寸限制：API 通常对请求体有大小限制
传输效率：大文件直接影响接口响应速度
安全性：敏感图片需要额外保护措施

技术原理 ：
将图片二进制流转换为 ASCII 字符串嵌入 JSON

优点：
– 实现简单，无需额外存储服务
– 适合 10MB 以下的小文件
– 一次请求完成全部数据传输

缺点：
– 数据体积增大 30% 左右
– 大文件容易触发 API 大小限制
– 无缓存机制，重复传输效率低

技术原理 ：
先将图片上传到 OSS/S3 等存储服务，再传递 URL

优点：
– 突破 API 大小限制
– 支持 CDN 加速和缓存
– 便于权限控制和生命周期管理

缺点：
– 需要集成存储服务 SDK
– 多了一次上传操作
– 产生额外存储费用

选型建议 ：
– 临时小图用 Base64
– 高频访问或大文件用云存储
– 敏感图片建议自建存储服务

# 安装依赖
pip install pillow boto3 python-magic

import base64
from io import BytesIO
from PIL import Image

def process_image_base64(image_path, max_size=1024):
    """
    图片预处理和 Base64 编码
    :param image_path: 图片路径
    :param max_size: 最大边长 (保持比例)
    :return: (mime_type, base64_str)
    """
    with Image.open(image_path) as img:
        # 自动旋转手机拍摄的照片
        if hasattr(img, '_getexif'):
            exif = img._getexif()
            if exif and 274 in exif:  # Orientation tag
                orientation = exif[274]
                # 处理不同旋转角度...

        # 等比例缩放
        width, height = img.size
        if width > max_size or height > max_size:
            ratio = min(max_size/width, max_size/height)
            new_size = (int(width*ratio), int(height*ratio))
            img = img.resize(new_size, Image.LANCZOS)

        # 转换为 JPEG 格式 (兼容性最好)
        if img.format != 'JPEG':
            output = BytesIO()
            img.convert('RGB').save(output, format='JPEG', quality=85)
            img_data = output.getvalue()
            mime_type = 'image/jpeg'
        else:
            with open(image_path, 'rb') as f:
                img_data = f.read()
            mime_type = f'image/{img.format.lower()}'

        return mime_type, base64.b64encode(img_data).decode('utf-8')

import boto3
from botocore.exceptions import ClientError

def upload_to_s3(file_path, bucket_name, object_name=None):
    """
    上传文件到 S3 存储桶
    :return: 公开访问 URL
    """s3 = boto3.client('s3')

    # 如果未指定 object_name 则使用文件名
    if object_name is None:
        object_name = file_path.split('/')[-1]

    try:
        # 上传时设置 ACL 和缓存控制
        s3.upload_file(
            file_path, bucket_name, object_name,
            ExtraArgs={
                'ACL': 'public-read',
                'CacheControl': 'max-age=2592000',  # 30 天缓存
                'ContentType': get_mime_type(file_path)
            }
        )
        return f"https://{bucket_name}.s3.amazonaws.com/{object_name}"
    except ClientError as e:
        print(f"上传失败: {e}")
        return None

import requests

# Base64 方式
mime_type, base64_data = process_image_base64("product.jpg")
payload = {
    "text": "请分析这张产品图",
    "image": {
        "mime_type": mime_type,
        "data": base64_data
    }
}

# 云存储 URL 方式
image_url = upload_to_s3("product.jpg", "my-bucket")
payload = {
    "text": "请分析这张产品图",
    "image_url": image_url
}

# 发送请求
headers = {"Authorization": f"Bearer {API_KEY}"}
response = requests.post(
    "https://api.deepseek.com/v1/chat",
    json=payload,
    headers=headers,
    timeout=30
)

分块上传 ：超过 5MB 的文件建议使用分块上传

# AWS S3 分块上传示例
config = boto3.s3.transfer.TransferConfig(
    multipart_threshold=5*1024*1024,
    max_concurrency=4
)
s3.upload_file(..., Config=config)

压缩策略 ：
PNG 图片使用 zopflipng 压缩
JPEG 质量控制在 75-85 之间
考虑使用 WebP 格式 (比 JPEG 小 30%)

敏感图片上传前添加水印
云存储 URL 设置有效期 (Pre-signed URL)
内容审核后再展示 (使用 AWS Rekognition 等)

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def safe_api_call(payload):
    response = requests.post(API_ENDPOINT, json=payload)
    response.raise_for_status()  # 触发重试
    return response