97 lines
3.5 KiB
Python
97 lines
3.5 KiB
Python
import json
|
|
import re
|
|
from copy import deepcopy
|
|
from typing import Any
|
|
|
|
from jsonschema import Draft202012Validator, ValidationError
|
|
|
|
from apps.skills.models import SkillDefinition, SkillVersion
|
|
|
|
VARIABLE_PATTERN = re.compile(r"{{\s*([^{}]+?)\s*}}")
|
|
ALLOWED_PATH = re.compile(r"^input(?:\.[A-Za-z_][A-Za-z0-9_]*)*$")
|
|
|
|
|
|
class SkillValidationError(ValueError):
|
|
pass
|
|
|
|
|
|
def validate_json_schema(schema: dict[str, Any], label: str) -> None:
|
|
if not isinstance(schema, dict) or not schema:
|
|
raise SkillValidationError(f"{label}不能为空")
|
|
try:
|
|
Draft202012Validator.check_schema(schema)
|
|
except Exception as error:
|
|
raise SkillValidationError(f"{label}不是合法的 JSON Schema: {error}") from error
|
|
|
|
|
|
def validate_prompt_template(template: str) -> None:
|
|
if not template.strip():
|
|
raise SkillValidationError("Prompt 模板不能为空")
|
|
for variable in VARIABLE_PATTERN.findall(template):
|
|
if not ALLOWED_PATH.fullmatch(variable.strip()):
|
|
raise SkillValidationError(f"不允许的模板变量: {variable.strip()}")
|
|
|
|
|
|
def validate_skill_version(version: SkillVersion) -> None:
|
|
validate_prompt_template(version.prompt_template)
|
|
validate_json_schema(version.input_schema, "输入 Schema")
|
|
validate_json_schema(version.output_schema, "输出 Schema")
|
|
if version.example_input:
|
|
try:
|
|
Draft202012Validator(version.input_schema).validate(version.example_input)
|
|
except ValidationError as error:
|
|
raise SkillValidationError(f"示例输入不符合输入 Schema: {error.message}") from error
|
|
|
|
|
|
def validate_payload(schema: dict[str, Any], payload: Any, label: str = "数据") -> None:
|
|
try:
|
|
Draft202012Validator(schema).validate(payload)
|
|
except ValidationError as error:
|
|
path = ".".join(str(item) for item in error.absolute_path)
|
|
location = f" ({path})" if path else ""
|
|
raise SkillValidationError(f"{label}不符合 Schema{location}: {error.message}") from error
|
|
|
|
|
|
def _resolve_path(payload: dict[str, Any], path: str) -> Any:
|
|
value: Any = payload
|
|
for part in path.split(".")[1:]:
|
|
if not isinstance(value, dict) or part not in value:
|
|
raise SkillValidationError(f"模板变量不存在: {path}")
|
|
value = value[part]
|
|
return value
|
|
|
|
|
|
def render_prompt(template: str, input_payload: dict[str, Any]) -> str:
|
|
validate_prompt_template(template)
|
|
|
|
def replace(match: re.Match[str]) -> str:
|
|
path = match.group(1).strip()
|
|
value = _resolve_path(input_payload, path)
|
|
if isinstance(value, (dict, list)):
|
|
return json.dumps(value, ensure_ascii=False)
|
|
if value is None:
|
|
return ""
|
|
return str(value)
|
|
|
|
return VARIABLE_PATTERN.sub(replace, template)
|
|
|
|
|
|
def ensure_draft(definition: SkillDefinition, user=None) -> SkillVersion:
|
|
existing = definition.versions.filter(status="draft").first()
|
|
if existing:
|
|
return existing
|
|
source = definition.active_version
|
|
next_version = (
|
|
definition.versions.order_by("-version").values_list("version", flat=True).first() or 0
|
|
) + 1
|
|
return SkillVersion.objects.create(
|
|
definition=definition,
|
|
version=next_version,
|
|
status="draft",
|
|
prompt_template=source.prompt_template if source else "",
|
|
input_schema=deepcopy(source.input_schema) if source else {"type": "object"},
|
|
output_schema=deepcopy(source.output_schema) if source else {"type": "object"},
|
|
example_input=deepcopy(source.example_input) if source else {},
|
|
created_by=user,
|
|
)
|