Python调用DeepSeek API全攻略：流式传输与高效开发指南

2025年9月24日 互联网

Python调用DeepSeek API详细教程：从流式传输到实战技巧——附代码示例

一、环境准备与基础配置

1.1 安装依赖库

使用DeepSeek API前需安装核心依赖库requests（HTTP请求）和websocket-client（流式传输）。推荐通过虚拟环境管理依赖：

  1. python -m venv deepseek_env
  2. source deepseek_env/bin/activate  # Linux/Mac
  3. # 或 deepseek_env\Scripts\activate (Windows)
  4. pip install requests websocket-client

1.2 获取API密钥

登录DeepSeek开发者平台，在「API管理」页面创建应用并获取API_KEY和API_SECRET。建议将密钥存储在环境变量中：

  1. import os
  2. os.environ['DEEPSEEK_API_KEY'] = 'your_api_key_here'
  3. os.environ['DEEPSEEK_API_SECRET'] = 'your_api_secret_here'

二、基础API调用方法

2.1 文本生成示例

通过POST /v1/completions接口实现基础文本生成：

  1. import requests
  2. import json
  4. def generate_text(prompt, model="deepseek-chat"):
  5.     url = "https://api.deepseek.com/v1/completions"
  6.     headers = {
  7.         "Content-Type": "application/json",
  8.         "Authorization": f"Bearer {os.getenv('DEEPSEEK_API_KEY')}"
  9.     }
  10.     data = {
  11.         "model": model,
  12.         "prompt": prompt,
  13.         "max_tokens": 200,
  14.         "temperature": 0.7
  15.     }
  16.     response = requests.post(url, headers=headers, data=json.dumps(data))
  17.     return response.json()
  19. # 示例调用
  20. result = generate_text("解释量子计算的基本原理")
  21. print(result['choices'][0]['text'])

2.2 参数详解

  • model: 指定模型版本（如deepseek-7b、deepseek-chat）
  • max_tokens: 控制生成文本长度（建议100-2000）
  • temperature: 创造力参数（0.1-1.0，值越高越随机）
  • top_p: 核采样阈值（0.8-0.95推荐）

三、流式传输实现

3.1 WebSocket流式传输

对于长文本生成，使用WebSocket实现实时输出：

  1. import websocket
  2. import json
  4. def stream_generate(prompt):
  5.     ws_url = "wss://api.deepseek.com/v1/stream/completions"
  6.     headers = {
  7.         "Authorization": f"Bearer {os.getenv('DEEPSEEK_API_KEY')}"
  8.     }
  10.     def on_message(ws, message):
  11.         data = json.loads(message)
  12.         if 'choices' in data:
  13.             delta = data['choices'][0]['delta']
  14.             if 'content' in delta:
  15.                 print(delta['content'], end='', flush=True)
  17.     def on_error(ws, error):
  18.         print(f"WebSocket错误: {error}")
  20.     ws = websocket.WebSocketApp(
  21.         ws_url,
  22.         header=headers,
  23.         on_message=on_message,
  24.         on_error=on_error
  25.     )
  27.     request_data = {
  28.         "model": "deepseek-chat",
  29.         "prompt": prompt,
  30.         "stream": True
  31.     }
  33.     ws.on_open = lambda ws: ws.send(json.dumps(request_data))
  34.     ws.run_forever()
  36. # 示例调用
  37. stream_generate("写一首关于春天的七言绝句")

3.2 SSE（Server-Sent Events）方案

若服务支持SSE，可使用更简单的实现：

  1. import requests
  3. def sse_stream(prompt):
  4.     url = "https://api.deepseek.com/v1/sse/completions"
  5.     headers = {
  6.         "Accept": "text/event-stream",
  7.         "Authorization": f"Bearer {os.getenv('DEEPSEEK_API_KEY')}"
  8.     }
  9.     params = {
  10.         "model": "deepseek-chat",
  11.         "prompt": prompt,
  12.         "stream": True
  13.     }
  15.     with requests.get(url, headers=headers, params=params, stream=True) as r:
  16.         for line in r.iter_lines():
  17.             if line:
  18.                 decoded = line.decode('utf-8')
  19.                 if 'data:' in decoded:
  20.                     content = decoded.split('data: ')[1].strip()
  21.                     if content != '[DONE]':
  22.                         print(json.loads(content)['choices'][0]['text'], end='', flush=True)
  24. # 示例调用
  25. sse_stream("用Python写一个快速排序算法")

四、实战优化技巧

4.1 请求重试机制

网络不稳定时实现自动重试：

  1. from requests.adapters import HTTPAdapter
  2. from urllib3.util.retry import Retry
  4. def create_session():
  5.     session = requests.Session()
  6.     retries = Retry(
  7.         total=3,
  8.         backoff_factor=1,
  9.         status_forcelist=[500, 502, 503, 504]
  10.     )
  11.     session.mount('https://', HTTPAdapter(max_retries=retries))
  12.     return session
  14. # 使用示例
  15. session = create_session()
  16. response = session.post(url, headers=headers, data=json.dumps(data))

4.2 批量请求处理

合并多个短请求提升效率：

  1. def batch_generate(prompts):
  2.     url = "https://api.deepseek.com/v1/batch/completions"
  3.     data = {
  4.         "requests": [
  5.             {"prompt": p, "max_tokens": 100} for p in prompts
  6.         ]
  7.     }
  8.     response = requests.post(url, headers=headers, data=json.dumps(data))
  9.     return response.json()
  11. # 示例调用
  12. results = batch_generate(["问题1", "问题2", "问题3"])

4.3 性能监控

添加请求耗时统计：

  1. import time
  3. def timed_generate(prompt):
  4.     start = time.time()
  5.     result = generate_text(prompt)
  6.     elapsed = time.time() - start
  7.     print(f"\n请求耗时: {elapsed:.2f}秒")
  8.     return result
  10. # 示例调用
  11. timed_generate("分析Python装饰器的使用场景")

五、常见问题解决方案

5.1 连接超时处理

  1. import socket
  2. from requests.packages.urllib3.util.connection import allowed_gai_family
  4. # 限制为IPv4
  5. socket.gai_family = socket.AF_INET
  6. requests.packages.urllib3.util.connection.allowed_gai_family = lambda: socket.AF_INET

5.2 速率限制应对

  1. from time import sleep
  3. def rate_limited_call(func, max_calls=10, period=60):
  4.     import time
  5.     from collections import deque
  7.     call_times = deque(maxlen=max_calls)
  9.     def wrapper(*args, **kwargs):
  10.         now = time.time()
  11.         call_times.append(now)
  13.         # 计算窗口内调用次数
  14.         oldest = now - period
  15.         while call_times and call_times[0] < oldest:
  16.             call_times.popleft()
  18.         if len(call_times) >= max_calls:
  19.             elapsed = now - call_times[0]
  20.             sleep_time = period - elapsed
  21.             if sleep_time > 0:
  22.                 sleep(sleep_time)
  24.         return func(*args, **kwargs)
  26.     return wrapper
  28. # 使用示例
  29. @rate_limited_call(max_calls=5, period=10)
  30. def limited_generate(prompt):
  31.     return generate_text(prompt)

六、完整项目示例

6.1 智能问答系统

  1. class DeepSeekQA:
  2.     def __init__(self):
  3.         self.api_key = os.getenv('DEEPSEEK_API_KEY')
  4.         self.session = create_session()
  6.     def ask(self, question, stream=False):
  7.         url = "https://api.deepseek.com/v1/completions"
  8.         if stream:
  9.             url = url.replace('/completions', '/stream/completions')
  11.         headers = {
  12.             "Content-Type": "application/json",
  13.             "Authorization": f"Bearer {self.api_key}"
  14.         }
  15.         data = {
  16.             "model": "deepseek-chat",
  17.             "prompt": f"问题: {question}\n答案:",
  18.             "max_tokens": 300,
  19.             "temperature": 0.3
  20.         }
  22.         if stream:
  23.             return self._stream_response(url, headers, data)
  24.         else:
  25.             return self._sync_response(url, headers, data)
  27.     def _sync_response(self, url, headers, data):
  28.         response = self.session.post(url, headers=headers, data=json.dumps(data))
  29.         return response.json()
  31.     def _stream_response(self, url, headers, data):
  32.         # 实现类似前面的流式处理逻辑
  33.         pass
  35. # 使用示例
  36. qa = DeepSeekQA()
  37. answer = qa.ask("Python中如何实现多线程？")
  38. print(answer)

七、最佳实践建议

  1. 模型选择：根据任务复杂度选择模型，简单问答用deepseek-7b，创意写作用deepseek-chat
  2. 参数调优：
    • 事实性查询：temperature=0.1-0.3
    • 创意写作：temperature=0.7-0.9
  3. 错误处理：实现完整的异常捕获链
  4. 日志记录：记录请求参数和响应状态
  5. 缓存机制：对重复问题实现结果缓存

通过本文的详细指导，开发者可以快速掌握DeepSeek API的调用技巧，从基础集成到高级流式传输实现，构建高效稳定的AI应用。实际开发中建议结合具体业务场景进行参数调优和架构设计。

最新文章