前言

最近在做一个智能对话项目时，我尝试了多个大模型 API，发现 DeepSeek 和豆包大模型的响应速度和质量都很出色。但在实际调用过程中，也踩了不少坑——认证流程、流式输出处理、上下文管理、错误重试……每一个环节都有讲究。

今天就把这些经验整理成文，手把手教你如何用 Python 高效调用大模型 API，涵盖 DeepSeek 和豆包两个主流平台。

📌 适用读者：有一定 Python 基础，想快速将大模型能力集成到项目中的开发者。

一、准备工作

1.1 获取 API Key

DeepSeek：

1. 访问 https://platform.deepseek.com/ 注册/登录账号
2. 进入“API Keys”页面（https://platform.deepseek.com/api_keys）
3. 点击“Create new API Key”，命名后生成
4. ⚠️ Key 仅显示一次，立即复制保存！

豆包（火山引擎）：

1. 访问火山引擎官网，完成开发者注册
2. 在控制台创建应用，获取 API Key

🔐 安全建议：将 API Key 存入环境变量，避免硬编码到代码中。

1.2 安装依赖

# 创建虚拟环境（推荐）
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# 安装依赖
pip install openai requests

DeepSeek API 完全兼容 OpenAI SDK，只需修改 base_url 即可无缝迁移。

二、基础调用：非流式对话

2.1 使用 OpenAI SDK（推荐方式）

DeepSeek API 兼容 OpenAI API 格式，这是最推荐的调用方式。

import os
from openai import OpenAI

# 初始化客户端
client = OpenAI(
    api_key=os.environ.get('DEEPSEEK_API_KEY'),
    base_url="https://api.deepseek.com"  # DeepSeek 的 API 地址
)

# 发起对话
response = client.chat.completions.create(
    model="deepseek-v4-pro",  # 或 "deepseek-chat"（2026/07/24 弃用）
    messages=[
        {"role": "system", "content": "你是一个专业的Python开发助手"},
        {"role": "user", "content": "请用Python写一个快速排序算法"}
    ],
    temperature=0.7,
    max_tokens=2048,
    stream=False
)

print(response.choices[0].message.content)

参数说明：

参数	说明	推荐值
model	模型版本	deepseek-v4-pro（最新）
temperature	生成多样性，0-1	0.7（创意任务）/ 0.1（确定性任务）
max_tokens	最大输出长度	根据需求设置
stream	是否流式输出	False（同步）/ True（流式）

2.2 使用 requests 库（HTTP 直接调用）

如果不想依赖 SDK，也可以用 requests 直接发送 HTTP 请求。

import os
import requests
import json

API_KEY = os.environ.get('DEEPSEEK_API_KEY')
URL = "https://api.deepseek.com/chat/completions"

headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}

data = {
    "model": "deepseek-v4-pro",
    "messages": [
        {"role": "system", "content": "你是一个专业的Python开发助手"},
        {"role": "user", "content": "请用Python写一个快速排序算法"}
    ],
    "temperature": 0.7,
    "max_tokens": 2048,
    "stream": False
}

response = requests.post(URL, headers=headers, json=data)
result = response.json()
print(result['choices'][0]['message']['content'])

三、流式输出（Streaming）：实现打字机效果

流式输出可以让 AI 的回复逐字显示，提升用户体验。

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ.get('DEEPSEEK_API_KEY'),
    base_url="https://api.deepseek.com"
)

# 流式输出：设置 stream=True
stream = client.chat.completions.create(
    model="deepseek-v4-pro",
    messages=[
        {"role": "system", "content": "你是一个友好的AI助手"},
        {"role": "user", "content": "请介绍一下Python的装饰器"}
    ],
    stream=True  # 关键参数
)

# 逐块打印输出
for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

运行效果： 文字会像打字机一样逐字出现，而不是一次性全部返回。

💡 小贴士：流式输出适合聊天机器人、实时问答等对响应体验要求高的场景。

四、多轮对话：上下文管理

DeepSeek 的 Chat API 是 “无状态” 的，即服务端不记录用户请求的上下文。每次请求时，需要将之前所有对话历史拼接好后，一起传递给 API。

4.1 手动维护对话历史

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ.get('DEEPSEEK_API_KEY'),
    base_url="https://api.deepseek.com"
)

# 初始化对话历史
messages = []

def chat(user_input):
    # 添加用户消息
    messages.append({"role": "user", "content": user_input})
    
    # 调用 API（带上完整历史）
    response = client.chat.completions.create(
        model="deepseek-v4-pro",
        messages=messages  # 传递完整对话历史
    )
    
    # 获取 AI 回复
    assistant_reply = response.choices[0].message.content
    
    # 将 AI 回复也加入历史
    messages.append({"role": "assistant", "content": assistant_reply})
    
    return assistant_reply

# 模拟多轮对话
print(chat("世界上最高的山是什么？"))
print(chat("那第二高的呢？"))  # AI 知道你在问"山"

4.2 对话历史结构解析

第一轮请求时，传递给 API 的 messages 为：

[
    {"role": "user", "content": "世界上最高的山是什么？"}
]

第二轮请求时，需要将第一轮的对话都加进去：

[
    {"role": "user", "content": "世界上最高的山是什么？"},
    {"role": "assistant", "content": "世界上最高的山是珠穆朗玛峰。"},
    {"role": "user", "content": "那第二高的呢？"}
]

4.3 封装为对话管理类（生产级）

import os
from openai import OpenAI
from typing import List, Dict

class DialogueManager:
    """对话管理器 - 自动维护上下文"""
    
    def __init__(self, system_prompt: str = None, max_history: int = 20):
        self.client = OpenAI(
            api_key=os.environ.get('DEEPSEEK_API_KEY'),
            base_url="https://api.deepseek.com"
        )
        self.messages: List[Dict[str, str]] = []
        self.max_history = max_history
        
        # 添加系统提示词（可选）
        if system_prompt:
            self.messages.append({"role": "system", "content": system_prompt})
    
    def chat(self, user_input: str, model: str = "deepseek-v4-pro") -> str:
        """发送消息并获取回复"""
        # 添加用户消息
        self.messages.append({"role": "user", "content": user_input})
        
        # 限制历史长度，避免超长
        if len(self.messages) > self.max_history:
            # 保留 system 消息（如果有）和最近的对话
            if self.messages[0]["role"] == "system":
                self.messages = [self.messages[0]] + self.messages[-(self.max_history-1):]
            else:
                self.messages = self.messages[-self.max_history:]
        
        # 调用 API
        response = self.client.chat.completions.create(
            model=model,
            messages=self.messages,
            temperature=0.7
        )
        
        # 获取回复并加入历史
        assistant_reply = response.choices[0].message.content
        self.messages.append({"role": "assistant", "content": assistant_reply})
        
        return assistant_reply
    
    def get_history(self) -> List[Dict[str, str]]:
        """获取完整对话历史"""
        return self.messages.copy()
    
    def clear(self):
        """清空对话历史（保留 system prompt）"""
        if self.messages and self.messages[0]["role"] == "system":
            self.messages = [self.messages[0]]
        else:
            self.messages = []

# 使用示例
bot = DialogueManager(system_prompt="你是一个耐心的Python编程老师")
print(bot.chat("什么是闭包？"))
print(bot.chat("能给我一个例子吗？"))  # AI 知道你在追问"闭包"的例子

五、Prompt 工程：让 AI 输出更可控

好的 Prompt 能大幅提升输出质量。以下是几条核心原则：

5.1 角色 + 上下文 + 约束 模板

def build_prompt(role: str, task: str, constraints: str, examples: str = None) -> str:
    """结构化构建 Prompt"""
    prompt = f"""
# 角色
你是一个{role}

# 任务
{task}

# 约束条件
{constraints}
"""
    if examples:
        prompt += f"\n# 示例\n{examples}"
    
    return prompt

# 使用示例
prompt = build_prompt(
    role="资深的Python代码审查专家",
    task="审查以下代码，指出潜在的性能问题和安全漏洞",
    constraints="1. 按严重程度排序\n2. 每个问题给出修改建议\n3. 用中文回复",
    examples="代码：def func(): pass"
)

response = client.chat.completions.create(
    model="deepseek-v4-pro",
    messages=[{"role": "user", "content": prompt}]
)

5.2 思维链（Chain-of-Thought）提示

对于复杂推理任务，引导模型“一步步思考”效果更好。

# 启用 DeepSeek 的思考模式
response = client.chat.completions.create(
    model="deepseek-v4-pro",  # 或 "deepseek-reasoner"（已弃用）
    messages=[
        {"role": "user", "content": "请一步步推理：一个水池有A、B两个进水管，A管单独注满需4小时，B管需6小时，两管同时开多久能注满？"}
    ],
    extra_body={"thinking": {"type": "enabled"}},  # 开启思考模式
    reasoning_effort="high"  # 推理强度
)
print(response.choices[0].message.content)

六、错误处理与重试机制（生产必备）

大模型 API 调用可能遇到网络超时、限流、服务端错误等问题，必须做好错误处理。

6.1 基础错误处理

import time
import requests
from typing import Optional

def call_api_with_retry(
    prompt: str,
    max_retries: int = 3,
    base_delay: float = 1.0
) -> Optional[str]:
    """
    带指数退避重试的 API 调用
    """
    url = "https://api.deepseek.com/chat/completions"
    headers = {
        "Authorization": f"Bearer {os.environ.get('DEEPSEEK_API_KEY')}",
        "Content-Type": "application/json"
    }
    data = {
        "model": "deepseek-v4-pro",
        "messages": [{"role": "user", "content": prompt}],
        "temperature": 0.7
    }
    
    for attempt in range(max_retries):
        try:
            response = requests.post(url, headers=headers, json=data, timeout=30)
            
            if response.status_code == 200:
                return response.json()['choices'][0]['message']['content']
            elif response.status_code == 429:  # 限流
                wait_time = base_delay * (2 ** attempt)  # 指数退避
                print(f"触发限流，{wait_time}秒后重试...")
                time.sleep(wait_time)
            else:
                print(f"请求失败，状态码: {response.status_code}")
                if attempt < max_retries - 1:
                    time.sleep(base_delay * (2 ** attempt))
                    
        except requests.exceptions.Timeout:
            print(f"请求超时，第{attempt+1}次重试...")
            time.sleep(base_delay * (2 ** attempt))
        except Exception as e:
            print(f"未知错误: {e}")
            if attempt < max_retries - 1:
                time.sleep(base_delay * (2 ** attempt))
    
    return None

6.2 使用装饰器实现重试（更优雅）

from functools import wraps
import time

def retry_on_failure(max_retries=3, base_delay=1.0):
    """API 调用重试装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise e
                    wait_time = base_delay * (2 ** attempt)
                    print(f"调用失败 ({e})，{wait_time}s 后重试...")
                    time.sleep(wait_time)
            return None
        return wrapper
    return decorator

# 使用
@retry_on_failure(max_retries=3, base_delay=1.0)
def call_deepseek(prompt: str) -> str:
    # API 调用逻辑...
    pass

七、豆包大模型 API 调用

豆包（Doubao）是字节跳动推出的大语言模型，在中文场景下表现优异。其调用方式与 DeepSeek 类似，也兼容 OpenAI 格式。

7.1 获取 Access Token

豆包 API 采用 Bearer Token 认证，需要先获取访问令牌。

import requests

def get_doubao_access_token(client_id: str, client_secret: str) -> str:
    """
    获取豆包 API 访问令牌
    """
    auth_url = "https://api.doubao.com/oauth/token"
    payload = {
        "grant_type": "client_credentials",
        "client_id": client_id,
        "client_secret": client_secret
    }
    
    try:
        response = requests.post(auth_url, data=payload, timeout=10)
        response.raise_for_status()
        return response.json()["access_token"]
    except requests.exceptions.RequestException as e:
        print(f"获取令牌失败: {e}")
        return None

7.2 调用豆包对话 API

def call_doubao_api(
    prompt: str,
    token: str,
    model: str = "doubao-pro",
    temperature: float = 0.7
) -> str:
    """
    调用豆包大模型 API
    """
    api_url = "https://api.doubao.com/v1/chat/completions"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    payload = {
        "model": model,
        "messages": [{"role": "user", "content": prompt}],
        "max_tokens": 1024,
        "temperature": temperature
    }
    
    try:
        response = requests.post(api_url, headers=headers, json=payload, timeout=30)
        response.raise_for_status()
        return response.json()["choices"][0]["message"]["content"]
    except Exception as e:
        print(f"调用豆包 API 失败: {e}")
        return None

7.3 统一封装：同时支持 DeepSeek 和豆包

from abc import ABC, abstractmethod

class BaseLLMClient(ABC):
    """大模型客户端基类"""
    
    @abstractmethod
    def chat(self, prompt: str, **kwargs) -> str:
        pass

class DeepSeekClient(BaseLLMClient):
    def __init__(self, api_key: str = None):
        self.api_key = api_key or os.environ.get('DEEPSEEK_API_KEY')
        self.client = OpenAI(
            api_key=self.api_key,
            base_url="https://api.deepseek.com"
        )
    
    def chat(self, prompt: str, **kwargs) -> str:
        response = self.client.chat.completions.create(
            model=kwargs.get('model', 'deepseek-v4-pro'),
            messages=[{"role": "user", "content": prompt}],
            temperature=kwargs.get('temperature', 0.7)
        )
        return response.choices[0].message.content

class DoubaoClient(BaseLLMClient):
    def __init__(self, client_id: str, client_secret: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.token = None
        self._refresh_token()
    
    def _refresh_token(self):
        self.token = get_doubao_access_token(self.client_id, self.client_secret)
    
    def chat(self, prompt: str, **kwargs) -> str:
        if not self.token:
            self._refresh_token()
        return call_doubao_api(prompt, self.token, **kwargs)

# 使用
deepseek = DeepSeekClient()
doubao = DoubaoClient(client_id="xxx", client_secret="xxx")

print(deepseek.chat("你好"))
print(doubao.chat("你好"))

八、完整实战项目：智能问答助手

下面是一个完整的 FastAPI 智能问答助手，整合了上述所有能力。

# app.py
import os
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from openai import OpenAI
from typing import Optional, List, Dict
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI(title="智能问答助手 API")

# ============ 配置 ============
DEEPSEEK_API_KEY = os.environ.get('DEEPSEEK_API_KEY')
if not DEEPSEEK_API_KEY:
    raise ValueError("请设置环境变量 DEEPSEEK_API_KEY")

client = OpenAI(
    api_key=DEEPSEEK_API_KEY,
    base_url="https://api.deepseek.com"
)

# ============ 数据模型 ============
class ChatRequest(BaseModel):
    messages: List[Dict[str, str]]
    model: Optional[str] = "deepseek-v4-pro"
    temperature: Optional[float] = 0.7
    stream: Optional[bool] = False

class ChatResponse(BaseModel):
    reply: str
    model: str

# ============ API 端点 ============
@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    """多轮对话接口"""
    try:
        logger.info(f"收到请求，消息数: {len(request.messages)}")
        
        response = client.chat.completions.create(
            model=request.model,
            messages=request.messages,
            temperature=request.temperature,
            stream=request.stream
        )
        
        return ChatResponse(
            reply=response.choices[0].message.content,
            model=request.model
        )
    except Exception as e:
        logger.error(f"API 调用失败: {e}")
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health():
    """健康检查"""
    return {"status": "ok"}

# ============ 启动 ============
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

测试：

# 启动服务
python app.py

# 测试调用
curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{
    "messages": [
      {"role": "system", "content": "你是一个Python专家"},
      {"role": "user", "content": "什么是异步编程？"}
    ],
    "temperature": 0.5
  }'

九、常见踩坑与解决方案

问题	原因	解决方案
401 Unauthorized	API Key 无效或过期	检查 Key 是否正确，重新生成
429 Too Many Requests	触发限流	实现指数退避重试
返回内容被截断	max_tokens 设置太小	增大 max_tokens 参数
上下文丢失	未正确传递历史消息	每次请求携带完整 messages
中文乱码	编码问题	确保使用 UTF-8 编码
响应速度慢	网络延迟或 stream=False	启用流式输出 stream=True

十、总结

本文从零开始，完整介绍了：

1. 基础调用：使用 OpenAI SDK 和 requests 两种方式
2. 流式输出：实现打字机效果，提升用户体验
3. 多轮对话：手动维护对话历史，封装对话管理类
4. Prompt 工程：角色+任务+约束模板，思维链提示
5. 错误处理：指数退避重试机制
6. 豆包 API：获取 Token 和调用方法
7. 完整项目：FastAPI 智能问答助手

📌 核心要点：DeepSeek API 完全兼容 OpenAI 格式，切换只需改 base_url；多轮对话需手动维护 messages 列表；生产环境务必做好错误处理和重试机制。

---

代码已全部可运行，复制即可使用。 如果有问题，欢迎在评论区交流！😊