1177 lines
41 KiB
Python
1177 lines
41 KiB
Python
from flask import Blueprint, render_template, request, redirect, url_for, flash, jsonify, current_app
|
||
from openai import OpenAI
|
||
from flask_prompt_master import db
|
||
from flask_prompt_master.models import User, Prompt, Feedback, PromptTemplate, WxUser
|
||
from flask_prompt_master.forms import PromptForm, FeedbackForm
|
||
from config import Config
|
||
import pymysql
|
||
from datetime import datetime
|
||
import requests
|
||
import hashlib
|
||
import time
|
||
import json
|
||
|
||
main_bp = Blueprint('main', __name__)
|
||
|
||
client = OpenAI(api_key=Config.LLM_API_KEY, base_url=Config.LLM_API_URL)
|
||
|
||
# 从配置中获取微信小程序配置
|
||
WX_APPID = Config.WX_APPID
|
||
WX_SECRET = Config.WX_SECRET
|
||
|
||
def get_system_prompt(template_id=None):
|
||
"""获取系统提示词模板"""
|
||
if template_id:
|
||
template = PromptTemplate.query.get(template_id)
|
||
if template:
|
||
return template.system_prompt
|
||
|
||
# 如果没有指定模板ID或模板不存在，返回默认模板
|
||
default_template = PromptTemplate.query.filter_by(is_default=True).first()
|
||
if default_template:
|
||
return default_template.system_prompt
|
||
|
||
# 如果数据库中没有模板，返回硬编码的默认模板
|
||
return """你是一个专业的提示词工程师，擅长将普通的描述转换为结构化、专业的 Prompt。
|
||
|
||
你需要：
|
||
1. 分析用户的需求和意图
|
||
2. 将其转换为清晰、详细的提示词
|
||
3. 添加必要的上下文和约束条件
|
||
4. 使用专业的术语和格式
|
||
5. 确保生成的提示词能够获得最佳的 AI 响应
|
||
|
||
请直接返回优化后的提示词，不要添加任何解释或其他内容。"""
|
||
|
||
def generate_with_llm(input_text, template_id=None):
|
||
"""调用大模型API生成提示词"""
|
||
try:
|
||
system_prompt = get_system_prompt(template_id)
|
||
|
||
# 打印参数
|
||
print("\n=== API 调用参数 ===")
|
||
print(f"模板ID: {template_id}")
|
||
print(f"输入文本: {input_text}")
|
||
print(f"系统提示: {system_prompt}")
|
||
print("==================\n")
|
||
|
||
response = client.chat.completions.create(
|
||
model="deepseek-chat",
|
||
messages=[
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": input_text}
|
||
],
|
||
temperature=0.7,
|
||
max_tokens=500
|
||
)
|
||
|
||
# 打印响应
|
||
generated_text = response.choices[0].message.content.strip()
|
||
print("\n=== API 响应结果 ===")
|
||
print(f"生成的提示词: {generated_text}")
|
||
print("==================\n")
|
||
|
||
return generated_text
|
||
except Exception as e:
|
||
current_app.logger.error(f'LLM API调用失败: {str(e)}')
|
||
return "提示词生成失败，请稍后重试"
|
||
|
||
def get_template_icon(category):
|
||
"""根据分类返回对应的Font Awesome图标类名"""
|
||
icons = {
|
||
'通用工具': 'fa-magic',
|
||
'内容创作': 'fa-pen-fancy',
|
||
'设计创意': 'fa-palette',
|
||
'技术研发': 'fa-code',
|
||
'商业营销': 'fa-chart-line',
|
||
'专业服务': 'fa-briefcase',
|
||
'教育培训': 'fa-graduation-cap',
|
||
'智慧城市': 'fa-city',
|
||
'工业制造': 'fa-industry',
|
||
'生活服务': 'fa-concierge-bell'
|
||
}
|
||
return icons.get(category, 'fa-star') # 默认返回星星图标
|
||
|
||
@main_bp.route('/', methods=['GET', 'POST'])
|
||
def index():
|
||
form = PromptForm()
|
||
templates = PromptTemplate.query.all()
|
||
|
||
# 获取所有可用的分类选项
|
||
industries = sorted(set(t.industry for t in templates if t.industry))
|
||
professions = sorted(set(t.profession for t in templates if t.profession))
|
||
categories = sorted(set(t.category for t in templates if t.category))
|
||
sub_categories = sorted(set(t.sub_category for t in templates if t.sub_category))
|
||
|
||
if form.validate_on_submit():
|
||
template_id = request.form.get('template_id')
|
||
generated_text = generate_with_llm(form.input_text.data, template_id)
|
||
|
||
# 获取默认用户的 uid
|
||
try:
|
||
conn = pymysql.connect(
|
||
host='localhost',
|
||
user='root',
|
||
password='123456',
|
||
database='food_db',
|
||
charset='utf8mb4'
|
||
)
|
||
cursor = conn.cursor()
|
||
cursor.execute("SELECT uid FROM user WHERE login_name = 'admin' LIMIT 1")
|
||
result = cursor.fetchone()
|
||
if result:
|
||
user_id = result[0]
|
||
else:
|
||
user_id = 1 # 如果没有找到用户，使用默认值
|
||
cursor.close()
|
||
conn.close()
|
||
except Exception as e:
|
||
print(f"获取用户ID失败: {str(e)}")
|
||
user_id = 1 # 如果查询失败，使用默认值
|
||
|
||
prompt = Prompt(
|
||
input_text=form.input_text.data,
|
||
generated_text=generated_text,
|
||
user_id=user_id # 使用查询到的用户ID
|
||
)
|
||
db.session.add(prompt)
|
||
db.session.commit()
|
||
return render_template('generate.html', form=form, prompt=prompt, templates=templates,
|
||
get_template_icon=get_template_icon, industries=industries,
|
||
professions=professions, categories=categories,
|
||
sub_categories=sub_categories)
|
||
return render_template('generate.html', form=form, prompt=None, templates=templates,
|
||
get_template_icon=get_template_icon, industries=industries,
|
||
professions=professions, categories=categories,
|
||
sub_categories=sub_categories)
|
||
|
||
@main_bp.route('/prompt/<int:prompt_id>')
|
||
def show_prompt(prompt_id):
|
||
prompt = Prompt.query.get_or_404(prompt_id)
|
||
return render_template('prompt.html', prompt=prompt)
|
||
|
||
@main_bp.route('/feedback/<int:prompt_id>', methods=['GET', 'POST'])
|
||
def submit_feedback(prompt_id):
|
||
prompt = Prompt.query.get_or_404(prompt_id)
|
||
form = FeedbackForm()
|
||
if form.validate_on_submit():
|
||
feedback = Feedback(
|
||
rating=form.rating.data,
|
||
comment=form.comment.data,
|
||
user_id=1, # 临时用户ID
|
||
prompt_id=prompt.id
|
||
)
|
||
db.session.add(feedback)
|
||
db.session.commit()
|
||
flash('感谢您的反馈！')
|
||
return redirect(url_for('main.show_prompt', prompt_id=prompt.id))
|
||
return render_template('feedback.html', form=form, prompt=prompt)
|
||
|
||
# 添加一个API端点来获取模板详情
|
||
@main_bp.route('/api/template/<int:template_id>')
|
||
def get_template_details(template_id):
|
||
template = PromptTemplate.query.get_or_404(template_id)
|
||
return jsonify({
|
||
'id': template.id,
|
||
'name': template.name,
|
||
'description': template.description,
|
||
'system_prompt': template.system_prompt
|
||
})
|
||
|
||
# 添加删除模板的API路由
|
||
@main_bp.route('/api/templates/<int:template_id>', methods=['DELETE'])
|
||
def delete_template(template_id):
|
||
try:
|
||
# 查找模板
|
||
template = PromptTemplate.query.get_or_404(template_id)
|
||
|
||
# 检查是否是默认模板
|
||
if template.is_default:
|
||
return jsonify({
|
||
'success': False,
|
||
'message': '默认模板不能删除'
|
||
}), 403
|
||
|
||
# 检查权限（可选：如果需要检查用户是否有权限删除）
|
||
# if template.user_id != current_user.id:
|
||
# return jsonify({
|
||
# 'success': False,
|
||
# 'message': '没有权限删除此模板'
|
||
# }), 403
|
||
|
||
# 删除模板
|
||
db.session.delete(template)
|
||
db.session.commit()
|
||
|
||
return jsonify({
|
||
'success': True,
|
||
'message': '模板删除成功'
|
||
})
|
||
|
||
except Exception as e:
|
||
# 回滚事务
|
||
db.session.rollback()
|
||
# 记录错误
|
||
current_app.logger.error(f'删除模板失败: {str(e)}')
|
||
return jsonify({
|
||
'success': False,
|
||
'message': '删除模板失败，请稍后重试'
|
||
}), 500
|
||
|
||
# 添加微信小程序API路由
|
||
@main_bp.route('/api/wx/generate', methods=['POST'])
|
||
def wx_generate_prompt():
|
||
"""微信小程序生成提示词接口"""
|
||
try:
|
||
data = request.get_json()
|
||
if not data or 'input_text' not in data or 'uid' not in data:
|
||
return jsonify({
|
||
'code': 400,
|
||
'message': '缺少必要参数',
|
||
'data': None
|
||
})
|
||
|
||
input_text = data.get('input_text')
|
||
template_id = data.get('template_id')
|
||
wx_user_id = data.get('uid') # 使用uid
|
||
|
||
# 调用大模型生成提示词
|
||
generated_text = generate_with_llm(input_text, template_id)
|
||
|
||
# 保存到数据库
|
||
prompt = Prompt(
|
||
input_text=input_text,
|
||
generated_text=generated_text,
|
||
wx_user_id=wx_user_id, # 使用wx_user_id
|
||
created_at=datetime.utcnow()
|
||
)
|
||
db.session.add(prompt)
|
||
db.session.commit()
|
||
|
||
return jsonify({
|
||
'code': 200,
|
||
'message': 'success',
|
||
'data': {
|
||
'prompt_id': prompt.id,
|
||
'input_text': prompt.input_text,
|
||
'generated_text': prompt.generated_text,
|
||
'created_at': prompt.created_at.strftime('%Y-%m-%d %H:%M:%S')
|
||
}
|
||
})
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"生成提示词失败: {str(e)}")
|
||
return jsonify({
|
||
'code': 500,
|
||
'message': str(e),
|
||
'data': None
|
||
})
|
||
|
||
@main_bp.route('/api/wx/templates', methods=['GET'])
|
||
def wx_get_templates():
|
||
"""获取提示词模板列表"""
|
||
try:
|
||
# 获取筛选参数
|
||
industry = request.args.get('industry')
|
||
profession = request.args.get('profession')
|
||
category = request.args.get('category')
|
||
sub_category = request.args.get('sub_category')
|
||
|
||
# 构建查询
|
||
query = PromptTemplate.query
|
||
|
||
if industry:
|
||
query = query.filter_by(industry=industry)
|
||
if profession:
|
||
query = query.filter_by(profession=profession)
|
||
if category:
|
||
query = query.filter_by(category=category)
|
||
if sub_category:
|
||
query = query.filter_by(sub_category=sub_category)
|
||
|
||
templates = query.all()
|
||
|
||
# 返回模板列表
|
||
return jsonify({
|
||
'code': 200,
|
||
'message': 'success',
|
||
'data': [{
|
||
'id': t.id,
|
||
'name': t.name,
|
||
'description': t.description,
|
||
'category': t.category,
|
||
'industry': t.industry,
|
||
'profession': t.profession,
|
||
'sub_category': t.sub_category,
|
||
'is_default': t.is_default
|
||
} for t in templates]
|
||
})
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"获取模板列表失败: {str(e)}")
|
||
return jsonify({
|
||
'code': 500,
|
||
'message': str(e),
|
||
'data': None
|
||
})
|
||
|
||
@main_bp.route('/api/wx/template/<int:template_id>', methods=['GET'])
|
||
def wx_get_template_detail(template_id):
|
||
"""获取模板详情"""
|
||
try:
|
||
template = PromptTemplate.query.get_or_404(template_id)
|
||
|
||
return jsonify({
|
||
'code': 200,
|
||
'message': 'success',
|
||
'data': {
|
||
'id': template.id,
|
||
'name': template.name,
|
||
'description': template.description,
|
||
'category': template.category,
|
||
'industry': template.industry,
|
||
'profession': template.profession,
|
||
'sub_category': template.sub_category,
|
||
'system_prompt': template.system_prompt,
|
||
'is_default': template.is_default
|
||
}
|
||
})
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"获取模板详情失败: {str(e)}")
|
||
return jsonify({
|
||
'code': 500,
|
||
'message': str(e),
|
||
'data': None
|
||
})
|
||
|
||
@main_bp.route('/api/wx/login', methods=['POST'])
|
||
def wx_login():
|
||
"""微信小程序登录接口"""
|
||
try:
|
||
# 添加调试日志
|
||
print("\n=== 微信登录配置 ===")
|
||
print(f"APPID: {WX_APPID}")
|
||
print(f"SECRET: {WX_SECRET}")
|
||
print("==================\n")
|
||
|
||
data = request.get_json()
|
||
if not data or 'code' not in data:
|
||
return jsonify({
|
||
'code': 400,
|
||
'message': '缺少code参数',
|
||
'data': None
|
||
})
|
||
|
||
code = data.get('code')
|
||
print(f"收到的code: {code}")
|
||
|
||
# 请求微信接口
|
||
wx_url = 'https://api.weixin.qq.com/sns/jscode2session'
|
||
params = {
|
||
'appid': WX_APPID,
|
||
'secret': WX_SECRET,
|
||
'js_code': code,
|
||
'grant_type': 'authorization_code'
|
||
}
|
||
|
||
response = requests.get(wx_url, params=params)
|
||
wx_data = response.json()
|
||
print(f"微信返回数据: {wx_data}")
|
||
|
||
if 'errcode' in wx_data:
|
||
return jsonify({
|
||
'code': 500,
|
||
'message': f"微信登录失败：{wx_data.get('errmsg')}",
|
||
'data': None
|
||
})
|
||
|
||
openid = wx_data.get('openid')
|
||
session_key = wx_data.get('session_key')
|
||
|
||
# 查找或创建微信用户
|
||
wx_user = WxUser.query.filter_by(openid=openid).first()
|
||
if not wx_user:
|
||
wx_user = WxUser(
|
||
openid=openid,
|
||
session_key=session_key
|
||
)
|
||
db.session.add(wx_user)
|
||
db.session.commit()
|
||
else:
|
||
wx_user.session_key = session_key
|
||
wx_user.last_login = datetime.utcnow()
|
||
db.session.commit()
|
||
|
||
# 生成登录态token
|
||
token = hashlib.md5(f'{openid}{int(time.time())}'.encode()).hexdigest()
|
||
|
||
return jsonify({
|
||
'code': 200,
|
||
'message': 'success',
|
||
'data': {
|
||
'token': token,
|
||
'openid': openid,
|
||
'uid': wx_user.id,
|
||
'user_info': {
|
||
'id': wx_user.id,
|
||
'nickname': wx_user.nickname,
|
||
'avatar_url': wx_user.avatar_url,
|
||
'gender': wx_user.gender,
|
||
'phone': wx_user.phone
|
||
}
|
||
}
|
||
})
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"微信登录失败: {str(e)}")
|
||
return jsonify({
|
||
'code': 500,
|
||
'message': str(e),
|
||
'data': None
|
||
})
|
||
|
||
@main_bp.route('/api/wx/update_userinfo', methods=['POST'])
|
||
def wx_update_userinfo():
|
||
"""更新微信用户信息"""
|
||
try:
|
||
data = request.get_json()
|
||
if not data or 'openid' not in data:
|
||
return jsonify({
|
||
'code': 400,
|
||
'message': '缺少必要参数',
|
||
'data': None
|
||
})
|
||
|
||
openid = data.get('openid')
|
||
wx_user = WxUser.query.filter_by(openid=openid).first()
|
||
|
||
if not wx_user:
|
||
return jsonify({
|
||
'code': 404,
|
||
'message': '用户不存在',
|
||
'data': None
|
||
})
|
||
|
||
# 更新用户信息
|
||
if 'nickName' in data:
|
||
wx_user.nickname = data['nickName']
|
||
if 'avatarUrl' in data:
|
||
wx_user.avatar_url = data['avatarUrl']
|
||
if 'gender' in data:
|
||
wx_user.gender = data['gender']
|
||
if 'country' in data:
|
||
wx_user.country = data['country']
|
||
if 'province' in data:
|
||
wx_user.province = data['province']
|
||
if 'city' in data:
|
||
wx_user.city = data['city']
|
||
if 'language' in data:
|
||
wx_user.language = data['language']
|
||
|
||
wx_user.updated_at = datetime.utcnow()
|
||
db.session.commit()
|
||
|
||
return jsonify({
|
||
'code': 200,
|
||
'message': 'success',
|
||
'data': {
|
||
'id': wx_user.id,
|
||
'nickname': wx_user.nickname,
|
||
'avatar_url': wx_user.avatar_url,
|
||
'gender': wx_user.gender,
|
||
'country': wx_user.country,
|
||
'province': wx_user.province,
|
||
'city': wx_user.city,
|
||
'language': wx_user.language,
|
||
'phone': wx_user.phone
|
||
}
|
||
})
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"更新用户信息失败: {str(e)}")
|
||
return jsonify({
|
||
'code': 500,
|
||
'message': str(e),
|
||
'data': None
|
||
})
|
||
|
||
@main_bp.route('/api/wx/prompts', methods=['GET'])
|
||
def wx_get_prompts():
|
||
"""获取用户的提示词历史记录"""
|
||
try:
|
||
# 获取参数
|
||
uid = request.args.get('uid')
|
||
page = request.args.get('page', 1, type=int)
|
||
per_page = request.args.get('per_page', 10, type=int)
|
||
|
||
if not uid:
|
||
return jsonify({
|
||
'code': 400,
|
||
'message': '缺少用户ID',
|
||
'data': None
|
||
})
|
||
|
||
# 查询该用户的所有提示词记录
|
||
query = Prompt.query.filter_by(wx_user_id=uid)\
|
||
.order_by(Prompt.created_at.desc())
|
||
|
||
# 分页
|
||
pagination = query.paginate(page=page, per_page=per_page, error_out=False)
|
||
prompts = pagination.items
|
||
|
||
# 返回数据
|
||
return jsonify({
|
||
'code': 200,
|
||
'message': 'success',
|
||
'data': {
|
||
'prompts': [{
|
||
'id': p.id,
|
||
'input_text': p.input_text,
|
||
'generated_text': p.generated_text,
|
||
'created_at': p.created_at.strftime('%Y-%m-%d %H:%M:%S')
|
||
} for p in prompts],
|
||
'pagination': {
|
||
'total': pagination.total, # 总记录数
|
||
'pages': pagination.pages, # 总页数
|
||
'current_page': page, # 当前页
|
||
'per_page': per_page, # 每页记录数
|
||
'has_next': pagination.has_next, # 是否有下一页
|
||
'has_prev': pagination.has_prev # 是否有上一页
|
||
}
|
||
}
|
||
})
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"获取提示词历史失败: {str(e)}")
|
||
return jsonify({
|
||
'code': 500,
|
||
'message': str(e),
|
||
'data': None
|
||
})
|
||
|
||
@main_bp.route('/api/wx/prompt/<int:prompt_id>', methods=['GET'])
|
||
def wx_get_prompt_detail(prompt_id):
|
||
"""获取提示词详情"""
|
||
try:
|
||
prompt = Prompt.query.get_or_404(prompt_id)
|
||
|
||
# 可以选择性地验证用户身份
|
||
# uid = request.args.get('uid')
|
||
# if str(prompt.wx_user_id) != str(uid):
|
||
# return jsonify({
|
||
# 'code': 403,
|
||
# 'message': '无权访问此记录',
|
||
# 'data': None
|
||
# })
|
||
|
||
return jsonify({
|
||
'code': 200,
|
||
'message': 'success',
|
||
'data': {
|
||
'id': prompt.id,
|
||
'input_text': prompt.input_text,
|
||
'generated_text': prompt.generated_text,
|
||
'created_at': prompt.created_at.strftime('%Y-%m-%d %H:%M:%S'),
|
||
'feedbacks': [{
|
||
'id': f.id,
|
||
'rating': f.rating,
|
||
'comment': f.comment,
|
||
'created_at': f.created_at.strftime('%Y-%m-%d %H:%M:%S')
|
||
} for f in prompt.feedbacks]
|
||
}
|
||
})
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"获取提示词详情失败: {str(e)}")
|
||
return jsonify({
|
||
'code': 500,
|
||
'message': str(e),
|
||
'data': None
|
||
})
|
||
|
||
@main_bp.route('/api/wx/prompts/count', methods=['GET'])
|
||
def wx_get_prompts_count():
|
||
"""获取用户的提示词历史记录数量"""
|
||
try:
|
||
# 获取用户ID
|
||
uid = request.args.get('uid')
|
||
|
||
if not uid:
|
||
return jsonify({
|
||
'code': 400,
|
||
'message': '缺少用户ID',
|
||
'data': None
|
||
})
|
||
|
||
# 查询该用户的提示词记录数量
|
||
count = Prompt.query.filter_by(wx_user_id=uid).count()
|
||
|
||
# 获取今日记录数量
|
||
today = datetime.now().date()
|
||
today_count = Prompt.query.filter_by(wx_user_id=uid)\
|
||
.filter(db.func.date(Prompt.created_at) == today)\
|
||
.count()
|
||
|
||
# 获取本月记录数量
|
||
this_month = today.replace(day=1)
|
||
month_count = Prompt.query.filter_by(wx_user_id=uid)\
|
||
.filter(db.func.date(Prompt.created_at) >= this_month)\
|
||
.count()
|
||
|
||
return jsonify({
|
||
'code': 200,
|
||
'message': 'success',
|
||
'data': {
|
||
'total_count': count, # 总记录数
|
||
'today_count': today_count, # 今日记录数
|
||
'month_count': month_count, # 本月记录数
|
||
'uid': uid
|
||
}
|
||
})
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"获取提示词历史数量失败: {str(e)}")
|
||
return jsonify({
|
||
'code': 500,
|
||
'message': str(e),
|
||
'data': None
|
||
})
|
||
|
||
@main_bp.route('/api/wx/prompt/<int:prompt_id>', methods=['DELETE'])
|
||
def wx_delete_prompt(prompt_id):
|
||
"""删除提示词记录"""
|
||
try:
|
||
# 从 URL 参数或请求体中获取用户ID
|
||
uid = request.args.get('uid') or request.get_json().get('uid')
|
||
if not uid:
|
||
return jsonify({
|
||
'code': 400,
|
||
'message': '缺少用户ID',
|
||
'data': None
|
||
})
|
||
|
||
# 查找记录
|
||
prompt = Prompt.query.get_or_404(prompt_id)
|
||
|
||
# 验证是否是用户自己的记录
|
||
if str(prompt.wx_user_id) != str(uid):
|
||
return jsonify({
|
||
'code': 403,
|
||
'message': '无权删除此记录',
|
||
'data': None
|
||
})
|
||
|
||
# 删除相关的反馈
|
||
Feedback.query.filter_by(prompt_id=prompt_id).delete()
|
||
|
||
# 删除提示词记录
|
||
db.session.delete(prompt)
|
||
db.session.commit()
|
||
|
||
return jsonify({
|
||
'code': 200,
|
||
'message': '删除成功',
|
||
'data': {
|
||
'id': prompt_id
|
||
}
|
||
})
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"删除提示词记录失败: {str(e)}")
|
||
db.session.rollback()
|
||
return jsonify({
|
||
'code': 500,
|
||
'message': str(e),
|
||
'data': None
|
||
})
|
||
|
||
@main_bp.route('/api/wx/prompts/search', methods=['GET'])
|
||
def wx_search_prompts():
|
||
"""搜索提示词接口"""
|
||
try:
|
||
# 获取参数
|
||
uid = request.args.get('uid')
|
||
keyword = request.args.get('keyword', '').strip()
|
||
page = request.args.get('page', 1, type=int)
|
||
per_page = request.args.get('per_page', 10, type=int)
|
||
|
||
if not uid:
|
||
return jsonify({
|
||
'code': 400,
|
||
'message': '缺少用户ID',
|
||
'data': None
|
||
})
|
||
|
||
# 构建查询
|
||
query = Prompt.query.filter_by(wx_user_id=uid)
|
||
|
||
# 如果有关键词，添加搜索条件
|
||
if keyword:
|
||
search_condition = (
|
||
Prompt.input_text.ilike(f'%{keyword}%') | # 搜索输入文本
|
||
Prompt.generated_text.ilike(f'%{keyword}%') # 搜索生成的提示词
|
||
)
|
||
query = query.filter(search_condition)
|
||
|
||
# 按时间倒序排序并分页
|
||
query = query.order_by(Prompt.created_at.desc())
|
||
pagination = query.paginate(page=page, per_page=per_page, error_out=False)
|
||
prompts = pagination.items
|
||
|
||
return jsonify({
|
||
'code': 200,
|
||
'message': 'success',
|
||
'data': {
|
||
'prompts': [{
|
||
'id': p.id,
|
||
'input_text': p.input_text,
|
||
'generated_text': p.generated_text,
|
||
'created_at': p.created_at.strftime('%Y-%m-%d %H:%M:%S')
|
||
} for p in prompts],
|
||
'pagination': {
|
||
'total': pagination.total,
|
||
'pages': pagination.pages,
|
||
'current_page': page,
|
||
'per_page': per_page,
|
||
'has_next': pagination.has_next,
|
||
'has_prev': pagination.has_prev
|
||
}
|
||
}
|
||
})
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"搜索提示词失败: {str(e)}")
|
||
return jsonify({
|
||
'code': 500,
|
||
'message': str(e),
|
||
'data': None
|
||
})
|
||
|
||
@main_bp.route('/api/wx/templates/search', methods=['GET'])
|
||
def wx_search_templates():
|
||
"""搜索提示词模板接口"""
|
||
try:
|
||
# 获取搜索参数
|
||
keyword = request.args.get('keyword', '').strip()
|
||
page = request.args.get('page', 1, type=int)
|
||
per_page = request.args.get('per_page', 10, type=int)
|
||
|
||
# 构建基础查询
|
||
query = PromptTemplate.query
|
||
|
||
# 添加搜索条件
|
||
if keyword:
|
||
search_condition = (
|
||
PromptTemplate.name.ilike(f'%{keyword}%') | # 搜索模板名称
|
||
PromptTemplate.description.ilike(f'%{keyword}%') | # 搜索模板描述
|
||
PromptTemplate.category.ilike(f'%{keyword}%') | # 搜索分类
|
||
PromptTemplate.industry.ilike(f'%{keyword}%') | # 搜索行业
|
||
PromptTemplate.profession.ilike(f'%{keyword}%') | # 搜索职业
|
||
PromptTemplate.system_prompt.ilike(f'%{keyword}%') # 搜索系统提示词
|
||
)
|
||
query = query.filter(search_condition)
|
||
|
||
# 获取筛选参数（可选）
|
||
industry = request.args.get('industry')
|
||
profession = request.args.get('profession')
|
||
category = request.args.get('category')
|
||
|
||
# 添加筛选条件
|
||
if industry:
|
||
query = query.filter_by(industry=industry)
|
||
if profession:
|
||
query = query.filter_by(profession=profession)
|
||
if category:
|
||
query = query.filter_by(category=category)
|
||
|
||
# 按是否默认模板和创建时间排序
|
||
query = query.order_by(PromptTemplate.is_default.desc(),
|
||
PromptTemplate.created_at.desc())
|
||
|
||
# 分页
|
||
pagination = query.paginate(page=page, per_page=per_page, error_out=False)
|
||
templates = pagination.items
|
||
|
||
return jsonify({
|
||
'code': 200,
|
||
'message': 'success',
|
||
'data': {
|
||
'templates': [{
|
||
'id': t.id,
|
||
'name': t.name,
|
||
'description': t.description,
|
||
'system_prompt': t.system_prompt, # 添加system_prompt字段
|
||
'category': t.category,
|
||
'industry': t.industry,
|
||
'profession': t.profession,
|
||
'sub_category': t.sub_category,
|
||
'is_default': t.is_default,
|
||
'created_at': t.created_at.strftime('%Y-%m-%d %H:%M:%S') if t.created_at else None
|
||
} for t in templates],
|
||
'pagination': {
|
||
'total': pagination.total,
|
||
'pages': pagination.pages,
|
||
'current_page': page,
|
||
'per_page': per_page,
|
||
'has_next': pagination.has_next,
|
||
'has_prev': pagination.has_prev
|
||
}
|
||
}
|
||
})
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"搜索模板失败: {str(e)}")
|
||
return jsonify({
|
||
'code': 500,
|
||
'message': str(e),
|
||
'data': None
|
||
})
|
||
|
||
@main_bp.route('/api/wx/templates/intent', methods=['POST'])
|
||
def wx_get_template_by_intent():
|
||
"""根据意图获取提示词模板"""
|
||
try:
|
||
# 获取参数
|
||
data = request.get_json()
|
||
user_input = data.get('input_text', '').strip()
|
||
|
||
# 意图识别系统提示词
|
||
intent_system_prompt = """你是一位出色的意图识别专家。请分析用户输入的意图，并仅返回以下类别之一：
|
||
- 新闻获取
|
||
- 生成图片
|
||
- 网站开发
|
||
- 文案创作
|
||
- 代码开发
|
||
- 数据分析
|
||
- 市场营销
|
||
- 产品设计
|
||
- 其它
|
||
|
||
只返回分类名称，不要其他任何内容。"""
|
||
|
||
# 调用意图识别
|
||
response = client.chat.completions.create(
|
||
model="deepseek-chat",
|
||
messages=[
|
||
{"role": "system", "content": intent_system_prompt},
|
||
{"role": "user", "content": user_input}
|
||
],
|
||
temperature=0.1
|
||
)
|
||
|
||
intent = response.choices[0].message.content.strip()
|
||
|
||
# 根据意图获取对应的模板提示词
|
||
intent_prompts = {
|
||
"新闻获取": "你是一位专业的新闻编辑，擅长整理和总结新闻信息。请帮助用户获取和理解新闻内容，注意：\n1. 确保信息的准确性和时效性\n2. 提供客观中立的视角\n3. 突出重要信息要点\n4. 适当添加背景信息解释",
|
||
|
||
"生成图片": "你是一位专业的图像生成提示词专家，擅长将文字需求转化为详细的图像生成提示词。请注意：\n1. 详细描述图像的视觉元素\n2. 指定图像的风格和氛围\n3. 添加技术参数说明\n4. 包含构图和视角建议",
|
||
|
||
"网站开发": "你是一位专业的网站开发专家，擅长将需求转化为具体的开发方案。请注意：\n1. 明确网站的目标用户和核心功能\n2. 建议合适的技术栈\n3. 考虑性能和安全性要求\n4. 提供响应式设计建议",
|
||
|
||
"文案创作": "你是一位专业的文案创作专家，擅长创作各类营销和品牌文案。请注意：\n1. 确定目标受众和传播渠道\n2. 突出产品/服务的核心价值\n3. 使用适当的语言风格\n4. 注意文案的节奏和结构",
|
||
|
||
"代码开发": "你是一位专业的软件开发工程师，擅长编写高质量的代码。请注意：\n1. 遵循编码规范和最佳实践\n2. 考虑代码的可维护性和扩展性\n3. 注重性能优化\n4. 添加适当的注释和文档",
|
||
|
||
"数据分析": "你是一位专业的数据分析师，擅长数据处理和分析。请注意：\n1. 明确分析目标和范围\n2. 选择合适的分析方法\n3. 关注数据质量和准确性\n4. 提供可操作的洞察建议",
|
||
|
||
"市场营销": "你是一位专业的市场营销专家，擅长制定营销策略。请注意：\n1. 分析目标市场和竞争环境\n2. 制定明确的营销目标\n3. 选择合适的营销渠道\n4. 设计有效的营销活动",
|
||
|
||
"产品设计": "你是一位专业的产品设计师，擅长用户体验和界面设计。请注意：\n1. 理解用户需求和痛点\n2. 遵循设计原则和规范\n3. 注重交互体验\n4. 考虑可实现性",
|
||
|
||
"其它": "你是一位专业的AI助手，擅长理解和解决各类问题。请注意：\n1. 仔细理解用户需求\n2. 提供清晰的解决方案\n3. 使用专业的语言表达\n4. 确保回答的实用性"
|
||
}
|
||
|
||
template_prompt = intent_prompts.get(intent, intent_prompts["其它"])
|
||
|
||
return jsonify({
|
||
'code': 200,
|
||
'message': 'success',
|
||
'data': {
|
||
'intent': intent,
|
||
'template_prompt': template_prompt
|
||
}
|
||
})
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"获取意图模板失败: {str(e)}")
|
||
return jsonify({
|
||
'code': 500,
|
||
'message': str(e),
|
||
'data': None
|
||
})
|
||
|
||
@main_bp.route('/api/wx/generate/expert', methods=['POST'])
|
||
def wx_generate_expert_prompt():
|
||
"""两阶段专家提示词生成系统"""
|
||
try:
|
||
# 检查请求数据
|
||
if not request.is_json:
|
||
return jsonify({
|
||
'code': 400,
|
||
'message': '请求必须是JSON格式',
|
||
'data': None
|
||
})
|
||
|
||
data = request.get_json()
|
||
if not data:
|
||
return jsonify({
|
||
'code': 400,
|
||
'message': '请求数据为空',
|
||
'data': None
|
||
})
|
||
|
||
# 验证必要参数
|
||
user_input = data.get('input_text')
|
||
uid = data.get('uid')
|
||
|
||
if not user_input or not uid:
|
||
return jsonify({
|
||
'code': 400,
|
||
'message': '缺少必要参数：input_text 或 uid',
|
||
'data': None
|
||
})
|
||
|
||
user_input = user_input.strip()
|
||
|
||
# 修改第一阶段：意图识别专家的提示词，使其更严格
|
||
intent_analyst_prompt = """你是一位资深的意图分析专家，请分析用户输入的意图和需求。
|
||
|
||
你必须严格按照以下JSON格式返回，不要添加任何其他内容：
|
||
{
|
||
"core_intent": "技术", // 必须是以下选项之一：技术、创意、分析、咨询
|
||
"domain": "web开发", // 具体的专业领域
|
||
"key_requirements": [ // 2-4个关键需求
|
||
"需求1",
|
||
"需求2"
|
||
],
|
||
"expected_output": "期望输出的具体形式", // 简短描述
|
||
"constraints": [ // 1-3个主要约束
|
||
"约束1",
|
||
"约束2"
|
||
],
|
||
"keywords": [ // 2-4个关键词
|
||
"关键词1",
|
||
"关键词2"
|
||
]
|
||
}
|
||
|
||
注意：
|
||
1. 严格遵守JSON格式
|
||
2. core_intent必须是四个选项之一
|
||
3. 数组至少包含1个元素
|
||
4. 所有字段都必须存在
|
||
5. 不要包含注释
|
||
6. 不要添加任何额外的文本"""
|
||
|
||
try:
|
||
# 获取意图分析结果
|
||
intent_response = client.chat.completions.create(
|
||
model="deepseek-chat",
|
||
messages=[
|
||
{"role": "system", "content": intent_analyst_prompt},
|
||
{"role": "user", "content": user_input}
|
||
],
|
||
temperature=0.1 # 降低温度，使输出更确定
|
||
)
|
||
|
||
intent_analysis_text = intent_response.choices[0].message.content.strip()
|
||
|
||
# 添加日志记录
|
||
current_app.logger.info(f"AI返回的意图分析结果: {intent_analysis_text}")
|
||
|
||
# 尝试清理和解析JSON
|
||
try:
|
||
# 移除可能的markdown代码块标记
|
||
intent_analysis_text = intent_analysis_text.replace('```json', '').replace('```', '').strip()
|
||
intent_analysis = json.loads(intent_analysis_text)
|
||
|
||
# 验证必要字段
|
||
required_fields = ['core_intent', 'domain', 'key_requirements',
|
||
'expected_output', 'constraints', 'keywords']
|
||
for field in required_fields:
|
||
if field not in intent_analysis:
|
||
raise ValueError(f"缺少必要字段: {field}")
|
||
|
||
# 验证core_intent是否为有效值
|
||
valid_intents = ['技术', '创意', '分析', '咨询']
|
||
if intent_analysis['core_intent'] not in valid_intents:
|
||
intent_analysis['core_intent'] = '技术' # 默认使用技术
|
||
|
||
# 确保数组字段非空
|
||
array_fields = ['key_requirements', 'constraints', 'keywords']
|
||
for field in array_fields:
|
||
if not isinstance(intent_analysis[field], list) or len(intent_analysis[field]) == 0:
|
||
intent_analysis[field] = ['未指定']
|
||
|
||
except json.JSONDecodeError as e:
|
||
current_app.logger.error(f"JSON解析失败: {str(e)}, 原始文本: {intent_analysis_text}")
|
||
return jsonify({
|
||
'code': 500,
|
||
'message': 'AI返回的格式有误，请重试',
|
||
'data': None
|
||
})
|
||
except ValueError as e:
|
||
current_app.logger.error(f"数据验证失败: {str(e)}")
|
||
return jsonify({
|
||
'code': 500,
|
||
'message': str(e),
|
||
'data': None
|
||
})
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"意图分析失败: {str(e)}")
|
||
return jsonify({
|
||
'code': 500,
|
||
'message': '意图分析过程出错，请重试',
|
||
'data': None
|
||
})
|
||
|
||
# 第二阶段：领域专家提示生成
|
||
domain_expert_templates = {
|
||
"技术": """你是一位专业的技术领域提示工程师。基于以下意图分析，生成一个专业的技术任务提示词：
|
||
|
||
意图分析：
|
||
{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. 效果评估标准
|
||
|
||
使用专业咨询术语，确保提示词的专业性和实用性。"""
|
||
}
|
||
|
||
# 选择领域专家模板
|
||
expert_prompt = domain_expert_templates.get(
|
||
intent_analysis['core_intent'],
|
||
"""你是一位专业的通用领域提示工程师。基于以下意图分析，生成一个专业的提示词：
|
||
|
||
意图分析：
|
||
{analysis}
|
||
|
||
请生成的提示词包含：
|
||
1. 明确的目标定义
|
||
2. 具体要求和规范
|
||
3. 质量标准
|
||
4. 约束条件
|
||
5. 预期输出
|
||
6. 评估标准
|
||
|
||
确保提示词的清晰性和可执行性。"""
|
||
)
|
||
|
||
try:
|
||
# 生成最终提示词
|
||
final_response = client.chat.completions.create(
|
||
model="deepseek-chat",
|
||
messages=[
|
||
{"role": "system", "content": expert_prompt.format(
|
||
analysis=json.dumps(intent_analysis, ensure_ascii=False, indent=2)
|
||
)},
|
||
{"role": "user", "content": user_input}
|
||
],
|
||
temperature=0.7
|
||
)
|
||
|
||
generated_prompt = final_response.choices[0].message.content.strip()
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"生成提示词失败: {str(e)}")
|
||
return jsonify({
|
||
'code': 500,
|
||
'message': '生成提示词过程出错',
|
||
'data': None
|
||
})
|
||
|
||
try:
|
||
# 保存到数据库
|
||
prompt = Prompt(
|
||
input_text=user_input,
|
||
generated_text=generated_prompt,
|
||
wx_user_id=uid,
|
||
#intent_analysis=json.dumps(intent_analysis, ensure_ascii=False),
|
||
created_at=datetime.utcnow()
|
||
)
|
||
db.session.add(prompt)
|
||
db.session.commit()
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"保存到数据库失败: {str(e)}")
|
||
db.session.rollback()
|
||
# 即使保存失败，也返回生成的结果
|
||
|
||
return jsonify({
|
||
'code': 200,
|
||
'message': 'success',
|
||
'data': {
|
||
'prompt_id': prompt.id if 'prompt' in locals() else None,
|
||
'intent_analysis': intent_analysis,
|
||
'generated_prompt': generated_prompt,
|
||
'created_at': prompt.created_at.strftime('%Y-%m-%d %H:%M:%S') if 'prompt' in locals() else None
|
||
}
|
||
})
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"生成专家提示词失败: {str(e)}")
|
||
return jsonify({
|
||
'code': 500,
|
||
'message': str(e),
|
||
'data': None
|
||
})
|
||
|
||
@main_bp.route('/expert_generate')
|
||
def expert_generate():
|
||
"""专家提示词生成页面"""
|
||
return render_template('expert_generate.html')
|
||
|
||
# ... 其他路由保持不变，但要把 @app 改成 @main_bp ... |