DeepSeek V4.1 Flash API 接入全解析：MLA 长上下文架构、视觉多模态与中转高可用落地

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI(
    api_key="sk-gpt345-your-actual-api-key",
    base_url="https://api.gpt345.com/v1"
)

async def stream_deepseek_multimodal():
    response = await client.chat.completions.create(
        model="deepseek-flash",
        messages=[
            {"role": "system", "content": "你是一位资深多模态架构专家，分析请先展开思考过程。"},
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": "分析此系统拓扑架构图中的单点故障与瓶颈："},
                    {
                        "type": "image_url",
                        "image_url": {"url": "https://images.unsplash.com/photo-1558494949-ef010cbdcc31?w=800"}
                    }
                ]
            }
        ],
        stream=True,
        max_tokens=2048
    )

    reasoning_buffer = []
    content_buffer = []

    async for chunk in response:
        delta = chunk.choices[0].delta
        # 捕获推理思考流
        if hasattr(delta, 'reasoning_content') and delta.reasoning_content:
            reasoning_buffer.append(delta.reasoning_content)
            print(f"[思考推导]: {delta.reasoning_content}", end="", flush=True)
        # 捕获最终产出流
        if hasattr(delta, 'content') and delta.content:
            content_buffer.append(delta.content)
            print(delta.content, end="", flush=True)

asyncio.run(stream_deepseek_multimodal())

import time
import random
import requests

class ResilientDeepSeekClient:
    def __init__(self, api_key: str, base_url: str = "https://api.gpt345.com/v1"):
        self.api_key = api_key
        self.base_url = base_url.rstrip('/')

    def send_chat_completion(self, messages: list, max_retries: int = 4):
        endpoint = f"{self.base_url}/chat/completions"
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        payload = {
            "model": "deepseek-flash",
            "messages": messages,
            "stream": False,
            "max_tokens": 1024
        }

        for attempt in range(max_retries):
            try:
                start_t = time.time()
                resp = requests.post(endpoint, json=payload, headers=headers, timeout=45)
                elapsed = time.time() - start_t

                if resp.status_code == 200:
                    data = resp.json()
                    usage = data.get("usage", {})
                    return {
                        "status": "success",
                        "latency": elapsed,
                        "usage": usage,
                        "content": data["choices"][0]["message"]["content"]
                    }
                elif resp.status_code in [429, 500, 502, 503]:
                    backoff = (2 ** attempt) + random.uniform(0.1, 0.8)
                    print(f"[警告] HTTP {resp.status_code}，将在 {backoff:.2f} 秒后执行第 {attempt+1} 次重试...")
                    time.sleep(backoff)
                elif resp.status_code == 401:
                    raise PermissionError("认证密钥无效，请核对 API Key 是否有效。")
                else:
                    resp.raise_for_status()
            except requests.exceptions.RequestException as e:
                if attempt == max_retries - 1:
                    raise RuntimeError(f"服务连接最终失败: {str(e)}")
                time.sleep(1.5)

# 使用示例
client = ResilientDeepSeekClient(api_key="sk-gpt345-your-actual-api-key")
result = client.send_chat_completion([{"role": "user", "content": "验证连接延迟与可用性"}])
print(f"响应成功，耗时: {result['latency']:.2f}s, 消耗 tokens: {result['usage']}")