imetting/backend/app/services/llm_service.py

import json
import dashscope
from http import HTTPStatus
from typing import Optional, Dict, List, Generator
import app.core.config as config_module
from app.core.database import get_db_connection
from app.services.system_config_service import SystemConfigService


class LLMService:
    """LLM服务 - 专注于大模型API调用和提示词管理"""

    def __init__(self):
        # 设置dashscope API key
        dashscope.api_key = config_module.QWEN_API_KEY

    def _get_llm_call_params(self) -> Dict[str, Any]:
        """
        获取 dashscope.Generation.call() 所需的参数字典

        Returns:
            Dict: 包含 model、timeout、temperature、top_p 的参数字典
        """
        return {
            'model': SystemConfigService.get_llm_model_name(),
            'timeout': SystemConfigService.get_llm_timeout(),
            'temperature': SystemConfigService.get_llm_temperature(),
            'top_p': SystemConfigService.get_llm_top_p(),
        }

    def get_task_prompt(self, task_type: str, cursor=None, prompt_id: Optional[int] = None) -> str:
        """
        统一的提示词获取方法

        Args:
            task_type: 任务类型，如 'MEETING_TASK', 'KNOWLEDGE_TASK' 等
            cursor: 数据库游标，如果传入则使用，否则创建新连接
            prompt_id: 可选的提示词ID，如果指定则使用该提示词，否则使用默认提示词

        Returns:
            str: 提示词内容，如果未找到返回默认提示词
        """
        # 如果指定了 prompt_id，直接获取该提示词
        if prompt_id:
            query = """
                SELECT content
                FROM prompts
                WHERE id = %s AND task_type = %s AND is_active = TRUE
                LIMIT 1
            """
            params = (prompt_id, task_type)
        else:
            # 否则获取默认提示词
            query = """
                SELECT content
                FROM prompts
                WHERE task_type = %s
                  AND is_default = TRUE
                  AND is_active = TRUE
                LIMIT 1
            """
            params = (task_type,)

        if cursor:
            cursor.execute(query, params)
            result = cursor.fetchone()
            if result:
                return result['content'] if isinstance(result, dict) else result[0]
        else:
            with get_db_connection() as connection:
                cursor = connection.cursor(dictionary=True)
                cursor.execute(query, params)
                result = cursor.fetchone()
                if result:
                    return result['content']

        # 返回默认提示词
        return self._get_default_prompt(task_type)

    def _get_default_prompt(self, task_name: str) -> str:
        """获取默认提示词"""
        system_prompt = config_module.LLM_CONFIG.get("system_prompt", "请根据提供的内容进行总结和分析。")
        default_prompts = {
            'MEETING_TASK': system_prompt,
            'KNOWLEDGE_TASK': "请根据提供的信息生成知识库文章。",
        }
        return default_prompts.get(task_name, "请根据提供的内容进行总结和分析。")

    def _call_llm_api_stream(self, prompt: str) -> Generator[str, None, None]:
        """流式调用阿里Qwen大模型API"""
        try:
            responses = dashscope.Generation.call(
                **self._get_llm_call_params(),
                prompt=prompt,
                stream=True,
                incremental_output=True
            )

            for response in responses:
                if response.status_code == HTTPStatus.OK:
                    # 增量输出内容
                    new_content = response.output.get('text', '')
                    if new_content:
                        yield new_content
                else:
                    error_msg = f"Request failed with status code: {response.status_code}, Error: {response.message}"
                    print(error_msg)
                    yield f"error: {error_msg}"
                    break

        except Exception as e:
            error_msg = f"流式调用大模型API错误: {e}"
            print(error_msg)
            yield f"error: {error_msg}"

    def _call_llm_api(self, prompt: str) -> Optional[str]:
        """调用阿里Qwen大模型API（非流式）"""
        try:
            response = dashscope.Generation.call(
                **self._get_llm_call_params(),
                prompt=prompt
            )

            if response.status_code == HTTPStatus.OK:
                return response.output.get('text', '')
            else:
                print(f"API调用失败: {response.status_code}, {response.message}")
                return None

        except Exception as e:
            print(f"调用大模型API错误: {e}")
            return None


# 测试代码
if __name__ == '__main__':
    print("--- 运行LLM服务测试 ---")
    llm_service = LLMService()

    # 测试获取任务提示词
    meeting_prompt = llm_service.get_task_prompt('MEETING_TASK')
    print(f"会议任务提示词: {meeting_prompt[:100]}...")

    knowledge_prompt = llm_service.get_task_prompt('KNOWLEDGE_TASK')
    print(f"知识库任务提示词: {knowledge_prompt[:100]}...")

    print("--- LLM服务测试完成 ---")