DeepSeek V4 API完整接入实战手册

import requests
import json
import os
from typing import Optional

class DeepSeekClient:
    def __init__(self, api_key: Optional[str] = None):
        self.api_key = api_key or os.getenv("DEEPSEEK_API_KEY")
        if not self.api_key:
            raise ValueError("API Key未设置，请通过参数或环境变量提供")
        self.api_url = "https://api.deepseek.com/v1/chat/completions"
    
    def chat_completion(
        self, 
        prompt: str, 
        system_prompt: str = "你是专业靠谱的AI开发助手",
        temperature: float = 0.7,
        max_tokens: int = 2000
    ) -> str:
        """
        单次对话请求（非流式）
        
        Args:
            prompt: 用户提问内容
            system_prompt: 系统提示词
            temperature: 随机性参数（0~2）
            max_tokens: 最大输出Token数
        
        Returns:
            str: 模型回复内容
        """
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.api_key}"
        }
        payload = {
            "model": "deepseek-chat",
            "messages": [
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": prompt}
            ],
            "temperature": temperature,
            "max_tokens": max_tokens
        }
        
        try:
            response = requests.post(
                self.api_url, 
                headers=headers, 
                json=payload,
                timeout=60  # 超时控制
            )
            response.raise_for_status()
            result = response.json()
            return result["choices"][0]["message"]["content"]
            
        except requests.exceptions.Timeout:
            return "请求超时，请检查网络或减少max_tokens数值"
        except requests.exceptions.HTTPError as e:
            return f"HTTP错误：{e.response.status_code} - {e.response.text}"
        except requests.exceptions.RequestException as e:
            return f"请求异常：{str(e)}"

# 使用示例
if __name__ == "__main__":
    client = DeepSeekClient()
    response = client.chat_completion("用动态规划实现斐波那契数列的Python代码")
    print("AI返回结果：\n", response)

import requests
import json
import sys
from typing import Generator

def stream_chat(prompt: str) -> Generator[str, None, None]:
    """
    流式对话生成器
    
    Yields:
        str: 逐步返回的文本片段
    """
    api_key = os.getenv("DEEPSEEK_API_KEY")
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {api_key}"
    }
    payload = {
        "model": "deepseek-chat",
        "messages": [{"role": "user", "content": prompt}],
        "stream": True,
        "temperature": 0.7
    }
    
    with requests.post(
        "https://api.deepseek.com/v1/chat/completions",
        headers=headers,
        json=payload,
        stream=True,
        timeout=30
    ) as response:
        response.raise_for_status()
        for line in response.iter_lines(decode_unicode=True):
            if not line:
                continue
            # 解析SSE格式数据
            if line.startswith("data: "):
                data_str = line[6:]  # 去除"data: "前缀
                if data_str == "[DONE]":
                    break
                try:
                    chunk = json.loads(data_str)
                    delta = chunk["choices"][0].get("delta", {})
                    content = delta.get("content", "")
                    if content:
                        yield content
                except json.JSONDecodeError:
                    continue

# 使用示例
if __name__ == "__main__":
    print("AI：", end="", flush=True)
    for text in stream_chat("写一首关于编程的短诗"):
        print(text, end="", flush=True)
        sys.stdout.flush()
    print()  # 换行

import requests
import json
import os
from typing import List, Dict

class ConversationBot:
    """支持上下文记忆的对话机器人"""
    
    def __init__(self, system_prompt: str = "你是一个智能助手"):
        self.api_key = os.getenv("DEEPSEEK_API_KEY")
        self.api_url = "https://api.deepseek.com/v1/chat/completions"
        self.messages: List[Dict[str, str]] = [
            {"role": "system", "content": system_prompt}
        ]
        self.max_history = 20  # 最大保留历史轮次
    
    def chat(self, user_input: str) -> str:
        """发送消息并获取回复"""
        # 添加用户消息
        self.messages.append({"role": "user", "content": user_input})
        
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.api_key}"
        }
        payload = {
            "model": "deepseek-chat",
            "messages": self.messages,
            "temperature": 0.7,
            "max_tokens": 2000
        }
        
        try:
            response = requests.post(self.api_url, headers=headers, json=payload, timeout=60)
            response.raise_for_status()
            result = response.json()
            reply = result["choices"][0]["message"]["content"]
            
            # 保存助手回复
            self.messages.append({"role": "assistant", "content": reply})
            
            # 控制历史长度，防止超出上下文窗口
            if len(self.messages) > self.max_history * 2:
                # 保留system提示 + 最近N轮对话
                self.messages = [self.messages[0]] + self.messages[-(self.max_history * 2):]
            
            return reply
            
        except requests.exceptions.RequestException as e:
            return f"请求失败：{str(e)}"
    
    def clear_history(self):
        """清空对话历史（保留系统提示）"""
        system_prompt = self.messages[0]["content"]
        self.messages = [{"role": "system", "content": system_prompt}]
    
    def get_history(self) -> List[Dict[str, str]]:
        """获取完整对话历史"""
        return self.messages.copy()

# 使用示例
if __name__ == "__main__":
    bot = ConversationBot(system_prompt="你是Python编程专家")
    
    print("AI：", bot.chat("Python是什么类型的语言？"))
    print("AI：", bot.chat("它的核心优势有哪些？"))
    print("AI：", bot.chat("请用一段Hello World代码演示基础语法"))
    
    # 查看历史
    print(f"\n当前对话共 {len(bot.get_history())} 条消息")

import time
from functools import wraps

def retry_on_failure(max_retries=3, base_delay=1, backoff_factor=2):
    """指数退避重试装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            delay = base_delay
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except requests.exceptions.RequestException as e:
                    if attempt == max_retries - 1:
                        raise e
                    print(f"请求失败，{delay}秒后重试（{attempt+1}/{max_retries}）")
                    time.sleep(delay)
                    delay *= backoff_factor
            return None
        return wrapper
    return decorator

# 使用示例
@retry_on_failure(max_retries=3)
def robust_api_call(prompt):
    # 调用DeepSeek API的代码
    pass