OpenCode技能系统基础：DSL解析器详细方案

我们专注于技能DSL解析器的详细设计方案，包括UML建模和具体实现。 考虑到时间限制，需要高效地设计一个可扩展的解析器，能够处理YAML和Markdown混合的DSL。

步骤：

将使用以下技术栈：

• Pydantic：用于数据验证和设置管理

• PyYAML：用于解析YAML

• Markdown：用于解析Markdown（如果需要提取结构）

• 自定义解析逻辑

由于技能DSL是YAML和Markdown的混合，假设结构如下：

name: skill–name
version: 1.0.0
description: A brief description
author: Your Name
tags:
– tag1
– tag2

inputs:
input1:
type: string
description: "输入描述"
required: true

outputs:
output1:
type: string
description: "输出描述"

steps:
– step1:
action: "action1"
parameters:
param1: value1
– step2:
action: "action2"
parameters:
param2: value2
—
这里是Markdown正文，用于详细描述技能的使用方法、示例等。
也可以包含一些代码块，用于表示技能的步骤或配置。

注意：使用—作为YAML和Markdown的分隔符。

设计思路：

但是，技能DSL需要被机器解析，因此步骤部分必须结构化。因此，我们将步骤定义在YAML部分，而Markdown部分仅作为补充文档。

接下来，将进行UML建模（类图）和详细设计。

OpenCode技能系统基础：DSL解析器详细方案

我将为你提供一个完整、深入的技能DSL解析器设计方案，涵盖UML建模、Schema设计、解析器实现和验证器创建。

一、系统架构设计：UML建模

1. 核心类图设计

#mermaid-svg-YW8G7fpEUeCxfz8l{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-YW8G7fpEUeCxfz8l .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-YW8G7fpEUeCxfz8l .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-YW8G7fpEUeCxfz8l .error-icon{fill:#552222;}#mermaid-svg-YW8G7fpEUeCxfz8l .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-YW8G7fpEUeCxfz8l .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-YW8G7fpEUeCxfz8l .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-YW8G7fpEUeCxfz8l .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-YW8G7fpEUeCxfz8l .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-YW8G7fpEUeCxfz8l .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-YW8G7fpEUeCxfz8l .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-YW8G7fpEUeCxfz8l .marker{fill:#333333;stroke:#333333;}#mermaid-svg-YW8G7fpEUeCxfz8l .marker.cross{stroke:#333333;}#mermaid-svg-YW8G7fpEUeCxfz8l svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-YW8G7fpEUeCxfz8l p{margin:0;}#mermaid-svg-YW8G7fpEUeCxfz8l g.classGroup text{fill:#9370DB;stroke:none;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:10px;}#mermaid-svg-YW8G7fpEUeCxfz8l g.classGroup text .title{font-weight:bolder;}#mermaid-svg-YW8G7fpEUeCxfz8l .nodeLabel,#mermaid-svg-YW8G7fpEUeCxfz8l .edgeLabel{color:#131300;}#mermaid-svg-YW8G7fpEUeCxfz8l .edgeLabel .label rect{fill:#ECECFF;}#mermaid-svg-YW8G7fpEUeCxfz8l .label text{fill:#131300;}#mermaid-svg-YW8G7fpEUeCxfz8l .labelBkg{background:#ECECFF;}#mermaid-svg-YW8G7fpEUeCxfz8l .edgeLabel .label span{background:#ECECFF;}#mermaid-svg-YW8G7fpEUeCxfz8l .classTitle{font-weight:bolder;}#mermaid-svg-YW8G7fpEUeCxfz8l .node rect,#mermaid-svg-YW8G7fpEUeCxfz8l .node circle,#mermaid-svg-YW8G7fpEUeCxfz8l .node ellipse,#mermaid-svg-YW8G7fpEUeCxfz8l .node polygon,#mermaid-svg-YW8G7fpEUeCxfz8l .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-YW8G7fpEUeCxfz8l .divider{stroke:#9370DB;stroke-width:1;}#mermaid-svg-YW8G7fpEUeCxfz8l g.clickable{cursor:pointer;}#mermaid-svg-YW8G7fpEUeCxfz8l g.classGroup rect{fill:#ECECFF;stroke:#9370DB;}#mermaid-svg-YW8G7fpEUeCxfz8l g.classGroup line{stroke:#9370DB;stroke-width:1;}#mermaid-svg-YW8G7fpEUeCxfz8l .classLabel .box{stroke:none;stroke-width:0;fill:#ECECFF;opacity:0.5;}#mermaid-svg-YW8G7fpEUeCxfz8l .classLabel .label{fill:#9370DB;font-size:10px;}#mermaid-svg-YW8G7fpEUeCxfz8l .relation{stroke:#333333;stroke-width:1;fill:none;}#mermaid-svg-YW8G7fpEUeCxfz8l .dashed-line{stroke-dasharray:3;}#mermaid-svg-YW8G7fpEUeCxfz8l .dotted-line{stroke-dasharray:1 2;}#mermaid-svg-YW8G7fpEUeCxfz8l #compositionStart,#mermaid-svg-YW8G7fpEUeCxfz8l .composition{fill:#333333!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-YW8G7fpEUeCxfz8l #compositionEnd,#mermaid-svg-YW8G7fpEUeCxfz8l .composition{fill:#333333!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-YW8G7fpEUeCxfz8l #dependencyStart,#mermaid-svg-YW8G7fpEUeCxfz8l .dependency{fill:#333333!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-YW8G7fpEUeCxfz8l #dependencyStart,#mermaid-svg-YW8G7fpEUeCxfz8l .dependency{fill:#333333!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-YW8G7fpEUeCxfz8l #extensionStart,#mermaid-svg-YW8G7fpEUeCxfz8l .extension{fill:transparent!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-YW8G7fpEUeCxfz8l #extensionEnd,#mermaid-svg-YW8G7fpEUeCxfz8l .extension{fill:transparent!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-YW8G7fpEUeCxfz8l #aggregationStart,#mermaid-svg-YW8G7fpEUeCxfz8l .aggregation{fill:transparent!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-YW8G7fpEUeCxfz8l #aggregationEnd,#mermaid-svg-YW8G7fpEUeCxfz8l .aggregation{fill:transparent!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-YW8G7fpEUeCxfz8l #lollipopStart,#mermaid-svg-YW8G7fpEUeCxfz8l .lollipop{fill:#ECECFF!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-YW8G7fpEUeCxfz8l #lollipopEnd,#mermaid-svg-YW8G7fpEUeCxfz8l .lollipop{fill:#ECECFF!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-YW8G7fpEUeCxfz8l .edgeTerminals{font-size:11px;line-height:initial;}#mermaid-svg-YW8G7fpEUeCxfz8l .classTitleText{text-anchor:middle;font-size:18px;fill:#333;}#mermaid-svg-YW8G7fpEUeCxfz8l .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#mermaid-svg-YW8G7fpEUeCxfz8l .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#mermaid-svg-YW8G7fpEUeCxfz8l :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

uses

uses

creates

has

has many

has many

has

uses

produces

SkillDSLParser

-SkillSchemaValidator validator

-SkillLoader loader

-TemplateEngine templateEngine

+parse(content: str) : Skill

+parse_file(path: Path) : Skill

+validate(content: str) : ValidationResult

+serialize(skill: Skill) : str

SkillSchemaValidator

-SchemaRegistry schemaRegistry

-List<Validator> validators

+validate_yaml(data: Dict) : ValidationResult

+validate_structure(skill: Skill) : ValidationResult

+validate_dependencies(skill: Skill) : ValidationResult

+register_custom_validator(validator: Validator)

SkillLoader

-YAMLLoader yamlLoader

-MarkdownExtractor markdownExtractor

-VariableResolver variableResolver

+load_yaml(content: str) : Dict

+extract_markdown(content: str) : str

+resolve_references(data: Dict, context: Dict) : Dict

Skill

-SkillMetadata metadata

-List<SkillInput> inputs

-List<SkillStep> steps

-Dict<str, any> outputs

-SkillDocumentation documentation

-SkillPermissions permissions

+validate() : bool

+execute(context: Dict) : ExecutionResult

+to_dict() : Dict

SkillMetadata

+String name

+String version

+String description

+String author

+List<String> tags

+DateTime created_at

+DateTime updated_at

+SkillRequirements requirements

SkillStep

+String name

+StepType type

+Dict<str, any> parameters

+String condition

+String template

+List<String> dependencies

+validate() : bool

+render(context: Dict) : Dict

SkillInput

+String name

+InputType type

+String description

+any default_value

+bool required

+List<any> choices

+String validation_pattern

+validate(value: any) : bool

SkillDocumentation

+String overview

+List<Example> examples

+List<Troubleshooting> troubleshooting

+String markdown_content

+generate_markdown() : str

ValidationResult

+bool is_valid

+List<ValidationError> errors

+List<ValidationWarning> warnings

+to_dict() : Dict

+merge(other: ValidationResult)

ValidationError

+String code

+String message

+String path

+Severity severity

+String suggestion

SkillSchema

+Dict<str, any> schema

+String version

+List<String> required_fields

+validate(data: Dict) : ValidationResult

+extend(extension: SchemaExtension)

2. 解析流程序列图

Skill

SkillSchema

SkillSchemaValidator

SkillLoader

SkillDSLParser

User

Skill

SkillSchema

SkillSchemaValidator

SkillLoader

SkillDSLParser

User

#mermaid-svg-qQcKrBsc2zCCNHTW{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-qQcKrBsc2zCCNHTW .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-qQcKrBsc2zCCNHTW .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-qQcKrBsc2zCCNHTW .error-icon{fill:#552222;}#mermaid-svg-qQcKrBsc2zCCNHTW .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-qQcKrBsc2zCCNHTW .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-qQcKrBsc2zCCNHTW .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-qQcKrBsc2zCCNHTW .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-qQcKrBsc2zCCNHTW .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-qQcKrBsc2zCCNHTW .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-qQcKrBsc2zCCNHTW .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-qQcKrBsc2zCCNHTW .marker{fill:#333333;stroke:#333333;}#mermaid-svg-qQcKrBsc2zCCNHTW .marker.cross{stroke:#333333;}#mermaid-svg-qQcKrBsc2zCCNHTW svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-qQcKrBsc2zCCNHTW p{margin:0;}#mermaid-svg-qQcKrBsc2zCCNHTW .actor{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-qQcKrBsc2zCCNHTW text.actor>tspan{fill:black;stroke:none;}#mermaid-svg-qQcKrBsc2zCCNHTW .actor-line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);}#mermaid-svg-qQcKrBsc2zCCNHTW .innerArc{stroke-width:1.5;stroke-dasharray:none;}#mermaid-svg-qQcKrBsc2zCCNHTW .messageLine0{stroke-width:1.5;stroke-dasharray:none;stroke:#333;}#mermaid-svg-qQcKrBsc2zCCNHTW .messageLine1{stroke-width:1.5;stroke-dasharray:2,2;stroke:#333;}#mermaid-svg-qQcKrBsc2zCCNHTW #arrowhead path{fill:#333;stroke:#333;}#mermaid-svg-qQcKrBsc2zCCNHTW .sequenceNumber{fill:white;}#mermaid-svg-qQcKrBsc2zCCNHTW #sequencenumber{fill:#333;}#mermaid-svg-qQcKrBsc2zCCNHTW #crosshead path{fill:#333;stroke:#333;}#mermaid-svg-qQcKrBsc2zCCNHTW .messageText{fill:#333;stroke:none;}#mermaid-svg-qQcKrBsc2zCCNHTW .labelBox{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-qQcKrBsc2zCCNHTW .labelText,#mermaid-svg-qQcKrBsc2zCCNHTW .labelText>tspan{fill:black;stroke:none;}#mermaid-svg-qQcKrBsc2zCCNHTW .loopText,#mermaid-svg-qQcKrBsc2zCCNHTW .loopText>tspan{fill:black;stroke:none;}#mermaid-svg-qQcKrBsc2zCCNHTW .loopLine{stroke-width:2px;stroke-dasharray:2,2;stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);}#mermaid-svg-qQcKrBsc2zCCNHTW .note{stroke:#aaaa33;fill:#fff5ad;}#mermaid-svg-qQcKrBsc2zCCNHTW .noteText,#mermaid-svg-qQcKrBsc2zCCNHTW .noteText>tspan{fill:black;stroke:none;}#mermaid-svg-qQcKrBsc2zCCNHTW .activation0{fill:#f4f4f4;stroke:#666;}#mermaid-svg-qQcKrBsc2zCCNHTW .activation1{fill:#f4f4f4;stroke:#666;}#mermaid-svg-qQcKrBsc2zCCNHTW .activation2{fill:#f4f4f4;stroke:#666;}#mermaid-svg-qQcKrBsc2zCCNHTW .actorPopupMenu{position:absolute;}#mermaid-svg-qQcKrBsc2zCCNHTW .actorPopupMenuPanel{position:absolute;fill:#ECECFF;box-shadow:0px 8px 16px 0px rgba(0,0,0,0.2);filter:drop-shadow(3px 5px 2px rgb(0 0 0 / 0.4));}#mermaid-svg-qQcKrBsc2zCCNHTW .actor-man line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-qQcKrBsc2zCCNHTW .actor-man circle,#mermaid-svg-qQcKrBsc2zCCNHTW line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;stroke-width:2px;}#mermaid-svg-qQcKrBsc2zCCNHTW :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

alt

[结构验证通过]

[结构验证失败]

alt

[验证通过]

[YAML验证失败]

alt

[YAML部分存在]

[无YAML部分]

parse_file("skill.yaml")

load_file("skill.yaml")

raw_content

split_yaml_markdown(raw_content)

load_yaml(yaml_part)

yaml_data

validate_yaml(yaml_data)

validate(yaml_data)

schema_validation_result

validation_result

resolve_references(yaml_data)

resolved_data

build_skill_object(resolved_data)

new Skill(resolved_data)

skill_instance

validate_structure(skill_instance)

structure_validation_result

attach_documentation(markdown_part)

skill_instance

ValidationError

ValidationError

ParseError("No YAML frontmatter found")

二、技能元数据Schema设计

1. 核心Schema定义（基于JSON Schema）

# schema/skill-schema-v1.0.0.yaml
$schema: "http://json-schema.org/draft-07/schema#"
$id: "https://opencode.ai/schemas/skill/v1.0.0"
title: "OpenCode Skill Schema"
description: "Schema for defining OpenCode skills"
type: "object"

definitions:
# 元数据定义
metadata:
type: "object"
properties:
name:
type: "string"
pattern: "^[a-z0-9-_]+$"
minLength: 3
maxLength: 50
description: "Unique identifier for the skill"

version:
type: "string"
pattern: "^\\\\d+\\\\.\\\\d+\\\\.\\\\d+$"
description: "Semantic version of the skill"

description:
type: "string"
minLength: 10
maxLength: 200
description: "Brief description of what the skill does"

author:
type: "string"
description: "Author or maintainer of the skill"

tags:
type: "array"
items:
type: "string"
pattern: "^[a-z0-9-]+$"
uniqueItems: true
maxItems: 10
description: "Keywords for categorization and search"

created_at:
type: "string"
format: "date-time"
description: "Creation timestamp"

updated_at:
type: "string"
format: "date-time"
description: "Last update timestamp"

requirements:
$ref: "#/definitions/requirements"

required: ["name", "version", "description"]
additionalProperties: false

# 依赖要求定义
requirements:
type: "object"
properties:
opencode_version:
type: "string"
pattern: "^\\\\d+\\\\.\\\\d+\\\\.\\\\d+$"
description: "Minimum OpenCode version required"

tools:
type: "array"
items:
type: "object"
properties:
name:
type: "string"
version:
type: "string"
optional:
type: "boolean"
default: false

permissions:
type: "array"
items:
type: "string"
enum: ["file.read", "file.write", "network", "process"]

python_packages:
type: "array"
items:
type: "object"
properties:
name:
type: "string"
version:
type: "string"
optional:
type: "boolean"
default: false

# 输入参数定义
input_schema:
type: "object"
patternProperties:
"^[a-z][a-z0-9_]*$":
type: "object"
properties:
type:
type: "string"
enum: ["string", "integer", "number", "boolean", "array", "object"]

description:
type: "string"
minLength: 5

required:
type: "boolean"
default: true

default:
oneOf:
– {type: "string"}
– {type: "number"}
– {type: "boolean"}
– {type: "array"}
– {type: "object"}

choices:
type: "array"
minItems: 1

validation:
type: "object"
properties:
pattern:
type: "string"
min:
type: "number"
max:
type: "number"
minLength:
type: "integer"
maxLength:
type: "integer"

examples:
type: "array"
items:
type: "object"
properties:
value:
oneOf:
– {type: "string"}
– {type: "number"}
– {type: "boolean"}
description:
type: "string"

required: ["type", "description"]
additionalProperties: false

# 步骤定义
step_schema:
type: "object"
properties:
name:
type: "string"
pattern: "^[a-z][a-z0-9_]*$"

type:
type: "string"
enum: ["command", "script", "template", "condition", "loop", "parallel"]

description:
type: "string"
minLength: 10

condition:
type: "string"
description: "JavaScript-like condition expression"

parameters:
type: "object"
additionalProperties: true

template:
type: "string"
description: "Jinja2 template for content generation"

script:
type: "string"
description: "JavaScript/Python script to execute"

outputs:
type: "object"
patternProperties:
"^[a-z][a-z0-9_]*$":
type: "string"
description: "Output variable mapping"

timeout:
type: "integer"
minimum: 1
maximum: 3600
default: 300

retry:
type: "object"
properties:
attempts:
type: "integer"
minimum: 0
maximum: 5
delay:
type: "integer"
minimum: 1
backoff_factor:
type: "number"
minimum: 1.0

required: ["name", "type"]
additionalProperties: false

# 输出定义
output_schema:
type: "object"
patternProperties:
"^[a-z][a-z0-9_]*$":
type: "object"
properties:
description:
type: "string"
type:
type: "string"
value:
type: "string"
description: "JavaScript expression to compute output"

additionalProperties: false

# 根级Schema
properties:
metadata:
$ref: "#/definitions/metadata"

inputs:
$ref: "#/definitions/input_schema"

steps:
type: "array"
items:
$ref: "#/definitions/step_schema"
minItems: 1

outputs:
$ref: "#/definitions/output_schema"

documentation:
type: "object"
properties:
overview:
type: "string"
examples:
type: "array"
items:
type: "object"
properties:
title:
type: "string"
description:
type: "string"
code:
type: "string"
troubleshooting:
type: "array"
items:
type: "object"
properties:
problem:
type: "string"
solution:
type: "string"

permissions:
type: "object"
properties:
read:
type: "array"
items:
type: "string"
write:
type: "array"
items:
type: "string"
execute:
type: "array"
items:
type: "string"

required: ["metadata", "steps"]
additionalProperties: false

2. Python数据模型（Pydantic）

# models/skill_models.py
from typing import Dict, List, Optional, Any, Union
from enum import Enum
from datetime import datetime
from pydantic import BaseModel, Field, validator, root_validator
import re

class StepType(str, Enum):
COMMAND = "command"
SCRIPT = "script"
TEMPLATE = "template"
CONDITION = "condition"
LOOP = "loop"
PARALLEL = "parallel"
VALIDATION = "validation"
HTTP_REQUEST = "http_request"
DATABASE_QUERY = "database_query"

class InputType(str, Enum):
STRING = "string"
INTEGER = "integer"
NUMBER = "number"
BOOLEAN = "boolean"
ARRAY = "array"
OBJECT = "object"
FILE = "file"
DIRECTORY = "directory"

class Severity(str, Enum):
INFO = "info"
WARNING = "warning"
ERROR = "error"
CRITICAL = "critical"

class SkillRequirements(BaseModel):
"""技能依赖要求"""
opencode_version: Optional[str] = Field(
None,
regex=r"^\\d+\\.\\d+\\.\\d+$",
description="Minimum OpenCode version required"
)
tools: List[Dict[str, Union[str, bool]]] = Field(
default_factory=list,
description="Required external tools"
)
permissions: List[str] = Field(
default_factory=list,
description="Required permissions"
)
python_packages: List[Dict[str, Union[str, bool]]] = Field(
default_factory=list,
description="Required Python packages"
)

@validator('permissions')
def validate_permissions(cls, v):
valid_permissions = {"file.read", "file.write", "network", "process", "env"}
for perm in v:
if perm not in valid_permissions:
raise ValueError(f"Invalid permission: {perm}. Valid permissions: {valid_permissions}")
return v

class SkillMetadata(BaseModel):
"""技能元数据"""
name: str = Field(
...,
min_length=3,
max_length=50,
regex=r"^[a-z0-9-_]+$",
description="Unique identifier for the skill"
)
version: str = Field(
...,
regex=r"^\\d+\\.\\d+\\.\\d+$",
description="Semantic version of the skill"
)
description: str = Field(
...,
min_length=10,
max_length=200,
description="Brief description of what the skill does"
)
author: Optional[str] = Field(None, description="Author or maintainer")
tags: List[str] = Field(
default_factory=list,
max_items=10,
description="Keywords for categorization"
)
created_at: datetime = Field(default_factory=datetime.now)
updated_at: datetime = Field(default_factory=datetime.now)
requirements: SkillRequirements = Field(default_factory=SkillRequirements)

@validator('tags')
def validate_tags(cls, v):
for tag in v:
if not re.match(r"^[a-z0-9-]+$", tag):
raise ValueError(f"Tag '{tag}' must contain only lowercase letters, numbers, and hyphens")
return v

class Config:
json_schema_extra = {
"example": {
"name": "git-release",
"version": "1.0.0",
"description": "Automates semantic versioning and release process",
"author": "OpenCode Team",
"tags": ["git", "release", "automation"]
}
}

class SkillInput(BaseModel):
"""技能输入参数定义"""
name: str = Field(
...,
regex=r"^[a-z][a-z0-9_]*$",
description="Input parameter name"
)
type: InputType = Field(..., description="Data type of the input")
description: str = Field(..., min_length=5, description="Description of the input")
required: bool = Field(True, description="Whether the input is required")
default: Optional[Any] = Field(None, description="Default value if not provided")
choices: Optional[List[Any]] = Field(None, description="Allowed values")
validation: Optional[Dict[str, Any]] = Field(
None,
description="Validation rules (pattern, min, max, etc.)"
)
examples: Optional[List[Dict[str, Any]]] = Field(
None,
description="Example values and descriptions"
)

@validator('default')
def validate_default_type(cls, v, values):
if v is not None and 'type' in values:
expected_type = values['type']
type_checks = {
InputType.STRING: isinstance(v, str),
InputType.INTEGER: isinstance(v, int),
InputType.NUMBER: isinstance(v, (int, float)),
InputType.BOOLEAN: isinstance(v, bool),
InputType.ARRAY: isinstance(v, list),
InputType.OBJECT: isinstance(v, dict),
InputType.FILE: isinstance(v, str),
InputType.DIRECTORY: isinstance(v, str),
}
if not type_checks.get(expected_type, True):
raise ValueError(f"Default value type {type(v)} doesn't match input type {expected_type}")
return v

@root_validator
def validate_choices(cls, values):
choices = values.get('choices')
default = values.get('default')
if choices and default and default not in choices:
raise ValueError(f"Default value {default} not in choices {choices}")
return values

class SkillStep(BaseModel):
"""技能执行步骤"""
name: str = Field(
...,
regex=r"^[a-z][a-z0-9_]*$",
description="Step identifier"
)
type: StepType = Field(..., description="Type of step")
description: Optional[str] = Field(None, min_length=10, description="Step description")
condition: Optional[str] = Field(
None,
description="JavaScript-like condition expression for conditional execution"
)
parameters: Dict[str, Any] = Field(
default_factory=dict,
description="Step-specific parameters"
)
template: Optional[str] = Field(None, description="Jinja2 template for content generation")
script: Optional[str] = Field(None, description="JavaScript/Python script to execute")
outputs: Dict[str, str] = Field(
default_factory=dict,
description="Output variable mappings"
)
timeout: int = Field(
300,
ge=1,
le=3600,
description="Maximum execution time in seconds"
)
retry: Optional[Dict[str, Any]] = Field(
None,
description="Retry configuration"
)
dependencies: List[str] = Field(
default_factory=list,
description="Steps that must complete before this step"
)

@validator('condition')
def validate_condition_syntax(cls, v):
if v:
# Basic validation for condition syntax
# In production, use a proper JavaScript parser
forbidden_keywords = ['import', 'exec', 'eval', '__']
for keyword in forbidden_keywords:
if keyword in v:
raise ValueError(f"Forbidden keyword '{keyword}' in condition")
return v

@validator('retry')
def validate_retry_config(cls, v):
if v:
if 'attempts' not in v:
raise ValueError("Retry config must include 'attempts'")
if v['attempts'] < 0 or v['attempts'] > 5:
raise ValueError("Retry attempts must be between 0 and 5")
return v

class SkillOutput(BaseModel):
"""技能输出定义"""
name: str = Field(
...,
regex=r"^[a-z][a-z0-9_]*$",
description="Output variable name"
)
description: str = Field(..., description="Description of the output")
type: str = Field(..., description="Data type of the output")
value: str = Field(
...,
description="JavaScript expression to compute the output value"
)

class SkillDocumentation(BaseModel):
"""技能文档"""
overview: Optional[str] = Field(None, description="High-level overview")
examples: List[Dict[str, str]] = Field(
default_factory=list,
description="Usage examples"
)
troubleshooting: List[Dict[str, str]] = Field(
default_factory=list,
description="Common problems and solutions"
)
markdown_content: Optional[str] = Field(
None,
description="Full Markdown documentation"
)

def generate_markdown(self) –> str:
"""生成完整的Markdown文档"""
sections = []

if self.overview:
sections.append(f"# Overview\\n\\n{self.overview}")

if self.examples:
sections.append("## Examples")
for i, example in enumerate(self.examples, 1):
sections.append(f"### Example {i}: {example.get('title', '')}")
if 'description' in example:
sections.append(f"{example['description']}\\n")
if 'code' in example:
sections.append(f"```yaml\\n{example['code']}\\n```")

if self.troubleshooting:
sections.append("## Troubleshooting")
for item in self.troubleshooting:
sections.append(f"### {item.get('problem', 'Problem')}")
sections.append(f"{item.get('solution', 'Solution')}")

if self.markdown_content:
sections.append(self.markdown_content)

return "\\n\\n".join(sections)

class SkillPermissions(BaseModel):
"""技能权限要求"""
read: List[str] = Field(default_factory=list, description="Files/directories to read")
write: List[str] = Field(default_factory=list, description="Files/directories to write")
execute: List[str] = Field(default_factory=list, description="Commands to execute")
network: List[str] = Field(default_factory=list, description="Network endpoints to access")

@validator('read', 'write')
def validate_path_patterns(cls, v):
for path in v:
if not re.match(r'^[a-zA-Z0-9_\\-./*?]+$', path):
raise ValueError(f"Invalid path pattern: {path}")
return v

class Skill(BaseModel):
"""完整的技能定义"""
metadata: SkillMetadata
inputs: Dict[str, SkillInput] = Field(default_factory=dict)
steps: List[SkillStep] = Field(..., min_items=1)
outputs: Dict[str, SkillOutput] = Field(default_factory=dict)
documentation: SkillDocumentation = Field(default_factory=SkillDocumentation)
permissions: SkillPermissions = Field(default_factory=SkillPermissions)

@validator('steps')
def validate_step_dependencies(cls, v):
"""验证步骤依赖关系"""
step_names = {step.name for step in v}

for step in v:
for dep in step.dependencies:
if dep not in step_names:
raise ValueError(f"Step '{step.name}' depends on unknown step '{dep}'")

# 检查循环依赖
visited = set()
recursion_stack = set()

def has_cycle(step_name, steps_dict):
visited.add(step_name)
recursion_stack.add(step_name)

step = steps_dict[step_name]
for dep in step.dependencies:
if dep not in visited:
if has_cycle(dep, steps_dict):
return True
elif dep in recursion_stack:
return True

recursion_stack.remove(step_name)
return False

steps_dict = {step.name: step for step in v}
for step in v:
if step.name not in visited:
if has_cycle(step.name, steps_dict):
raise ValueError("Circular dependency detected in steps")

return v

@root_validator
def validate_output_references(cls, values):
"""验证输出引用是否有效"""
steps = values.get('steps', [])
outputs = values.get('outputs', {})

# 收集所有可用的输出变量
available_outputs = set()
for step in steps:
available_outputs.update(step.outputs.keys())

# 验证输出定义中的引用
for output_name, output_def in outputs.items():
# 这里可以添加更复杂的引用解析逻辑
# 暂时只检查基本的语法
if not output_def.value:
raise ValueError(f"Output '{output_name}' must have a value expression")

return values

class Config:
json_schema_extra = {
"example": {
"metadata": {
"name": "example-skill",
"version": "1.0.0",
"description": "An example skill"
},
"steps": [
{
"name": "step1",
"type": "command",
"description": "First step"
}
]
}
}

三、技能DSL解析器实现

1. 主解析器类实现

# parsers/skill_parser.py
import re
import yaml
import json
from pathlib import Path
from typing import Dict, List, Optional, Tuple, Any, Union
from dataclasses import dataclass
from enum import Enum
import jsonschema
from jsonschema import validate, ValidationError
import jinja2
from jinja2.sandbox import SandboxedEnvironment

from models.skill_models import Skill, SkillMetadata, SkillStep, SkillInput, SkillOutput

class ParseError(Exception):
"""解析错误异常"""
def __init__(self, message: str, line: Optional[int] = None, column: Optional[int] = None):
self.message = message
self.line = line
self.column = column
super().__init__(f"{message} (line {line}, column {column})" if line else message)

class ValidationSeverity(str, Enum):
INFO = "info"
WARNING = "warning"
ERROR = "error"
CRITICAL = "critical"

@dataclass
class ValidationIssue:
"""验证问题"""
code: str
message: str
severity: ValidationSeverity
path: str
line: Optional[int] = None
column: Optional[int] = None
suggestion: Optional[str] = None

@dataclass
class ValidationResult:
"""验证结果"""
is_valid: bool
issues: List[ValidationIssue]
schema_version: str

def add_issue(self, issue: ValidationIssue):
self.issues.append(issue)
if issue.severity in [ValidationSeverity.ERROR, ValidationSeverity.CRITICAL]:
self.is_valid = False

def merge(self, other: 'ValidationResult'):
self.issues.extend(other.issues)
self.is_valid = self.is_valid and other.is_valid

def to_dict(self) –> Dict:
return {
"is_valid": self.is_valid,
"schema_version": self.schema_version,
"issues": [
{
"code": i.code,
"message": i.message,
"severity": i.severity,
"path": i.path,
"line": i.line,
"column": i.column,
"suggestion": i.suggestion
}
for i in self.issues
],
"summary": {
"total_issues": len(self.issues),
"errors": len([i for i in self.issues if i.severity == ValidationSeverity.ERROR]),
"warnings": len([i for i in self.issues if i.severity == ValidationSeverity.WARNING]),
"infos": len([i for i in self.issues if i.severity == ValidationSeverity.INFO])
}
}

class SkillSchemaValidator:
"""技能Schema验证器"""

def __init__(self, schema_dir: Optional[Path] = None):
self.schema_dir = schema_dir or Path(__file__).parent / "schemas"
self.schemas = self._load_schemas()
self.custom_validators = []

def _load_schemas(self) –> Dict[str, Dict]:
"""加载所有Schema文件"""
schemas = {}

for schema_file in self.schema_dir.glob("skill-schema-*.yaml"):
try:
with open(schema_file, 'r', encoding='utf-8') as f:
schema = yaml.safe_load(f)
version = schema.get('$id', '').split('/')[–1]
schemas[version] = schema
except Exception as e:
print(f"Warning: Failed to load schema {schema_file}: {e}")

return schemas

def validate_yaml(self, yaml_data: Dict, schema_version: str = "v1.0.0") –> ValidationResult:
"""验证YAML数据是否符合Schema"""
result = ValidationResult(
is_valid=True,
issues=[],
schema_version=schema_version
)

if schema_version not in self.schemas:
result.add_issue(ValidationIssue(
code="SCHEMA_NOT_FOUND",
message=f"Schema version {schema_version} not found",
severity=ValidationSeverity.ERROR,
path="$"
))
return result

schema = self.schemas[schema_version]

try:
# 基础Schema验证
validate(instance=yaml_data, schema=schema)
except ValidationError as e:
result.add_issue(ValidationIssue(
code="SCHEMA_VALIDATION_FAILED",
message=e.message,
severity=ValidationSeverity.ERROR,
path=e.json_path or "$",
line=getattr(e, 'line', None),
column=getattr(e, 'column', None)
))

# 自定义验证规则
self._validate_custom_rules(yaml_data, result)

return result

def _validate_custom_rules(self, data: Dict, result: ValidationResult):
"""应用自定义验证规则"""

# 1. 验证技能名称唯一性模式
if 'metadata' in data and 'name' in data['metadata']:
name = data['metadata']['name']
if not re.match(r'^[a-z0-9-_]+$', name):
result.add_issue(ValidationIssue(
code="INVALID_SKILL_NAME",
message=f"Skill name '{name}' must contain only lowercase letters, numbers, hyphens, and underscores",
severity=ValidationSeverity.ERROR,
path="metadata.name"
))

# 2. 验证版本格式
if 'metadata' in data and 'version' in data['metadata']:
version = data['metadata']['version']
if not re.match(r'^\\d+\\.\\d+\\.\\d+$', version):
result.add_issue(ValidationIssue(
code="INVALID_VERSION_FORMAT",
message=f"Version '{version}' must follow semantic versioning (e.g., 1.0.0)",
severity=ValidationSeverity.ERROR,
path="metadata.version"
))

# 3. 验证步骤名称唯一性
if 'steps' in data:
step_names = []
for i, step in enumerate(data['steps']):
if 'name' in step:
if step['name'] in step_names:
result.add_issue(ValidationIssue(
code="DUPLICATE_STEP_NAME",
message=f"Duplicate step name '{step['name']}'",
severity=ValidationSeverity.ERROR,
path=f"steps[{i}].name"
))
step_names.append(step['name'])

# 4. 验证输入参数名称
if 'inputs' in data:
for input_name in data['inputs'].keys():
if not re.match(r'^[a-z][a-z0-9_]*$', input_name):
result.add_issue(ValidationIssue(
code="INVALID_INPUT_NAME",
message=f"Input name '{input_name}' must start with a letter and contain only lowercase letters, numbers, and underscores",
severity=ValidationSeverity.ERROR,
path=f"inputs.{input_name}"
))

# 5. 验证依赖关系
if 'steps' in data:
all_step_names = {step.get('name', f'step_{i}') for i, step in enumerate(data['steps'])}

for i, step in enumerate(data['steps']):
if 'dependencies' in step:
for dep in step['dependencies']:
if dep not in all_step_names:
result.add_issue(ValidationIssue(
code="UNKNOWN_DEPENDENCY",
message=f"Step '{step.get('name', f'step_{i}')}' depends on unknown step '{dep}'",
severity=ValidationSeverity.ERROR,
path=f"steps[{i}].dependencies"
))

def validate_skill_structure(self, skill: Skill) –> ValidationResult:
"""验证技能结构完整性"""
result = ValidationResult(
is_valid=True,
issues=[],
schema_version="v1.0.0"
)

# 验证步骤执行顺序
self._validate_execution_order(skill, result)

# 验证输入默认值
self._validate_input_defaults(skill, result)

# 验证模板语法
self._validate_templates(skill, result)

# 验证条件表达式
self._validate_conditions(skill, result)

return result

def _validate_execution_order(self, skill: Skill, result: ValidationResult):
"""验证执行顺序"""
# 检查循环依赖已在Skill模型中验证
# 这里可以添加其他执行顺序相关的验证

step_names = [step.name for step in skill.steps]

# 验证输出引用
for output_name, output_def in skill.outputs.items():
# 检查输出表达式是否引用有效的步骤输出
# 这里可以添加更复杂的分析
pass

def _validate_input_defaults(self, skill: Skill, result: ValidationResult):
"""验证输入默认值"""
for input_name, input_def in skill.inputs.items():
if input_def.default is not None:
# 验证默认值类型
try:
input_def.validate(input_def.default)
except ValueError as e:
result.add_issue(ValidationIssue(
code="INVALID_DEFAULT_VALUE",
message=f"Invalid default value for input '{input_name}': {str(e)}",
severity=ValidationSeverity.ERROR,
path=f"inputs.{input_name}.default"
))

# 验证默认值在choices中（如果定义了choices）
if input_def.choices and input_def.default not in input_def.choices:
result.add_issue(ValidationIssue(
code="DEFAULT_NOT_IN_CHOICES",
message=f"Default value '{input_def.default}' not in choices {input_def.choices}",
severity=ValidationSeverity.ERROR,
path=f"inputs.{input_name}.default"
))

def _validate_templates(self, skill: Skill, result: ValidationResult):
"""验证模板语法"""
env = SandboxedEnvironment()

for i, step in enumerate(skill.steps):
if step.template:
try:
template = env.from_string(step.template)
# 尝试编译模板
template.render()
except jinja2.TemplateError as e:
result.add_issue(ValidationIssue(
code="INVALID_TEMPLATE",
message=f"Invalid template in step '{step.name}': {str(e)}",
severity=ValidationSeverity.ERROR,
path=f"steps[{i}].template",
suggestion="Check Jinja2 template syntax"
))

def _validate_conditions(self, skill: Skill, result: ValidationResult):
"""验证条件表达式语法"""
# 这里可以集成JavaScript表达式解析器
# 暂时只进行基本验证

forbidden_patterns = [
(r'import\\s+', "Import statements not allowed in conditions"),
(r'__\\w+__', "Dunder methods not allowed in conditions"),
(r'eval\\(', "eval() function not allowed in conditions"),
(r'exec\\(', "exec() function not allowed in conditions"),
(r'open\\(', "File operations not allowed in conditions"),
]

for i, step in enumerate(skill.steps):
if step.condition:
condition = step.condition

for pattern, message in forbidden_patterns:
if re.search(pattern, condition):
result.add_issue(ValidationIssue(
code="FORBIDDEN_EXPRESSION",
message=f"{message} in condition: {condition[:50]}…",
severity=ValidationSeverity.ERROR,
path=f"steps[{i}].condition"
))

# 检查语法基本有效性
try:
# 这里可以添加更复杂的语法检查
# 例如使用pyparsing或类似的库
if not condition.strip():
result.add_issue(ValidationIssue(
code="EMPTY_CONDITION",
message="Condition expression is empty",
severity=ValidationSeverity.WARNING,
path=f"steps[{i}].condition"
))
except Exception as e:
result.add_issue(ValidationIssue(
code="CONDITION_SYNTAX_ERROR",
message=f"Condition syntax error: {str(e)}",
severity=ValidationSeverity.ERROR,
path=f"steps[{i}].condition"
))

def register_custom_validator(self, validator_func):
"""注册自定义验证器"""
self.custom_validators.append(validator_func)

class SkillLoader:
"""技能加载器"""

def __init__(self):
self.yaml_loader = yaml.SafeLoader
self.template_env = SandboxedEnvironment()

def load_yaml(self, content: str) –> Dict:
"""加载YAML内容"""
try:
data = yaml.safe_load(content)
if data is None:
return {}
return data
except yaml.YAMLError as e:
# 尝试提供更详细的错误信息
if hasattr(e, 'problem_mark'):
mark = e.problem_mark
raise ParseError(
f"YAML parse error: {e.problem}",
line=mark.line + 1,
column=mark.column + 1
)
else:
raise ParseError(f"YAML parse error: {str(e)}")

def extract_markdown(self, content: str) –> Tuple[str, str]:
"""提取YAML和Markdown部分"""
# 支持三种分隔符格式
separators = [
r'^—\\s*$', # 标准分隔符
r'^\\.\\.\\.\\s*$', # 替代分隔符
r'^```yaml\\s*$', # 代码块风格
]

lines = content.split('\\n')
yaml_lines = []
markdown_lines = []
in_yaml = True
separator_found = False

for i, line in enumerate(lines):
if in_yaml and any(re.match(pattern, line) for pattern in separators):
if yaml_lines: # 只有已经有YAML内容时才认为是分隔符
separator_found = True
in_yaml = False
continue

if in_yaml:
yaml_lines.append(line)
else:
markdown_lines.append(line)

yaml_content = '\\n'.join(yaml_lines)
markdown_content = '\\n'.join(markdown_lines)

return yaml_content, markdown_content

def resolve_references(self, data: Dict, context: Dict = None) –> Dict:
"""解析引用和变量"""
if context is None:
context = {}

def _resolve(obj, path=""):
if isinstance(obj, dict):
resolved = {}
for key, value in obj.items():
new_path = f"{path}.{key}" if path else key
resolved[key] = _resolve(value, new_path)
return resolved
elif isinstance(obj, list):
return [_resolve(item, f"{path}[{i}]") for i, item in enumerate(obj)]
elif isinstance(obj, str):
# 解析变量引用 {{variable}}
return self._resolve_variables(obj, context)
else:
return obj

return _resolve(data)

def _resolve_variables(self, text: str, context: Dict) –> str:
"""解析变量引用"""
if not isinstance(text, str):
return text

pattern = r'\\{\\{\\s*([a-zA-Z_][a-zA-Z0-9_]*(\\.[a-zA-Z_][a-zA-Z0-9_]*)*)\\s*\\}\\}'

def replace_match(match):
var_path = match.group(1)
try:
# 简单的点号分割路径解析
value = context
for part in var_path.split('.'):
if isinstance(value, dict) and part in value:
value = value[part]
else:
# 如果找不到变量，保持原样（将在执行时解析）
return match.group(0)
return str(value)
except:
return match.group(0)

return re.sub(pattern, replace_match, text)

class SkillDSLParser:
"""技能DSL主解析器"""

def __init__(self, schema_dir: Optional[Path] = None):
self.validator = SkillSchemaValidator(schema_dir)
self.loader = SkillLoader()
self.skills_cache = {}

def parse(self, content: str, validate: bool = True) –> Skill:
"""解析技能DSL内容"""
# 提取YAML和Markdown部分
yaml_content, markdown_content = self.loader.extract_markdown(content)

if not yaml_content.strip():
raise ParseError("No YAML content found in skill definition")

# 加载YAML
yaml_data = self.loader.load_yaml(yaml_content)

# 验证YAML结构
if validate:
validation_result = self.validator.validate_yaml(yaml_data)
if not validation_result.is_valid:
errors = "\\n".join(
f"- {issue.message} (path: {issue.path})"
for issue in validation_result.issues
if issue.severity in [ValidationSeverity.ERROR, ValidationSeverity.CRITICAL]
)
raise ParseError(f"Skill validation failed:\\n{errors}")

# 解析变量引用
resolved_data = self.loader.resolve_references(yaml_data)

# 构建技能对象
try:
skill = self._build_skill_object(resolved_data, markdown_content)
except Exception as e:
raise ParseError(f"Failed to build skill object: {str(e)}")

# 验证技能结构
if validate:
structure_result = self.validator.validate_skill_structure(skill)
if not structure_result.is_valid:
errors = "\\n".join(
f"- {issue.message} (path: {issue.path})"
for issue in structure_result.issues
if issue.severity in [ValidationSeverity.ERROR, ValidationSeverity.CRITICAL]
)
raise ParseError(f"Skill structure validation failed:\\n{errors}")

return skill

def parse_file(self, file_path: Union[str, Path], validate: bool = True) –> Skill:
"""从文件解析技能"""
file_path = Path(file_path)

# 检查缓存
cache_key = (str(file_path), file_path.stat().st_mtime if file_path.exists() else 0)
if cache_key in self.skills_cache:
return self.skills_cache[cache_key]

if not file_path.exists():
raise FileNotFoundError(f"Skill file not found: {file_path}")

try:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
except UnicodeDecodeError:
# 尝试其他编码
with open(file_path, 'r', encoding='latin-1') as f:
content = f.read()

skill = self.parse(content, validate)

# 缓存结果
self.skills_cache[cache_key] = skill

return skill

def validate(self, content: str) –> ValidationResult:
"""验证技能DSL内容"""
try:
# 尝试解析但不构建完整对象
yaml_content, _ = self.loader.extract_markdown(content)
yaml_data = self.loader.load_yaml(yaml_content)

# Schema验证
schema_result = self.validator.validate_yaml(yaml_data)

# 如果Schema验证通过，尝试构建对象进行结构验证
if schema_result.is_valid:
try:
skill = self._build_skill_object(yaml_data)
structure_result = self.validator.validate_skill_structure(skill)
schema_result.merge(structure_result)
except Exception as e:
schema_result.add_issue(ValidationIssue(
code="STRUCTURE_BUILD_FAILED",
message=f"Failed to build skill structure: {str(e)}",
severity=ValidationSeverity.ERROR,
path="$"
))

return schema_result

except Exception as e:
return ValidationResult(
is_valid=False,
issues=[ValidationIssue(
code="PARSE_ERROR",
message=f"Parse error: {str(e)}",
severity=ValidationSeverity.ERROR,
path="$"
)],
schema_version="v1.0.0"
)

def serialize(self, skill: Skill, format: str = "yaml") –> str:
"""序列化技能对象为DSL格式"""
if format.lower() == "yaml":
return self._serialize_to_yaml(skill)
elif format.lower() == "json":
return self._serialize_to_json(skill)
else:
raise ValueError(f"Unsupported format: {format}. Use 'yaml' or 'json'.")

def _build_skill_object(self, data: Dict, markdown_content: str = "") –> Skill:
"""从字典数据构建技能对象"""

# 处理元数据
metadata_data = data.get('metadata', {})
metadata = SkillMetadata(**metadata_data)

# 处理输入
inputs = {}
for name, input_data in data.get('inputs', {}).items():
inputs[name] = SkillInput(name=name, **input_data)

# 处理步骤
steps = []
for step_data in data.get('steps', []):
steps.append(SkillStep(**step_data))

# 处理输出
outputs = {}
for name, output_data in data.get('outputs', {}).items():
outputs[name] = SkillOutput(name=name, **output_data)

# 处理文档
doc_data = data.get('documentation', {})
if markdown_content:
doc_data['markdown_content'] = markdown_content
documentation = SkillDocumentation(**doc_data)

# 处理权限
permissions_data = data.get('permissions', {})
from models.skill_models import SkillPermissions
permissions = SkillPermissions(**permissions_data)

# 构建完整技能对象
skill_data = {
'metadata': metadata,
'inputs': inputs,
'steps': steps,
'outputs': outputs,
'documentation': documentation,
'permissions': permissions
}

return Skill(**skill_data)

def _serialize_to_yaml(self, skill: Skill) –> str:
"""序列化为YAML格式"""
import yaml

def skill_to_dict(skill_obj: Skill) –> Dict:
"""将技能对象转换为字典"""
data = skill_obj.dict(exclude_none=True)

# 处理特殊字段
if 'documentation' in data and 'markdown_content' in data['documentation']:
markdown_content = data['documentation'].pop('markdown_content')
else:
markdown_content = ""

return data, markdown_content

data, markdown_content = skill_to_dict(skill)

# 生成YAML
yaml_content = yaml.dump(data, default_flow_style=False, sort_keys=False, allow_unicode=True)

# 添加分隔符和Markdown
if markdown_content:
yaml_content = f"{yaml_content}\\n—\\n{markdown_content}"

return yaml_content

def _serialize_to_json(self, skill: Skill) –> str:
"""序列化为JSON格式"""
import json

data = skill.dict(exclude_none=True)

# 添加Schema信息
data['$schema'] = 'https://opencode.ai/schemas/skill/v1.0.0'

return json.dumps(data, indent=2, ensure_ascii=False, default=str)

# 快捷函数
def parse_skill(content: str, **kwargs) –> Skill:
"""快捷解析函数"""
parser = SkillDSLParser()
return parser.parse(content, **kwargs)

def parse_skill_file(file_path: Union[str, Path], **kwargs) –> Skill:
"""快捷文件解析函数"""
parser = SkillDSLParser()
return parser.parse_file(file_path, **kwargs)

def validate_skill(content: str) –> ValidationResult:
"""快捷验证函数"""
parser = SkillDSLParser()
return parser.validate(content)

2. 扩展验证器实现

# parsers/extended_validators.py
import re
from typing import Dict, List, Any
from pathlib import Path
import ast
import json

class ExtendedSkillValidator:
"""扩展的技能验证器"""

def __init__(self):
self.validators = {
'security': self.validate_security,
'performance': self.validate_performance,
'best_practices': self.validate_best_practices,
'compatibility': self.validate_compatibility
}

def validate_security(self, skill_data: Dict) –> List[Dict]:
"""安全验证"""
issues = []

# 检查危险操作
dangerous_patterns = [
(r'eval\\s*\\(', 'eval() usage', 'high'),
(r'exec\\s*\\(', 'exec() usage', 'high'),
(r'__import__\\s*\\(', 'dynamic import', 'medium'),
(r'subprocess\\.Popen', 'subprocess execution', 'medium'),
(r'os\\.system', 'system command execution', 'medium'),
]

# 检查步骤中的脚本和模板
for step in skill_data.get('steps', []):
text_to_check = []

if 'script' in step:
text_to_check.append(('script', step['script']))
if 'template' in step:
text_to_check.append(('template', step['template']))
if 'condition' in step:
text_to_check.append(('condition', step['condition']))

for field_type, text in text_to_check:
for pattern, description, severity in dangerous_patterns:
if re.search(pattern, text, re.IGNORECASE):
issues.append({
'type': 'security',
'severity': severity,
'description': f'Potential security issue in {field_type}: {description}',
'location': f"steps[{step.get('name', 'unnamed')}].{field_type}",
'suggestion': 'Consider using safer alternatives or adding proper validation'
})

# 检查文件权限
permissions = skill_data.get('permissions', {})
write_permissions = permissions.get('write', [])

for path in write_permissions:
if path in ['/', '/etc', '/usr', '/bin', '/sbin', '/var']:
issues.append({
'type': 'security',
'severity': 'high',
'description': f'Writing to system directory: {path}',
'location': 'permissions.write',
'suggestion': 'Restrict write permissions to application directories only'
})

return issues

def validate_performance(self, skill_data: Dict) –> List[Dict]:
"""性能验证"""
issues = []

# 检查步骤超时设置
for step in skill_data.get('steps', []):
timeout = step.get('timeout', 300)

if timeout > 3600: # 1小时
issues.append({
'type': 'performance',
'severity': 'medium',
'description': f'Step timeout too long: {timeout} seconds',
'location': f"steps[{step.get('name', 'unnamed')}].timeout",
'suggestion': 'Consider breaking long-running steps into smaller units'
})

# 检查潜在的无限循环模式
if 'script' in step and 'while True' in step['script']:
issues.append({
'type': 'performance',
'severity': 'high',
'description': 'Potential infinite loop detected',
'location': f"steps[{step.get('name', 'unnamed')}].script",
'suggestion': 'Add proper loop termination conditions'
})

# 检查资源密集型操作
resource_intensive_patterns = [
(r'for.*in.*range\\(\\d{6,}\\)', 'Large loop iteration'),
(r'\\.read\\(\\)', 'Reading entire file to memory'),
(r'SELECT \\*', 'Selecting all columns without limit'),
]

for step in skill_data.get('steps', []):
if 'script' in step:
for pattern, description in resource_intensive_patterns:
if re.search(pattern, step['script']):
issues.append({
'type': 'performance',
'severity': 'medium',
'description': f'Potential performance issue: {description}',
'location': f"steps[{step.get('name', 'unnamed')}].script",
'suggestion': 'Optimize for memory usage and processing time'
})

return issues

def validate_best_practices(self, skill_data: Dict) –> List[Dict]:
"""最佳实践验证"""
issues = []

metadata = skill_data.get('metadata', {})

# 检查文档完整性
if not metadata.get('description') or len(metadata.get('description', '')) < 20:
issues.append({
'type': 'best_practices',
'severity': 'low',
'description': 'Skill description is too brief',
'location': 'metadata.description',
'suggestion': 'Provide a more detailed description of what the skill does'
})

if not metadata.get('tags'):
issues.append({
'type': 'best_practices',
'severity': 'low',
'description': 'No tags specified',
'location': 'metadata.tags',
'suggestion': 'Add relevant tags for better discoverability'
})

# 检查输入验证
inputs = skill_data.get('inputs', {})
for input_name, input_def in inputs.items():
if input_def.get('required', False) and 'validation' not in input_def:
issues.append({
'type': 'best_practices',
'severity': 'medium',
'description': f'Required input "{input_name}" has no validation rules',
'location': f'inputs.{input_name}',
'suggestion': 'Add validation rules to ensure input quality'
})

# 检查错误处理
steps = skill_data.get('steps', [])
has_error_handling = False

for step in steps:
if 'retry' in step:
has_error_handling = True
break

if not has_error_handling and len(steps) > 1:
issues.append({
'type': 'best_practices',
'severity': 'medium',
'description': 'No error handling or retry logic specified',
'location': 'steps',
'suggestion': 'Add retry configuration for network or transient failures'
})

return issues

def validate_compatibility(self, skill_data: Dict) –> List[Dict]:
"""兼容性验证"""
issues = []

requirements = skill_data.get('metadata', {}).get('requirements', {})

# 检查OpenCode版本兼容性
opencode_version = requirements.get('opencode_version')
if opencode_version:
try:
from packaging import version
current_version = version.parse("1.0.0") # 假设当前版本
required_version = version.parse(opencode_version)

if required_version > current_version:
issues.append({
'type': 'compatibility',
'severity': 'high',
'description': f'Requires OpenCode {opencode_version}, but current is 1.0.0',
'location': 'metadata.requirements.opencode_version',
'suggestion': 'Update OpenCode or adjust skill requirements'
})
except:
pass

# 检查工具依赖
tools = requirements.get('tools', [])
for tool in tools:
if isinstance(tool, dict) and tool.get('optional', False) is False:
# 检查工具是否常见
common_tools = {'git', 'docker', 'kubectl', 'aws', 'terraform'}
if tool.get('name') not in common_tools:
issues.append({
'type': 'compatibility',
'severity': 'medium',
'description': f'Requires uncommon tool: {tool.get("name")}',
'location': 'metadata.requirements.tools',
'suggestion': 'Consider making this tool optional or providing installation instructions'
})

return issues

def validate_all(self, skill_data: Dict) –> Dict[str, List[Dict]]:
"""执行所有验证"""
results = {}

for validator_name, validator_func in self.validators.items():
try:
results[validator_name] = validator_func(skill_data)
except Exception as e:
results[validator_name] = [{
'type': 'validator_error',
'severity': 'high',
'description': f'Validator {validator_name} failed: {str(e)}',
'location': '$',
'suggestion': 'Check skill data format'
}]

return results

四、测试用例与验证

1. 测试数据准备

# tests/test_data/sample_skill.yaml
metadata:
name: git–release
version: 1.0.0
description: Automates semantic versioning and GitHub release creation
author: OpenCode Team
tags: [git, release, automation, ci–cd]
requirements:
opencode_version: "1.0.0"
tools:
– name: git
version: ">=2.25"
optional: false
– name: gh
version: ">=2.0"
optional: true
permissions: [file.read, file.write, network]
python_packages:
– name: requests
version: ">=2.25"
optional: false

inputs:
release_type:
type: string
description: Type of release (major, minor, patch)
required: true
choices: [major, minor, patch]
default: patch
examples:
– value: major
description: Major version bump for breaking changes
– value: minor
description: Minor version bump for new features
– value: patch
description: Patch version bump for bug fixes

dry_run:
type: boolean
description: Perform a dry run without creating release
required: false
default: false

changelog_path:
type: string
description: Path to changelog file
required: false
default: CHANGELOG.md
validation:
pattern: "^[a-zA-Z0-9_\\\\-./]+\\\\.md$"

steps:
– name: validate_environment
type: validation
description: Validate Git environment and permissions
condition: "git.status === 'clean' && git.branch === 'main'"
script: |
// Check Git status
const isClean = await git.isClean();
const currentBranch = await git.currentBranch();

if (!isClean) {
throw new Error('Working directory is not clean');
}

if (currentBranch !== 'main') {
throw new Error('Must be on main branch for releases');
}

outputs:
git_info: "context.git"

– name: analyze_commits
type: command
description: Analyze Git commits since last release
dependencies: [validate_environment]
parameters:
command: git
args:
– log
– ––oneline
– ––no–merges
– "{{git_info.latest_tag}}..HEAD"
outputs:
commits: "result.stdout"
commit_count: "result.stdout.split('\\\\n').length"

– name: determine_version
type: script
description: Determine next version based on commit analysis
dependencies: [analyze_commits]
script: |
const commits = context.commits;
let bumpType = 'patch';

// Analyze commit messages
for (const commit of commits.split('\\n')) {
if (commit.includes('BREAKING CHANGE')) {
bumpType = 'major';
break;
}
if (commit.startsWith('feat:')) {
bumpType = 'minor';
}
}

// Use user input if provided
if (inputs.release_type) {
bumpType = inputs.release_type;
}

// Calculate new version
const currentVersion = context.git_info.latest_tag.replace(/^v/, '');
const [major, minor, patch] = currentVersion.split('.').map(Number);

let newVersion;
switch (bumpType) {
case 'major':
newVersion = `${major + 1}.0.0`;
break;
case 'minor':
newVersion = `${major}.${minor + 1}.0`;
break;
case 'patch':
newVersion = `${major}.${minor}.${patch + 1}`;
break;
}

return { bumpType, newVersion };

timeout: 30
retry:
attempts: 2
delay: 5

– name: generate_changelog
type: template
description: Generate changelog from commits
dependencies: [analyze_commits, determine_version]
template: |
# Changelog

## Version {{determine_version.new_version}} ({{now | date('%Y-%m-%d')}})

### Features
{% for commit in commits %}
{% if commit.startswith('feat:') %}
– {{commit[6:]}}
{% endif %}
{% endfor %}