目录
简介1. 环境准备1.1 依赖安装1.2 项目结构2. 完整代码实现3. 代码详解3.1 核心功能文件记录功能API 配置流式请求处理3.2 配置参数说明4.错误处理5. 使用方法5.1 修改配置5.2 运行程序5.3 交互方式6. 性能优化建议总结简介
本文将详细介绍如何使用 Python 调用 DeepSeek API,实现流式对话并保存对话记录。相比 Go 版本,Python 实现更加简洁优雅,适合快速开发和原型验证。https://cloud.siliconflow.cn/i/vnCCfVaQ
1. 环境准备
1.1 依赖安装
pip install requests
1.2 项目结构
deepseek-project/├── main.py # 主程序└── conversation.txt # 对话记录文件
2. 完整代码实现
import osimport jsonimport timeimport requestsfrom datetime import datetimedef 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 运行程序
python main.py
5.3 交互方式
输入问题进行对话输入 ‘q’ 退出程序查看 conversation.txt 获取对话记录6. 性能优化建议
文件操作
使用适当的缓冲区大小定期刷新文件缓冲正确关闭文件句柄网络请求
设置适当的超时使用会话(Session)复用连接处理网络异常内存管理
及时释放资源避免大量数据积累使用生成器处理流式数据总结
Python 版本的 DeepSeek API 调用实现简单直观,适合快速开发和测试。通过流式处理和文件记录,提供了完整的对话体验。