人工智能大模型专栏：基于Pyqt与Api接口创造一个自己的AI助手
用 PyQt5 + DeepSeek API 实现桌面 AI 对话助手：QThread 流式输出、密钥配置界面与完整可运行源码讲解。
提示：文章写完后，目录可以自动生成，如何生成可参考右边的帮助文档
前言
在AI接口日益普及的今天，我们常常需要一个轻量化的工具来快速实现“本地UI+AI能力”的交互——不需要复杂的后端部署，也不用依赖网页环境，一个可直接运行的桌面应用就足够。
这篇文章要分享的，正是一个基于Python打造的极简AI对话助手。它用PyQt5搭建了直观的交互界面，集成了DeepSeek API实现智能对话，还通过多线程处理实现了流式响应——就像我们日常使用的聊天工具一样，AI的回复会逐字实时显示，而非等待完整响应。我们会从界面设计到功能实现，一步步拆解核心逻辑，让你既能直接用起来，也能明白背后的工作原理。
一、原理介绍
1、API介绍
简单来说，AI 的 API 本质是一套预先定义好的通信规则：
开发者通过代码发送特定格式的请求（比如输入一段文字、一张图片，或一个问题），API 会将这个请求传递给背后的 AI 模型；AI 模型处理后，再通过 API 将结果（比如回答、识别结果、生成的内容）返回给开发者的程序。整个过程中，开发者不用关心 AI 模型的内部原理（比如神经网络结构、训练数据），只需专注于 “如何调用” 和 “如何使用结果”。
举几个常见的例子：
1、对话 API：比如我们之前代码中用到的 DeepSeek 对话 API，开发者传入文字问题，API 返回 AI 生成的回答，能快速给 App 加上 “智能聊天” 功能；
2、图像识别 API：传入一张图片，API 返回 “这是一只猫”“图片中有 3 个人” 等识别结果，可用于相册分类、安防监控等场景；
2、Pyqt介绍
PyQt 是一套用于创建桌面应用界面的 Python 工具库：它就像一个 “界面搭建工具箱”—— 开发者不需要自己编写复杂的底层界面渲染代码（比如窗口绘制、按钮交互逻辑），只需调用 PyQt 提供的现成组件（如窗口、按钮、输入框）和布局工具，就能快速搭建出可视化的桌面应用界面。
二、API Key获取
网络上也有很多免费的api-key，但是我们这边采用的是Deepseek的一个试用版的API（挺便宜的，正常情况下十块钱试用就可以用挺久的）。获取网址如下：Deepseek API Key获取网址
充值之后在API keys那边创建密钥既可
需要注意的是，API密钥信息只会在创建时候提供给你，在那之后是无法查询的，所以需要在一开始就做好记录，某则只能重新创建一个API keys(虽然问题也不大就是了)
如果有需要暂时试用API进行测试，同时不想浪费钱购买的话可以私信借用我的API密钥
三、代码实现：
1、API通信类代码
# 导入必要的系统模块
import sys
# 导入JSON处理模块，用于解析API返回的JSON数据
import json
# 导入requests模块，用于发送HTTP请求
import requests
# 从PyQt5库导入所需的UI组件
from PyQt5.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout,
QHBoxLayout, QLabel, QPushButton, QTextEdit,
QGroupBox, QLineEdit, QMessageBox)
# 从PyQt5库导入核心功能模块
from PyQt5.QtCore import Qt, QThread, pyqtSignal
# 从PyQt5库导入GUI相关模块
from PyQt5.QtGui import QFont
class DeepSeekClient(QThread):
"""AI对话核心处理线程，负责与API交互，继承自QThread以实现多线程操作"""
# 定义信号：当完整响应接收完毕时发射
response_received = pyqtSignal(str)
# 定义信号：当接收到流式响应片段时发射
stream_chunk_received = pyqtSignal(str)
# 定义信号：当API调用发生错误时发射
api_error = pyqtSignal(str)
def __init__(self, api_key):
# 调用父类QThread的构造函数
super().__init__()
# 存储API密钥
self.api_key = api_key
# 设置DeepSeek API的URL
self.api_url = "https://api.deepseek.com/v1/chat/completions"
# 存储用户的提问内容
self.prompt = ""
# 线程运行状态标志
self.running = False
# 是否启用流式输出（实时显示回复内容）
self.streaming = True
def set_prompt(self, prompt):
"""设置用户的提问内容"""
self.prompt = prompt
def run(self):
"""线程执行函数，负责发送API请求并处理响应"""
# 设置线程运行状态为True
self.running = True
# 构建HTTP请求头，包含内容类型和认证信息
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}" # 使用Bearer认证方式
}
# 构建API请求数据
data = {
"model": "deepseek-chat", # 指定使用的AI模型
"messages": [{"role": "user", "content": self.prompt}], # 构建对话消息
"stream": self.streaming # 指定是否使用流式响应
}
try:
# 如果启用了流式输出
if self.streaming:
# 发送POST请求，开启流式响应模式
response = requests.post(self.api_url, headers=headers, json=data, stream=True)
# 检查请求是否成功（HTTP状态码200表示成功）
if response.status_code == 200:
# 用于存储完整的响应内容
full_response = ""
# 迭代处理流式响应的每一行数据
for line in response.iter_lines():
# 检查线程是否仍在运行且接收到有效数据
if line and self.running:
# 将字节数据转换为字符串
line_text = line.decode('utf-8')
# 检查是否是有效的数据行
if line_text.startswith("data: "):
# 提取JSON数据部分（去除"data: "前缀）
json_str = line_text[6:]
# 检查是否是流式响应结束标志
if json_str.strip() == "[DONE]":
break
try:
# 解析JSON数据
chunk_data = json.loads(json_str)
# 检查响应数据结构是否正确
if "choices" in chunk_data and len(chunk_data["choices"]) > 0:
# 提取AI回复的内容片段
chunk = chunk_data["choices"][0].get("delta", {}).get("content", "")
# 如果有内容，则添加到完整响应并发射信号
if chunk:
full_response += chunk
self.stream_chunk_received.emit(chunk)
# 处理JSON解析错误
except json.JSONDecodeError:
continue
# 发射完整响应信号
self.response_received.emit(full_response)
# 处理API返回的错误状态码
else:
self.api_error.emit(f"API错误: {response.status_code} - {response.text}")
# 如果未启用流式输出
else:
# 发送普通POST请求
response = requests.post(self.api_url, headers=headers, json=data)
# 检查请求是否成功
if response.status_code == 200:
# 解析JSON响应
response_data = response.json()
# 检查响应数据结构是否正确
if "choices" in response_data and len(response_data["choices"]) > 0:
# 提取AI回复内容
message = response_data["choices"][0]["message"]["content"]
# 发射完整响应信号
self.response_received.emit(message)
# 处理API返回的错误状态码
else:
self.api_error.emit(f"API错误: {response.status_code} - {response.text}")
# 处理其他可能的异常（如网络错误等）
except Exception as e:
self.api_error.emit(f"请求失败: {str(e)}")
# 线程运行结束，更新状态标志
self.running = False
def stop(self):
"""停止线程运行"""
self.running = False
代码解释
一、信号定义：通过pyqtSignal定义 3 种信号，用于向主线程传递信息：
1、stream_chunk_received：传递 AI 流式回复的内容片段（实时显示用）
2、response_received：传递完整的 AI 回复（对话记录存档用）
3、api_error：传递错误信息（如网络问题、API 调用失败等）
二、初始化与基础配置：
1、接收 API 密钥，指定 DeepSeek API 的请求地址
2、存储用户提问内容（prompt），设置是否启用流式响应（streaming）
3、维护线程运行状态标志（running）
三、核心运行逻辑（run方法）：
1、构建 API 请求的必要参数：包括请求头（含认证信息Bearer Token）、请求数据（指定 AI 模型、用户提问、响应模式）
分两种模式处理 API 交互：
（1） 流式响应（默认）：发送带stream=True的请求，逐行接收 AI 返回的片段，解析后通过stream_chunk_received实时发射，直到收到[DONE]结束标志
（2） 普通响应：发送stream=False的请求，接收完整回复后一次性发射
2、异常处理：捕获网络错误、JSON 解析错误、API 非 200 状态码等问题，通过api_error发射错误信息
3、辅助方法：
（1） set_prompt：用于外部设置用户提问内容
（2） stop：通过修改running标志停止线程运行，避免资源泄漏
提示：这里对文章进行总结：
例如：以上就是今天要讲的内容，本文仅仅简单介绍了pandas的使用，而pandas提供了大量能使我们快速便捷地处理数据的函数和方法。
2、Pyqt界面端代码
class AIChatApp(QMainWindow):
"""AI对话应用主窗口，继承自QMainWindow"""
def __init__(self):
# 调用父类QMainWindow的构造函数
super().__init__()
# 存储API密钥
self.api_key = ""
# AI客户端实例
self.deepseek_client = None
# 初始化用户界面
self.init_ui()
def init_ui(self):
# 设置窗口标题
self.setWindowTitle("AI对话助手")
# 设置窗口最小尺寸
self.setMinimumSize(800, 600)
# 设置全局字体为微软雅黑，大小为10
font = QFont("微软雅黑", 10)
self.setFont(font)
# 创建中央部件并设置为主窗口的中央部件
central_widget = QWidget()
self.setCentralWidget(central_widget)
# 创建主布局（垂直布局）
main_layout = QVBoxLayout(central_widget)
# 设置布局内控件间距
main_layout.setSpacing(15)
# 设置布局边缘的边距
main_layout.setContentsMargins(20, 20, 20, 20)
# 创建标题标签
title_label = QLabel("AI对话助手")
# 设置标签内容居中对齐
title_label.setAlignment(Qt.AlignCenter)
# 设置标题样式
title_label.setStyleSheet("font-size: 20px; font-weight: bold; margin-bottom: 10px;")
# 将标题添加到主布局
main_layout.addWidget(title_label)
# 创建API设置分组框
api_group = QGroupBox("API设置")
# 创建API设置区域的水平布局
api_layout = QHBoxLayout()
# 创建API密钥输入框
self.api_key_input = QLineEdit()
# 设置输入框提示文本
self.api_key_input.setPlaceholderText("请输入DeepSeek API密钥")
# 设置输入框为密码模式（输入内容显示为圆点）
self.api_key_input.setEchoMode(QLineEdit.Password)
# 设置输入框最小高度
self.api_key_input.setMinimumHeight(35)
# 创建保存密钥按钮
self.save_api_key_button = QPushButton("保存密钥")
# 设置按钮最小高度
self.save_api_key_button.setMinimumHeight(35)
# 绑定按钮点击事件到保存密钥函数
self.save_api_key_button.clicked.connect(self.save_api_key)
# 将控件添加到API设置布局
api_layout.addWidget(QLabel("API密钥:"), 0) # 标签，伸缩因子0
api_layout.addWidget(self.api_key_input, 1) # 输入框，伸缩因子1（占更多空间）
api_layout.addWidget(self.save_api_key_button, 0) # 按钮，伸缩因子0
# 设置API设置分组框的布局
api_group.setLayout(api_layout)
# 将API设置分组框添加到主布局
main_layout.addWidget(api_group)
# 创建对话历史分组框
chat_history_group = QGroupBox("对话历史")
# 创建对话历史区域的垂直布局
chat_history_layout = QVBoxLayout()
# 创建对话历史文本编辑框
self.chat_history = QTextEdit()
# 设置文本框为只读模式
self.chat_history.setReadOnly(True)
# 设置文本框样式
self.chat_history.setStyleSheet("""
background-color: #f9f9f9;
border: 1px solid #ddd;
border-radius: 5px;
padding: 10px;
""")
# 将对话历史文本框添加到布局
chat_history_layout.addWidget(self.chat_history)
# 设置对话历史分组框的布局
chat_history_group.setLayout(chat_history_layout)
# 将对话历史分组框添加到主布局，伸缩因子1（占主要空间）
main_layout.addWidget(chat_history_group, 1)
# 创建输入区域分组框
input_group = QGroupBox("发送消息")
# 创建输入区域的垂直布局
input_layout = QVBoxLayout()
# 创建用户输入文本框
self.user_input = QTextEdit()
# 设置输入框提示文本
self.user_input.setPlaceholderText("请输入您的问题...")
# 设置输入框最小高度
self.user_input.setMinimumHeight(80)
# 设置输入框样式
self.user_input.setStyleSheet("""
border: 1px solid #ddd;
border-radius: 5px;
padding: 10px;
""")
# 创建发送按钮
self.send_button = QPushButton("发送")
# 设置按钮最小高度
self.send_button.setMinimumHeight(40)
# 设置按钮样式
self.send_button.setStyleSheet("""
background-color: #27ae60;
color: white;
font-weight: bold;
border-radius: 5px;
""")
# 绑定按钮点击事件到发送消息函数
self.send_button.clicked.connect(self.send_message)
# 未设置密钥前禁用发送按钮
self.send_button.setEnabled(False)
# 将控件添加到输入区域布局
input_layout.addWidget(self.user_input)
input_layout.addWidget(self.send_button)
# 设置输入区域分组框的布局
input_group.setLayout(input_layout)
# 将输入区域分组框添加到主布局
main_layout.addWidget(input_group)
def save_api_key(self):
"""保存API密钥并初始化AI客户端"""
# 获取输入的API密钥并去除首尾空格
self.api_key = self.api_key_input.text().strip()
# 检查API密钥是否有效（非空）
if self.api_key:
# 初始化AI客户端实例
self.deepseek_client = DeepSeekClient(self.api_key)
# 连接AI客户端的信号到对应的槽函数
self.deepseek_client.response_received.connect(self.handle_complete_response)
self.deepseek_client.stream_chunk_received.connect(self.handle_stream_chunk)
self.deepseek_client.api_error.connect(self.handle_api_error)
# 显示成功提示
QMessageBox.information(self, "成功", "API密钥已保存，现在可以开始对话")
# 启用发送按钮
self.send_button.setEnabled(True)
# 如果API密钥为空
else:
# 显示错误提示
QMessageBox.warning(self, "错误", "请输入有效的API密钥")
# 保持发送按钮禁用状态
self.send_button.setEnabled(False)
def send_message(self):
"""发送用户消息到AI"""
# 检查AI客户端是否已初始化
if not self.deepseek_client:
# 显示错误提示
QMessageBox.warning(self, "错误", "请先设置并保存API密钥")
return
# 获取用户输入的消息并去除首尾空格
user_message = self.user_input.toPlainText().strip()
# 检查消息是否为空
if not user_message:
return # 空消息不发送
# 在对话历史中显示用户消息（使用HTML格式加粗显示"您:"）
self.chat_history.append(f"<b>您:</b> {user_message}")
self.chat_history.append("") # 添加空行分隔
self.chat_history.append("<b>AI:</b> ") # 准备显示AI回复
# 清空输入框
self.user_input.clear()
# 禁用发送按钮防止重复发送
self.send_button.setEnabled(False)
# 更新按钮文本显示正在处理
self.send_button.setText("正在回复...")
# 设置AI客户端的提问内容
self.deepseek_client.set_prompt(user_message)
# 启动AI客户端线程（发送请求）
self.deepseek_client.start()
def handle_stream_chunk(self, chunk):
"""处理流式返回的内容片段，实时显示到对话历史"""
# 在对话历史中插入接收到的内容片段
self.chat_history.textCursor().insertText(chunk)
# 自动滚动到最底部，确保最新内容可见
self.chat_history.verticalScrollBar().setValue(
self.chat_history.verticalScrollBar().maximum()
)
def handle_complete_response(self, response):
"""处理完整响应（流式结束时）"""
# 添加空行分隔不同的对话
self.chat_history.append("")
# 重新启用发送按钮
self.send_button.setEnabled(True)
# 恢复按钮文本
self.send_button.setText("发送")
def handle_api_error(self, error_msg):
"""处理API错误，显示错误信息"""
# 在对话历史中显示错误信息（使用红色字体）
self.chat_history.append(f"<span style='color: #e74c3c;'><b>错误:</b> {error_msg}</span>")
self.chat_history.append("") # 添加空行分隔
# 重新启用发送按钮
self.send_button.setEnabled(True)
# 恢复按钮文本
self.send_button.setText("发送")
def closeEvent(self, event):
"""窗口关闭时的处理函数，确保线程正确停止"""
# 检查AI客户端线程是否正在运行
if self.deepseek_client and self.deepseek_client.isRunning():
# 停止线程
self.deepseek_client.stop()
# 等待线程结束
self.deepseek_client.wait()
# 接受窗口关闭事件
event.accept()
一、界面设计（init_ui方法）：通过 PyQt5 组件搭建完整交互界面，分为 4 个核心区域：
1、标题区：显示 “AI 对话助手” 标题，居中突出显示；
2、API 设置区：包含 “API 密钥输入框”（密码模式隐藏输入）和 “保存密钥按钮”，用于配置调用 AI 所需的认证信息；
3、对话历史区：只读文本框（QTextEdit），用于展示用户提问和 AI 回复的完整记录，设置浅色背景和边框样式提升可读性；
4、输入区：包含 “消息输入框”（用于输入提问）和 “发送按钮”（默认禁用，需先保存 API 密钥），按钮样式使用绿色突出显示。
布局上通过QVBoxLayout（垂直）和QHBoxLayout（水平）嵌套，控制控件排列和占比（如对话历史区占最大空间），同时设置字体、边距、样式等美化界面。
二、 核心交互逻辑
1、API 密钥管理（save_api_key方法）：
接收用户输入的 API 密钥，验证非空后初始化DeepSeekClient实例，并将其信号（流式片段、完整响应、错误）与自身槽函数绑定；通过弹窗提示保存结果，同时启用 “发送按钮”（未输入密钥时保持禁用）。
2、消息发送（send_message方法）：
验证用户输入（非空、已初始化 AI 客户端）后，将提问内容添加到 “对话历史区”（标记 “您:” 并加粗）；清空输入框、禁用发送按钮（显示 “正在回复”），随后调用DeepSeekClient的set_prompt设置提问，并启动线程发送请求。
3、响应处理（3 个槽函数）：
（1）handle_stream_chunk：接收 AI 流式回复的片段，实时插入对话历史并自动滚动到底部（模拟 “逐字输出” 效果）；
（2）handle_complete_response：流式回复结束后，添加空行分隔对话，恢复发送按钮状态（启用并显示 “发送”）；
（3）handle_api_error：接收错误信息（如网络问题、API 调用失败），以红色字体显示在对话历史中，同时恢复发送按钮可用状态。
三、 资源与状态管理
1、窗口关闭处理（closeEvent方法）：关闭窗口时检查DeepSeekClient线程是否运行，若运行则停止并等待线程结束，避免资源泄漏；
2、状态反馈：通过按钮文本（“发送”→“正在回复”）、弹窗提示（密钥保存成功 / 失败）、对话历史标记（“您:”“AI:”“错误”），让用户清晰感知当前操作状态。
3、完整代码
# 导入必要的系统模块
import sys
# 导入JSON处理模块，用于解析API返回的JSON数据
import json
# 导入requests模块，用于发送HTTP请求
import requests
# 从PyQt5库导入所需的UI组件
from PyQt5.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout,
QHBoxLayout, QLabel, QPushButton, QTextEdit,
QGroupBox, QLineEdit, QMessageBox)
# 从PyQt5库导入核心功能模块
from PyQt5.QtCore import Qt, QThread, pyqtSignal
# 从PyQt5库导入GUI相关模块
from PyQt5.QtGui import QFont
class DeepSeekClient(QThread):
"""AI对话核心处理线程，负责与API交互，继承自QThread以实现多线程操作"""
# 定义信号：当完整响应接收完毕时发射
response_received = pyqtSignal(str)
# 定义信号：当接收到流式响应片段时发射
stream_chunk_received = pyqtSignal(str)
# 定义信号：当API调用发生错误时发射
api_error = pyqtSignal(str)
def __init__(self, api_key):
# 调用父类QThread的构造函数
super().__init__()
# 存储API密钥
self.api_key = api_key
# 设置DeepSeek API的URL
self.api_url = "https://api.deepseek.com/v1/chat/completions"
# 存储用户的提问内容
self.prompt = ""
# 线程运行状态标志
self.running = False
# 是否启用流式输出（实时显示回复内容）
self.streaming = True
def set_prompt(self, prompt):
"""设置用户的提问内容"""
self.prompt = prompt
def run(self):
"""线程执行函数，负责发送API请求并处理响应"""
# 设置线程运行状态为True
self.running = True
# 构建HTTP请求头，包含内容类型和认证信息
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}" # 使用Bearer认证方式
}
# 构建API请求数据
data = {
"model": "deepseek-chat", # 指定使用的AI模型
"messages": [{"role": "user", "content": self.prompt}], # 构建对话消息
"stream": self.streaming # 指定是否使用流式响应
}
try:
# 如果启用了流式输出
if self.streaming:
# 发送POST请求，开启流式响应模式
response = requests.post(self.api_url, headers=headers, json=data, stream=True)
# 检查请求是否成功（HTTP状态码200表示成功）
if response.status_code == 200:
# 用于存储完整的响应内容
full_response = ""
# 迭代处理流式响应的每一行数据
for line in response.iter_lines():
# 检查线程是否仍在运行且接收到有效数据
if line and self.running:
# 将字节数据转换为字符串
line_text = line.decode('utf-8')
# 检查是否是有效的数据行
if line_text.startswith("data: "):
# 提取JSON数据部分（去除"data: "前缀）
json_str = line_text[6:]
# 检查是否是流式响应结束标志
if json_str.strip() == "[DONE]":
break
try:
# 解析JSON数据
chunk_data = json.loads(json_str)
# 检查响应数据结构是否正确
if "choices" in chunk_data and len(chunk_data["choices"]) > 0:
# 提取AI回复的内容片段
chunk = chunk_data["choices"][0].get("delta", {}).get("content", "")
# 如果有内容，则添加到完整响应并发射信号
if chunk:
full_response += chunk
self.stream_chunk_received.emit(chunk)
# 处理JSON解析错误
except json.JSONDecodeError:
continue
# 发射完整响应信号
self.response_received.emit(full_response)
# 处理API返回的错误状态码
else:
self.api_error.emit(f"API错误: {response.status_code} - {response.text}")
# 如果未启用流式输出
else:
# 发送普通POST请求
response = requests.post(self.api_url, headers=headers, json=data)
# 检查请求是否成功
if response.status_code == 200:
# 解析JSON响应
response_data = response.json()
# 检查响应数据结构是否正确
if "choices" in response_data and len(response_data["choices"]) > 0:
# 提取AI回复内容
message = response_data["choices"][0]["message"]["content"]
# 发射完整响应信号
self.response_received.emit(message)
# 处理API返回的错误状态码
else:
self.api_error.emit(f"API错误: {response.status_code} - {response.text}")
# 处理其他可能的异常（如网络错误等）
except Exception as e:
self.api_error.emit(f"请求失败: {str(e)}")
# 线程运行结束，更新状态标志
self.running = False
def stop(self):
"""停止线程运行"""
self.running = False
class AIChatApp(QMainWindow):
"""AI对话应用主窗口，继承自QMainWindow"""
def __init__(self):
# 调用父类QMainWindow的构造函数
super().__init__()
# 存储API密钥
self.api_key = ""
# AI客户端实例
self.deepseek_client = None
# 初始化用户界面
self.init_ui()
def init_ui(self):
# 设置窗口标题
self.setWindowTitle("AI对话助手")
# 设置窗口最小尺寸
self.setMinimumSize(800, 600)
# 设置全局字体为微软雅黑，大小为10
font = QFont("微软雅黑", 10)
self.setFont(font)
# 创建中央部件并设置为主窗口的中央部件
central_widget = QWidget()
self.setCentralWidget(central_widget)
# 创建主布局（垂直布局）
main_layout = QVBoxLayout(central_widget)
# 设置布局内控件间距
main_layout.setSpacing(15)
# 设置布局边缘的边距
main_layout.setContentsMargins(20, 20, 20, 20)
# 创建标题标签
title_label = QLabel("AI对话助手")
# 设置标签内容居中对齐
title_label.setAlignment(Qt.AlignCenter)
# 设置标题样式
title_label.setStyleSheet("font-size: 20px; font-weight: bold; margin-bottom: 10px;")
# 将标题添加到主布局
main_layout.addWidget(title_label)
# 创建API设置分组框
api_group = QGroupBox("API设置")
# 创建API设置区域的水平布局
api_layout = QHBoxLayout()
# 创建API密钥输入框
self.api_key_input = QLineEdit()
# 设置输入框提示文本
self.api_key_input.setPlaceholderText("请输入DeepSeek API密钥")
# 设置输入框为密码模式（输入内容显示为圆点）
self.api_key_input.setEchoMode(QLineEdit.Password)
# 设置输入框最小高度
self.api_key_input.setMinimumHeight(35)
# 创建保存密钥按钮
self.save_api_key_button = QPushButton("保存密钥")
# 设置按钮最小高度
self.save_api_key_button.setMinimumHeight(35)
# 绑定按钮点击事件到保存密钥函数
self.save_api_key_button.clicked.connect(self.save_api_key)
# 将控件添加到API设置布局
api_layout.addWidget(QLabel("API密钥:"), 0) # 标签，伸缩因子0
api_layout.addWidget(self.api_key_input, 1) # 输入框，伸缩因子1（占更多空间）
api_layout.addWidget(self.save_api_key_button, 0) # 按钮，伸缩因子0
# 设置API设置分组框的布局
api_group.setLayout(api_layout)
# 将API设置分组框添加到主布局
main_layout.addWidget(api_group)
# 创建对话历史分组框
chat_history_group = QGroupBox("对话历史")
# 创建对话历史区域的垂直布局
chat_history_layout = QVBoxLayout()
# 创建对话历史文本编辑框
self.chat_history = QTextEdit()
# 设置文本框为只读模式
self.chat_history.setReadOnly(True)
# 设置文本框样式
self.chat_history.setStyleSheet("""
background-color: #f9f9f9;
border: 1px solid #ddd;
border-radius: 5px;
padding: 10px;
""")
# 将对话历史文本框添加到布局
chat_history_layout.addWidget(self.chat_history)
# 设置对话历史分组框的布局
chat_history_group.setLayout(chat_history_layout)
# 将对话历史分组框添加到主布局，伸缩因子1（占主要空间）
main_layout.addWidget(chat_history_group, 1)
# 创建输入区域分组框
input_group = QGroupBox("发送消息")
# 创建输入区域的垂直布局
input_layout = QVBoxLayout()
# 创建用户输入文本框
self.user_input = QTextEdit()
# 设置输入框提示文本
self.user_input.setPlaceholderText("请输入您的问题...")
# 设置输入框最小高度
self.user_input.setMinimumHeight(80)
# 设置输入框样式
self.user_input.setStyleSheet("""
border: 1px solid #ddd;
border-radius: 5px;
padding: 10px;
""")
# 创建发送按钮
self.send_button = QPushButton("发送")
# 设置按钮最小高度
self.send_button.setMinimumHeight(40)
# 设置按钮样式
self.send_button.setStyleSheet("""
background-color: #27ae60;
color: white;
font-weight: bold;
border-radius: 5px;
""")
# 绑定按钮点击事件到发送消息函数
self.send_button.clicked.connect(self.send_message)
# 未设置密钥前禁用发送按钮
self.send_button.setEnabled(False)
# 将控件添加到输入区域布局
input_layout.addWidget(self.user_input)
input_layout.addWidget(self.send_button)
# 设置输入区域分组框的布局
input_group.setLayout(input_layout)
# 将输入区域分组框添加到主布局
main_layout.addWidget(input_group)
def save_api_key(self):
"""保存API密钥并初始化AI客户端"""
# 获取输入的API密钥并去除首尾空格
self.api_key = self.api_key_input.text().strip()
# 检查API密钥是否有效（非空）
if self.api_key:
# 初始化AI客户端实例
self.deepseek_client = DeepSeekClient(self.api_key)
# 连接AI客户端的信号到对应的槽函数
self.deepseek_client.response_received.connect(self.handle_complete_response)
self.deepseek_client.stream_chunk_received.connect(self.handle_stream_chunk)
self.deepseek_client.api_error.connect(self.handle_api_error)
# 显示成功提示
QMessageBox.information(self, "成功", "API密钥已保存，现在可以开始对话")
# 启用发送按钮
self.send_button.setEnabled(True)
# 如果API密钥为空
else:
# 显示错误提示
QMessageBox.warning(self, "错误", "请输入有效的API密钥")
# 保持发送按钮禁用状态
self.send_button.setEnabled(False)
def send_message(self):
"""发送用户消息到AI"""
# 检查AI客户端是否已初始化
if not self.deepseek_client:
# 显示错误提示
QMessageBox.warning(self, "错误", "请先设置并保存API密钥")
return
# 获取用户输入的消息并去除首尾空格
user_message = self.user_input.toPlainText().strip()
# 检查消息是否为空
if not user_message:
return # 空消息不发送
# 在对话历史中显示用户消息（使用HTML格式加粗显示"您:"）
self.chat_history.append(f"<b>您:</b> {user_message}")
self.chat_history.append("") # 添加空行分隔
self.chat_history.append("<b>AI:</b> ") # 准备显示AI回复
# 清空输入框
self.user_input.clear()
# 禁用发送按钮防止重复发送
self.send_button.setEnabled(False)
# 更新按钮文本显示正在处理
self.send_button.setText("正在回复...")
# 设置AI客户端的提问内容
self.deepseek_client.set_prompt(user_message)
# 启动AI客户端线程（发送请求）
self.deepseek_client.start()
def handle_stream_chunk(self, chunk):
"""处理流式返回的内容片段，实时显示到对话历史"""
# 在对话历史中插入接收到的内容片段
self.chat_history.textCursor().insertText(chunk)
# 自动滚动到最底部，确保最新内容可见
self.chat_history.verticalScrollBar().setValue(
self.chat_history.verticalScrollBar().maximum()
)
def handle_complete_response(self, response):
"""处理完整响应（流式结束时）"""
# 添加空行分隔不同的对话
self.chat_history.append("")
# 重新启用发送按钮
self.send_button.setEnabled(True)
# 恢复按钮文本
self.send_button.setText("发送")
def handle_api_error(self, error_msg):
"""处理API错误，显示错误信息"""
# 在对话历史中显示错误信息（使用红色字体）
self.chat_history.append(f"<span style='color: #e74c3c;'><b>错误:</b> {error_msg}</span>")
self.chat_history.append("") # 添加空行分隔
# 重新启用发送按钮
self.send_button.setEnabled(True)
# 恢复按钮文本
self.send_button.setText("发送")
def closeEvent(self, event):
"""窗口关闭时的处理函数，确保线程正确停止"""
# 检查AI客户端线程是否正在运行
if self.deepseek_client and self.deepseek_client.isRunning():
# 停止线程
self.deepseek_client.stop()
# 等待线程结束
self.deepseek_client.wait()
# 接受窗口关闭事件
event.accept()
# 程序入口点
if __name__ == "__main__":
# 创建Qt应用实例
app = QApplication(sys.argv)
# 创建主窗口实例
window = AIChatApp()
# 显示主窗口
window.show()
# 进入应用主循环，等待用户交互
sys.exit(app.exec_())w1
四、效果演示
运行代码，弹出pyqt窗口
在API密钥中输入获取的Deekseek API密钥，并进行提问。效果如下
人工智能大模型專欄：基於Pyqt與Api接口創造一個自己的AI助手
用 PyQt5 + DeepSeek API 實現桌面 AI 對話助手：QThread 流式輸出、密鑰配置界面與完整可運行源碼講解。
來源：https://blog.csdn.net/2403_87969572/article/details/149724535
抓取時間（ISO本地）：2026-05-18 05:17:00
提示：文章寫完後，目錄可以自動生成，如何生成可參考右邊的幫助文檔
文章目錄
前言
在AI接口日益普及的今天，我們常常需要一個輕量化的工具來快速實現“本地UI+AI能力”的交互——不需要複雜的後端部署，也不用依賴網頁環境，一個可直接運行的桌面應用就足夠。
這篇文章要分享的，正是一個基於Python打造的極簡AI對話助手。它用PyQt5搭建了直觀的交互界面，集成了DeepSeek API實現智能對話，還通過多線程處理實現了流式響應——就像我們日常使用的聊天工具一樣，AI的回覆會逐字實時顯示，而非等待完整響應。我們會從界面設計到功能實現，一步步拆解核心邏輯，讓你既能直接用起來，也能明白背後的工作原理。
一、原理介紹
1、API介紹
簡單來說，AI 的 API 本質是一套預先定義好的通信規則：
開發者通過代碼發送特定格式的請求（比如輸入一段文字、一張圖片，或一個問題），API 會將這個請求傳遞給背後的 AI 模型；AI 模型處理後，再通過 API 將結果（比如回答、識別結果、生成的內容）返回給開發者的程序。整個過程中，開發者不用關心 AI 模型的內部原理（比如神經網絡結構、訓練數據），只需專注於 “如何調用” 和 “如何使用結果”。
舉幾個常見的例子：
1、對話 API：比如我們之前代碼中用到的 DeepSeek 對話 API，開發者傳入文字問題，API 返回 AI 生成的回答，能快速給 App 加上 “智能聊天” 功能；
2、圖像識別 API：傳入一張圖片，API 返回 “這是一隻貓”“圖片中有 3 個人” 等識別結果，可用於相冊分類、安防監控等場景；
2、Pyqt介紹
PyQt 是一套用於創建桌面應用界面的 Python 工具庫：它就像一個 “界面搭建工具箱”—— 開發者不需要自己編寫複雜的底層界面渲染代碼（比如窗口繪製、按鈕交互邏輯），只需調用 PyQt 提供的現成組件（如窗口、按鈕、輸入框）和佈局工具，就能快速搭建出可視化的桌面應用界面。
二、API Key獲取
網絡上也有很多免費的api-key，但是我們這邊採用的是Deepseek的一個試用版的API（挺便宜的，正常情況下十塊錢試用就可以用挺久的）。獲取網址如下：Deepseek API Key獲取網址
充值之後在API keys那邊創建密鑰既可
需要注意的是，API密鑰信息只會在創建時候提供給你，在那之後是無法查詢的，所以需要在一開始就做好記錄，某則只能重新創建一個API keys(雖然問題也不大就是了)
如果有需要暫時試用API進行測試，同時不想浪費錢購買的話可以私信借用我的API密鑰
三、代碼實現：
1、API通信類代碼
# 導入必要的系統模塊
import sys
# 導入JSON處理模塊，用於解析API返回的JSON數據
import json
# 導入requests模塊，用於發送HTTP請求
import requests
# 從PyQt5庫導入所需的UI組件
from PyQt5.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout,
QHBoxLayout, QLabel, QPushButton, QTextEdit,
QGroupBox, QLineEdit, QMessageBox)
# 從PyQt5庫導入核心功能模塊
from PyQt5.QtCore import Qt, QThread, pyqtSignal
# 從PyQt5庫導入GUI相關模塊
from PyQt5.QtGui import QFont
class DeepSeekClient(QThread):
"""AI對話核心處理線程，負責與API交互，繼承自QThread以實現多線程操作"""
# 定義信號：當完整響應接收完畢時發射
response_received = pyqtSignal(str)
# 定義信號：當接收到流式響應片段時發射
stream_chunk_received = pyqtSignal(str)
# 定義信號：當API調用發生錯誤時發射
api_error = pyqtSignal(str)
def __init__(self, api_key):
# 調用父類QThread的構造函數
super().__init__()
# 存儲API密鑰
self.api_key = api_key
# 設置DeepSeek API的URL
self.api_url = "https://api.deepseek.com/v1/chat/completions"
# 存儲用戶的提問內容
self.prompt = ""
# 線程運行狀態標誌
self.running = False
# 是否啟用流式輸出（實時顯示回覆內容）
self.streaming = True
def set_prompt(self, prompt):
"""設置用戶的提問內容"""
self.prompt = prompt
def run(self):
"""線程執行函數，負責發送API請求並處理響應"""
# 設置線程運行狀態為True
self.running = True
# 構建HTTP請求頭，包含內容類型和認證信息
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}" # 使用Bearer認證方式
}
# 構建API請求數據
data = {
"model": "deepseek-chat", # 指定使用的AI模型
"messages": [{"role": "user", "content": self.prompt}], # 構建對話消息
"stream": self.streaming # 指定是否使用流式響應
}
try:
# 如果啟用了流式輸出
if self.streaming:
# 發送POST請求，開啟流式響應模式
response = requests.post(self.api_url, headers=headers, json=data, stream=True)
# 檢查請求是否成功（HTTP狀態碼200表示成功）
if response.status_code == 200:
# 用於存儲完整的響應內容
full_response = ""
# 迭代處理流式響應的每一行數據
for line in response.iter_lines():
# 檢查線程是否仍在運行且接收到有效數據
if line and self.running:
# 將字節數據轉換為字符串
line_text = line.decode('utf-8')
# 檢查是否是有效的數據行
if line_text.startswith("data: "):
# 提取JSON數據部分（去除"data: "前綴）
json_str = line_text[6:]
# 檢查是否是流式響應結束標誌
if json_str.strip() == "[DONE]":
break
try:
# 解析JSON數據
chunk_data = json.loads(json_str)
# 檢查響應數據結構是否正確
if "choices" in chunk_data and len(chunk_data["choices"]) > 0:
# 提取AI回覆的內容片段
chunk = chunk_data["choices"][0].get("delta", {}).get("content", "")
# 如果有內容，則添加到完整響應併發射信號
if chunk:
full_response += chunk
self.stream_chunk_received.emit(chunk)
# 處理JSON解析錯誤
except json.JSONDecodeError:
continue
# 發射完整響應信號
self.response_received.emit(full_response)
# 處理API返回的錯誤狀態碼
else:
self.api_error.emit(f"API錯誤: {response.status_code} - {response.text}")
# 如果未啟用流式輸出
else:
# 發送普通POST請求
response = requests.post(self.api_url, headers=headers, json=data)
# 檢查請求是否成功
if response.status_code == 200:
# 解析JSON響應
response_data = response.json()
# 檢查響應數據結構是否正確
if "choices" in response_data and len(response_data["choices"]) > 0:
# 提取AI回覆內容
message = response_data["choices"][0]["message"]["content"]
# 發射完整響應信號
self.response_received.emit(message)
# 處理API返回的錯誤狀態碼
else:
self.api_error.emit(f"API錯誤: {response.status_code} - {response.text}")
# 處理其他可能的異常（如網絡錯誤等）
except Exception as e:
self.api_error.emit(f"請求失敗: {str(e)}")
# 線程運行結束，更新狀態標誌
self.running = False
def stop(self):
"""停止線程運行"""
self.running = False
代碼解釋
一、信號定義：通過pyqtSignal定義 3 種信號，用於向主線程傳遞信息：
1、stream_chunk_received：傳遞 AI 流式回覆的內容片段（實時顯示用）
2、response_received：傳遞完整的 AI 回覆（對話記錄存檔用）
3、api_error：傳遞錯誤信息（如網絡問題、API 調用失敗等）
二、初始化與基礎配置：
1、接收 API 密鑰，指定 DeepSeek API 的請求地址
2、存儲用戶提問內容（prompt），設置是否啟用流式響應（streaming）
3、維護線程運行狀態標誌（running）
三、核心運行邏輯（run方法）：
1、構建 API 請求的必要參數：包括請求頭（含認證信息Bearer Token）、請求數據（指定 AI 模型、用戶提問、響應模式）
分兩種模式處理 API 交互：
（1） 流式響應（默認）：發送帶stream=True的請求，逐行接收 AI 返回的片段，解析後通過stream_chunk_received實時發射，直到收到[DONE]結束標誌
（2） 普通響應：發送stream=False的請求，接收完整回覆後一次性發射
2、異常處理：捕獲網絡錯誤、JSON 解析錯誤、API 非 200 狀態碼等問題，通過api_error發射錯誤信息
3、輔助方法：
（1） set_prompt：用於外部設置用戶提問內容
（2） stop：通過修改running標誌停止線程運行，避免資源洩漏
提示：這裡對文章進行總結：
例如：以上就是今天要講的內容，本文僅僅簡單介紹了pandas的使用，而pandas提供了大量能使我們快速便捷地處理數據的函數和方法。
2、Pyqt界面端代碼
class AIChatApp(QMainWindow):
"""AI對話應用主窗口，繼承自QMainWindow"""
def __init__(self):
# 調用父類QMainWindow的構造函數
super().__init__()
# 存儲API密鑰
self.api_key = ""
# AI客戶端實例
self.deepseek_client = None
# 初始化用戶界面
self.init_ui()
def init_ui(self):
# 設置窗口標題
self.setWindowTitle("AI對話助手")
# 設置窗口最小尺寸
self.setMinimumSize(800, 600)
# 設置全局字體為微軟雅黑，大小為10
font = QFont("微軟雅黑", 10)
self.setFont(font)
# 創建中央部件並設置為主窗口的中央部件
central_widget = QWidget()
self.setCentralWidget(central_widget)
# 創建主佈局（垂直佈局）
main_layout = QVBoxLayout(central_widget)
# 設置佈局內控件間距
main_layout.setSpacing(15)
# 設置佈局邊緣的邊距
main_layout.setContentsMargins(20, 20, 20, 20)
# 創建標題標籤
title_label = QLabel("AI對話助手")
# 設置標籤內容居中對齊
title_label.setAlignment(Qt.AlignCenter)
# 設置標題樣式
title_label.setStyleSheet("font-size: 20px; font-weight: bold; margin-bottom: 10px;")
# 將標題添加到主佈局
main_layout.addWidget(title_label)
# 創建API設置分組框
api_group = QGroupBox("API設置")
# 創建API設置區域的水平佈局
api_layout = QHBoxLayout()
# 創建API密鑰輸入框
self.api_key_input = QLineEdit()
# 設置輸入框提示文本
self.api_key_input.setPlaceholderText("請輸入DeepSeek API密鑰")
# 設置輸入框為密碼模式（輸入內容顯示為圓點）
self.api_key_input.setEchoMode(QLineEdit.Password)
# 設置輸入框最小高度
self.api_key_input.setMinimumHeight(35)
# 創建保存密鑰按鈕
self.save_api_key_button = QPushButton("保存密鑰")
# 設置按鈕最小高度
self.save_api_key_button.setMinimumHeight(35)
# 綁定按鈕點擊事件到保存密鑰函數
self.save_api_key_button.clicked.connect(self.save_api_key)
# 將控件添加到API設置佈局
api_layout.addWidget(QLabel("API密鑰:"), 0) # 標籤，伸縮因子0
api_layout.addWidget(self.api_key_input, 1) # 輸入框，伸縮因子1（佔更多空間）
api_layout.addWidget(self.save_api_key_button, 0) # 按鈕，伸縮因子0
# 設置API設置分組框的佈局
api_group.setLayout(api_layout)
# 將API設置分組框添加到主佈局
main_layout.addWidget(api_group)
# 創建對話歷史分組框
chat_history_group = QGroupBox("對話歷史")
# 創建對話歷史區域的垂直佈局
chat_history_layout = QVBoxLayout()
# 創建對話歷史文本編輯框
self.chat_history = QTextEdit()
# 設置文本框為只讀模式
self.chat_history.setReadOnly(True)
# 設置文本框樣式
self.chat_history.setStyleSheet("""
background-color: #f9f9f9;
border: 1px solid #ddd;
border-radius: 5px;
padding: 10px;
""")
# 將對話歷史文本框添加到佈局
chat_history_layout.addWidget(self.chat_history)
# 設置對話歷史分組框的佈局
chat_history_group.setLayout(chat_history_layout)
# 將對話歷史分組框添加到主佈局，伸縮因子1（佔主要空間）
main_layout.addWidget(chat_history_group, 1)
# 創建輸入區域分組框
input_group = QGroupBox("發送消息")
# 創建輸入區域的垂直佈局
input_layout = QVBoxLayout()
# 創建用戶輸入文本框
self.user_input = QTextEdit()
# 設置輸入框提示文本
self.user_input.setPlaceholderText("請輸入您的問題...")
# 設置輸入框最小高度
self.user_input.setMinimumHeight(80)
# 設置輸入框樣式
self.user_input.setStyleSheet("""
border: 1px solid #ddd;
border-radius: 5px;
padding: 10px;
""")
# 創建發送按鈕
self.send_button = QPushButton("發送")
# 設置按鈕最小高度
self.send_button.setMinimumHeight(40)
# 設置按鈕樣式
self.send_button.setStyleSheet("""
background-color: #27ae60;
color: white;
font-weight: bold;
border-radius: 5px;
""")
# 綁定按鈕點擊事件到發送消息函數
self.send_button.clicked.connect(self.send_message)
# 未設置密鑰前禁用發送按鈕
self.send_button.setEnabled(False)
# 將控件添加到輸入區域佈局
input_layout.addWidget(self.user_input)
input_layout.addWidget(self.send_button)
# 設置輸入區域分組框的佈局
input_group.setLayout(input_layout)
# 將輸入區域分組框添加到主佈局
main_layout.addWidget(input_group)
def save_api_key(self):
"""保存API密鑰並初始化AI客戶端"""
# 獲取輸入的API密鑰並去除首尾空格
self.api_key = self.api_key_input.text().strip()
# 檢查API密鑰是否有效（非空）
if self.api_key:
# 初始化AI客戶端實例
self.deepseek_client = DeepSeekClient(self.api_key)
# 連接AI客戶端的信號到對應的槽函數
self.deepseek_client.response_received.connect(self.handle_complete_response)
self.deepseek_client.stream_chunk_received.connect(self.handle_stream_chunk)
self.deepseek_client.api_error.connect(self.handle_api_error)
# 顯示成功提示
QMessageBox.information(self, "成功", "API密鑰已保存，現在可以開始對話")
# 啟用發送按鈕
self.send_button.setEnabled(True)
# 如果API密鑰為空
else:
# 顯示錯誤提示
QMessageBox.warning(self, "錯誤", "請輸入有效的API密鑰")
# 保持發送按鈕禁用狀態
self.send_button.setEnabled(False)
def send_message(self):
"""發送用戶消息到AI"""
# 檢查AI客戶端是否已初始化
if not self.deepseek_client:
# 顯示錯誤提示
QMessageBox.warning(self, "錯誤", "請先設置並保存API密鑰")
return
# 獲取用戶輸入的消息並去除首尾空格
user_message = self.user_input.toPlainText().strip()
# 檢查消息是否為空
if not user_message:
return # 空消息不發送
# 在對話歷史中顯示用戶消息（使用HTML格式加粗顯示"您:"）
self.chat_history.append(f"<b>您:</b> {user_message}")
self.chat_history.append("") # 添加空行分隔
self.chat_history.append("<b>AI:</b> ") # 準備顯示AI回覆
# 清空輸入框
self.user_input.clear()
# 禁用發送按鈕防止重複發送
self.send_button.setEnabled(False)
# 更新按鈕文本顯示正在處理
self.send_button.setText("正在回覆...")
# 設置AI客戶端的提問內容
self.deepseek_client.set_prompt(user_message)
# 啟動AI客戶端線程（發送請求）
self.deepseek_client.start()
def handle_stream_chunk(self, chunk):
"""處理流式返回的內容片段，實時顯示到對話歷史"""
# 在對話歷史中插入接收到的內容片段
self.chat_history.textCursor().insertText(chunk)
# 自動滾動到最底部，確保最新內容可見
self.chat_history.verticalScrollBar().setValue(
self.chat_history.verticalScrollBar().maximum()
)
def handle_complete_response(self, response):
"""處理完整響應（流式結束時）"""
# 添加空行分隔不同的對話
self.chat_history.append("")
# 重新啟用發送按鈕
self.send_button.setEnabled(True)
# 恢復按鈕文本
self.send_button.setText("發送")
def handle_api_error(self, error_msg):
"""處理API錯誤，顯示錯誤信息"""
# 在對話歷史中顯示錯誤信息（使用紅色字體）
self.chat_history.append(f"<span style='color: #e74c3c;'><b>錯誤:</b> {error_msg}</span>")
self.chat_history.append("") # 添加空行分隔
# 重新啟用發送按鈕
self.send_button.setEnabled(True)
# 恢復按鈕文本
self.send_button.setText("發送")
def closeEvent(self, event):
"""窗口關閉時的處理函數，確保線程正確停止"""
# 檢查AI客戶端線程是否正在運行
if self.deepseek_client and self.deepseek_client.isRunning():
# 停止線程
self.deepseek_client.stop()
# 等待線程結束
self.deepseek_client.wait()
# 接受窗口關閉事件
event.accept()
一、界面設計（init_ui方法）：通過 PyQt5 組件搭建完整交互界面，分為 4 個核心區域：
1、標題區：顯示 “AI 對話助手” 標題，居中突出顯示；
2、API 設置區：包含 “API 密鑰輸入框”（密碼模式隱藏輸入）和 “保存密鑰按鈕”，用於配置調用 AI 所需的認證信息；
3、對話歷史區：只讀文本框（QTextEdit），用於展示用戶提問和 AI 回覆的完整記錄，設置淺色背景和邊框樣式提升可讀性；
4、輸入區：包含 “消息輸入框”（用於輸入提問）和 “發送按鈕”（默認禁用，需先保存 API 密鑰），按鈕樣式使用綠色突出顯示。
佈局上通過QVBoxLayout（垂直）和QHBoxLayout（水平）嵌套，控制控件排列和佔比（如對話歷史區佔最大空間），同時設置字體、邊距、樣式等美化界面。
二、 核心交互邏輯
1、API 密鑰管理（save_api_key方法）：
接收用戶輸入的 API 密鑰，驗證非空後初始化DeepSeekClient實例，並將其信號（流式片段、完整響應、錯誤）與自身槽函數綁定；通過彈窗提示保存結果，同時啟用 “發送按鈕”（未輸入密鑰時保持禁用）。
2、消息發送（send_message方法）：
驗證用戶輸入（非空、已初始化 AI 客戶端）後，將提問內容添加到 “對話歷史區”（標記 “您:” 並加粗）；清空輸入框、禁用發送按鈕（顯示 “正在回覆”），隨後調用DeepSeekClient的set_prompt設置提問，並啟動線程發送請求。
3、響應處理（3 個槽函數）：
（1）handle_stream_chunk：接收 AI 流式回覆的片段，實時插入對話歷史並自動滾動到底部（模擬 “逐字輸出” 效果）；
（2）handle_complete_response：流式回覆結束後，添加空行分隔對話，恢復發送按鈕狀態（啟用並顯示 “發送”）；
（3）handle_api_error：接收錯誤信息（如網絡問題、API 調用失敗），以紅色字體顯示在對話歷史中，同時恢復發送按鈕可用狀態。
三、 資源與狀態管理
1、窗口關閉處理（closeEvent方法）：關閉窗口時檢查DeepSeekClient線程是否運行，若運行則停止並等待線程結束，避免資源洩漏；
2、狀態反饋：通過按鈕文本（“發送”→“正在回覆”）、彈窗提示（密鑰保存成功 / 失敗）、對話歷史標記（“您:”“AI:”“錯誤”），讓用戶清晰感知當前操作狀態。
3、完整代碼
# 導入必要的系統模塊
import sys
# 導入JSON處理模塊，用於解析API返回的JSON數據
import json
# 導入requests模塊，用於發送HTTP請求
import requests
# 從PyQt5庫導入所需的UI組件
from PyQt5.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout,
QHBoxLayout, QLabel, QPushButton, QTextEdit,
QGroupBox, QLineEdit, QMessageBox)
# 從PyQt5庫導入核心功能模塊
from PyQt5.QtCore import Qt, QThread, pyqtSignal
# 從PyQt5庫導入GUI相關模塊
from PyQt5.QtGui import QFont
class DeepSeekClient(QThread):
"""AI對話核心處理線程，負責與API交互，繼承自QThread以實現多線程操作"""
# 定義信號：當完整響應接收完畢時發射
response_received = pyqtSignal(str)
# 定義信號：當接收到流式響應片段時發射
stream_chunk_received = pyqtSignal(str)
# 定義信號：當API調用發生錯誤時發射
api_error = pyqtSignal(str)
def __init__(self, api_key):
# 調用父類QThread的構造函數
super().__init__()
# 存儲API密鑰
self.api_key = api_key
# 設置DeepSeek API的URL
self.api_url = "https://api.deepseek.com/v1/chat/completions"
# 存儲用戶的提問內容
self.prompt = ""
# 線程運行狀態標誌
self.running = False
# 是否啟用流式輸出（實時顯示回覆內容）
self.streaming = True
def set_prompt(self, prompt):
"""設置用戶的提問內容"""
self.prompt = prompt
def run(self):
"""線程執行函數，負責發送API請求並處理響應"""
# 設置線程運行狀態為True
self.running = True
# 構建HTTP請求頭，包含內容類型和認證信息
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}" # 使用Bearer認證方式
}
# 構建API請求數據
data = {
"model": "deepseek-chat", # 指定使用的AI模型
"messages": [{"role": "user", "content": self.prompt}], # 構建對話消息
"stream": self.streaming # 指定是否使用流式響應
}
try:
# 如果啟用了流式輸出
if self.streaming:
# 發送POST請求，開啟流式響應模式
response = requests.post(self.api_url, headers=headers, json=data, stream=True)
# 檢查請求是否成功（HTTP狀態碼200表示成功）
if response.status_code == 200:
# 用於存儲完整的響應內容
full_response = ""
# 迭代處理流式響應的每一行數據
for line in response.iter_lines():
# 檢查線程是否仍在運行且接收到有效數據
if line and self.running:
# 將字節數據轉換為字符串
line_text = line.decode('utf-8')
# 檢查是否是有效的數據行
if line_text.startswith("data: "):
# 提取JSON數據部分（去除"data: "前綴）
json_str = line_text[6:]
# 檢查是否是流式響應結束標誌
if json_str.strip() == "[DONE]":
break
try:
# 解析JSON數據
chunk_data = json.loads(json_str)
# 檢查響應數據結構是否正確
if "choices" in chunk_data and len(chunk_data["choices"]) > 0:
# 提取AI回覆的內容片段
chunk = chunk_data["choices"][0].get("delta", {}).get("content", "")
# 如果有內容，則添加到完整響應併發射信號
if chunk:
full_response += chunk
self.stream_chunk_received.emit(chunk)
# 處理JSON解析錯誤
except json.JSONDecodeError:
continue
# 發射完整響應信號
self.response_received.emit(full_response)
# 處理API返回的錯誤狀態碼
else:
self.api_error.emit(f"API錯誤: {response.status_code} - {response.text}")
# 如果未啟用流式輸出
else:
# 發送普通POST請求
response = requests.post(self.api_url, headers=headers, json=data)
# 檢查請求是否成功
if response.status_code == 200:
# 解析JSON響應
response_data = response.json()
# 檢查響應數據結構是否正確
if "choices" in response_data and len(response_data["choices"]) > 0:
# 提取AI回覆內容
message = response_data["choices"][0]["message"]["content"]
# 發射完整響應信號
self.response_received.emit(message)
# 處理API返回的錯誤狀態碼
else:
self.api_error.emit(f"API錯誤: {response.status_code} - {response.text}")
# 處理其他可能的異常（如網絡錯誤等）
except Exception as e:
self.api_error.emit(f"請求失敗: {str(e)}")
# 線程運行結束，更新狀態標誌
self.running = False
def stop(self):
"""停止線程運行"""
self.running = False
class AIChatApp(QMainWindow):
"""AI對話應用主窗口，繼承自QMainWindow"""
def __init__(self):
# 調用父類QMainWindow的構造函數
super().__init__()
# 存儲API密鑰
self.api_key = ""
# AI客戶端實例
self.deepseek_client = None
# 初始化用戶界面
self.init_ui()
def init_ui(self):
# 設置窗口標題
self.setWindowTitle("AI對話助手")
# 設置窗口最小尺寸
self.setMinimumSize(800, 600)
# 設置全局字體為微軟雅黑，大小為10
font = QFont("微軟雅黑", 10)
self.setFont(font)
# 創建中央部件並設置為主窗口的中央部件
central_widget = QWidget()
self.setCentralWidget(central_widget)
# 創建主佈局（垂直佈局）
main_layout = QVBoxLayout(central_widget)
# 設置佈局內控件間距
main_layout.setSpacing(15)
# 設置佈局邊緣的邊距
main_layout.setContentsMargins(20, 20, 20, 20)
# 創建標題標籤
title_label = QLabel("AI對話助手")
# 設置標籤內容居中對齊
title_label.setAlignment(Qt.AlignCenter)
# 設置標題樣式
title_label.setStyleSheet("font-size: 20px; font-weight: bold; margin-bottom: 10px;")
# 將標題添加到主佈局
main_layout.addWidget(title_label)
# 創建API設置分組框
api_group = QGroupBox("API設置")
# 創建API設置區域的水平佈局
api_layout = QHBoxLayout()
# 創建API密鑰輸入框
self.api_key_input = QLineEdit()
# 設置輸入框提示文本
self.api_key_input.setPlaceholderText("請輸入DeepSeek API密鑰")
# 設置輸入框為密碼模式（輸入內容顯示為圓點）
self.api_key_input.setEchoMode(QLineEdit.Password)
# 設置輸入框最小高度
self.api_key_input.setMinimumHeight(35)
# 創建保存密鑰按鈕
self.save_api_key_button = QPushButton("保存密鑰")
# 設置按鈕最小高度
self.save_api_key_button.setMinimumHeight(35)
# 綁定按鈕點擊事件到保存密鑰函數
self.save_api_key_button.clicked.connect(self.save_api_key)
# 將控件添加到API設置佈局
api_layout.addWidget(QLabel("API密鑰:"), 0) # 標籤，伸縮因子0
api_layout.addWidget(self.api_key_input, 1) # 輸入框，伸縮因子1（佔更多空間）
api_layout.addWidget(self.save_api_key_button, 0) # 按鈕，伸縮因子0
# 設置API設置分組框的佈局
api_group.setLayout(api_layout)
# 將API設置分組框添加到主佈局
main_layout.addWidget(api_group)
# 創建對話歷史分組框
chat_history_group = QGroupBox("對話歷史")
# 創建對話歷史區域的垂直佈局
chat_history_layout = QVBoxLayout()
# 創建對話歷史文本編輯框
self.chat_history = QTextEdit()
# 設置文本框為只讀模式
self.chat_history.setReadOnly(True)
# 設置文本框樣式
self.chat_history.setStyleSheet("""
background-color: #f9f9f9;
border: 1px solid #ddd;
border-radius: 5px;
padding: 10px;
""")
# 將對話歷史文本框添加到佈局
chat_history_layout.addWidget(self.chat_history)
# 設置對話歷史分組框的佈局
chat_history_group.setLayout(chat_history_layout)
# 將對話歷史分組框添加到主佈局，伸縮因子1（佔主要空間）
main_layout.addWidget(chat_history_group, 1)
# 創建輸入區域分組框
input_group = QGroupBox("發送消息")
# 創建輸入區域的垂直佈局
input_layout = QVBoxLayout()
# 創建用戶輸入文本框
self.user_input = QTextEdit()
# 設置輸入框提示文本
self.user_input.setPlaceholderText("請輸入您的問題...")
# 設置輸入框最小高度
self.user_input.setMinimumHeight(80)
# 設置輸入框樣式
self.user_input.setStyleSheet("""
border: 1px solid #ddd;
border-radius: 5px;
padding: 10px;
""")
# 創建發送按鈕
self.send_button = QPushButton("發送")
# 設置按鈕最小高度
self.send_button.setMinimumHeight(40)
# 設置按鈕樣式
self.send_button.setStyleSheet("""
background-color: #27ae60;
color: white;
font-weight: bold;
border-radius: 5px;
""")
# 綁定按鈕點擊事件到發送消息函數
self.send_button.clicked.connect(self.send_message)
# 未設置密鑰前禁用發送按鈕
self.send_button.setEnabled(False)
# 將控件添加到輸入區域佈局
input_layout.addWidget(self.user_input)
input_layout.addWidget(self.send_button)
# 設置輸入區域分組框的佈局
input_group.setLayout(input_layout)
# 將輸入區域分組框添加到主佈局
main_layout.addWidget(input_group)
def save_api_key(self):
"""保存API密鑰並初始化AI客戶端"""
# 獲取輸入的API密鑰並去除首尾空格
self.api_key = self.api_key_input.text().strip()
# 檢查API密鑰是否有效（非空）
if self.api_key:
# 初始化AI客戶端實例
self.deepseek_client = DeepSeekClient(self.api_key)
# 連接AI客戶端的信號到對應的槽函數
self.deepseek_client.response_received.connect(self.handle_complete_response)
self.deepseek_client.stream_chunk_received.connect(self.handle_stream_chunk)
self.deepseek_client.api_error.connect(self.handle_api_error)
# 顯示成功提示
QMessageBox.information(self, "成功", "API密鑰已保存，現在可以開始對話")
# 啟用發送按鈕
self.send_button.setEnabled(True)
# 如果API密鑰為空
else:
# 顯示錯誤提示
QMessageBox.warning(self, "錯誤", "請輸入有效的API密鑰")
# 保持發送按鈕禁用狀態
self.send_button.setEnabled(False)
def send_message(self):
"""發送用戶消息到AI"""
# 檢查AI客戶端是否已初始化
if not self.deepseek_client:
# 顯示錯誤提示
QMessageBox.warning(self, "錯誤", "請先設置並保存API密鑰")
return
# 獲取用戶輸入的消息並去除首尾空格
user_message = self.user_input.toPlainText().strip()
# 檢查消息是否為空
if not user_message:
return # 空消息不發送
# 在對話歷史中顯示用戶消息（使用HTML格式加粗顯示"您:"）
self.chat_history.append(f"<b>您:</b> {user_message}")
self.chat_history.append("") # 添加空行分隔
self.chat_history.append("<b>AI:</b> ") # 準備顯示AI回覆
# 清空輸入框
self.user_input.clear()
# 禁用發送按鈕防止重複發送
self.send_button.setEnabled(False)
# 更新按鈕文本顯示正在處理
self.send_button.setText("正在回覆...")
# 設置AI客戶端的提問內容
self.deepseek_client.set_prompt(user_message)
# 啟動AI客戶端線程（發送請求）
self.deepseek_client.start()
def handle_stream_chunk(self, chunk):
"""處理流式返回的內容片段，實時顯示到對話歷史"""
# 在對話歷史中插入接收到的內容片段
self.chat_history.textCursor().insertText(chunk)
# 自動滾動到最底部，確保最新內容可見
self.chat_history.verticalScrollBar().setValue(
self.chat_history.verticalScrollBar().maximum()
)
def handle_complete_response(self, response):
"""處理完整響應（流式結束時）"""
# 添加空行分隔不同的對話
self.chat_history.append("")
# 重新啟用發送按鈕
self.send_button.setEnabled(True)
# 恢復按鈕文本
self.send_button.setText("發送")
def handle_api_error(self, error_msg):
"""處理API錯誤，顯示錯誤信息"""
# 在對話歷史中顯示錯誤信息（使用紅色字體）
self.chat_history.append(f"<span style='color: #e74c3c;'><b>錯誤:</b> {error_msg}</span>")
self.chat_history.append("") # 添加空行分隔
# 重新啟用發送按鈕
self.send_button.setEnabled(True)
# 恢復按鈕文本
self.send_button.setText("發送")
def closeEvent(self, event):
"""窗口關閉時的處理函數，確保線程正確停止"""
# 檢查AI客戶端線程是否正在運行
if self.deepseek_client and self.deepseek_client.isRunning():
# 停止線程
self.deepseek_client.stop()
# 等待線程結束
self.deepseek_client.wait()
# 接受窗口關閉事件
event.accept()
# 程序入口點
if __name__ == "__main__":
# 創建Qt應用實例
app = QApplication(sys.argv)
# 創建主窗口實例
window = AIChatApp()
# 顯示主窗口
window.show()
# 進入應用主循環，等待用戶交互
sys.exit(app.exec_())w1
四、效果演示
運行代碼，彈出pyqt窗口
在API密鑰中輸入獲取的Deekseek API密鑰，並進行提問。效果如下
AI Column: Build Your Own Assistant with PyQt5 + API
Desktop AI chat with PyQt5 and DeepSeek API: threaded streaming UI, API key setup, and full runnable source walkthrough.
Captured at (local ISO): 2026-05-18 05:17:00
Preface
A lightweight desktop chat app: PyQt5 UI + DeepSeek API + streaming replies in a background thread.
I. Concepts
1. API
HTTP rules: send prompt JSON → model returns text. No need to host the model yourself.
2. PyQt5
Widget toolkit for native desktop UIs.
II. API Key
Get a key at DeepSeek platform. Save it when created — it is shown only once.
III. Implementation
1. API worker (DeepSeekClient)
# 导入必要的系统模块
import sys
# 导入JSON处理模块，用于解析API返回的JSON数据
import json
# 导入requests模块，用于发送HTTP请求
import requests
# 从PyQt5库导入所需的UI组件
from PyQt5.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout,
QHBoxLayout, QLabel, QPushButton, QTextEdit,
QGroupBox, QLineEdit, QMessageBox)
# 从PyQt5库导入核心功能模块
from PyQt5.QtCore import Qt, QThread, pyqtSignal
# 从PyQt5库导入GUI相关模块
from PyQt5.QtGui import QFont
class DeepSeekClient(QThread):
"""AI对话核心处理线程，负责与API交互，继承自QThread以实现多线程操作"""
# 定义信号：当完整响应接收完毕时发射
response_received = pyqtSignal(str)
# 定义信号：当接收到流式响应片段时发射
stream_chunk_received = pyqtSignal(str)
# 定义信号：当API调用发生错误时发射
api_error = pyqtSignal(str)
def __init__(self, api_key):
# 调用父类QThread的构造函数
super().__init__()
# 存储API密钥
self.api_key = api_key
# 设置DeepSeek API的URL
self.api_url = "https://api.deepseek.com/v1/chat/completions"
# 存储用户的提问内容
self.prompt = ""
# 线程运行状态标志
self.running = False
# 是否启用流式输出（实时显示回复内容）
self.streaming = True
def set_prompt(self, prompt):
"""设置用户的提问内容"""
self.prompt = prompt
def run(self):
"""线程执行函数，负责发送API请求并处理响应"""
# 设置线程运行状态为True
self.running = True
# 构建HTTP请求头，包含内容类型和认证信息
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}" # 使用Bearer认证方式
}
# 构建API请求数据
data = {
"model": "deepseek-chat", # 指定使用的AI模型
"messages": [{"role": "user", "content": self.prompt}], # 构建对话消息
"stream": self.streaming # 指定是否使用流式响应
}
try:
# 如果启用了流式输出
if self.streaming:
# 发送POST请求，开启流式响应模式
response = requests.post(self.api_url, headers=headers, json=data, stream=True)
# 检查请求是否成功（HTTP状态码200表示成功）
if response.status_code == 200:
# 用于存储完整的响应内容
full_response = ""
# 迭代处理流式响应的每一行数据
for line in response.iter_lines():
# 检查线程是否仍在运行且接收到有效数据
if line and self.running:
# 将字节数据转换为字符串
line_text = line.decode('utf-8')
# 检查是否是有效的数据行
if line_text.startswith("data: "):
# 提取JSON数据部分（去除"data: "前缀）
json_str = line_text[6:]
# 检查是否是流式响应结束标志
if json_str.strip() == "[DONE]":
break
try:
# 解析JSON数据
chunk_data = json.loads(json_str)
# 检查响应数据结构是否正确
if "choices" in chunk_data and len(chunk_data["choices"]) > 0:
# 提取AI回复的内容片段
chunk = chunk_data["choices"][0].get("delta", {}).get("content", "")
# 如果有内容，则添加到完整响应并发射信号
if chunk:
full_response += chunk
self.stream_chunk_received.emit(chunk)
# 处理JSON解析错误
except json.JSONDecodeError:
continue
# 发射完整响应信号
self.response_received.emit(full_response)
# 处理API返回的错误状态码
else:
self.api_error.emit(f"API错误: {response.status_code} - {response.text}")
# 如果未启用流式输出
else:
# 发送普通POST请求
response = requests.post(self.api_url, headers=headers, json=data)
# 检查请求是否成功
if response.status_code == 200:
# 解析JSON响应
response_data = response.json()
# 检查响应数据结构是否正确
if "choices" in response_data and len(response_data["choices"]) > 0:
# 提取AI回复内容
message = response_data["choices"][0]["message"]["content"]
# 发射完整响应信号
self.response_received.emit(message)
# 处理API返回的错误状态码
else:
self.api_error.emit(f"API错误: {response.status_code} - {response.text}")
# 处理其他可能的异常（如网络错误等）
except Exception as e:
self.api_error.emit(f"请求失败: {str(e)}")
# 线程运行结束，更新状态标志
self.running = False
def stop(self):
"""停止线程运行"""
self.running = False
2. UI (AIChatApp)
init_ui: title, API key group (password field), read-only chat log, input + send (disabled until key saved). Layouts: QVBoxLayout / QHBoxLayout.
Logic:
save_api_key— wire signalsstream_chunk_received,response_received,api_errorsend_message— append user text, start threadhandle_stream_chunk— incremental displayhandle_complete_response/handle_api_error— re-enable sendcloseEvent— stopQThreadcleanly
3. Full code
# 导入必要的系统模块
import sys
# 导入JSON处理模块，用于解析API返回的JSON数据
import json
# 导入requests模块，用于发送HTTP请求
import requests
# 从PyQt5库导入所需的UI组件
from PyQt5.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout,
QHBoxLayout, QLabel, QPushButton, QTextEdit,
QGroupBox, QLineEdit, QMessageBox)
# 从PyQt5库导入核心功能模块
from PyQt5.QtCore import Qt, QThread, pyqtSignal
# 从PyQt5库导入GUI相关模块
from PyQt5.QtGui import QFont
class DeepSeekClient(QThread):
"""AI对话核心处理线程，负责与API交互，继承自QThread以实现多线程操作"""
# 定义信号：当完整响应接收完毕时发射
response_received = pyqtSignal(str)
# 定义信号：当接收到流式响应片段时发射
stream_chunk_received = pyqtSignal(str)
# 定义信号：当API调用发生错误时发射
api_error = pyqtSignal(str)
def __init__(self, api_key):
# 调用父类QThread的构造函数
super().__init__()
# 存储API密钥
self.api_key = api_key
# 设置DeepSeek API的URL
self.api_url = "https://api.deepseek.com/v1/chat/completions"
# 存储用户的提问内容
self.prompt = ""
# 线程运行状态标志
self.running = False
# 是否启用流式输出（实时显示回复内容）
self.streaming = True
def set_prompt(self, prompt):
"""设置用户的提问内容"""
self.prompt = prompt
def run(self):
"""线程执行函数，负责发送API请求并处理响应"""
# 设置线程运行状态为True
self.running = True
# 构建HTTP请求头，包含内容类型和认证信息
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}" # 使用Bearer认证方式
}
# 构建API请求数据
data = {
"model": "deepseek-chat", # 指定使用的AI模型
"messages": [{"role": "user", "content": self.prompt}], # 构建对话消息
"stream": self.streaming # 指定是否使用流式响应
}
try:
# 如果启用了流式输出
if self.streaming:
# 发送POST请求，开启流式响应模式
response = requests.post(self.api_url, headers=headers, json=data, stream=True)
# 检查请求是否成功（HTTP状态码200表示成功）
if response.status_code == 200:
# 用于存储完整的响应内容
full_response = ""
# 迭代处理流式响应的每一行数据
for line in response.iter_lines():
# 检查线程是否仍在运行且接收到有效数据
if line and self.running:
# 将字节数据转换为字符串
line_text = line.decode('utf-8')
# 检查是否是有效的数据行
if line_text.startswith("data: "):
# 提取JSON数据部分（去除"data: "前缀）
json_str = line_text[6:]
# 检查是否是流式响应结束标志
if json_str.strip() == "[DONE]":
break
try:
# 解析JSON数据
chunk_data = json.loads(json_str)
# 检查响应数据结构是否正确
if "choices" in chunk_data and len(chunk_data["choices"]) > 0:
# 提取AI回复的内容片段
chunk = chunk_data["choices"][0].get("delta", {}).get("content", "")
# 如果有内容，则添加到完整响应并发射信号
if chunk:
full_response += chunk
self.stream_chunk_received.emit(chunk)
# 处理JSON解析错误
except json.JSONDecodeError:
continue
# 发射完整响应信号
self.response_received.emit(full_response)
# 处理API返回的错误状态码
else:
self.api_error.emit(f"API错误: {response.status_code} - {response.text}")
# 如果未启用流式输出
else:
# 发送普通POST请求
response = requests.post(self.api_url, headers=headers, json=data)
# 检查请求是否成功
if response.status_code == 200:
# 解析JSON响应
response_data = response.json()
# 检查响应数据结构是否正确
if "choices" in response_data and len(response_data["choices"]) > 0:
# 提取AI回复内容
message = response_data["choices"][0]["message"]["content"]
# 发射完整响应信号
self.response_received.emit(message)
# 处理API返回的错误状态码
else:
self.api_error.emit(f"API错误: {response.status_code} - {response.text}")
# 处理其他可能的异常（如网络错误等）
except Exception as e:
self.api_error.emit(f"请求失败: {str(e)}")
# 线程运行结束，更新状态标志
self.running = False
def stop(self):
"""停止线程运行"""
self.running = False
class AIChatApp(QMainWindow):
"""AI对话应用主窗口，继承自QMainWindow"""
def __init__(self):
# 调用父类QMainWindow的构造函数
super().__init__()
# 存储API密钥
self.api_key = ""
# AI客户端实例
self.deepseek_client = None
# 初始化用户界面
self.init_ui()
def init_ui(self):
# 设置窗口标题
self.setWindowTitle("AI对话助手")
# 设置窗口最小尺寸
self.setMinimumSize(800, 600)
# 设置全局字体为微软雅黑，大小为10
font = QFont("微软雅黑", 10)
self.setFont(font)
# 创建中央部件并设置为主窗口的中央部件
central_widget = QWidget()
self.setCentralWidget(central_widget)
# 创建主布局（垂直布局）
main_layout = QVBoxLayout(central_widget)
# 设置布局内控件间距
main_layout.setSpacing(15)
# 设置布局边缘的边距
main_layout.setContentsMargins(20, 20, 20, 20)
# 创建标题标签
title_label = QLabel("AI对话助手")
# 设置标签内容居中对齐
title_label.setAlignment(Qt.AlignCenter)
# 设置标题样式
title_label.setStyleSheet("font-size: 20px; font-weight: bold; margin-bottom: 10px;")
# 将标题添加到主布局
main_layout.addWidget(title_label)
# 创建API设置分组框
api_group = QGroupBox("API设置")
# 创建API设置区域的水平布局
api_layout = QHBoxLayout()
# 创建API密钥输入框
self.api_key_input = QLineEdit()
# 设置输入框提示文本
self.api_key_input.setPlaceholderText("请输入DeepSeek API密钥")
# 设置输入框为密码模式（输入内容显示为圆点）
self.api_key_input.setEchoMode(QLineEdit.Password)
# 设置输入框最小高度
self.api_key_input.setMinimumHeight(35)
# 创建保存密钥按钮
self.save_api_key_button = QPushButton("保存密钥")
# 设置按钮最小高度
self.save_api_key_button.setMinimumHeight(35)
# 绑定按钮点击事件到保存密钥函数
self.save_api_key_button.clicked.connect(self.save_api_key)
# 将控件添加到API设置布局
api_layout.addWidget(QLabel("API密钥:"), 0) # 标签，伸缩因子0
api_layout.addWidget(self.api_key_input, 1) # 输入框，伸缩因子1（占更多空间）
api_layout.addWidget(self.save_api_key_button, 0) # 按钮，伸缩因子0
# 设置API设置分组框的布局
api_group.setLayout(api_layout)
# 将API设置分组框添加到主布局
main_layout.addWidget(api_group)
# 创建对话历史分组框
chat_history_group = QGroupBox("对话历史")
# 创建对话历史区域的垂直布局
chat_history_layout = QVBoxLayout()
# 创建对话历史文本编辑框
self.chat_history = QTextEdit()
# 设置文本框为只读模式
self.chat_history.setReadOnly(True)
# 设置文本框样式
self.chat_history.setStyleSheet("""
background-color: #f9f9f9;
border: 1px solid #ddd;
border-radius: 5px;
padding: 10px;
""")
# 将对话历史文本框添加到布局
chat_history_layout.addWidget(self.chat_history)
# 设置对话历史分组框的布局
chat_history_group.setLayout(chat_history_layout)
# 将对话历史分组框添加到主布局，伸缩因子1（占主要空间）
main_layout.addWidget(chat_history_group, 1)
# 创建输入区域分组框
input_group = QGroupBox("发送消息")
# 创建输入区域的垂直布局
input_layout = QVBoxLayout()
# 创建用户输入文本框
self.user_input = QTextEdit()
# 设置输入框提示文本
self.user_input.setPlaceholderText("请输入您的问题...")
# 设置输入框最小高度
self.user_input.setMinimumHeight(80)
# 设置输入框样式
self.user_input.setStyleSheet("""
border: 1px solid #ddd;
border-radius: 5px;
padding: 10px;
""")
# 创建发送按钮
self.send_button = QPushButton("发送")
# 设置按钮最小高度
self.send_button.setMinimumHeight(40)
# 设置按钮样式
self.send_button.setStyleSheet("""
background-color: #27ae60;
color: white;
font-weight: bold;
border-radius: 5px;
""")
# 绑定按钮点击事件到发送消息函数
self.send_button.clicked.connect(self.send_message)
# 未设置密钥前禁用发送按钮
self.send_button.setEnabled(False)
# 将控件添加到输入区域布局
input_layout.addWidget(self.user_input)
input_layout.addWidget(self.send_button)
# 设置输入区域分组框的布局
input_group.setLayout(input_layout)
# 将输入区域分组框添加到主布局
main_layout.addWidget(input_group)
def save_api_key(self):
"""保存API密钥并初始化AI客户端"""
# 获取输入的API密钥并去除首尾空格
self.api_key = self.api_key_input.text().strip()
# 检查API密钥是否有效（非空）
if self.api_key:
# 初始化AI客户端实例
self.deepseek_client = DeepSeekClient(self.api_key)
# 连接AI客户端的信号到对应的槽函数
self.deepseek_client.response_received.connect(self.handle_complete_response)
self.deepseek_client.stream_chunk_received.connect(self.handle_stream_chunk)
self.deepseek_client.api_error.connect(self.handle_api_error)
# 显示成功提示
QMessageBox.information(self, "成功", "API密钥已保存，现在可以开始对话")
# 启用发送按钮
self.send_button.setEnabled(True)
# 如果API密钥为空
else:
# 显示错误提示
QMessageBox.warning(self, "错误", "请输入有效的API密钥")
# 保持发送按钮禁用状态
self.send_button.setEnabled(False)
def send_message(self):
"""发送用户消息到AI"""
# 检查AI客户端是否已初始化
if not self.deepseek_client:
# 显示错误提示
QMessageBox.warning(self, "错误", "请先设置并保存API密钥")
return
# 获取用户输入的消息并去除首尾空格
user_message = self.user_input.toPlainText().strip()
# 检查消息是否为空
if not user_message:
return # 空消息不发送
# 在对话历史中显示用户消息（使用HTML格式加粗显示"您:"）
self.chat_history.append(f"<b>您:</b> {user_message}")
self.chat_history.append("") # 添加空行分隔
self.chat_history.append("<b>AI:</b> ") # 准备显示AI回复
# 清空输入框
self.user_input.clear()
# 禁用发送按钮防止重复发送
self.send_button.setEnabled(False)
# 更新按钮文本显示正在处理
self.send_button.setText("正在回复...")
# 设置AI客户端的提问内容
self.deepseek_client.set_prompt(user_message)
# 启动AI客户端线程（发送请求）
self.deepseek_client.start()
def handle_stream_chunk(self, chunk):
"""处理流式返回的内容片段，实时显示到对话历史"""
# 在对话历史中插入接收到的内容片段
self.chat_history.textCursor().insertText(chunk)
# 自动滚动到最底部，确保最新内容可见
self.chat_history.verticalScrollBar().setValue(
self.chat_history.verticalScrollBar().maximum()
)
def handle_complete_response(self, response):
"""处理完整响应（流式结束时）"""
# 添加空行分隔不同的对话
self.chat_history.append("")
# 重新启用发送按钮
self.send_button.setEnabled(True)
# 恢复按钮文本
self.send_button.setText("发送")
def handle_api_error(self, error_msg):
"""处理API错误，显示错误信息"""
# 在对话历史中显示错误信息（使用红色字体）
self.chat_history.append(f"<span style='color: #e74c3c;'><b>错误:</b> {error_msg}</span>")
self.chat_history.append("") # 添加空行分隔
# 重新启用发送按钮
self.send_button.setEnabled(True)
# 恢复按钮文本
self.send_button.setText("发送")
def closeEvent(self, event):
"""窗口关闭时的处理函数，确保线程正确停止"""
# 检查AI客户端线程是否正在运行
if self.deepseek_client and self.deepseek_client.isRunning():
# 停止线程
self.deepseek_client.stop()
# 等待线程结束
self.deepseek_client.wait()
# 接受窗口关闭事件
event.accept()
# 程序入口点
if __name__ == "__main__":
# 创建Qt应用实例
app = QApplication(sys.argv)
# 创建主窗口实例
window = AIChatApp()
# 显示主窗口
window.show()
# 进入应用主循环，等待用户交互
sys.exit(app.exec_())w1
IV. Demo
Run the script → enter DeepSeek API key → chat with streaming output.