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

1次阅读

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

在 API 对接过程中，图片传输是一个常见但充满挑战的需求。无论是内容审核、智能客服还是数据分析场景，都需要高效可靠地处理图片数据。Claude 和 DeepSeek 作为两个流行的 AI 服务平台，它们的对接过程中图片传输往往会遇到三个主要技术难点：数据格式兼容性、传输效率问题以及安全合规要求。

实现原理 ：将二进制图片数据转换为 ASCII 字符串格式

优点：

实现简单，无需额外存储服务
适合小尺寸图片（<1MB）
单次请求完成数据传输

缺点：

数据体积膨胀约 33%
大文件容易导致请求超时
缺乏访问控制机制

实现原理 ：先将图片上传到 OSS/S3 等云存储，再传递 URL

优点：

支持任意大小的文件
可复用 CDN 加速
具备完善的权限管理

缺点：

需要集成存储 SDK
产生额外存储费用
存在 URL 失效风险

import base64
from PIL import Image
import io

def process_image_for_claude(file_path, max_size=1024):
    """图片预处理与 Base64 编码"""
    try:
        # 打开并自动旋转图片
        img = Image.open(file_path)
        if hasattr(img, '_getexif'):
            exif = img._getexif()
            if exif and 274 in exif:  # Orientation tag
                orientation = exif[274]
                # 处理不同旋转角度
                if orientation == 3:
                    img = img.rotate(180, expand=True)
                elif orientation == 6:
                    img = img.rotate(270, expand=True)
                elif orientation == 8:
                    img = img.rotate(90, expand=True)

        # 等比例缩放
        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)

        # 转换为 JPEG 格式并压缩
        buffer = io.BytesIO()
        img.convert('RGB').save(buffer, format='JPEG', quality=85)

        # Base64 编码
        encoded_str = base64.b64encode(buffer.getvalue()).decode('utf-8')
        return f"data:image/jpeg;base64,{encoded_str}"

    except Exception as e:
        print(f"图片处理失败: {str(e)}")
        raise

import oss2
from datetime import datetime, timedelta

def upload_to_oss(file_path, config):
    """上传图片到 OSS 并生成签名 URL"""
    try:
        # 初始化 OSS 客户端
        auth = oss2.Auth(config['access_key'], config['secret_key'])
        bucket = oss2.Bucket(auth, config['endpoint'], config['bucket_name'])

        # 生成唯一对象名
        object_name = f"images/{datetime.now().strftime('%Y%m%d')}/{uuid.uuid4().hex}.jpg"

        # 断点续传上传（适合大文件）oss2.resumable_upload(bucket, object_name, file_path,
                            store=oss2.ResumableStore(root='/tmp'),
                            multipart_threshold=100*1024,
                            part_size=100*1024,
                            num_threads=3)

        # 生成带签名的访问 URL（7 天有效期）return bucket.sign_url('GET', object_name, 60*60*24*7)

    except oss2.exceptions.OssError as e:
        print(f"OSS 上传错误: {e.status}, {e.message}")
        raise

对于 Base64 方案：
强制压缩超过 500KB 的图片
采用分块编码传输（需服务端支持）
设置合理的请求超时时间（建议 30-60 秒）
对于云存储方案：
启用分片上传（如上例中的 resumable_upload）
使用异步上传队列
考虑边缘加速上传节点

内容安全扫描（推荐阿里云内容安全 API）
敏感图片脱敏处理（人脸打码 / 关键信息模糊）
短期有效的访问令牌（JWT 临时签名）

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3),
       wait=wait_exponential(multiplier=1, min=4, max=10))
def safe_api_call(payload):
    """带指数退避的重试机制"""
    response = requests.post(API_ENDPOINT, json=payload)
    response.raise_for_status()
    return response.json()