基于PyQt5的基因序列分析图形用户界面应用程序。以下为代码的各个部分：
1. 整体架构
这个应用基于"Genos"模型，进行基因序列分析：
使用PyQt5构建GUI界面
支持单序列和批量分析
采用多线程处理避免界面卡顿
可配置多种分析参数
2. 核心组件
2.1 GenosAnalysisThread类
class GenosAnalysisThread(QThread):作用：后台工作线程，防止长时间的分析任务阻塞UI
使用PyQt5的QThread实现
通过信号(signals)与主线程通信
支持进度报告、完成通知和错误处理
关键信号：
finished：分析完成时发射结果error：分析出错时发射错误信息progress：报告分析进度
2.2 GenosApp类
class GenosApp(QMainWindow):作用：主应用程序窗口，管理所有UI组件和业务逻辑
3. 界面布局
3.1 主窗口结构
GenosApp (QMainWindow) ├── 中心部件 (Central Widget) │ ├── 主布局 (QVBoxLayout) │ │ ├── 标题标签 │ │ ├── 选项卡控件 (QTabWidget) │ │ │ ├── 序列分析标签页 │ │ │ ├── 批量分析标签页 │ │ │ └── 信息标签页 │ │ └── 状态栏3.2 三个主要标签页
1) 序列分析标签页
序列分析标签页 ├── 输入基因序列区域 │ ├── 多行文本输入框 (QTextEdit) │ └── 工具按钮 (清空、粘贴、从文件加载) ├── 参数设置区域 │ ├── 任务类型下拉框 (QComboBox) │ ├── Temperature调节框 (QSpinBox) │ ├── Top P调节框 │ └── 最大Token调节框 ├── 开始分析按钮 ├── 进度条 └── 分析结果显示区域2) 批量分析标签页
支持多行输入，每行一个序列
批量处理所有序列
显示汇总结果
3) 信息标签页
显示模型信息
使用说明文档
4. 主要功能流程
4.1 启动流程
1. main()函数启动应用 2. 创建GenosApp实例 3. init_ui()初始化界面 4. load_model()加载AI模型 5. 显示窗口，进入事件循环4.2 分析流程
用户点击"开始分析"按钮： 1. start_analysis()被调用 2. 验证模型是否加载 3. 验证输入序列是否为空 4. 收集参数设置 5. 创建并启动工作线程 6. 线程在后台调用模型进行分析 7. 通过信号返回结果 8. 在UI中显示结果4.3 线程通信机制
主线程(UI) <---> 工作线程(分析) ↓ ↓ 接收用户输入 执行分析任务 更新UI状态 ← progress信号 显示结果 ← finished信号 处理错误 ← error信号5. 关键功能
5.1 模型加载 (load_model)
尝试从当前目录加载模型
优先使用Transformers版本（支持CUDA加速）
提供详细的错误处理和状态反馈
5.2 文件加载 (load_from_file)
支持多种文件格式：
.txt,.fasta,.fa自动识别FASTA格式（以
>开头的序列文件）提取纯序列内容
5.3 分析参数配置
Temperature：控制输出的随机性（0.01-2.0）
Top P：核采样概率（0.5-1.0）
最大Token：限制生成长度（1-2000）
6. 设计特点
6.1 用户体验
响应式设计：通过线程防止UI卡顿
进度反馈：显示分析进度
错误处理：友好的错误提示
批量处理：高效处理多个序列
6.2 代码结构
模块化设计：不同功能分离到不同方法
信号槽机制：松耦合的线程通信
异常处理：全面的错误捕获和处理
6.3 扩展性
插件式设计：易于添加新的分析功能
参数化配置：支持多种分析参数
模型抽象：可替换不同的AI模型后端
7. 使用方法
7.1 单个序列分析
在"序列分析"标签页输入基因序列
选择任务类型（analyze/predict/complete）
调整参数设置
点击"开始分析"
7.2 批量分析
在"批量分析"标签页每行输入一个序列
点击"批量分析"按钮
查看汇总结果
8. 代码
genos_app.py
""" Genos 基因序列分析应用 - PyQt5 GUI 提供友好的用户界面用于基因组序列分析 """ import sys import os from PyQt5.QtWidgets import ( QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QTextEdit, QLabel, QPushButton, QComboBox, QSpinBox, QGroupBox, QFileDialog, QMessageBox, QProgressBar, QTabWidget, QSplitter ) from PyQt5.QtCore import Qt, QThread, pyqtSignal from PyQt5.QtGui import QFont, QTextCharFormat, QColor, QSyntaxHighlighter import json class GenosAnalysisThread(QThread): """后台分析线程""" finished = pyqtSignal(dict) error = pyqtSignal(str) progress = pyqtSignal(int) def __init__(self, model, sequence, task_type, params): super().__init__() self.model = model self.sequence = sequence self.task_type = task_type self.params = params def run(self): try: self.progress.emit(10) if self.task_type == "batch": # 批量分析 sequences = [s.strip() for s in self.sequence.split('\n') if s.strip()] result = { "type": "batch", "results": self.model.batch_analyze(sequences, self.task_type, **self.params) } else: # 单个序列分析 result = self.model.analyze_sequence(self.sequence, self.task_type, **self.params) result["type"] = "single" self.progress.emit(100) self.finished.emit(result) except Exception as e: self.error.emit(f"分析失败: {str(e)}") class GenosApp(QMainWindow): """Genos 基因序列分析主窗口""" def __init__(self): super().__init__() self.model = None self.worker = None self.init_ui() self.load_model() def init_ui(self): """初始化用户界面""" self.setWindowTitle("Genos - 基因序列分析系统") self.setGeometry(100, 100, 1200, 800) # 创建中心部件 central_widget = QWidget() self.setCentralWidget(central_widget) # 主布局 main_layout = QVBoxLayout() central_widget.setLayout(main_layout) # 标题 title_label = QLabel("Genos-1.2B 基因序列分析系统") title_label.setFont(QFont("Arial", 18, QFont.Bold)) title_label.setAlignment(Qt.AlignCenter) main_layout.addWidget(title_label) # 创建选项卡 self.tab_widget = QTabWidget() main_layout.addWidget(self.tab_widget) # 添加选项卡 self.create_analysis_tab() self.create_batch_tab() self.create_info_tab() # 状态栏 self.statusBar().showMessage("准备就绪") def create_analysis_tab(self): """创建序列分析选项卡""" tab = QWidget() layout = QVBoxLayout() tab.setLayout(layout) # 输入区域 input_group = QGroupBox("输入基因序列") input_layout = QVBoxLayout() # 序列输入框 self.sequence_input = QTextEdit() self.sequence_input.setPlaceholderText("请输入基因序列 (A, C, G, T, N)\n例如: ATGCGATCGTAGCTAGCTAG") self.sequence_input.setFont(QFont("Consolas", 12)) self.sequence_input.setMaximumHeight(150) input_layout.addWidget(self.sequence_input) # 工具按钮 button_layout = QHBoxLayout() clear_btn = QPushButton("清空") clear_btn.clicked.connect(self.clear_input) button_layout.addWidget(clear_btn) paste_btn = QPushButton("粘贴") paste_btn.clicked.connect(self.paste_input) button_layout.addWidget(paste_btn) load_btn = QPushButton("从文件加载") load_btn.clicked.connect(self.load_from_file) button_layout.addWidget(load_btn) input_layout.addLayout(button_layout) input_group.setLayout(input_layout) layout.addWidget(input_group) # 参数设置区域 param_group = QGroupBox("参数设置") param_layout = QHBoxLayout() param_layout.addWidget(QLabel("任务类型:")) self.task_combo = QComboBox() self.task_combo.addItems(["analyze", "predict", "complete"]) param_layout.addWidget(self.task_combo) param_layout.addWidget(QLabel("Temperature:")) self.temp_spin = QSpinBox() self.temp_spin.setRange(1, 200) self.temp_spin.setValue(70) self.temp_spin.setSuffix(" (x0.01)") param_layout.addWidget(self.temp_spin) param_layout.addWidget(QLabel("Top P:")) self.topp_spin = QSpinBox() self.topp_spin.setRange(50, 100) self.topp_spin.setValue(95) self.topp_spin.setSuffix(" (x0.01)") param_layout.addWidget(self.topp_spin) param_layout.addWidget(QLabel("最大 Token:")) self.maxtoken_spin = QSpinBox() self.maxtoken_spin.setRange(1, 2000) self.maxtoken_spin.setValue(512) param_layout.addWidget(self.maxtoken_spin) param_layout.addStretch() param_group.setLayout(param_layout) layout.addWidget(param_group) # 分析按钮 self.analyze_btn = QPushButton("开始分析") self.analyze_btn.setFont(QFont("Arial", 12, QFont.Bold)) self.analyze_btn.setStyleSheet("QPushButton { background-color: #4CAF50; color: white; padding: 10px; } QPushButton:hover { background-color: #45a049; }") self.analyze_btn.clicked.connect(self.start_analysis) layout.addWidget(self.analyze_btn) # 进度条 self.progress_bar = QProgressBar() self.progress_bar.setVisible(False) layout.addWidget(self.progress_bar) # 结果显示区域 result_group = QGroupBox("分析结果") result_layout = QVBoxLayout() self.result_output = QTextEdit() self.result_output.setReadOnly(True) self.result_output.setFont(QFont("Consolas", 10)) result_layout.addWidget(self.result_output) result_group.setLayout(result_layout) layout.addWidget(result_group) self.tab_widget.addTab(tab, "序列分析") def create_batch_tab(self): """创建批量分析选项卡""" tab = QWidget() layout = QVBoxLayout() tab.setLayout(layout) # 输入说明 info_label = QLabel("每行输入一个基因序列，系统将批量分析所有序列") info_label.setStyleSheet("color: #666; padding: 5px;") layout.addWidget(info_label) # 批量输入框 self.batch_input = QTextEdit() self.batch_input.setPlaceholderText("ATGCGATCG\nAGCTAGCTAG\n...") self.batch_input.setFont(QFont("Consolas", 11)) layout.addWidget(self.batch_input) # 批量分析按钮 batch_btn = QPushButton("批量分析") batch_btn.setFont(QFont("Arial", 12)) batch_btn.clicked.connect(self.start_batch_analysis) layout.addWidget(batch_btn) # 批量结果 self.batch_result = QTextEdit() self.batch_result.setReadOnly(True) self.batch_result.setFont(QFont("Consolas", 9)) layout.addWidget(self.batch_result) self.tab_widget.addTab(tab, "批量分析") def create_info_tab(self): """创建信息选项卡""" tab = QWidget() layout = QVBoxLayout() tab.setLayout(layout) # 模型信息 info_group = QGroupBox("模型信息") info_layout = QVBoxLayout() self.model_info_text = QTextEdit() self.model_info_text.setReadOnly(True) self.model_info_text.setFont(QFont("Consolas", 10)) info_layout.addWidget(self.model_info_text) info_group.setLayout(info_layout) layout.addWidget(info_group) # 使用说明 usage_group = QGroupBox("使用说明") usage_layout = QVBoxLayout() usage_text = """ 1. 序列分析：输入单个基因序列，选择任务类型和参数，点击"开始分析" 2. 批量分析：每行输入一个基因序列，点击"批量分析" 3. 任务类型： - analyze: 分析序列特征 - predict: 预测序列后续内容 - complete: 补全序列 4. 参数说明： - Temperature: 控制输出的随机性 (0.01-2.0) - Top P: 核采样概率 (0.5-1.0) - 最大 Token: 限制生成长度 """ usage_output = QTextEdit() usage_output.setPlainText(usage_text) usage_output.setReadOnly(True) usage_layout.addWidget(usage_output) usage_group.setLayout(usage_layout) layout.addWidget(usage_group) self.tab_widget.addTab(tab, "信息") def load_model(self): """加载模型""" try: model_path = os.path.dirname(os.path.abspath(__file__)) print(f"[INFO] 正在从路径加载模型: {model_path}") # 优先尝试 Transformers 版本（支持CUDA加速） try: from genos_model_transformers import GenosInference print("[INFO] 使用 Transformers 版本...") self.model = GenosInference(model_path=model_path) self.update_model_info() self.statusBar().showMessage("模型加载成功 (Transformers)") print("[INFO] 模型加载成功") return except Exception as e: print(f"[ERROR] Transformers 版本加载失败: {str(e)}") import traceback traceback.print_exc() raise except Exception as e: error_msg = f"模型加载失败: {str(e)}" self.statusBar().showMessage(error_msg) print(f"[ERROR] {error_msg}") import traceback traceback.print_exc() QMessageBox.critical(self, "错误", f"{error_msg}\n\n详细信息请查看控制台输出") QMessageBox.warning(self, "警告", f"模型加载失败: {str(e)}") def update_model_info(self): """更新模型信息显示""" if self.model: info = self.model.get_model_info() info_text = f""" 模型路径: {info['model_path']} 词汇表大小: {info['vocab_size']} 最大序列长度: {info['max_position_embeddings']:,} """ self.model_info_text.setPlainText(info_text) def clear_input(self): """清空输入""" self.sequence_input.clear() self.result_output.clear() def paste_input(self): """粘贴输入""" clipboard = QApplication.clipboard() self.sequence_input.setText(clipboard.text()) def load_from_file(self): """从文件加载序列""" file_path, _ = QFileDialog.getOpenFileName( self, "选择文件", "", "文本文件 (*.txt);;FASTA 文件 (*.fasta *.fa);;所有文件 (*)" ) if file_path: try: with open(file_path, 'r', encoding='utf-8') as f: content = f.read() # 如果是 FASTA 格式，只提取序列部分 if content.startswith('>'): lines = content.split('\n') sequence = ''.join([line.strip() for line in lines if not line.startswith('>')]) self.sequence_input.setPlainText(sequence) else: self.sequence_input.setPlainText(content) self.statusBar().showMessage(f"已加载文件: {os.path.basename(file_path)}") except Exception as e: QMessageBox.warning(self, "错误", f"加载文件失败: {str(e)}") def start_analysis(self): """开始分析""" if not self.model: QMessageBox.warning(self, "错误", "模型未加载") return sequence = self.sequence_input.toPlainText().strip() if not sequence: QMessageBox.warning(self, "警告", "请输入基因序列") return # 获取参数 task_type = self.task_combo.currentText() params = { 'temperature': self.temp_spin.value() / 100.0, 'top_p': self.topp_spin.value() / 100.0, 'max_tokens': self.maxtoken_spin.value() } # 禁用按钮并显示进度 self.analyze_btn.setEnabled(False) self.progress_bar.setVisible(True) self.progress_bar.setValue(0) self.statusBar().showMessage("正在分析...") # 创建并启动工作线程 self.worker = GenosAnalysisThread(self.model, sequence, task_type, params) self.worker.progress.connect(self.update_progress) self.worker.finished.connect(self.analysis_finished) self.worker.error.connect(self.analysis_error) self.worker.start() def start_batch_analysis(self): """开始批量分析""" if not self.model: QMessageBox.warning(self, "错误", "模型未加载") return sequences = self.batch_input.toPlainText().strip() if not sequences: QMessageBox.warning(self, "警告", "请输入基因序列") return params = { 'temperature': 0.7, 'top_p': 0.95, 'max_tokens': 512 } self.worker = GenosAnalysisThread(self.model, sequences, "batch", params) self.worker.finished.connect(self.batch_finished) self.worker.error.connect(self.analysis_error) self.worker.start() def update_progress(self, value): """更新进度""" self.progress_bar.setValue(value) def analysis_finished(self, result): """分析完成""" self.analyze_btn.setEnabled(True) self.progress_bar.setVisible(False) self.statusBar().showMessage("分析完成") # 显示结果 result_text = f""" 输入序列 ({len(result['input_sequence'])} bp): {result['input_sequence'][:200]}{'...' if len(result['input_sequence']) > 200 else ''} 任务类型: {result['task_type']} 生成 Token 数: {result['tokens_generated']} 完成原因: {result['finish_reason']} 分析结果: {result['output']} """ self.result_output.setPlainText(result_text) def batch_finished(self, result): """批量分析完成""" self.statusBar().showMessage(f"批量分析完成，共 {len(result['results'])} 条结果") results = [] for i, r in enumerate(result['results']): results.append(f""" --- 序列 {i+1} ({len(r['input_sequence'])} bp) --- 输入: {r['input_sequence'][:100]}{'...' if len(r['input_sequence']) > 100 else ''} 输出: {r['output'][:200]}{'...' if len(r['output']) > 200 else ''} """) self.batch_result.setPlainText('\n'.join(results)) def analysis_error(self, error_msg): """分析错误""" self.analyze_btn.setEnabled(True) self.progress_bar.setVisible(False) self.statusBar().showMessage("分析失败") QMessageBox.critical(self, "错误", error_msg) def main(): """主函数""" app = QApplication(sys.argv) app.setStyle('Fusion') window = GenosApp() window.show() sys.exit(app.exec_()) if __name__ == "__main__": main()genos_model.py
""" Genos 模型加载和推理模块 使用 VLLM 框架进行高效的基因组序列分析 """ import os from typing import List, Optional, Dict from vllm import LLM, SamplingParams from transformers import PreTrainedTokenizer class GenosInference: """Genos 模型推理类""" def __init__(self, model_path: str = None, **kwargs): """ 初始化 Genos 模型 Args: model_path: 模型路径，默认使用当前目录 **kwargs: VLLM 的其他参数 """ if model_path is None: model_path = os.path.dirname(os.path.abspath(__file__)) self.model_path = model_path # 初始化 LLM 模型 self.llm = LLM( model=model_path, trust_remote_code=True, tensor_parallel_size=kwargs.get('tensor_parallel_size', 1), gpu_memory_utilization=kwargs.get('gpu_memory_utilization', 0.9), max_model_len=kwargs.get('max_model_len', 8192), dtype=kwargs.get('dtype', 'float16'), **kwargs ) # 设置采样参数 self.default_sampling_params = SamplingParams( temperature=0.7, top_p=0.95, max_tokens=kwargs.get('max_tokens', 512), stop_token_ids=[16], # </s> token ) # 加载 tokenizer from transformers import AutoTokenizer self.tokenizer = AutoTokenizer.from_pretrained( model_path, trust_remote_code=True ) print(f"Genos 模型已加载，路径: {model_path}") def analyze_sequence(self, sequence: str, task_type: str = "analyze", **kwargs) -> Dict: """ 分析基因序列 Args: sequence: 基因序列（A, C, G, T, N） task_type: 任务类型（analyze, predict, complete） **kwargs: 采样参数 Returns: 分析结果字典 """ # 验证序列 sequence = self._validate_sequence(sequence) if sequence is None: return {"error": "无效的基因序列"} # 根据任务类型构建提示 prompt = self._build_prompt(sequence, task_type) # 设置采样参数 sampling_params = SamplingParams( temperature=kwargs.get('temperature', 0.7), top_p=kwargs.get('top_p', 0.95), max_tokens=kwargs.get('max_tokens', 512), stop_token_ids=[16], ) # 执行推理 outputs = self.llm.generate([prompt], sampling_params) # 解析结果 result = { "input_sequence": sequence, "task_type": task_type, "output": outputs[0].outputs[0].text, "tokens_generated": len(outputs[0].outputs[0].token_ids), "finish_reason": outputs[0].outputs[0].finish_reason, } return result def _validate_sequence(self, sequence: str) -> Optional[str]: """验证基因序列""" if not sequence: return None # 转大写并移除空白 sequence = sequence.upper().strip().replace('\n', '').replace(' ', '') # 检查是否只包含有效碱基 valid_bases = {'A', 'C', 'G', 'T', 'N'} for base in sequence: if base not in valid_bases: return None return sequence def _build_prompt(self, sequence: str, task_type: str) -> str: """构建提示词""" if task_type == "analyze": prompt = f"<s>{sequence}<SEP>" elif task_type == "predict": prompt = f"<s>{sequence}<MASK>" elif task_type == "complete": prompt = f"<s>{sequence}" else: prompt = f"<s>{sequence}" return prompt def batch_analyze(self, sequences: List[str], task_type: str = "analyze", **kwargs) -> List[Dict]: """ 批量分析基因序列 Args: sequences: 基因序列列表 task_type: 任务类型 **kwargs: 采样参数 Returns: 分析结果列表 """ # 验证和准备序列 valid_sequences = [] for seq in sequences: validated = self._validate_sequence(seq) if validated: valid_sequences.append(validated) if not valid_sequences: return [] # 构建提示词 prompts = [self._build_prompt(seq, task_type) for seq in valid_sequences] # 设置采样参数 sampling_params = SamplingParams( temperature=kwargs.get('temperature', 0.7), top_p=kwargs.get('top_p', 0.95), max_tokens=kwargs.get('max_tokens', 512), stop_token_ids=[16], ) # 批量推理 outputs = self.llm.generate(prompts, sampling_params) # 解析结果 results = [] for i, output in enumerate(outputs): results.append({ "input_sequence": valid_sequences[i], "task_type": task_type, "output": output.outputs[0].text, "tokens_generated": len(output.outputs[0].token_ids), "finish_reason": output.outputs[0].finish_reason, }) return results def get_token_count(self, sequence: str) -> int: """获取序列的 token 数量""" validated = self._validate_sequence(sequence) if validated is None: return 0 encoded = self.tokenizer.encode(validated) return len(encoded) def get_model_info(self) -> Dict: """获取模型信息""" return { "model_path": self.model_path, "vocab_size": self.tokenizer.vocab_size, "max_position_embeddings": self.llm.llm_engine.model_config.max_model_len, } if __name__ == "__main__": # 测试代码 model = GenosInference() # 测试序列分析 test_sequence = "ATGCGATCGTAGCTAGCTAG" result = model.analyze_sequence(test_sequence, task_type="analyze") print(f"分析结果: {result}")
