aitsc/src/flask_prompt_master/routes/expert_generate_3.py

# -*- coding: utf-8 -*-
"""
智能提示词优化3号专家 - 独立实现，与专家模式/2号专家功能一致但代码完全分离。
便于后续对比优化，不复用任何其他专家代码。
含提示词生成历史记录功能。
"""
from flask import Blueprint, render_template, request, jsonify
from openai import OpenAI
import hashlib
import json
import logging
import os
import time
from src.flask_prompt_master import db
from src.flask_prompt_master.models.models import User, Prompt
from src.flask_prompt_master.models.history_models import PromptHistory, UserStatistics

logger = logging.getLogger(__name__)
expert_generate_3_bp = Blueprint('expert_generate_3', __name__)
_dedup_cache = {}

_llm_client = OpenAI(
    api_key=os.environ.get('LLM_API_KEY') or 'sk-fdf7cc1c73504e628ec0119b7e11b8cc',
    base_url=os.environ.get('LLM_API_URL') or 'https://api.deepseek.com/v1'
)

INTENT_PROMPT_3 = """你是一位资深的意图分析专家，请分析用户输入的意图和需求。

你必须严格按照以下JSON格式返回，不要添加任何其他内容：
{
    "core_intent": "技术",
    "domain": "具体专业领域",
    "key_requirements": ["需求1", "需求2"],
    "expected_output": "期望输出的具体形式",
    "constraints": ["约束1", "约束2"],
    "keywords": ["关键词1", "关键词2"]
}

注意：
1. 严格遵守JSON格式
2. core_intent必须是以下之一：技术、创意、分析、咨询
3. 数组至少包含1个元素
4. 所有字段都必须存在
5. 不要包含注释
6. 不要添加任何额外的文本"""

EXPERT_TEMPLATES_3 = {
    "技术": """你是一位专业的技术领域提示工程师。基于以下意图分析，生成一个专业的技术任务提示词：

意图分析：
{analysis}

请生成的提示词包含：
1. 明确的技术背景和上下文
2. 具体的技术要求和规范
3. 性能和质量标准
4. 技术约束条件
5. 预期交付成果
6. 评估标准

使用专业技术术语，确保提示词的可执行性和可验证性。""",
    "创意": """你是一位专业的创意领域提示工程师。基于以下意图分析，生成一个创意设计提示词：

意图分析：
{analysis}

请生成的提示词包含：
1. 创意方向和灵感来源
2. 风格和氛围要求
3. 目标受众定义
4. 设计元素规范
5. 创意表现形式
6. 评估标准

使用专业创意术语，确保提示词的创新性和可执行性。""",
    "分析": """你是一位专业的数据分析提示工程师。基于以下意图分析，生成一个数据分析提示词：

意图分析：
{analysis}

请生成的提示词包含：
1. 分析目标和范围
2. 数据要求和规范
3. 分析方法和工具
4. 输出格式要求
5. 关键指标定义
6. 质量控制标准

使用专业分析术语，确保提示词的科学性和可操作性。""",
    "咨询": """你是一位专业的咨询领域提示工程师。基于以下意图分析，生成一个咨询服务提示词：

意图分析：
{analysis}

请生成的提示词包含：
1. 咨询问题界定
2. 背景信息要求
3. 分析框架设定
4. 建议输出格式
5. 实施考虑因素
6. 效果评估标准

使用专业咨询术语，确保提示词的专业性和实用性。"""
}

FALLBACK_TEMPLATE_3 = """你是一位专业的通用领域提示工程师。基于以下意图分析，生成一个专业的提示词：

意图分析：
{analysis}

请生成的提示词包含：
1. 明确的目标定义
2. 具体要求和规范
3. 质量标准
4. 约束条件
5. 预期输出
6. 评估标准

确保提示词的清晰性和可执行性。"""

TEMPLATE_NAME_3 = '智能提示词优化3号专家'


def _resolve_user_id():
    try:
        from flask_login import current_user
        if current_user.is_authenticated:
            return getattr(current_user, 'id', None) or getattr(current_user, 'uid', None)
    except Exception:
        pass
    from flask import session
    sid = session.get('user_id')
    if sid is not None:
        return sid
    try:
        u = User.query.filter_by(login_name='admin').first()
        return u.uid if u else 1
    except Exception as ex:
        logger.warning("3号专家 获取默认用户失败: %s", ex)
        return 1


@expert_generate_3_bp.route('/expert-generate-3', methods=['GET'])
def expert_generate_3_page():
    """智能提示词优化3号专家 - 页面"""
    return render_template('expert_generate_3.html')


@expert_generate_3_bp.route('/api/expert-generate-3/generate', methods=['POST'])
def expert_generate_3_api():
    """智能提示词优化3号专家 - 两阶段生成 API，含历史记录"""
    try:
        if not request.is_json:
            return jsonify({'code': 400, 'message': '请求必须是JSON格式', 'data': None})
        payload = request.get_json() or {}
        raw_input = (payload.get('input_text') or '').strip()
        if not raw_input:
            return jsonify({'code': 400, 'message': '请输入您的需求', 'data': None})

        temperature = payload.get('temperature')
        temperature = float(temperature) if temperature is not None else 0.7
        temperature = max(0.0, min(2.0, temperature))
        max_tokens = payload.get('max_tokens')
        max_tokens = int(max_tokens) if max_tokens is not None else 1000
        max_tokens = max(100, min(4000, max_tokens))
        timeout = payload.get('timeout')
        timeout = int(timeout) if timeout is not None else 60
        timeout = max(10, min(300, timeout))

        uid = _resolve_user_id()
        req_key = (uid, hashlib.md5(raw_input.encode()).hexdigest())
        now_ts = time.time()
        if req_key in _dedup_cache and (now_ts - _dedup_cache[req_key]) < 8:
            return jsonify({'code': 429, 'message': '请勿重复提交，请稍后再试', 'data': None})
        _dedup_cache[req_key] = now_ts
        if len(_dedup_cache) > 500:
            _dedup_cache.clear()

        resp1 = _llm_client.chat.completions.create(
            model="deepseek-chat",
            messages=[
                {"role": "system", "content": INTENT_PROMPT_3},
                {"role": "user", "content": raw_input}
            ],
            temperature=0.1,
            timeout=timeout
        )
        intent_raw = (resp1.choices[0].message.content or "").strip()
        intent_raw = intent_raw.replace('```json', '').replace('```', '').strip()
        try:
            intent_data = json.loads(intent_raw)
            for field in ['core_intent', 'domain', 'key_requirements', 'expected_output', 'constraints', 'keywords']:
                if field not in intent_data:
                    raise ValueError("缺少字段: " + field)
            if intent_data['core_intent'] not in ('技术', '创意', '分析', '咨询'):
                intent_data['core_intent'] = '技术'
            for arr_field in ['key_requirements', 'constraints', 'keywords']:
                v = intent_data.get(arr_field)
                if not isinstance(v, list) or len(v) == 0:
                    intent_data[arr_field] = ['未指定']
        except json.JSONDecodeError as e:
            logger.error("3号专家 JSON解析失败: %s", e)
            return jsonify({'code': 500, 'message': '意图分析格式有误，请重试', 'data': None})
        except ValueError as e:
            logger.error("3号专家 数据验证失败: %s", e)
            return jsonify({'code': 500, 'message': str(e), 'data': None})
        tpl = EXPERT_TEMPLATES_3.get(intent_data['core_intent'], FALLBACK_TEMPLATE_3)
        analysis_str = json.dumps(intent_data, ensure_ascii=False, indent=2)
        resp2 = _llm_client.chat.completions.create(
            model="deepseek-chat",
            messages=[
                {"role": "system", "content": tpl.format(analysis=analysis_str)},
                {"role": "user", "content": raw_input}
            ],
            temperature=temperature,
            max_tokens=max_tokens,
            timeout=timeout
        )
        result_prompt = (resp2.choices[0].message.content or "").strip()
        if not result_prompt:
            return jsonify({'code': 500, 'message': '生成失败，请重试', 'data': None})
        try:
            db.session.add(Prompt(input_text=raw_input, generated_text=result_prompt, user_id=uid))
            db.session.commit()
        except Exception as e:
            db.session.rollback()
            logger.warning("3号专家 保存 Prompt 失败: %s", e)
        try:
            PromptHistory.add_history(
                user_id=uid,
                original_input=raw_input,
                generated_prompt=result_prompt,
                template_name=TEMPLATE_NAME_3
            )
            UserStatistics.update_statistics(uid)
        except Exception as e:
            logger.warning("3号专家 保存历史失败: %s", e)
        return jsonify({
            'code': 200,
            'message': 'success',
            'data': {'intent_analysis': intent_data, 'generated_prompt': result_prompt}
        })
    except Exception as e:
        logger.exception("智能提示词优化3号专家失败")
        return jsonify({'code': 500, 'message': str(e) or '生成失败，请重试', 'data': None})