LLM辅助数据库Schema设计：从自然语言需求到DDL的智能转换与审查-育师

LLM辅助数据库Schema设计：从自然语言需求到DDL的智能转换与审查

一、PRD到DDL的"翻译损耗"：一个被严重低估的工程问题

产品经理在 PRD 中写下"需要一个用户积分系统，支持积分获取、消费和过期"，但到技术方案评审时发现，最终的表结构设计遗漏了积分流水溯源、并发扣减、过期策略三个核心需求。这不是个别现象，而是 Schema 设计中的"翻译损耗"问题——需求文档到数据库设计的转换过程存在严重的信息衰减。

传统的 Schema 设计流程是：需求文档 → 架构师理解 → 手动设计 ER 图 → 编写 DDL。每一步都可能引入理解偏差。以积分系统为例，架构师设计的是user_points (user_id, balance, updated_at)的单表单字段方案，但实际业务需要的是：

积分来源追溯（签到、消费返利、活动奖励等不同来源）
并发安全（多线程/多服务同时对同一用户加减积分）
过期策略（签到积分 30 天过期，活动积分 7 天过期）

这些问题在人工设计时极易被遗漏，而大语言模型恰好在"从非结构化需求中提取结构化设计"方面表现出色。本文将探讨如何利用 LLM 辅助 Schema 设计，并建立一套从需求分析到 DDL 自动生成与审查的完整工具链。

二、需求到Schema的智能映射与质量卡控体系

flowchart TB A[PRD / 自然语言需求] --> B[LLM 需求解析] B --> C[实体识别<br/>Entity Extraction] C --> D[关系推断<br/>Relationship Deduction] D --> E[约束推导<br/>Constraint Derivation] E --> F[初始 DDL 生成] F --> G{Schema 质量审查} G -->|归一范式检查| H[范式合规] G -->|命名规范检查| I[命名规范] G -->|性能预检| J[索引合理性] G -->|安全审查| K[敏感字段标识] H --> L{审查结果} I --> L J --> L K --> L L -->|通过| M[输出最终 DDL] L -->|不通过| N[生成修改建议] N --> F

核心流程包含三个阶段：

阶段一：需求解析——LLM 从自然语言中提取实体、属性、关系和业务约束。关键挑战在于让模型理解"隐含需求"：PRD 中不会明确写"积分流水需要分表存储防止单表过大"，但专业的设计者应当预见到这一点。

阶段二：DDL 生成——基于解析结果生成符合规范的建表语句，包括字段类型选择、索引设计、分区策略、字符集等。

阶段三：质量审查——对生成的 DDL 进行多维度的自动审查，这是整个流程中最关键的一环。因为 LLM 虽然擅长"生成看起来合理的东西"，但不擅长检查约束条件的完备性。

三、端到端的工程实现

3.1 需求解析与 Schema 生成

from typing import List, Dict, Optional from dataclasses import dataclass, field import json import re @dataclass class Entity: name: str comment: str columns: List[Dict] = field(default_factory=list) indexes: List[Dict] = field(default_factory=list) relationships: List[Dict] = field(default_factory=list) @dataclass class SchemaDesign: entities: List[Entity] version: str generated_at: str class SchemaGenerator: """从自然语言需求生成数据库 Schema""" SYSTEM_PROMPT = """你是一个数据库设计专家。请根据需求文档，生成数据库表结构设计。 输出格式为严格的 JSON，包含以下字段： - entities: 实体列表 - name: 表名（snake_case） - comment: 表注释 - columns: 列定义 [{name, type, nullable, default, comment, is_primary_key}] - indexes: 索引定义 [{name, columns, type(UNIQUE/NORMAL), comment}] - relationships: 关系定义 [{from, to, type(ONE_TO_ONE/ONE_TO_MANY), foreign_key}] 需要特别关注： 1. 高并发场景下的数据一致性设计（乐观锁、版本号） 2. 审计字段（created_at, updated_at） 3. 数据增长的预估（分区策略建议） 4. 敏感字段标识（需要加密或脱敏的字段）""" def __init__(self, llm_client): self.llm = llm_client def generate(self, requirement: str) -> SchemaDesign: """生成 Schema 设计""" prompt = f"需求文档：\n{requirement}\n\n请生成数据库表结构设计。" response = self.llm.chat( system=self.SYSTEM_PROMPT, user_message=prompt, temperature=0.2 # 低温度以确保一致性 ) schema_dict = self._parse_json_response(response) return self._validate_and_build(schema_dict) def _parse_json_response(self, response: str) -> dict: """从 LLM 响应中提取 JSON""" # 处理 LLM 可能包裹在 ```json 代码块中的响应 json_match = re.search(r'```json\s*(.*?)```', response, re.DOTALL) if json_match: json_str = json_match.group(1) else: json_str = response try: return json.loads(json_str) except json.JSONDecodeError as e: raise ValueError(f"LLM 返回的 JSON 解析失败: {e}") def _validate_and_build(self, schema_dict: dict) -> SchemaDesign: """验证并构建 Schema 对象""" entities = [] for e in schema_dict.get('entities', []): # 验证必填字段 if not e.get('name') or not e.get('columns'): continue entity = Entity( name=e['name'], comment=e.get('comment', ''), columns=e['columns'], indexes=e.get('indexes', []), relationships=e.get('relationships', []) ) entities.append(entity) return SchemaDesign( entities=entities, version="1.0", generated_at=datetime.now().isoformat() )

3.2 DDL 编译与质量审查

class DDLCompiler: """将 Schema 设计编译为 MySQL DDL 语句""" TYPE_MAPPING = { 'string': 'VARCHAR(255)', 'text': 'TEXT', 'long_text': 'LONGTEXT', 'integer': 'INT', 'big_integer': 'BIGINT', 'decimal': 'DECIMAL(18,2)', 'boolean': 'TINYINT(1)', 'datetime': 'DATETIME', 'date': 'DATE', 'json': 'JSON', } def compile(self, design: SchemaDesign) -> str: """生成完整的 DDL 脚本""" ddl_parts = [] for entity in design.entities: ddl = self._compile_entity(entity) ddl_parts.append(ddl) return "\n\n".join(ddl_parts) def _compile_entity(self, entity: Entity) -> str: """编译单个实体为 CREATE TABLE 语句""" # 列定义 col_defs = [] primary_keys = [] for col in entity.columns: col_sql = self._compile_column(col) col_defs.append(col_sql) if col.get('is_primary_key'): primary_keys.append(col['name']) # 主键 if primary_keys: col_defs.append(f"PRIMARY KEY ({', '.join(primary_keys)})") # 索引 for idx in entity.indexes: idx_sql = self._compile_index(idx) col_defs.append(idx_sql) # 组装 columns_sql = ",\n ".join(col_defs) ddl = f"""CREATE TABLE `{entity.name}` ( {columns_sql} ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='{entity.comment}';""" # 分区建议作为注释 partition_note = self._suggest_partition(entity) if partition_note: ddl += f"\n-- 分区建议: {partition_note}" return ddl def _compile_column(self, col: dict) -> str: """编译单个列定义""" col_type = self.TYPE_MAPPING.get(col.get('type', 'string'), 'VARCHAR(255)') parts = [f"`{col['name']}`", col_type] if not col.get('nullable', True): parts.append("NOT NULL") if col.get('default') is not None: default_val = col['default'] if isinstance(default_val, str): parts.append(f"DEFAULT '{default_val}'") else: parts.append(f"DEFAULT {default_val}") if col.get('comment'): parts.append(f"COMMENT '{col['comment']}'") return " ".join(parts) def _suggest_partition(self, entity: Entity) -> Optional[str]: """根据列名和类型建议分区策略""" time_columns = ['created_at', 'create_time', 'event_time', 'order_time'] for col in entity.columns: if col['name'] in time_columns: return f"PARTITION BY RANGE (TO_DAYS({col['name']})) 按月分区" return None class SchemaReviewer: """Schema 质量审查引擎""" def __init__(self): self.checks = [ self._check_naming_convention, self._check_required_columns, self._check_index_coverage, self._check_field_types, self._check_sensitive_fields, ] def review(self, ddl: str) -> List[Dict]: """执行全部审查并返回问题列表""" issues = [] for check_fn in self.checks: result = check_fn(ddl) issues.extend(result) return issues def _check_naming_convention(self, ddl: str) -> List[Dict]: """命名规范检查""" issues = [] # 表名必须 snake_case table_pattern = r'CREATE TABLE\s+`(\w+)`' tables = re.findall(table_pattern, ddl, re.IGNORECASE) for table in tables: if not re.match(r'^[a-z][a-z0-9_]*$', table): issues.append({ 'level': 'ERROR', 'type': 'naming', 'message': f'表名 "{table}" 不符合 snake_case 规范' }) # 索引名规范 idx_pattern = r'(?:INDEX|KEY)\s+`(\w+)`' indexes = re.findall(idx_pattern, ddl, re.IGNORECASE) for idx in indexes: if not idx.startswith('idx_') and idx != 'PRIMARY': issues.append({ 'level': 'WARN', 'type': 'naming', 'message': f'索引名 "{idx}" 建议以 idx_ 开头' }) return issues def _check_required_columns(self, ddl: str) -> List[Dict]: """必填字段检查""" issues = [] required = { 'id': '主键', 'created_at': '创建时间', 'updated_at': '更新时间' } for col, desc in required.items(): if f'`{col}`' not in ddl: issues.append({ 'level': 'WARN', 'type': 'missing_column', 'message': f'缺少建议字段: {col} ({desc})' }) return issues def _check_sensitive_fields(self, ddl: str) -> List[Dict]: """敏感字段检测""" sensitive_keywords = ['password', 'phone', 'email', 'id_card', 'bank_card'] issues = [] for keyword in sensitive_keywords: if keyword in ddl.lower(): issues.append({ 'level': 'INFO', 'type': 'sensitive_field', 'message': f'检测到敏感字段 "{keyword}"，建议加密存储或脱敏' }) return issues def _check_index_coverage(self, ddl: str) -> List[Dict]: """索引覆盖率检查""" issues = [] # 外键列必须有索引 fk_pattern = r'FOREIGN KEY\s*\(`(\w+)`\)' fk_columns = re.findall(fk_pattern, ddl, re.IGNORECASE) for fk_col in fk_columns: if f'`{fk_col}`' not in ddl.split('PRIMARY KEY')[0]: issues.append({ 'level': 'WARN', 'type': 'missing_index', 'message': f'外键列 "{fk_col}" 缺少索引' }) return issues def _check_field_types(self, ddl: str) -> List[Dict]: """字段类型合理性检查""" issues = [] # 检查是否有 VARCHAR 没有指定长度 if 'VARCHAR()' in ddl.upper() or 'VARCHAR(0)' in ddl.upper(): issues.append({ 'level': 'ERROR', 'type': 'invalid_type', 'message': 'VARCHAR 必须指定长度' }) # 金额字段建议使用 DECIMAL 而非 FLOAT if re.search(r'`\w*amount\w*`.*FLOAT', ddl, re.IGNORECASE): issues.append({ 'level': 'WARN', 'type': 'type_suggestion', 'message': '金额字段建议使用 DECIMAL 而非 FLOAT，避免精度丢失' }) return issues