feat: add versioned skill configuration

backend/apps/skills/services.py

@@ -0,0 +1,96 @@
+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,
+    )