You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
34 KiB
34 KiB
AI 提示词系统服务
文档版本：v1.1
最后更新：2026-02-03
变更说明：修复技术栈合规性问题（日志格式统一）
符合规范：jointo-tech-stack v1.0
技术规范说明
本文档遵循 jointo-tech-stack 规范：
- UUID 规范：所有 UUID 字段使用 UUID v7（应用层生成），符合 ADR 001 规范
- 时间戳规范：所有时间字段使用 TIMESTAMPTZ 类型，符合 ADR 006 规范
- 枚举类型：使用 SMALLINT 存储，Python 使用 IntEnum
- 无物理外键：应用层保证引用完整性
- 日志格式：使用 %-formatting，错误日志包含 exc_info=True
- 异步编程：所有数据库操作使用 async/await
目录
服务概述
AI 提示词系统服务负责管理系统级的 AI 提示词模板，为各种 AI 生成场景提供统一的提示词管理。
职责
- 提示词 CRUD 操作（系统管理员专用）
- 提示词版本管理
- Skills 配置管理
- 提示词查询和获取
- 默认提示词管理
设计原则
- 系统级管理：仅供系统管理员维护，不对普通用户开放
- 版本控制：支持提示词版本管理和回滚
- Skills 集成：支持配置多个 skill 文件
- 类型分类：按使用场景分类（剧本、分镜、角色、场景、道具等）
- 应用层验证：遵循 jointo-tech-stack 规范，无物理外键约束
核心功能
1. 提示词管理
- 创建提示词：创建新的提示词模板
- 更新提示词：更新提示词内容和配置
- 删除提示词：禁用提示词（is_active = false）
- 查询提示词：按类型、版本、状态查询
2. 版本管理
- 创建新版本：基于现有提示词创建新版本
- 版本激活：激活指定版本的提示词
- 版本历史：查看提示词的所有版本
- 版本对比：对比不同版本的差异
3. Skills 配置
- 配置 Skills：为提示词配置使用的 skill 文件
- Skills 验证：验证 skill 文件路径有效性
- Skills 优先级：设置 skills 的加载顺序
4. 默认提示词
- 设置默认：为每种类型设置默认提示词
- 获取默认：根据类型获取默认提示词
- 回退机制：当指定提示词不存在时使用默认提示词
数据库设计
ai_prompts_system 表结构
-- Python 枚举定义（app/models/ai_prompt_system.py）
-- class PromptType(IntEnum):
-- SCREENPLAY = 1 # 剧本解析
-- STORYBOARD = 2 # 分镜生成
-- CHARACTER = 3 # 角色生成
-- SCENE = 4 # 场景生成
-- PROP = 5 # 道具生成
-- RESOURCE = 6 # 资源生成
-- GENERAL = 7 # 通用
CREATE TABLE ai_prompts_system (
prompt_id UUID PRIMARY KEY,
name TEXT NOT NULL,
prompt_type SMALLINT NOT NULL, -- 应用层校验：1-7，便于后续扩展
prompt_content TEXT NOT NULL,
skills_data JSONB NOT NULL DEFAULT '{"skills": []}',
version TEXT NOT NULL,
description TEXT,
is_active BOOLEAN NOT NULL DEFAULT true,
is_default BOOLEAN NOT NULL DEFAULT false,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
-- 唯一约束：同一名称的不同版本
CONSTRAINT ai_prompts_system_name_version_unique UNIQUE (name, version) NULLS NOT DISTINCT,
-- 唯一约束：每种类型只能有一个默认提示词
CONSTRAINT ai_prompts_system_default_unique UNIQUE (prompt_type, is_default)
WHERE is_default = true AND is_active = true
);
-- 索引
CREATE INDEX idx_ai_prompts_system_type ON ai_prompts_system (prompt_type);
CREATE INDEX idx_ai_prompts_system_is_active ON ai_prompts_system (is_active) WHERE is_active = true;
CREATE INDEX idx_ai_prompts_system_is_default ON ai_prompts_system (is_default) WHERE is_default = true;
CREATE INDEX idx_ai_prompts_system_name ON ai_prompts_system (name);
CREATE INDEX idx_ai_prompts_system_type_active ON ai_prompts_system (prompt_type, is_active)
WHERE is_active = true;
CREATE INDEX idx_ai_prompts_system_skills_data_gin ON ai_prompts_system USING GIN (skills_data);
CREATE INDEX idx_ai_prompts_system_name_trgm ON ai_prompts_system USING GIN (name gin_trgm_ops);
-- 触发器
CREATE TRIGGER update_ai_prompts_system_updated_at
BEFORE UPDATE ON ai_prompts_system
FOR EACH ROW
EXECUTE FUNCTION update_updated_at_column();
-- 列注释
COMMENT ON TABLE ai_prompts_system IS 'AI 提示词系统表（系统级管理）';
COMMENT ON COLUMN ai_prompts_system.prompt_id IS '提示词唯一标识符（UUID v7）';
COMMENT ON COLUMN ai_prompts_system.name IS '提示词名称';
COMMENT ON COLUMN ai_prompts_system.prompt_type IS '提示词类型：1=剧本 2=分镜 3=角色 4=场景 5=道具 6=资源 7=通用';
COMMENT ON COLUMN ai_prompts_system.prompt_content IS '提示词内容';
COMMENT ON COLUMN ai_prompts_system.skills_data IS 'Skills 配置（JSONB 数组）';
COMMENT ON COLUMN ai_prompts_system.version IS '版本号（如 1.0.0）';
COMMENT ON COLUMN ai_prompts_system.description IS '提示词说明';
COMMENT ON COLUMN ai_prompts_system.is_active IS '是否启用';
COMMENT ON COLUMN ai_prompts_system.is_default IS '是否为默认提示词';
COMMENT ON COLUMN ai_prompts_system.created_at IS '创建时间';
COMMENT ON COLUMN ai_prompts_system.updated_at IS '最后更新时间';
skills_data 结构
{
"skills": [
{
"skill_name": "jointo-tech-stack",
"file_path": ".claude/skills/jointo-tech-stack/references/database.md",
"url": "https://example.com/skills/jointo-tech-stack",
"version": "1.0",
"enabled": true,
"priority": 1,
"description": "项目技术栈规范"
},
{
"skill_name": "screenplay-parser",
"file_path": ".claude/skills/screenplay-parser/SKILL.md",
"url": null,
"version": "2.0",
"enabled": true,
"priority": 2,
"description": "剧本解析规则"
}
]
}
关联表改造
storyboard_images 表
-- 添加 ai_prompt_id 字段
ALTER TABLE storyboard_images
ADD COLUMN ai_prompt_id UUID;
CREATE INDEX idx_storyboard_images_ai_prompt_id ON storyboard_images (ai_prompt_id)
WHERE ai_prompt_id IS NOT NULL;
COMMENT ON COLUMN storyboard_images.ai_prompt_id IS 'AI 提示词 ID（应用层验证，可选）';
设计说明
- UUID v7 主键：应用层生成，符合 jointo-tech-stack 规范
- 无物理外键：应用层保证引用完整性
- 版本管理：name + version 唯一约束，支持多版本共存
- 默认提示词：每种类型只能有一个激活的默认提示词
- Skills 配置：JSONB 数组，支持多个 skill 组合
- 软禁用：使用 is_active 字段，不物理删除
- 全文搜索：使用 pg_trgm 扩展支持提示词名称模糊搜索
服务实现
AIPromptSystemService 类
# app/services/ai_prompt_system_service.py
from typing import List, Optional, Dict, Any
from uuid import UUID
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.ai_prompt_system import AIPromptSystem, PromptType
from app.repositories.ai_prompt_system_repository import AIPromptSystemRepository
from app.schemas.ai_prompt_system import (
AIPromptSystemCreate,
AIPromptSystemUpdate,
AIPromptSystemResponse
)
from app.core.exceptions import NotFoundError, ValidationError
from app.core.logging import get_logger
logger = get_logger(__name__)
class AIPromptSystemService:
def __init__(self, db: AsyncSession):
self.repository = AIPromptSystemRepository(db)
self.db = db
async def create_prompt(
self,
prompt_data: AIPromptSystemCreate
) -> AIPromptSystem:
"""创建提示词"""
logger.info(
"创建提示词: 名称=%s, 类型=%s, 版本=%s",
prompt_data.name, prompt_data.prompt_type, prompt_data.version
)
try:
# 1. 验证 skills_data 格式
self._validate_skills_data(prompt_data.skills_data)
# 2. 检查名称+版本是否已存在
existing = await self.repository.get_by_name_version(
prompt_data.name,
prompt_data.version
)
if existing:
raise ValidationError(
f"提示词 {prompt_data.name} 版本 {prompt_data.version} 已存在"
)
# 3. 如果设置为默认，取消同类型其他默认提示词
if prompt_data.is_default:
await self._unset_default_prompts(prompt_data.prompt_type)
# 4. 创建提示词
prompt = AIPromptSystem(
name=prompt_data.name,
prompt_type=prompt_data.prompt_type,
prompt_content=prompt_data.prompt_content,
skills_data=prompt_data.skills_data,
version=prompt_data.version,
description=prompt_data.description,
is_active=prompt_data.is_active,
is_default=prompt_data.is_default
)
created_prompt = await self.repository.create(prompt)
logger.info(
"提示词创建成功: ID=%s, 名称=%s",
created_prompt.prompt_id, created_prompt.name
)
return created_prompt
except ValidationError:
raise
except Exception as e:
logger.error(
"创建提示词失败: 名称=%s, 错误=%s",
prompt_data.name, str(e),
exc_info=True
)
raise
async def get_prompt_by_id(
self,
prompt_id: UUID
) -> AIPromptSystem:
"""获取提示词详情"""
logger.info("获取提示词详情: ID=%s", prompt_id)
prompt = await self.repository.get_by_id(prompt_id)
if not prompt:
raise NotFoundError(f"提示词不存在: {prompt_id}")
return prompt
async def get_default_prompt(
self,
prompt_type: PromptType
) -> Optional[AIPromptSystem]:
"""获取默认提示词"""
logger.info("获取默认提示词: 类型=%s", prompt_type)
prompt = await self.repository.get_default_by_type(prompt_type)
if not prompt:
logger.warning("未找到默认提示词: 类型=%s", prompt_type)
return prompt
async def list_prompts(
self,
prompt_type: Optional[PromptType] = None,
is_active: Optional[bool] = None,
name: Optional[str] = None,
page: int = 1,
page_size: int = 20
) -> Dict[str, Any]:
"""获取提示词列表"""
logger.info(
"获取提示词列表: 类型=%s, 激活=%s, 名称=%s, 页码=%d",
prompt_type, is_active, name, page
)
try:
prompts = await self.repository.list_prompts(
prompt_type=prompt_type,
is_active=is_active,
name=name,
page=page,
page_size=page_size
)
total = await self.repository.count_prompts(
prompt_type=prompt_type,
is_active=is_active,
name=name
)
logger.info(
"提示词列表获取成功: 总数=%d, 当前页=%d",
total, page
)
return {
'items': prompts,
'total': total,
'page': page,
'page_size': page_size,
'total_pages': (total + page_size - 1) // page_size
}
except Exception as e:
logger.error(
"获取提示词列表失败: 错误=%s",
str(e),
exc_info=True
)
raise
async def update_prompt(
self,
prompt_id: UUID,
prompt_data: AIPromptSystemUpdate
) -> AIPromptSystem:
"""更新提示词"""
logger.info("更新提示词: ID=%s", prompt_id)
try:
# 1. 检查提示词是否存在
prompt = await self.repository.get_by_id(prompt_id)
if not prompt:
raise NotFoundError(f"提示词不存在: {prompt_id}")
# 2. 验证 skills_data 格式（如果提供）
if prompt_data.skills_data is not None:
self._validate_skills_data(prompt_data.skills_data)
# 3. 如果设置为默认，取消同类型其他默认提示词
if prompt_data.is_default and not prompt.is_default:
await self._unset_default_prompts(prompt.prompt_type)
# 4. 更新提示词
update_data = prompt_data.model_dump(exclude_unset=True)
updated_prompt = await self.repository.update(prompt_id, update_data)
logger.info(
"提示词更新成功: ID=%s, 名称=%s",
updated_prompt.prompt_id, updated_prompt.name
)
return updated_prompt
except (NotFoundError, ValidationError):
raise
except Exception as e:
logger.error(
"更新提示词失败: ID=%s, 错误=%s",
prompt_id, str(e),
exc_info=True
)
raise
async def delete_prompt(
self,
prompt_id: UUID
) -> None:
"""删除提示词（软删除，设置 is_active = false）"""
logger.info("删除提示词: ID=%s", prompt_id)
try:
# 检查提示词是否存在
prompt = await self.repository.get_by_id(prompt_id)
if not prompt:
raise NotFoundError(f"提示词不存在: {prompt_id}")
# 如果是默认提示词，不允许删除
if prompt.is_default:
raise ValidationError("默认提示词不能删除，请先取消默认设置")
# 软删除
await self.repository.update(prompt_id, {'is_active': False})
logger.info("提示词删除成功: ID=%s", prompt_id)
except (NotFoundError, ValidationError):
raise
except Exception as e:
logger.error(
"删除提示词失败: ID=%s, 错误=%s",
prompt_id, str(e),
exc_info=True
)
raise
async def create_version(
self,
base_prompt_id: UUID,
new_version: str,
prompt_content: Optional[str] = None,
skills_data: Optional[Dict[str, Any]] = None,
description: Optional[str] = None
) -> AIPromptSystem:
"""基于现有提示词创建新版本"""
logger.info(
"创建提示词新版本: 基础ID=%s, 新版本=%s",
base_prompt_id, new_version
)
try:
# 1. 获取基础提示词
base_prompt = await self.repository.get_by_id(base_prompt_id)
if not base_prompt:
raise NotFoundError(f"基础提示词不存在: {base_prompt_id}")
# 2. 检查新版本是否已存在
existing = await self.repository.get_by_name_version(
base_prompt.name,
new_version
)
if existing:
raise ValidationError(
f"提示词 {base_prompt.name} 版本 {new_version} 已存在"
)
# 3. 验证 skills_data 格式（如果提供）
if skills_data is not None:
self._validate_skills_data(skills_data)
# 4. 创建新版本
new_prompt = AIPromptSystem(
name=base_prompt.name,
prompt_type=base_prompt.prompt_type,
prompt_content=prompt_content or base_prompt.prompt_content,
skills_data=skills_data or base_prompt.skills_data,
version=new_version,
description=description or f"基于版本 {base_prompt.version} 创建",
is_active=True,
is_default=False # 新版本默认不是默认提示词
)
created_prompt = await self.repository.create(new_prompt)
logger.info(
"提示词新版本创建成功: ID=%s, 版本=%s",
created_prompt.prompt_id, created_prompt.version
)
return created_prompt
except (NotFoundError, ValidationError):
raise
except Exception as e:
logger.error(
"创建提示词新版本失败: 基础ID=%s, 错误=%s",
base_prompt_id, str(e),
exc_info=True
)
raise
async def get_versions(
self,
name: str
) -> List[AIPromptSystem]:
"""获取提示词的所有版本"""
logger.info("获取提示词版本历史: 名称=%s", name)
versions = await self.repository.get_versions_by_name(name)
logger.info(
"版本历史获取成功: 名称=%s, 版本数=%d",
name, len(versions)
)
return versions
async def set_default(
self,
prompt_id: UUID
) -> AIPromptSystem:
"""设置为默认提示词"""
logger.info("设置默认提示词: ID=%s", prompt_id)
try:
# 1. 检查提示词是否存在
prompt = await self.repository.get_by_id(prompt_id)
if not prompt:
raise NotFoundError(f"提示词不存在: {prompt_id}")
# 2. 检查是否已激活
if not prompt.is_active:
raise ValidationError("只能将激活的提示词设置为默认")
# 3. 取消同类型其他默认提示词
await self._unset_default_prompts(prompt.prompt_type)
# 4. 设置为默认
updated_prompt = await self.repository.update(
prompt_id,
{'is_default': True}
)
logger.info(
"默认提示词设置成功: ID=%s, 类型=%s",
prompt_id, prompt.prompt_type
)
return updated_prompt
except (NotFoundError, ValidationError):
raise
except Exception as e:
logger.error(
"设置默认提示词失败: ID=%s, 错误=%s",
prompt_id, str(e),
exc_info=True
)
raise
# 私有方法
def _validate_skills_data(self, skills_data: Dict[str, Any]) -> None:
"""验证 skills_data 格式"""
if not isinstance(skills_data, dict):
raise ValidationError("skills_data 必须是字典类型")
if 'skills' not in skills_data:
raise ValidationError("skills_data 必须包含 'skills' 字段")
if not isinstance(skills_data['skills'], list):
raise ValidationError("skills_data.skills 必须是数组类型")
for skill in skills_data['skills']:
if not isinstance(skill, dict):
raise ValidationError("每个 skill 必须是字典类型")
required_fields = ['skill_name', 'enabled']
for field in required_fields:
if field not in skill:
raise ValidationError(f"skill 缺少必需字段: {field}")
async def _unset_default_prompts(self, prompt_type: PromptType) -> None:
"""取消同类型其他默认提示词"""
await self.repository.unset_default_by_type(prompt_type)
API 接口
1. 创建提示词
POST /api/v1/admin/ai-prompts
权限：系统管理员
请求体：
{
"name": "剧本解析提示词",
"prompt_type": 1,
"prompt_content": "你是一个专业的剧本解析助手...",
"skills_data": {
"skills": [
{
"skill_name": "jointo-tech-stack",
"file_path": ".claude/skills/jointo-tech-stack/references/database.md",
"url": null,
"version": "1.0",
"enabled": true,
"priority": 1
}
]
},
"version": "1.0.0",
"description": "用于解析剧本的提示词模板",
"is_active": true,
"is_default": true
}
响应：
{
"code": 200,
"message": "Success",
"data": {
"prompt_id": "019d1234-5678-7abc-def0-111111111111",
"name": "剧本解析提示词",
"prompt_type": 1,
"prompt_content": "你是一个专业的剧本解析助手...",
"skills_data": {
"skills": [...]
},
"version": "1.0.0",
"description": "用于解析剧本的提示词模板",
"is_active": true,
"is_default": true,
"created_at": "2026-01-30T10:00:00Z",
"updated_at": "2026-01-30T10:00:00Z"
}
}
2. 获取提示词列表
GET /api/v1/admin/ai-prompts?prompt_type=1&is_active=true&page=1&page_size=20
权限：系统管理员
查询参数：
prompt_type：提示词类型（可选）is_active：是否激活（可选）name：名称搜索（可选）page：页码（默认 1）page_size：每页数量（默认 20）
响应：
{
"code": 200,
"message": "Success",
"data": {
"items": [
{
"prompt_id": "019d1234-5678-7abc-def0-111111111111",
"name": "剧本解析提示词",
"prompt_type": 1,
"version": "1.0.0",
"is_active": true,
"is_default": true,
"created_at": "2026-01-30T10:00:00Z"
}
],
"total": 10,
"page": 1,
"page_size": 20,
"total_pages": 1
}
}
3. 获取提示词详情
GET /api/v1/admin/ai-prompts/{prompt_id}
权限：系统管理员
响应：同创建提示词响应
4. 更新提示词
PATCH /api/v1/admin/ai-prompts/{prompt_id}
权限：系统管理员
请求体：
{
"prompt_content": "更新后的提示词内容...",
"description": "更新说明",
"is_active": true
}
5. 删除提示词
DELETE /api/v1/admin/ai-prompts/{prompt_id}
权限：系统管理员
响应：
{
"code": 200,
"message": "提示词已删除"
}
6. 创建新版本
POST /api/v1/admin/ai-prompts/{prompt_id}/versions
权限：系统管理员
请求体：
{
"new_version": "1.1.0",
"prompt_content": "新版本的提示词内容...",
"description": "版本 1.1.0 更新说明"
}
7. 获取版本历史
GET /api/v1/admin/ai-prompts/versions?name=剧本解析提示词
权限：系统管理员
响应：
{
"code": 200,
"message": "Success",
"data": [
{
"prompt_id": "019d1234-5678-7abc-def0-111111111111",
"name": "剧本解析提示词",
"version": "1.0.0",
"is_active": true,
"is_default": false,
"created_at": "2026-01-30T10:00:00Z"
},
{
"prompt_id": "019d1234-5678-7abc-def0-222222222222",
"name": "剧本解析提示词",
"version": "1.1.0",
"is_active": true,
"is_default": true,
"created_at": "2026-01-30T11:00:00Z"
}
]
}
8. 设置默认提示词
POST /api/v1/admin/ai-prompts/{prompt_id}/set-default
权限：系统管理员
9. 获取默认提示词（公开接口）
GET /api/v1/ai-prompts/default?prompt_type=1
权限：无需认证（内部服务调用）
响应：
{
"code": 200,
"message": "Success",
"data": {
"prompt_id": "019d1234-5678-7abc-def0-111111111111",
"name": "剧本解析提示词",
"prompt_type": 1,
"prompt_content": "你是一个专业的剧本解析助手...",
"skills_data": {
"skills": [...]
},
"version": "1.0.0"
}
}
数据模型
Pydantic Schema
# app/schemas/ai_prompt_system.py
from pydantic import BaseModel, Field, field_validator
from uuid import UUID
from typing import Optional, Dict, Any, List
from datetime import datetime
from enum import IntEnum
class PromptType(IntEnum):
"""提示词类型"""
SCREENPLAY = 1
STORYBOARD = 2
CHARACTER = 3
SCENE = 4
PROP = 5
RESOURCE = 6
GENERAL = 7
class SkillConfig(BaseModel):
"""Skill 配置"""
skill_name: str = Field(..., description="Skill 名称")
file_path: Optional[str] = Field(None, description="文件路径")
url: Optional[str] = Field(None, description="URL")
version: Optional[str] = Field(None, description="版本号")
enabled: bool = Field(True, description="是否启用")
priority: int = Field(1, description="优先级")
description: Optional[str] = Field(None, description="说明")
class SkillsData(BaseModel):
"""Skills 数据"""
skills: List[SkillConfig] = Field(default_factory=list, description="Skills 列表")
class AIPromptSystemBase(BaseModel):
"""提示词基础模型"""
name: str = Field(..., description="提示词名称", min_length=1, max_length=200)
prompt_type: PromptType = Field(..., description="提示词类型")
prompt_content: str = Field(..., description="提示词内容", min_length=1)
skills_data: Dict[str, Any] = Field(
default_factory=lambda: {"skills": []},
description="Skills 配置"
)
version: str = Field(..., description="版本号", min_length=1, max_length=50)
description: Optional[str] = Field(None, description="提示词说明")
is_active: bool = Field(True, description="是否启用")
is_default: bool = Field(False, description="是否为默认提示词")
class AIPromptSystemCreate(AIPromptSystemBase):
"""创建提示词请求"""
pass
class AIPromptSystemUpdate(BaseModel):
"""更新提示词请求"""
prompt_content: Optional[str] = Field(None, description="提示词内容")
skills_data: Optional[Dict[str, Any]] = Field(None, description="Skills 配置")
description: Optional[str] = Field(None, description="提示词说明")
is_active: Optional[bool] = Field(None, description="是否启用")
is_default: Optional[bool] = Field(None, description="是否为默认提示词")
class AIPromptSystemResponse(AIPromptSystemBase):
"""提示词响应"""
prompt_id: UUID = Field(..., description="提示词 ID")
created_at: datetime = Field(..., description="创建时间")
updated_at: datetime = Field(..., description="更新时间")
class Config:
from_attributes = True
class AIPromptSystemListItem(BaseModel):
"""提示词列表项"""
prompt_id: UUID
name: str
prompt_type: PromptType
version: str
is_active: bool
is_default: bool
created_at: datetime
class Config:
from_attributes = True
class CreateVersionRequest(BaseModel):
"""创建版本请求"""
new_version: str = Field(..., description="新版本号")
prompt_content: Optional[str] = Field(None, description="提示词内容")
skills_data: Optional[Dict[str, Any]] = Field(None, description="Skills 配置")
description: Optional[str] = Field(None, description="版本说明")
SQLModel Model
# app/models/ai_prompt_system.py
from sqlmodel import SQLModel, Field, Column
from sqlalchemy import Text, TIMESTAMP, Index
from sqlalchemy.dialects.postgresql import UUID as PG_UUID, JSONB
from uuid import UUID
from datetime import datetime
from typing import Optional, Dict, Any
from enum import IntEnum
class PromptType(IntEnum):
"""提示词类型"""
SCREENPLAY = 1
STORYBOARD = 2
CHARACTER = 3
SCENE = 4
PROP = 5
RESOURCE = 6
GENERAL = 7
class AIPromptSystem(SQLModel, table=True):
"""AI 提示词系统表"""
__tablename__ = "ai_prompts_system"
prompt_id: UUID = Field(
sa_column=Column(PG_UUID(as_uuid=True), primary_key=True)
)
name: str = Field(sa_column=Column(Text, nullable=False))
prompt_type: int = Field(sa_column=Column("prompt_type", nullable=False))
prompt_content: str = Field(sa_column=Column(Text, nullable=False))
skills_data: Dict[str, Any] = Field(
default_factory=lambda: {"skills": []},
sa_column=Column(JSONB, nullable=False, server_default='{"skills": []}')
)
version: str = Field(sa_column=Column(Text, nullable=False))
description: Optional[str] = Field(default=None, sa_column=Column(Text))
is_active: bool = Field(default=True, sa_column=Column("is_active", nullable=False))
is_default: bool = Field(default=False, sa_column=Column("is_default", nullable=False))
created_at: datetime = Field(
sa_column=Column(TIMESTAMP(timezone=True), nullable=False)
)
updated_at: datetime = Field(
sa_column=Column(TIMESTAMP(timezone=True), nullable=False)
)
__table_args__ = (
Index('idx_ai_prompts_system_type', 'prompt_type'),
Index('idx_ai_prompts_system_is_active', 'is_active'),
Index('idx_ai_prompts_system_is_default', 'is_default'),
Index('idx_ai_prompts_system_name', 'name'),
Index('idx_ai_prompts_system_type_active', 'prompt_type', 'is_active'),
)
使用示例
1. AI Service 调用提示词
# app/services/ai_service.py
from app.services.ai_prompt_system_service import AIPromptSystemService
from app.models.ai_prompt_system import PromptType
class AIService:
async def parse_screenplay(
self,
screenplay_id: UUID,
screenplay_content: str,
prompt_id: Optional[UUID] = None
):
"""解析剧本"""
# 1. 获取提示词
prompt_service = AIPromptSystemService(self.db)
if prompt_id:
# 使用指定的提示词
prompt = await prompt_service.get_prompt_by_id(prompt_id)
else:
# 使用默认提示词
prompt = await prompt_service.get_default_prompt(PromptType.SCREENPLAY)
if not prompt:
raise ValidationError("未找到剧本解析的默认提示词")
# 2. 构建完整提示词（包含 skills）
full_prompt = self._build_prompt_with_skills(
prompt.prompt_content,
prompt.skills_data
)
# 3. 调用 AI 模型
result = await self._call_ai_model(full_prompt, screenplay_content)
return result
def _build_prompt_with_skills(
self,
prompt_content: str,
skills_data: Dict[str, Any]
) -> str:
"""构建包含 skills 的完整提示词"""
skills = skills_data.get('skills', [])
enabled_skills = [s for s in skills if s.get('enabled', True)]
# 按优先级排序
enabled_skills.sort(key=lambda x: x.get('priority', 999))
# 构建 skills 引用
skills_text = "\n\n".join([
f"参考文档 {i+1}: {skill['skill_name']}\n"
f"路径: {skill.get('file_path', 'N/A')}"
for i, skill in enumerate(enabled_skills)
])
return f"{prompt_content}\n\n{skills_text}"
2. 管理员创建提示词
# 示例：创建剧本解析提示词
prompt_data = AIPromptSystemCreate(
name="剧本解析提示词",
prompt_type=PromptType.SCREENPLAY,
prompt_content="""
你是一个专业的剧本解析助手。请分析以下剧本内容，提取以下信息：
1. 角色列表（包括角色名、描述、类型）
2. 场景列表（包括场景编号、标题、地点、时间）
3. 道具列表（包括道具名、描述、重要性）
4. 角色标签（如年龄段、状态等）
5. 场景标签（如时代、氛围等）
6. 道具标签（如状态、版本等）
7. 分镜脚本（包括镜号、描述、对白、景别、运镜等）
请以 JSON 格式返回结果。
""",
skills_data={
"skills": [
{
"skill_name": "jointo-tech-stack",
"file_path": ".claude/skills/jointo-tech-stack/references/database.md",
"version": "1.0",
"enabled": True,
"priority": 1,
"description": "数据库设计规范"
}
]
},
version="1.0.0",
description="用于解析剧本的提示词模板",
is_active=True,
is_default=True
)
prompt_service = AIPromptSystemService(db)
prompt = await prompt_service.create_prompt(prompt_data)
3. 创建新版本
# 基于现有提示词创建新版本
new_prompt = await prompt_service.create_version(
base_prompt_id=prompt.prompt_id,
new_version="1.1.0",
prompt_content="优化后的提示词内容...",
description="版本 1.1.0：优化了角色识别逻辑"
)
4. 设置默认提示词
# 将新版本设置为默认
await prompt_service.set_default(new_prompt.prompt_id)
相关文档
变更历史
v1.1 (2026-02-03)
- ✅ 修复日志格式为 %-formatting
- ✅ 统一错误处理，添加 exc_info=True
- ✅ 完善技术规范说明
v1.0 (2026-01-30)
- ✅ 初始版本
- ✅ 系统级提示词管理
- ✅ 版本管理功能
- ✅ Skills 配置支持
- ✅ 默认提示词机制
- ✅ 符合 jointo-tech-stack 规范
- ✅ 无物理外键约束，应用层验证
- ✅ UUID v7 主键
- ✅ SMALLINT 枚举类型
- ✅ 完整的日志记录
文档版本：v1.1
最后更新：2026-02-03
符合规范：jointo-tech-stack v1.0