SDD 规范驱动开发详解
SDD（Specification-Driven Development）是一种规范驱动的开发方法。如何通过规范提高 AI 应用质量？如何实现文档先行的开发流程？本文详解 SDD 规范驱动开发方法。
一、SDD 概述
1.1 核心理念
SDD 核心原则：
┌─────────────────────────────────────┐
│ 1. 规范先行 │
│ - 先写规范后写代码 │
│ - 规范即文档 │
│ - 规范可验证 │
├─────────────────────────────────────┤
│ 2. 机器可读 │
│ - 结构化规范 │
│ - 自动化验证 │
│ - 代码生成 │
├─────────────────────────────────────┤
│ 3. 持续演进 │
│ - 版本管理 │
│ - 变更追踪 │
│ - 影响分析 │
└─────────────────────────────────────┘1.2 SDD 流程
# sdd_process.py
from enum import Enum
from typing import Dict, List
class SDDPhase(Enum):
"""SDD 阶段"""
SPECIFICATION = "specification" # 规范定义
VALIDATION = "validation" # 规范验证
IMPLEMENTATION = "implementation" # 实现
VERIFICATION = "verification" # 验证
MAINTENANCE = "maintenance" # 维护
class SDDProcess:
"""SDD 流程"""
def __init__(self):
self.current_phase = SDDPhase.SPECIFICATION
self.artifacts: Dict[str, any] = {}
def execute_phase(self, phase: SDDPhase) -> bool:
"""执行阶段"""
if phase == SDDPhase.SPECIFICATION:
return self._specification_phase()
elif phase == SDDPhase.VALIDATION:
return self._validation_phase()
elif phase == SDDPhase.IMPLEMENTATION:
return self._implementation_phase()
elif phase == SDDPhase.VERIFICATION:
return self._verification_phase()
elif phase == SDDPhase.MAINTENANCE:
return self._maintenance_phase()
return False
def _specification_phase(self) -> bool:
"""规范定义阶段"""
# 1. 定义需求规范
# 2. 定义接口规范
# 3. 定义数据规范
# 4. 定义行为规范的
return True
def _validation_phase(self) -> bool:
"""规范验证阶段"""
# 1. 语法验证
# 2. 语义验证
# 3. 一致性验证
# 4. 完整性验证
return True
def _implementation_phase(self) -> bool:
"""实现阶段"""
# 1. 代码生成
# 2. 手动补充
# 3. 单元测试生成
return True
def _verification_phase(self) -> bool:
"""验证阶段"""
# 1. 自动化测试
# 2. 规范符合性检查
# 3. 性能验证
return True
def _maintenance_phase(self) -> bool:
"""维护阶段"""
# 1. 规范更新
# 2. 影响分析
# 3. 版本管理
return True二、规范定义
2.1 需求规范
# specification/requirements.yaml
# 需求规范示例
specification:
version: 1.0.0
created_at: 2026-02-25
author: AI Team
agent_requirements:
name: ResearchAgent
description: 研究助手 Agent
capabilities:
- name: web_search
description: 网络搜索能力
required: true
- name: document_analysis
description: 文档分析能力
required: true
- name: summary_generation
description: 摘要生成能力
required: false
performance_requirements:
response_time_ms: 3000
accuracy_threshold: 0.85
availability: 0.99
constraints:
max_context_length: 4000
supported_languages:
- zh
- en
cost_per_query_max: 0.01
input_spec:
type: object
properties:
query:
type: string
description: 研究问题
required: true
context:
type: object
description: 上下文信息
required: false
options:
type: object
description: 配置选项
required: false
output_spec:
type: object
properties:
answer:
type: string
description: 回答内容
required: true
sources:
type: array
description: 参考来源
required: false
confidence:
type: number
description: 置信度
required: true2.2 接口规范
# specification/interface.yaml
# 接口规范示例
specification:
version: 1.0.0
type: interface
api_spec:
name: AgentAPI
version: v1
base_path: /api/v1/agents
endpoints:
- path: /query
method: POST
description: 执行查询
request:
content_type: application/json
schema:
type: object
properties:
agent_id:
type: string
query:
type: string
options:
type: object
response:
content_type: application/json
schema:
type: object
properties:
result:
type: object
error:
type: string
latency_ms:
type: number
errors:
- code: 400
description: 无效请求
- code: 404
description: Agent 不存在
- code: 500
description: 内部错误
- path: /status/{agent_id}
method: GET
description: 获取 Agent 状态
response:
schema:
type: object
properties:
status:
type: string
enum: [idle, busy, error]
last_active:
type: string
format: date-time2.3 行为规范
# specification/behavior.yaml
# 行为规范示例
specification:
version: 1.0.0
type: behavior
behavior_specs:
- name: query_processing
description: 查询处理流程
preconditions:
- agent.status == "idle"
- query is not empty
- query.length <= 1000
postconditions:
- response is not empty
- response.latency_ms <= 3000
- response.confidence >= 0.5
invariants:
- agent.state is consistent
- no data leakage between queries
flow:
- step: validate_input
description: 验证输入
on_error: return_error
- step: retrieve_context
description: 检索上下文
timeout: 1000
- step: generate_response
description: 生成响应
timeout: 2000
- step: validate_output
description: 验证输出
on_error: retry
- step: return_response
description: 返回响应
error_handling:
timeout:
action: return_error
message: "请求超时"
validation_error:
action: return_error
message: "输入验证失败"
system_error:
action: log_and_return_error
message: "系统错误"三、规范验证
3.1 语法验证
# spec_validation.py
import jsonschema
from typing import Dict, List
class SpecValidator:
"""规范验证器"""
def __init__(self, spec_schema: Dict):
self.spec_schema = spec_schema
def validate_syntax(self, spec: Dict) -> Dict:
"""验证语法"""
result = {
'valid': True,
'errors': [],
'warnings': []
}
try:
# JSON Schema 验证
jsonschema.validate(spec, self.spec_schema)
except jsonschema.ValidationError as e:
result['valid'] = False
result['errors'].append({
'type': 'schema_validation',
'message': str(e.message),
'path': list(e.path)
})
# 额外验证规则
self._validate_required_fields(spec, result)
self._validate_field_types(spec, result)
return result
def _validate_required_fields(
self,
spec: Dict,
result: Dict
):
"""验证必填字段"""
required_fields = ['version', 'type', 'specification']
for field in required_fields:
if field not in spec:
result['valid'] = False
result['errors'].append({
'type': 'missing_field',
'message': f'缺少必填字段：{field}'
})
def _validate_field_types(
self,
spec: Dict,
result: Dict
):
"""验证字段类型"""
if 'version' in spec:
version = spec['version']
if not isinstance(version, str):
result['warnings'].append({
'type': 'type_mismatch',
'message': 'version 应该是字符串'
})3.2 语义验证
# semantic_validation.py
from typing import Dict, List
class SemanticValidator:
"""语义验证器"""
def validate_semantics(self, spec: Dict) -> Dict:
"""验证语义"""
result = {
'valid': True,
'errors': [],
'warnings': []
}
# 验证一致性
self._check_consistency(spec, result)
# 验证完整性
self._check_completeness(spec, result)
# 验证可行性
self._check_feasibility(spec, result)
return result
def _check_consistency(self, spec: Dict, result: Dict):
"""检查一致性"""
# 检查输入输出是否一致
if 'input_spec' in spec and 'output_spec' in spec:
input_spec = spec['input_spec']
output_spec = spec['output_spec']
# 检查是否有冲突
# ...
# 检查性能要求是否合理
if 'performance_requirements' in spec:
perf = spec['performance_requirements']
if perf.get('response_time_ms', 0) < 100:
result['warnings'].append({
'type': 'unrealistic_requirement',
'message': '响应时间要求可能过于严格'
})
def _check_completeness(self, spec: Dict, result: Dict):
"""检查完整性"""
# 检查是否有遗漏
if 'error_handling' not in spec:
result['warnings'].append({
'type': 'missing_section',
'message': '缺少错误处理规范'
})
def _check_feasibility(self, spec: Dict, result: Dict):
"""检查可行性"""
# 检查成本要求是否可行
if 'constraints' in spec:
constraints = spec['constraints']
if constraints.get('cost_per_query_max', 1) < 0.001:
result['warnings'].append({
'type': 'cost_constraint',
'message': '成本要求可能过低'
})四、代码生成
4.1 骨架代码生成
# code_generator.py
from typing import Dict, List
import jinja2
class CodeGenerator:
"""代码生成器"""
def __init__(self, template_dir: str):
self.template_dir = template_dir
self.env = jinja2.Environment(
loader=jinja2.FileSystemLoader(template_dir)
)
def generate_agent_skeleton(self, spec: Dict) -> str:
"""生成 Agent 骨架代码"""
template = self.env.get_template('agent_skeleton.py.j2')
context = {
'agent_name': spec['agent_requirements']['name'],
'capabilities': spec['agent_requirements']['capabilities'],
'input_spec': spec['agent_requirements']['input_spec'],
'output_spec': spec['agent_requirements']['output_spec']
}
return template.render(context)
def generate_api_skeleton(self, spec: Dict) -> str:
"""生成 API 骨架代码"""
template = self.env.get_template('api_skeleton.py.j2')
context = {
'api_name': spec['api_spec']['name'],
'version': spec['api_spec']['version'],
'endpoints': spec['api_spec']['endpoints']
}
return template.render(context)
def generate_test_skeleton(self, spec: Dict) -> str:
"""生成测试骨架代码"""
template = self.env.get_template('test_skeleton.py.j2')
context = {
'spec_name': spec.get('name', 'spec'),
'test_cases': self._extract_test_cases(spec)
}
return template.render(context)
def _extract_test_cases(self, spec: Dict) -> List[Dict]:
"""提取测试用例"""
test_cases = []
# 从规范中提取测试用例
if 'behavior_specs' in spec:
for behavior in spec['behavior_specs']:
test_cases.append({
'name': behavior['name'],
'preconditions': behavior.get('preconditions', []),
'postconditions': behavior.get('postconditions', [])
})
return test_cases4.2 模板示例
{# templates/agent_skeleton.py.j2 #}
"""
{{ agent_name }} - 自动生成的 Agent 骨架
此代码由 SDD 规范自动生成。
请勿直接修改，修改规范后重新生成。
"""
from typing import Dict, Optional
from abc import ABC, abstractmethod
class {{ agent_name }}(ABC):
"""{{ agent_name }} 基类"""
def __init__(self, config: Dict = None):
self.config = config or {}
self._initialized = False
@abstractmethod
async def process(self, input_data: Dict) -> Dict:
"""
处理输入
Args:
input_data: 输入数据
Returns:
处理结果
"""
pass
def validate_input(self, input_data: Dict) -> bool:
"""验证输入"""
# TODO: 实现输入验证
return True
def validate_output(self, output_data: Dict) -> bool:
"""验证输出"""
# TODO: 实现输出验证
return True
def _log(self, message: str):
"""日志记录"""
print(f"[{{ agent_name }}] {message}")
# 使用示例
if __name__ == "__main__":
class Concrete{{ agent_name }}({{ agent_name }}):
async def process(self, input_data: Dict) -> Dict:
self.validate_input(input_data)
# TODO: 实现业务逻辑
result = {
'answer': 'TODO',
'confidence': 1.0
}
self.validate_output(result)
return result
agent = Concrete{{ agent_name }}()五、持续维护
5.1 版本管理
# spec_versioning.py
from typing import Dict, List
from datetime import datetime
class SpecVersionManager:
"""规范版本管理器"""
def __init__(self):
self.versions: Dict[str, List[Dict]] = {}
def create_version(
self,
spec_id: str,
spec: Dict,
changes: List[str]
) -> Dict:
"""创建新版本"""
version = {
'version': self._get_next_version(spec_id),
'spec': spec,
'changes': changes,
'created_at': datetime.now().isoformat(),
'status': 'draft'
}
if spec_id not in self.versions:
self.versions[spec_id] = []
self.versions[spec_id].append(version)
return version
def _get_next_version(self, spec_id: str) -> str:
"""获取下一个版本号"""
if spec_id not in self.versions:
return '1.0.0'
last_version = self.versions[spec_id][-1]['version']
major, minor, patch = map(int, last_version.split('.'))
return f"{major}.{minor}.{patch + 1}"
def get_version_history(self, spec_id: str) -> List[Dict]:
"""获取版本历史"""
return self.versions.get(spec_id, [])
def compare_versions(
self,
spec_id: str,
version1: str,
version2: str
) -> Dict:
"""比较版本差异"""
# 实现版本比较逻辑
return {
'added': [],
'removed': [],
'modified': []
}5.2 变更管理
# change_management.py
from typing import Dict, List
class ChangeManager:
"""变更管理器"""
def __init__(self):
self.pending_changes: List[Dict] = []
def propose_change(
self,
spec_id: str,
change_type: str,
description: str,
impact_analysis: Dict
) -> Dict:
"""提议变更"""
change = {
'id': self._generate_change_id(),
'spec_id': spec_id,
'type': change_type,
'description': description,
'impact_analysis': impact_analysis,
'status': 'pending',
'approvers': []
}
self.pending_changes.append(change)
return change
def approve_change(self, change_id: str, approver: str) -> bool:
"""批准变更"""
for change in self.pending_changes:
if change['id'] == change_id:
if approver not in change['approvers']:
change['approvers'].append(approver)
# 如果所有审批人都批准了
if len(change['approvers']) >= 2:
change['status'] = 'approved'
return True
return False
def implement_change(self, change_id: str) -> bool:
"""实施变更"""
for change in self.pending_changes:
if change['id'] == change_id and change['status'] == 'approved':
# 实施变更
change['status'] = 'implemented'
return True
return False
def _generate_change_id(self) -> str:
"""生成变更 ID"""
import uuid
return str(uuid.uuid4())[:8]六、SDD 工具链
6.1 工具集成
# sdd_toolchain.py
SDD_TOOLCHAIN = {
'specification': {
'editor': 'SpecEditor',
'validator': 'SpecValidator',
'linter': 'SpecLinter'
},
'generation': {
'code_generator': 'CodeGenerator',
'test_generator': 'TestGenerator',
'doc_generator': 'DocGenerator'
},
'verification': {
'test_runner': 'TestRunner',
'spec_checker': 'SpecChecker',
'performance_tester': 'PerfTester'
},
'maintenance': {
'version_manager': 'SpecVersionManager',
'change_manager': 'ChangeManager',
'impact_analyzer': 'ImpactAnalyzer'
}
}
class SDDToolchain:
"""SDD 工具链"""
def __init__(self):
self.tools = {}
def setup(self):
"""设置工具链"""
# 初始化工具
# ...
pass
def run_pipeline(self, spec: Dict) -> Dict:
"""运行完整流程"""
results = {
'validation': None,
'generation': None,
'verification': None
}
# 1. 验证规范
validator = SpecValidator()
results['validation'] = validator.validate(spec)
if not results['validation']['valid']:
return results
# 2. 生成代码
generator = CodeGenerator()
results['generation'] = generator.generate(spec)
# 3. 验证生成结果
verifier = SpecChecker()
results['verification'] = verifier.verify(spec, results['generation'])
return results七、总结
7.1 核心要点
-
规范先行
- 先写规范后写代码
- 规范即文档
- 规范可验证
-
自动化
- 代码自动生成
- 测试自动生成
- 文档自动生成
-
持续演进
- 版本管理
- 变更控制
- 影响分析
7.2 实施建议
-
渐进式采用
- 从核心模块开始
- 逐步扩大范围
- 持续改进流程
-
工具支持
- 建立工具链
- 集成 CI/CD
- 自动化验证
-
团队培训
- 规范编写培训
- 工具使用培训
- 最佳实践分享
参考资料