259 lines
9.2 KiB
Python
259 lines
9.2 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
"""
|
|
提示词构建服务
|
|
负责根据选题信息构建完整的提示词
|
|
"""
|
|
|
|
import logging
|
|
from typing import Dict, Any, Optional, Tuple
|
|
from pathlib import Path
|
|
|
|
from core.config import ConfigManager, GenerateContentConfig, GenerateTopicConfig, PosterConfig
|
|
from utils.prompts import PromptTemplate
|
|
from api.services.prompt_service import PromptService
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class PromptBuilderService:
|
|
"""提示词构建服务类"""
|
|
|
|
def __init__(self, config_manager: ConfigManager, prompt_service: PromptService):
|
|
"""
|
|
初始化提示词构建服务
|
|
|
|
Args:
|
|
config_manager: 配置管理器
|
|
prompt_service: 提示词服务
|
|
"""
|
|
self.config_manager = config_manager
|
|
self.prompt_service = prompt_service
|
|
|
|
def _ensure_content_config(self) -> GenerateContentConfig:
|
|
"""确保内容生成配置已加载"""
|
|
# 按需加载内容生成配置
|
|
if not self.config_manager.load_task_config('content_gen'):
|
|
logger.warning("未找到内容生成配置，将使用默认配置")
|
|
|
|
return self.config_manager.get_config('content_gen', GenerateContentConfig)
|
|
|
|
def _ensure_topic_config(self) -> GenerateTopicConfig:
|
|
"""确保选题生成配置已加载"""
|
|
# 按需加载选题生成配置
|
|
if not self.config_manager.load_task_config('topic_gen'):
|
|
logger.warning("未找到选题生成配置，将使用默认配置")
|
|
|
|
return self.config_manager.get_config('topic_gen', GenerateTopicConfig)
|
|
|
|
def _ensure_poster_config(self) -> PosterConfig:
|
|
"""确保海报生成配置已加载"""
|
|
# 按需加载海报生成配置
|
|
if not self.config_manager.load_task_config('poster_gen'):
|
|
logger.warning("未找到海报生成配置，将使用默认配置")
|
|
|
|
return self.config_manager.get_config('poster_gen', PosterConfig)
|
|
|
|
def build_content_prompt(self, topic: Dict[str, Any], step: str = "content") -> Tuple[str, str]:
|
|
"""
|
|
构建内容生成提示词
|
|
|
|
Args:
|
|
topic: 选题信息
|
|
step: 当前步骤，用于过滤参考内容
|
|
|
|
Returns:
|
|
系统提示词和用户提示词的元组
|
|
"""
|
|
# 获取内容生成配置
|
|
content_config = self._ensure_content_config()
|
|
|
|
# 加载系统提示词和用户提示词模板
|
|
system_prompt_path = content_config.content_system_prompt
|
|
user_prompt_path = content_config.content_user_prompt
|
|
|
|
# 创建提示词模板
|
|
template = PromptTemplate(system_prompt_path, user_prompt_path)
|
|
|
|
# 获取风格内容
|
|
style_filename = topic.get("style", "")
|
|
style_content = self.prompt_service.get_style_content(style_filename)
|
|
|
|
# 获取目标受众内容
|
|
demand_filename = topic.get("target_audience", "")
|
|
demand_content = self.prompt_service.get_audience_content(demand_filename)
|
|
|
|
# 获取景区信息
|
|
object_name = topic.get("object", "")
|
|
object_content = self.prompt_service.get_scenic_spot_info(object_name)
|
|
|
|
# 获取产品信息
|
|
product_name = topic.get("product", "")
|
|
product_content = self.prompt_service.get_product_info(product_name)
|
|
|
|
# 获取参考内容
|
|
refer_content = self.prompt_service.get_refer_content(step)
|
|
|
|
# 构建系统提示词
|
|
system_prompt = template.get_system_prompt()
|
|
|
|
# 构建用户提示词
|
|
user_prompt = template.build_user_prompt(
|
|
style_content=f"{style_filename}\n{style_content}",
|
|
demand_content=f"{demand_filename}\n{demand_content}",
|
|
object_content=f"{object_name}\n{object_content}",
|
|
product_content=f"{product_name}\n{product_content}",
|
|
refer_content=refer_content
|
|
)
|
|
|
|
return system_prompt, user_prompt
|
|
|
|
def build_poster_prompt(self, topic: Dict[str, Any], content: Dict[str, Any]) -> Tuple[str, str]:
|
|
"""
|
|
构建海报生成提示词
|
|
|
|
Args:
|
|
topic: 选题信息
|
|
content: 生成的内容
|
|
|
|
Returns:
|
|
系统提示词和用户提示词的元组
|
|
"""
|
|
# 获取海报生成配置
|
|
poster_config = self._ensure_poster_config()
|
|
|
|
# 从配置中获取海报提示词模板路径
|
|
system_prompt_path = poster_config.poster_system_prompt
|
|
user_prompt_path = poster_config.poster_user_prompt
|
|
|
|
if not system_prompt_path or not user_prompt_path:
|
|
raise ValueError("海报提示词模板路径不完整")
|
|
|
|
# 创建提示词模板
|
|
template = PromptTemplate(system_prompt_path, user_prompt_path)
|
|
|
|
# 获取景区信息
|
|
object_name = topic.get("object", "")
|
|
object_content = self.prompt_service.get_scenic_spot_info(object_name)
|
|
|
|
# 构建系统提示词
|
|
system_prompt = template.get_system_prompt()
|
|
|
|
# 构建用户提示词
|
|
user_prompt = template.build_user_prompt(
|
|
content=content.get("content", ""),
|
|
title=content.get("title", ""),
|
|
object_name=object_name,
|
|
object_content=object_content
|
|
)
|
|
|
|
return system_prompt, user_prompt
|
|
|
|
def build_topic_prompt(self, num_topics: int, month: str) -> Tuple[str, str]:
|
|
"""
|
|
构建选题生成提示词
|
|
|
|
Args:
|
|
num_topics: 要生成的选题数量
|
|
month: 月份
|
|
|
|
Returns:
|
|
系统提示词和用户提示词的元组
|
|
"""
|
|
# 获取选题生成配置
|
|
topic_config = self._ensure_topic_config()
|
|
|
|
system_prompt_path = topic_config.topic_system_prompt
|
|
user_prompt_path = topic_config.topic_user_prompt
|
|
|
|
if not system_prompt_path or not user_prompt_path:
|
|
raise ValueError("选题提示词模板路径不完整")
|
|
|
|
# 创建提示词模板
|
|
template = PromptTemplate(system_prompt_path, user_prompt_path)
|
|
|
|
# 获取风格列表
|
|
styles = self.prompt_service.get_all_styles()
|
|
style_content = "Style文件列表:\n" + "\n".join([f"- {style['name']}" for style in styles])
|
|
|
|
# 获取目标受众列表
|
|
audiences = self.prompt_service.get_all_audiences()
|
|
demand_content = "Demand文件列表:\n" + "\n".join([f"- {audience['name']}" for audience in audiences])
|
|
|
|
# 获取参考内容
|
|
refer_content = self.prompt_service.get_refer_content("topic")
|
|
|
|
# 获取景区信息列表
|
|
spots = self.prompt_service.get_all_scenic_spots()
|
|
object_content = "Object信息:\n" + "\n".join([f"- {spot['name']}" for spot in spots])
|
|
|
|
# 构建系统提示词
|
|
system_prompt = template.get_system_prompt()
|
|
|
|
# 构建创作资料
|
|
creative_materials = (
|
|
f"你拥有的创作资料如下：\n"
|
|
f"{style_content}\n\n"
|
|
f"{demand_content}\n\n"
|
|
f"{refer_content}\n\n"
|
|
f"{object_content}"
|
|
)
|
|
|
|
# 构建用户提示词
|
|
user_prompt = template.build_user_prompt(
|
|
creative_materials=creative_materials,
|
|
num_topics=num_topics,
|
|
month=month
|
|
)
|
|
|
|
return system_prompt, user_prompt
|
|
|
|
def build_judge_prompt(self, topic: Dict[str, Any], content: Dict[str, Any]) -> Tuple[str, str]:
|
|
"""
|
|
构建内容审核提示词
|
|
|
|
Args:
|
|
topic: 选题信息
|
|
content: 生成的内容
|
|
|
|
Returns:
|
|
系统提示词和用户提示词的元组
|
|
"""
|
|
# 获取内容生成配置
|
|
content_config = self._ensure_content_config()
|
|
|
|
# 从配置中获取审核提示词模板路径
|
|
system_prompt_path = content_config.judger_system_prompt
|
|
user_prompt_path = content_config.judger_user_prompt
|
|
|
|
# 创建提示词模板
|
|
template = PromptTemplate(system_prompt_path, user_prompt_path)
|
|
|
|
# 获取景区信息
|
|
object_name = topic.get("object", "")
|
|
object_content = self.prompt_service.get_scenic_spot_info(object_name)
|
|
|
|
# 获取产品信息
|
|
product_name = topic.get("product", "")
|
|
product_content = self.prompt_service.get_product_info(product_name)
|
|
|
|
# 获取参考内容
|
|
refer_content = self.prompt_service.get_refer_content("judge")
|
|
|
|
# 构建系统提示词
|
|
system_prompt = template.get_system_prompt()
|
|
|
|
# 格式化内容
|
|
import json
|
|
tweet_content = json.dumps(content, ensure_ascii=False, indent=4)
|
|
|
|
# 构建用户提示词
|
|
user_prompt = template.build_user_prompt(
|
|
tweet_content=tweet_content,
|
|
object_content=object_content,
|
|
product_content=product_content,
|
|
refer_content=refer_content
|
|
)
|
|
|
|
return system_prompt, user_prompt |