from __future__ import annotations import os import re from typing import Optional from openai import AsyncOpenAI class DeepSeekService: """OpenAI-compatible DeepSeek refinement service.""" def __init__(self, api_key: Optional[str] = None): self.api_key = api_key or os.environ.get("DEEPSEEK_API_KEY") self.base_url = "https://api.deepseek.com" self.model = "deepseek-chat" self.client = None self.is_loaded = False print("DeepSeek API service initializing...") print(f"API Base URL: {self.base_url}") async def load(self): try: if not self.api_key: print("DeepSeek API key not provided") self.is_loaded = False return self.client = AsyncOpenAI(api_key=self.api_key, base_url=self.base_url) self.is_loaded = True print("DeepSeek API service is ready!") except Exception as exc: print(f"DeepSeek API service initialization failed: {exc}") self.is_loaded = False async def refine_diagnosis( self, raw_answer: str, raw_thinking: Optional[str] = None, language: str = "zh", ) -> dict: if not self.is_loaded or self.client is None: error_msg = ( "API not initialized, cannot generate analysis" if language == "en" else "API未初始化，无法生成分析过程" ) print("DeepSeek API not initialized, returning original result") return { "success": False, "description": "", "analysis_process": raw_thinking or error_msg, "diagnosis_result": raw_answer, "original_diagnosis": raw_answer, "error": "DeepSeek API not initialized", } try: prompt = self._build_refine_prompt(raw_answer, raw_thinking, language) system_content = ( "You are a professional medical text editor. Your task is to polish and organize " "medical diagnostic text to make it flow smoothly while preserving the original " "meaning. Output ONLY the formatted result. Do NOT add any explanations, comments, " "or thoughts. Just follow the format exactly." if language == "en" else "你是医学文本整理专家，按照用户要求将用户输入的文本整理成用户想要的格式，不要改写或总结。" ) response = await self.client.chat.completions.create( model=self.model, messages=[ {"role": "system", "content": system_content}, {"role": "user", "content": prompt}, ], temperature=0.1, max_tokens=2048, top_p=0.8, ) generated_text = response.choices[0].message.content parsed = self._parse_refined_output(generated_text, raw_answer, raw_thinking, language) return { "success": True, "description": parsed["description"], "analysis_process": parsed["analysis_process"], "diagnosis_result": parsed["diagnosis_result"], "original_diagnosis": raw_answer, "raw_refined": generated_text, } except Exception as exc: print(f"DeepSeek API call failed: {exc}") error_msg = ( "API call failed, cannot generate analysis" if language == "en" else "API调用失败，无法生成分析过程" ) return { "success": False, "description": "", "analysis_process": raw_thinking or error_msg, "diagnosis_result": raw_answer, "original_diagnosis": raw_answer, "error": str(exc), } def _build_refine_prompt( self, raw_answer: str, raw_thinking: Optional[str] = None, language: str = "zh", ) -> str: thinking_text = raw_thinking if raw_thinking else "No analysis process available." if language == "en": return f"""You are a text organization expert. There are two texts that need to be organized. Text 1 is the thinking process of the SkinGPT model, and Text 2 is the diagnosis result given by SkinGPT. 【Requirements】 - Preserve the original tone and expression style - Text 1 contains the thinking process, Text 2 contains the diagnosis result - Extract the image observation part from the thinking process as Description. This should include all factual observations about what was seen in the image, not just a brief summary. - For Diagnostic Reasoning: refine and condense the remaining thinking content. Remove redundancies, self-doubt, circular reasoning, and unnecessary repetition. Keep it concise and not too long. Keep the logical chain clear and enhance readability. IMPORTANT: DO NOT include any image description or visual observations in Diagnostic Reasoning. Only include reasoning, analysis, and diagnostic thought process. - If [Text 1] content is NOT: No analysis process available. Then organize [Text 1] content accordingly, DO NOT confuse [Text 1] and [Text 2] - If [Text 1] content IS: No analysis process available. Then extract the analysis process and description from [Text 2] - DO NOT infer or add new medical information, DO NOT output any meta-commentary - You may adjust unreasonable statements or remove redundant content to improve clarity [Text 1] {thinking_text} [Text 2] {raw_answer} 【Output】Only output three sections, do not output anything else: ## Description (Extract all image observation content from the thinking process - include all factual descriptions of what was seen) ## Analysis Process (Refined and condensed diagnostic reasoning: remove self-doubt, circular logic, and redundancies. Keep it concise and not too long. Keep logical flow clear. Do NOT include image observations) ## Diagnosis Result (The organized diagnosis result from Text 2) """ return f"""你是一个文本整理专家。有两段文本需要整理，文本1是SkinGPT模型的思考过程的文本，文本2是SkinGPT给出的诊断结果的文本。 【要求】 - 保留原文的语气和表达方式 - 文本1是思考过程，文本2是诊断结果 - 从思考过程中提取图像观察部分作为图像描述。需要包含所有关于图片中观察到的事实内容，不要简化或缩短。 - 对于分析过程：提炼并精简剩余的思考内容，去除冗余、自我怀疑、兜圈子的内容。保持简洁，不要太长。保持逻辑链条清晰，增强可读性。重要：分析过程中不要包含任何图像描述或视觉观察内容，只包含推理、分析和诊断思考过程。 - 如果【文本1】内容不是：No analysis process available.那么按要求整理【文本1】的内容，不要混淆【文本1】和【文本2】。 - 如果【文本1】内容是：No analysis process available.那么从【文本2】提炼分析过程和描述。 - 【文本1】和【文本2】需要翻译成简体中文 - 禁止推断或添加新的医学信息，禁止输出任何元评论 - 可以调整不合理的语句或去除冗余内容以提高清晰度 【文本1】 {thinking_text} 【文本2】 {raw_answer} 【输出】只输出三个部分，不要输出其他任何内容： ## 图像描述 （从思考过程中提取所有图像观察内容，包含所有关于图片的事实描述） ## 分析过程 （提炼并精简后的诊断推理：去除自我怀疑、兜圈逻辑和冗余内容。保持简洁，不要太长。保持逻辑流畅。不包含图像观察） ## 诊断结果 （整理后的诊断结果） """ def _parse_refined_output( self, generated_text: str, raw_answer: str, raw_thinking: Optional[str] = None, language: str = "zh", ) -> dict: description = "" analysis_process = None diagnosis_result = None if language == "en": desc_match = re.search( r"##\s*Description\s*\n([\s\S]*?)(?=##\s*Analysis\s*Process|$)", generated_text, re.IGNORECASE, ) analysis_match = re.search( r"##\s*Analysis\s*Process\s*\n([\s\S]*?)(?=##\s*Diagnosis\s*Result|$)", generated_text, re.IGNORECASE, ) result_match = re.search( r"##\s*Diagnosis\s*Result\s*\n([\s\S]*?)$", generated_text, re.IGNORECASE, ) desc_header = "## Description" analysis_header = "## Analysis Process" result_header = "## Diagnosis Result" else: desc_match = re.search(r"##\s*图像描述\s*\n([\s\S]*?)(?=##\s*分析过程|$)", generated_text) analysis_match = re.search(r"##\s*分析过程\s*\n([\s\S]*?)(?=##\s*诊断结果|$)", generated_text) result_match = re.search(r"##\s*诊断结果\s*\n([\s\S]*?)$", generated_text) desc_header = "## 图像描述" analysis_header = "## 分析过程" result_header = "## 诊断结果" if desc_match: description = desc_match.group(1).strip() else: description = "" if analysis_match: analysis_process = analysis_match.group(1).strip() else: result_pos = generated_text.find(result_header) if result_pos > 0: analysis_process = generated_text[:result_pos].strip() for header in [desc_header, analysis_header]: analysis_process = re.sub(f"{re.escape(header)}\\s*\\n?", "", analysis_process).strip() else: analysis_process = generated_text[: len(generated_text) // 2].strip() if not analysis_process and raw_thinking: analysis_process = raw_thinking if result_match: diagnosis_result = result_match.group(1).strip() else: result_pos = generated_text.find(result_header) if result_pos > 0: diagnosis_result = generated_text[result_pos:].strip() diagnosis_result = re.sub( f"^{re.escape(result_header)}\\s*\\n?", "", diagnosis_result, ).strip() else: diagnosis_result = generated_text[len(generated_text) // 2 :].strip() if not diagnosis_result: diagnosis_result = raw_answer return { "description": description, "analysis_process": analysis_process, "diagnosis_result": diagnosis_result, } _deepseek_service: Optional[DeepSeekService] = None async def get_deepseek_service(api_key: Optional[str] = None) -> Optional[DeepSeekService]: global _deepseek_service if _deepseek_service is None: try: _deepseek_service = DeepSeekService(api_key=api_key) await _deepseek_service.load() if not _deepseek_service.is_loaded: print("DeepSeek API service initialization failed, will use fallback mode") return _deepseek_service except Exception as exc: print(f"DeepSeek service initialization failed: {exc}") return None return _deepseek_service