简介
本文将详细介绍如何使用 Python 调用 DeepSeek API,实现流式对话并保存对话记录。相比 Go 版本,Python 实现更加简洁优雅,适合快速开发和原型验证。https://cloud.siliconflow.cn/i/vnCCfVaQ
1. 环境准备
1.1 依赖安装
1.2 项目结构
- deepseek-project/
- ├── main.py # 主程序
- └── conversation.txt # 对话记录文件
复制代码
2. 完整代码实现
- import os
- import json
- import time
- import requests
- from datetime import datetime
-
- def save_to_file(file, content, is_question=False):
- """保存对话内容到文件"""
- timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
- if is_question:
- file.write(f"\n[{timestamp}] Question:\n{content}\n\n[{timestamp}] Answer:\n")
- else:
- file.write(content)
-
- def main():
- # 配置
- url = "https://api.siliconflow.cn/v1/chat/completions"
- headers = {
- "Content-Type": "application/json",
- "Authorization": "Bearer YOUR_API_KEY" # 替换为你的 API Key
- }
-
- # 打开文件用于保存对话
- with open("conversation.txt", "a", encoding="utf-8") as file:
- while True:
- # 获取用户输入
- question = input("\n请输入您的问题 (输入 q 退出): ").strip()
-
- if question.lower() == 'q':
- print("程序已退出")
- break
-
- # 保存问题
- save_to_file(file, question, is_question=True)
-
- # 准备请求数据
- data = {
- "model": "deepseek-ai/DeepSeek-V3",
- "messages": [
- {
- "role": "user",
- "content": question
- }
- ],
- "stream": True,
- "max_tokens": 2048,
- "temperature": 0.7,
- "top_p": 0.7,
- "top_k": 50,
- "frequency_penalty": 0.5,
- "n": 1,
- "response_format": {
- "type": "text"
- }
- }
-
- try:
- # 发送流式请求
- response = requests.post(url, json=data, headers=headers, stream=True)
- response.raise_for_status() # 检查响应状态
-
- # 处理流式响应
- for line in response.iter_lines():
- if line:
- line = line.decode('utf-8')
- if line.startswith('data: '):
- if line == 'data: [DONE]':
- continue
-
- try:
- content = json.loads(line[6:]) # 去掉 'data: ' 前缀
- if content['choices'][0]['delta'].get('content'):
- chunk = content['choices'][0]['delta']['content']
- print(chunk, end='', flush=True)
- file.write(chunk)
- file.flush()
- except json.JSONDecodeError:
- continue
-
- # 添加分隔符
- print("\n----------------------------------------")
- file.write("\n----------------------------------------\n")
- file.flush()
-
- except requests.RequestException as e:
- error_msg = f"请求错误: {str(e)}\n"
- print(error_msg)
- file.write(error_msg)
- file.flush()
-
- if __name__ == "__main__":
- main()
复制代码
3. 代码详解
3.1 核心功能
文件记录功能
save_to_file 函数负责:
API 配置
- headers = {
- "Content-Type": "application/json",
- "Authorization": "Bearer YOUR_API_KEY" # 替换为你的 API Key
- }
复制代码
流式请求处理
程序使用 requests 库的流式处理功能:
- 使用 stream=True 启用流式传输
- 逐行处理响应数据
- 实时显示和保存内容
3.2 配置参数说明
API 请求参数:
- model: 使用的模型名称
- stream: 启用流式输出
- max_tokens: 最大输出长度 (2048)
- temperature: 控制随机性 (0.7)
- top_p, top_k: 采样参数
- frequency_penalty: 重复惩罚系数
4. 错误处理
代码包含完整的错误处理机制:
- 检查 HTTP 响应状态
- 捕获网络异常
- 处理 JSON 解析错误
- 文件操作错误处理
5. 使用方法
5.1 修改配置
在代码中替换 YOUR_API_KEY 为你的实际 API Key。
5.2 运行程序
5.3 交互方式
- 输入问题进行对话
- 输入 ‘q’ 退出程序
- 查看 conversation.txt 获取对话记录
6. 性能优化建议
-
文件操作
- 使用适当的缓冲区大小
- 定期刷新文件缓冲
- 正确关闭文件句柄
-
网络请求
- 设置适当的超时
- 使用会话(Session)复用连接
- 处理网络异常
-
内存管理
- 及时释放资源
- 避免大量数据积累
- 使用生成器处理流式数据
总结
Python 版本的 DeepSeek API 调用实现简单直观,适合快速开发和测试。通过流式处理和文件记录,提供了完整的对话体验。
立即体验
现在就来体验 DeepSeek 的强大功能!
快来体验 DeepSeek:https://cloud.siliconflow.cn/i/vnCCfVaQ
快来体验 DeepSeek:https://cloud.siliconflow.cn/i/vnCCfVaQ
快来体验 DeepSeek:https://cloud.siliconflow.cn/i/vnCCfVaQ |