shudao-main/shudao-chat-py/services/qwen_service.py

"""
Qwen AI 服务
"""
import httpx
import json
from typing import AsyncGenerator
from utils.config import settings
from utils.logger import logger
from utils.prompt_loader import load_prompt
from services.deepseek_service import deepseek_service


class QwenService:
    def __init__(self):
        # 确保 API URL 包含完整路径
        base_url = settings.qwen3.api_url.rstrip('/')
        self.api_url = f"{base_url}/v1/chat/completions"
        self.model = settings.qwen3.model
        
        # 意图识别使用专门的配置
        intent_base_url = settings.intent.api_url.rstrip('/')
        self.intent_api_url = f"{intent_base_url}/v1/chat/completions"
        self.intent_model = settings.intent.model

    def _should_fallback(self, status_code: int) -> bool:
        return status_code in (429, 500, 502, 503, 504)

    async def _fallback_deepseek(self, messages: list) -> str:
        try:
            logger.warning("[Qwen API] Falling back to DeepSeek due to upstream error")
            return await deepseek_service.chat(messages)
        except Exception as e:
            error_msg = str(e).strip() if str(e).strip() else type(e).__name__
            logger.error(f"[Qwen API] DeepSeek fallback failed: {type(e).__name__}: {error_msg}")
            raise RuntimeError(f"AI服务暂时不可用，主模型和备用模型均无法响应（{type(e).__name__}），请稍后重试") from e
    
    async def extract_keywords(self, question: str) -> str:
        """从问题中提炼搜索关键词"""
        # 使用prompt加载器加载关键词提取prompt（如果配置了的话）
        # 这里暂时保留原有逻辑，可以后续添加到prompt配置中
        keyword_prompt = """你是一个关键词提取助手。请从用户的问题中提炼出最核心的搜索关键词。
要求：
1. 提取2-5个最关键的词语
2. 去除语气词、助词等无意义词汇
3. 保留专业术语和核心概念
4. 以空格分隔多个关键词

直接返回关键词，不要其他说明。

用户问题："""
        
        messages = [
            {"role": "system", "content": keyword_prompt},
            {"role": "user", "content": question}
        ]
        
        try:
            keywords = await self.chat(messages)
            return keywords.strip()
        except Exception as e:
            logger.error(f"关键词提取失败: {e}")
            # 失败时返回原问题
            return question
    
    async def intent_recognition(self, message: str) -> dict:
        """意图识别"""
        # 使用prompt加载器加载意图识别prompt
        intent_prompt = load_prompt("intent_recognition", userMessage=message)
        
        messages = [
            {"role": "user", "content": intent_prompt}
        ]
        
        try:
            # 使用专门的意图识别API和模型
            response = await self.chat(messages, model=self.intent_model, api_url=self.intent_api_url)
            logger.info(f"意图识别原始响应: {response[:500]}")
            # 尝试解析JSON - 使用支持多行和嵌套的正则
            import re
            # 先去除 markdown 代码块标记
            cleaned = re.sub(r'```(?:json)?\s*', '', response)
            cleaned = cleaned.strip()
            # 匹配最外层的 { ... }（支持多行和嵌套）
            json_match = re.search(r'\{.*\}', cleaned, re.DOTALL)
            if json_match:
                result = json.loads(json_match.group())
                # 兼容模板输出的 "intent" 和 "intent_type" 两种字段名
                intent_type = (result.get("intent_type") or result.get("intent") or "").lower()
                # 统一设置 intent_type 字段，确保下游一致
                result["intent_type"] = intent_type
                
                # 优先使用模型返回的 direct_answer，否则使用预设回复
                direct_answer = result.get("direct_answer", "")
                
                if intent_type in ("greeting", "问候"):
                    result["response"] = direct_answer if direct_answer else "您好！我是蜀道集团智能助手，很高兴为您服务。"
                elif intent_type in ("faq", "常见问题"):
                    result["response"] = direct_answer if direct_answer else "我可以帮您解答常见问题，请告诉我您想了解什么。"
                else:
                    result["response"] = direct_answer or ""
                
                return result
            logger.warning(f"意图识别JSON解析失败，原始响应: {response[:300]}")
            return {"intent_type": "general_chat", "confidence": 0.5, "reason": "无法解析JSON", "response": ""}
        except Exception as e:
            logger.error(f"意图识别失败: {e}")
            return {"intent_type": "general_chat", "confidence": 0.5, "reason": str(e), "response": ""}
    
    async def chat(self, messages: list, model: str = None, api_url: str = None) -> str:
        """同步聊天"""
        data = {
            "model": model or self.model,
            "messages": messages,
            "stream": False  # 明确指定非流式
        }
        
        # 使用指定的API URL，默认使用qwen3的URL
        target_url = api_url or self.api_url
        normalized_target = target_url.rstrip("/")
        is_qwen3_target = normalized_target == self.api_url.rstrip("/")
        
        # 详细请求日志
        logger.info(f"[Qwen API] 请求 URL: {target_url}")
        logger.info(f"[Qwen API] 使用模型: {data['model']}")
        logger.info(f"[Qwen API] 消息数量: {len(messages)}")
        
        try:
            # 准备请求头
            headers = {
                "Content-Type": "application/json"
            }
            
            # 如果配置中有 token，添加到请求头（兼容需要认证的场景）
            if hasattr(settings, 'intent') and hasattr(settings.intent, 'token') and normalized_target == self.intent_api_url.rstrip("/"):
                if settings.intent.token:
                    headers["Authorization"] = f"Bearer {settings.intent.token}"
                    logger.info("[Qwen API] 已添加 Intent API Authorization header")
            elif hasattr(settings, 'qwen3') and hasattr(settings.qwen3, 'token') and normalized_target == self.api_url.rstrip("/"):
                if settings.qwen3.token:
                    headers["Authorization"] = f"Bearer {settings.qwen3.token}"
                    logger.info("[Qwen API] 已添加 Qwen3 API Authorization header")
            
            async with httpx.AsyncClient(timeout=120.0) as client:
                response = await client.post(
                    target_url,
                    json=data,
                    headers=headers
                )
                
                logger.info(f"[Qwen API] 响应状态码: {response.status_code}")
                logger.info(f"[Qwen API] 响应头: {dict(response.headers)}")
                
                # 记录响应内容（前500字符用于调试）
                response_preview = response.text[:500] if response.text else "(空响应)"
                logger.info(f"[Qwen API] 响应预览: {response_preview}")
                
                response.raise_for_status()
                
                # 检查响应是否为空
                if not response.text:
                    logger.error("[Qwen API] 返回空响应")
                    return ""
                
                # 检查是否是流式响应（以 data: 开头）
                if response.text.startswith("data:"):
                    logger.info("[Qwen API] 检测到流式响应，解析 SSE 格式")
                    # 解析 SSE 格式
                    content_parts = []
                    for line in response.text.split('\n'):
                        if line.startswith("data:"):
                            data_str = line[5:].strip()
                            if data_str and data_str != "[DONE]":
                                try:
                                    data_json = json.loads(data_str)
                                    delta_content = data_json.get('choices', [{}])[0].get('delta', {}).get('content', '')
                                    if delta_content:
                                        content_parts.append(delta_content)
                                except json.JSONDecodeError:
                                    continue
                    final_content = ''.join(content_parts)
                    logger.info(f"[Qwen API] SSE 解析完成，内容长度: {len(final_content)}")
                    return final_content
                
                # 普通 JSON 响应
                try:
                    result = response.json()
                    content = result.get('response', result.get('choices', [{}])[0].get('message', {}).get('content', ''))
                    logger.info(f"[Qwen API] JSON 解析成功，内容长度: {len(content)}")
                    return content
                except json.JSONDecodeError as je:
                    logger.error(f"[Qwen API] 响应不是有效的 JSON: {response.text[:200]}")
                    raise ValueError(f"无效的 JSON 响应: {str(je)}")
                
        except httpx.HTTPStatusError as e:
            logger.error(f"[Qwen API] HTTP 错误 - 状态码: {e.response.status_code}, URL: {target_url}")
            logger.error(f"[Qwen API] HTTP 错误响应: {e.response.text[:500]}")
            if is_qwen3_target and self._should_fallback(e.response.status_code):
                return await self._fallback_deepseek(messages)
            raise
        except httpx.RequestError as e:
            logger.error(f"[Qwen API] 请求错误 - URL: {target_url}, 错误: {type(e).__name__}: {str(e)}")
            if is_qwen3_target:
                return await self._fallback_deepseek(messages)
            raise
        except Exception as e:
            logger.error(f"[Qwen API] 未知错误 - URL: {target_url}, 模型: {data['model']}, 错误: {type(e).__name__}: {str(e)}")
            raise
    
    async def stream_chat(self, messages: list) -> AsyncGenerator[str, None]:
        """流式聊天"""
        data = {
            "model": self.model,
            "messages": messages,
            "stream": True
        }
        
        try:
            async with httpx.AsyncClient(timeout=120.0) as client:
                async with client.stream(
                    "POST",
                    self.api_url,
                    json=data
                ) as response:
                    response.raise_for_status()
                    async for line in response.aiter_lines():
                        if line.startswith("data: "):
                            data_str = line[6:]
                            if data_str == "[DONE]":
                                break
                            try:
                                data_json = json.loads(data_str)
                                # 兼容 OpenAI 格式 choices[0].delta.content
                                choices = data_json.get('choices', [])
                                if choices:
                                    content = choices[0].get('delta', {}).get('content', '') or choices[0].get('message', {}).get('content', '')
                                else:
                                    content = data_json.get('content', '')
                                if content:
                                    yield content
                            except json.JSONDecodeError:
                                continue
        except httpx.HTTPStatusError as e:
            status_code = e.response.status_code if e.response else 0
            logger.error(f"Qwen stream HTTP error: {status_code}")
            if self._should_fallback(status_code):
                logger.warning("[Qwen API] Stream fallback to DeepSeek")
                async for chunk in deepseek_service.stream_chat(messages):
                    yield chunk
                return
            raise
        except httpx.RequestError as e:
            logger.error(f"Qwen stream request error: {type(e).__name__}: {e}")
            logger.warning("[Qwen API] Stream fallback to DeepSeek")
            async for chunk in deepseek_service.stream_chat(messages):
                yield chunk
            return
        except Exception as e:
            logger.error(f"Qwen 流式 API 调用失败: {e}")
            raise


# 全局实例
qwen_service = QwenService()