第3讲我们实现了代码补全——当开发者敲代码时，MiniCopilot 能预测接下来要写什么。但有时候，开发者不想逐行敲代码，而是想用自然语言描述需求，让 AI 直接生成代码。

这一讲，我们要实现 MiniCopilot 的第二个核心能力：自然语言转代码（NL2Code）。

一、NL2Code 的核心挑战

NL2Code 流程

用户输入："写一个函数，读取 CSV 文件并返回平均值" ↓ 1. 意图分析（Orchestrator Agent） ├── 任务类型：函数生成 ├── 输入：CSV 文件路径 ├── 输出：数值平均值 └── 约束：需要错误处理 ↓ 2. 上下文收集（Research Agent） ├── 项目中已有的 CSV 处理方式 ├── 常用的库（pandas / csv） └── 代码风格约定 ↓ 3. 代码生成（Coding Agent） ├── 生成初始版本 ├── Review Agent 审查 └── 迭代修改 ↓ 4. 返回结果 + 插入位置建议

二、意图分析模块

2.1 任务分类器

# engine/nl2code/intent_classifier.py from enum import Enum from typing import Optional import re class TaskType(Enum): CREATE_FUNCTION = "create_function" CREATE_CLASS = "create_class" WRITE_TEST = "write_test" FIX_BUG = "fix_bug" REFACTOR = "refactor" ADD_COMMENT = "add_comment" EXPLAIN_CODE = "explain_code" QUERY_DATA = "query_data" OTHER = "other" class IntentClassifier: """意图分类器：识别用户想要做什么""" # 关键词模式 PATTERNS = { TaskType.CREATE_FUNCTION: [ r"写(一?个)?函数", r"创建(一?个)?函数", r"定义(一?个)?函数", r"实现.*功能", r"写一段代码.*(用于|来)", r"生成.*函数", r"编写.*方法", ], TaskType.CREATE_CLASS: [ r"写(一?个)?类", r"创建(一?个)?类", r"定义(一?个)?类", r"实现.*类", r"设计.*类", ], TaskType.WRITE_TEST: [ r"(写|生成|创建)(单元)?测试", r"test.*用例", r"为.*写测试", r"测试.*函数", ], TaskType.FIX_BUG: [ r"(修|改|修复).*(bug|错误|问题|异常)", r"这个.*不对", r"出错了", r"报错", ], TaskType.REFACTOR: [ r"重构", r"优化", r"改写", r"重写", r"改进.*代码", r"让.*更好", ], TaskType.ADD_COMMENT: [ r"加(注?释|评论)", r"写(注?释|文档)", r"生成.*文档", r"添加.*说明", ], TaskType.EXPLAIN_CODE: [ r"解释.*代码", r"这段代码.*意思", r"说明.*作用", r"分析.*代码", ], } def classify(self, user_input: str) -> tuple[TaskType, dict]: """ 分类用户意图 返回：(任务类型, 提取的参数) """ user_input = user_input.lower() for task_type, patterns in self.PATTERNS.items(): for pattern in patterns: if re.search(pattern, user_input): params = self._extract_params(user_input, task_type) return task_type, params return TaskType.OTHER, {} def _extract_params(self, text: str, task_type: TaskType) -> dict: """提取任务参数""" params = {} # 提取函数名 func_match = re.search(r"(?:函数|方法|功能)\s*[叫称]?\s*[`'\"《]?(\w+)[`'\"》]?", text) if func_match: params["function_name"] = func_match.group(1) # 提取类名 class_match = re.search(r"(?:类)\s*[叫称]?\s*[`'\"《]?(\w+)[`'\"》]?", text) if class_match: params["class_name"] = class_match.group(1) # 提取输入输出 input_match = re.search(r"(?:输入|传入|接收|参数)[：:\s]*(\w+)", text) if input_match: params["input"] = input_match.group(1) output_match = re.search(r"(?:输出|返回|结果)[：:\s]*(\w+)", text) if output_match: params["output"] = output_match.group(1) # 提取文件名 file_match = re.search(r"[`'\"《]([\w./\\]+)[`'\"》]", text) if file_match: params["file"] = file_match.group(1) return params

2.2 需求细化器

# engine/nl2code/requirement_refiner.py import requests import json class RequirementRefiner: """需求细化器：将模糊的自然语言转化为结构化需求""" def __init__(self, api_key: str): self.api_key = api_key def refine(self, user_input: str, context: dict = None) -> dict: """ 细化用户需求 返回结构化需求： { "task": "create_function", "name": "calculate_average", "inputs": [{"name": "file_path", "type": "str"}], "output": {"type": "float"}, "logic": "读取 CSV 文件，计算某一列的平均值", "constraints": ["处理文件不存在的情况", "支持指定列名"], "language": "python", "libraries": ["csv", "pandas"] } """ prompt = self._build_prompt(user_input, context) response = requests.post( "https://api.deepseek.com/v1/chat/completions", headers={"Authorization": f"Bearer {self.api_key}"}, json={ "model": "deepseek-chat", "messages": [ {"role": "system", "content": "你是一个软件需求分析师。将用户的自然语言需求转化为结构化的代码生成规格说明。输出 JSON 格式。"}, {"role": "user", "content": prompt} ], "response_format": {"type": "json_object"}, "temperature": 0.2 } ) result = response.json() content = result["choices"][0]["message"]["content"] try: return json.loads(content) except: return { "task": "unknown", "raw_description": user_input, "error": "无法解析需求" } def _build_prompt(self, user_input: str, context: dict = None) -> str: """构建细化需求的 prompt""" context_str = "" if context: context_str = f""" 项目上下文： - 语言: {context.get('language', 'python')} - 框架: {context.get('framework', '无')} - 已有代码风格: {context.get('style', '未知')} """ return f"""请将以下用户需求细化为结构化的代码生成规格。 用户需求：{user_input} {context_str} 请输出 JSON 格式，包含： - task: 任务类型 (create_function/create_class/write_test/fix_bug/refactor) - name: 生成的代码实体名称 - inputs: 输入参数列表 [{name, type, description}] - output: 输出 {{type, description}} - logic: 核心逻辑描述 - constraints: 约束条件列表 - language: 编程语言 - libraries: 需要用到的库列表 - edge_cases: 需要考虑的边界情况 """

三、代码生成器

3.1 基础代码生成器

# engine/nl2code/generator.py import requests import json from typing import Optional class CodeGenerator: """代码生成器""" def __init__(self, api_key: str): self.api_key = api_key def generate(self, requirement: dict, context_code: str = "") -> dict: """ 根据需求生成代码 返回： { "code": "生成的代码", "explanation": "代码说明", "insertion_point": "建议插入位置", "dependencies": ["需要的依赖"] } """ prompt = self._build_generation_prompt(requirement, context_code) response = requests.post( "https://api.deepseek.com/v1/chat/completions", headers={"Authorization": f"Bearer {self.api_key}"}, json={ "model": "deepseek-coder", "messages": [ {"role": "system", "content": self._get_system_prompt(requirement['language'])}, {"role": "user", "content": prompt} ], "temperature": 0.3, "max_tokens": 2048 } ) result = response.json() content = result["choices"][0]["message"]["content"] return self._parse_response(content) def _get_system_prompt(self, language: str) -> str: """获取系统提示词""" return f"""你是一个专业的 {language} 开发者。请根据需求生成高质量的代码。 要求： 1. 代码必须语法正确，可直接运行 2. 遵循 {language} 的最佳实践和命名规范 3. 包含完整的类型注解（如果语言支持） 4. 包含 docstring/注释说明 5. 处理边界情况和错误 6. 代码风格统一 输出格式：

{language}

// 代码

--- 简要说明代码的设计思路和用法 --- 需要的依赖包（如果有）""" def _build_generation_prompt(self, requirement: dict, context_code: str) -> str: """构建生成 prompt""" parts = [f"请生成以下需求的代码：\n"] parts.append(f"任务类型：{requirement.get('task', 'unknown')}") parts.append(f"名称：{requirement.get('name', 'unnamed')}") if requirement.get('inputs'): parts.append("\n输入参数：") for inp in requirement['inputs']: parts.append(f" - {inp.get('name')}: {inp.get('type')} ({inp.get('description', '')})") if requirement.get('output'): parts.append(f"\n输出：{requirement['output'].get('type')} - {requirement['output'].get('description', '')}") parts.append(f"\n核心逻辑：{requirement.get('logic', '')}") if requirement.get('constraints'): parts.append("\n约束条件：") for c in requirement['constraints']: parts.append(f" - {c}") if requirement.get('libraries'): parts.append(f"\n推荐库：{', '.join(requirement['libraries'])}") if requirement.get('edge_cases'): parts.append("\n边界情况：") for e in requirement['edge_cases']: parts.append(f" - {e}") if context_code: parts.append(f"\n\n现有代码上下文：\n```{requirement.get('language', 'python')}\n{context_code}\n```") return '\n'.join(parts) def _parse_response(self, content: str) -> dict: """解析 LLM 响应""" result = { "code": "", "explanation": "", "dependencies": [] } # 提取代码块 code_match = re.search(r'```(?:\w+)?\n(.*?)```', content, re.DOTALL) if code_match: result["code"] = code_match.group(1).strip() # 提取说明（在 --- 之后） parts = content.split('---') if len(parts) > 1: result["explanation"] = parts[-1].strip() # 提取依赖 dep_match = re.search(r'(?:依赖|requirements|dependencies)[：:]\s*(.+?)(?:\n|$)', content) if dep_match: deps = dep_match.group(1).strip() result["dependencies"] = [d.strip() for d in deps.split(',')] return result

3.2 代码生成器增强版（带审查）

# engine/nl2code/enhanced_generator.py from .generator import CodeGenerator class EnhancedCodeGenerator: """增强版代码生成器（带自我审查）""" def __init__(self, api_key: str): self.generator = CodeGenerator(api_key) def generate_with_review(self, requirement: dict, context_code: str = "", max_iterations: int = 3) -> dict: """ 生成代码并进行自我审查和改进 流程： 1. 生成初始版本 2. 审查代码（检查 Bug、安全问题、性能问题） 3. 如果需要改进，重新生成 4. 重复直到满意或达到最大迭代次数 """ best_result = None best_score = -1 for i in range(max_iterations): print(f"🔄 第 {i+1} 轮生成...") # 生成代码 result = self.generator.generate(requirement, context_code) # 审查代码 review = self._review_code(result["code"], requirement['language']) result["review"] = review result["iteration"] = i + 1 # 评分 score = self._calculate_score(review) result["score"] = score if score > best_score: best_score = score best_result = result # 如果评分足够高，提前结束 if score >= 90: print(f"✅ 代码质量优秀（{score}分），停止迭代") break # 将审查意见加入需求，用于下一轮改进 if i < max_iterations - 1: requirement["_previous_feedback"] = review["issues"] print(f"📝 发现 {len(review['issues'])} 个问题，继续改进...") return best_result def _review_code(self, code: str, language: str) -> dict: """审查代码质量""" issues = [] suggestions = [] # 1. 语法检查（简单） if not code or len(code.strip()) == 0: issues.append("代码为空") return {"issues": issues, "suggestions": [], "quality": 0} # 2. 缺少 docstring if language == "python" and '"""' not in code and "'''" not in code: suggestions.append("建议添加 docstring 说明函数用途") # 3. 缺少类型注解 if language == "python": func_defs = re.findall(r'def \w+\((.*?)\):', code) for params in func_defs: if params and ':' not in params: suggestions.append("建议为参数添加类型注解") # 4. 硬编码值 hardcoded = re.findall(r'(?<!["\'])\b\d{4,}\b(?!["\'])', code) if hardcoded: suggestions.append(f"存在硬编码数值: {hardcoded[:3]}，建议定义为常量") # 5. 空的 except if re.search(r'except\s*:', code): issues.append("使用裸 except，建议指定异常类型") # 6. 过长的函数 lines = code.split('\n') if len(lines) > 50: suggestions.append(f"函数过长（{len(lines)}行），建议拆分为多个小函数") # 计算质量分数 quality = max(0, 100 - len(issues) * 15 - len(suggestions) * 5) return { "issues": issues, "suggestions": suggestions, "quality": quality } def _calculate_score(self, review: dict) -> int: """计算综合评分""" return review.get("quality", 0)

四、上下文集成

4.1 项目上下文收集

# engine/nl2code/context_collector.py import os from typing import Optional class ProjectContextCollector: """项目上下文收集器""" def __init__(self, project_root: str): self.project_root = project_root def collect(self, current_file: str = "") -> dict: """ 收集项目上下文 返回： { "language": "python", "framework": "fastapi", "style": "google style", "imports": ["os", "sys", "fastapi"], "recent_files": [...], "similar_functions": [...] } """ context = { "language": self._detect_language(current_file), "framework": self._detect_framework(), "style": self._detect_code_style(), "imports": [], "recent_files": [], "similar_functions": [] } # 收集 imports if current_file and os.path.exists(current_file): with open(current_file, 'r') as f: content = f.read() context["imports"] = self._extract_imports(content) # 收集最近修改的文件 context["recent_files"] = self._get_recent_files() return context def _detect_language(self, file_path: str) -> str: """检测编程语言""" ext_map = { '.py': 'python', '.js': 'javascript', '.ts': 'typescript', '.java': 'java', '.go': 'go', '.rs': 'rust', '.cpp': 'cpp', '.c': 'c', } _, ext = os.path.splitext(file_path) return ext_map.get(ext, 'python') def _detect_framework(self) -> str: """检测使用的框架""" indicators = { 'fastapi': ['fastapi', 'FastAPI'], 'flask': ['flask', 'Flask'], 'django': ['django', 'Django'], 'react': ['react', 'React', 'jsx'], 'vue': ['vue', 'Vue'], } for framework, keywords in indicators.items(): for keyword in keywords: if self._search_in_project(keyword): return framework return "unknown" def _detect_code_style(self) -> str: """检测代码风格""" # 检查是否有类型注解 py_files = self._find_files('*.py') has_types = False has_google_docstring = False for f in py_files[:10]: with open(f, 'r') as fh: content = fh.read() if ': ' in content and 'def ' in content: has_types = True if 'Args:' in content or 'Returns:' in content: has_google_docstring = True if has_google_docstring: return "google style" elif has_types: return "modern python" else: return "simple" def _extract_imports(self, content: str) -> list[str]: """提取 import 语句""" imports = [] for line in content.split('\n'): line = line.strip() if line.startswith('import ') or line.startswith('from '): imports.append(line) return imports def _get_recent_files(self, count: int = 5) -> list[str]: """获取最近修改的文件""" files = [] for root, dirs, filenames in os.walk(self.project_root): for filename in filenames: if filename.endswith(('.py', '.js', '.ts', '.java')): filepath = os.path.join(root, filename) files.append((filepath, os.path.getmtime(filepath))) files.sort(key=lambda x: x[1], reverse=True) return [f[0] for f in files[:count]] def _search_in_project(self, keyword: str) -> bool: """在项目中搜索关键字""" for root, dirs, filenames in os.walk(self.project_root): for filename in filenames: if filename.endswith(('.py', '.txt', '.cfg', '.toml')): filepath = os.path.join(root, filename) try: with open(filepath, 'r', errors='ignore') as f: if keyword in f.read(): return True except: continue return False def _find_files(self, pattern: str) -> list[str]: """查找匹配的文件""" import glob return glob.glob(os.path.join(self.project_root, '**', pattern), recursive=True)

五、完整 NL2Code 管道

# engine/nl2code/pipeline.py from .intent_classifier import IntentClassifier from .requirement_refiner import RequirementRefiner from .generator import CodeGenerator from .enhanced_generator import EnhancedCodeGenerator from .context_collector import ProjectContextCollector class NL2CodePipeline: """自然语言转代码完整管道""" def __init__(self, api_key: str, project_root: str = "."): self.classifier = IntentClassifier() self.refiner = RequirementRefiner(api_key) self.generator = EnhancedCodeGenerator(api_key) self.context_collector = ProjectContextCollector(project_root) def execute(self, user_input: str, current_file: str = "", selected_code: str = "") -> dict: """ 执行 NL2Code 完整流程 参数： user_input: 用户的自然语言描述 current_file: 当前编辑的文件路径 selected_code: 用户选中的代码（如果有） 返回： { "requirement": {...}, # 结构化需求 "code": "...", # 生成的代码 "explanation": "...", # 代码说明 "review": {...}, # 审查结果 "insertion_point": "...", # 建议插入位置 "confidence": 0.95 # 置信度 } """ print(f"\n🎯 用户需求: {user_input}") # 1. 意图分类 task_type, params = self.classifier.classify(user_input) print(f"📋 识别任务: {task_type.value}") # 2. 收集上下文 context = self.context_collector.collect(current_file) if selected_code: context["selected_code"] = selected_code print(f"📁 项目语言: {context['language']}, 框架: {context['framework']}") # 3. 需求细化 requirement = self.refiner.refine(user_input, context) requirement["task"] = task_type.value requirement["language"] = context["language"] print(f"📝 需求规格: {requirement.get('name', '未命名')}") # 4. 代码生成（带审查） result = self.generator.generate_with_review(requirement, selected_code) # 5. 确定插入位置 result["insertion_point"] = self._determine_insertion_point( requirement, current_file, selected_code ) # 6. 计算置信度 result["confidence"] = self._calculate_confidence(result) return result def _determine_insertion_point(self, requirement: dict, current_file: str, selected_code: str) -> str: """确定代码插入位置""" if selected_code: return "replace_selection" if requirement.get("task") == "create_function": return "end_of_file" elif requirement.get("task") == "create_class": return "end_of_file" elif requirement.get("task") == "fix_bug": return "replace_current" else: return "cursor_position" def _calculate_confidence(self, result: dict) -> float: """计算置信度""" base = 0.8 # 有审查且问题少则加分 review = result.get("review", {}) issues_count = len(review.get("issues", [])) base -= issues_count * 0.05 # 有解释则加分 if result.get("explanation"): base += 0.05 # 有依赖信息则加分 if result.get("dependencies"): base += 0.02 return max(0.1, min(1.0, base))

六、完整测试

# test_nl2code.py from engine.nl2code.pipeline import NL2CodePipeline import json # 初始化 api_key = "your-api-key" pipeline = NL2CodePipeline(api_key, project_root="./my_project") # 测试1：创建函数 print("=" * 60) print("测试1：创建函数") print("=" * 60) result = pipeline.execute( "写一个函数，读取 CSV 文件并计算某一列的平均值，需要处理文件不存在的情况", current_file="./my_project/data_processor.py" ) print(f"\n📊 置信度: {result['confidence']:.2%}") print(f"📌 插入位置: {result['insertion_point']}") print(f"\n📝 生成的代码:\n{result['code']}") if result.get('explanation'): print(f"\n💡 说明:\n{result['explanation']}") if result.get('review', {}).get('issues'): print(f"\n⚠️ 审查发现的问题:") for issue in result['review']['issues']: print(f" - {issue}") # 测试2：修复 Bug print("\n" + "=" * 60) print("测试2：修复 Bug") print("=" * 60) buggy_code = """ def divide(a, b): return a / b """ result = pipeline.execute( "这个函数除以零会崩溃，帮我加上错误处理", selected_code=buggy_code ) print(f"\n📊 置信度: {result['confidence']:.2%}") print(f"\n📝 修复后的代码:\n{result['code']}") # 测试3：生成测试 print("\n" + "=" * 60) print("测试3：生成测试") print("=" * 60) target_code = """ def fibonacci(n): if n <= 0: return 0 elif n == 1: return 1 else: return fibonacci(n-1) + fibonacci(n-2) """ result = pipeline.execute( "为这个斐波那契函数生成单元测试，覆盖正常情况和边界情况", selected_code=target_code ) print(f"\n📊 置信度: {result['confidence']:.2%}") print(f"\n📝 生成的测试:\n{result['code']}")

七、性能优化与最佳实践

7.1 流式输出

对于较长的代码生成，可以使用流式输出让用户看到生成过程：

# engine/nl2code/stream_generator.py import requests import json from typing import Generator class StreamCodeGenerator: """流式代码生成器""" def __init__(self, api_key: str): self.api_key = api_key def generate_stream(self, requirement: dict) -> Generator[str, None, None]: """流式生成代码""" prompt = self._build_prompt(requirement) response = requests.post( "https://api.deepseek.com/v1/chat/completions", headers={"Authorization": f"Bearer {self.api_key}"}, json={ "model": "deepseek-coder", "messages": [ {"role": "system", "content": "你是一个代码生成器。生成完整可运行的代码。"}, {"role": "user", "content": prompt} ], "stream": True, "temperature": 0.3 }, stream=True ) for line in response.iter_lines(): if line: line = line.decode('utf-8') if line.startswith('data: '): data = line[6:] if data != '[DONE]': try: chunk = json.loads(data) if chunk['choices'][0]['delta'].get('content'): yield chunk['choices'][0]['delta']['content'] except: continue

7.2 代码模板缓存

# engine/nl2code/template_cache.py import hashlib import json from pathlib import Path class TemplateCache: """代码模板缓存""" def __init__(self, cache_dir: str = ".cache/templates"): self.cache_dir = Path(cache_dir) self.cache_dir.mkdir(parents=True, exist_ok=True) def _make_key(self, requirement: dict) -> str: content = json.dumps(requirement, sort_keys=True) return hashlib.md5(content.encode()).hexdigest() def get(self, requirement: dict) -> dict: """获取缓存的代码""" key = self._make_key(requirement) cache_file = self.cache_dir / f"{key}.json" if cache_file.exists(): with open(cache_file, 'r') as f: return json.load(f) return None def set(self, requirement: dict, result: dict): """缓存代码""" key = self._make_key(requirement) cache_file = self.cache_dir / f"{key}.json" with open(cache_file, 'w') as f: json.dump(result, f)

八、常见错误 & 排坑指南

1. 生成的代码不符合项目风格

   • 原因：没有提供足够的项目上下文

   • 解决：在 prompt 中包含更多项目代码示例

2. 生成的代码过于冗长

   • 原因：LLM 倾向于生成完整的、带大量注释的代码

   • 解决：在 prompt 中指定“保持简洁，只生成必要的代码”

3. 多轮对话中遗忘上下文

   • 原因：每次调用都是独立的，没有维护历史

   • 解决：将对话历史压缩后加入 prompt

4. 生成的代码使用了不存在的库

   • 原因：LLM 的知识截止日期之后的新库

   • 解决：在 prompt 中指定允许使用的库列表

九、课后作业

1. 实现代码解释器：让用户选中一段代码，AI 用自然语言解释它的作用。

2. 添加代码转换功能：支持“把这段 Python 代码转换成 JavaScript”。

3. 挑战题：实现“多步代码生成”——用户说“创建一个 REST API，包含用户 CRUD 操作”，系统自动生成完整的项目结构（路由、模型、控制器）。

十、总结

这一讲我们实现了自然语言转代码的完整管道：

• 意图分类：识别用户想做什么（创建函数、修复 Bug、生成测试等）

• 需求细化：将模糊的自然语言转化为结构化需求规格

• 代码生成：利用 LLM 生成高质量代码

• 自我审查：自动检查代码质量问题并迭代改进

• 上下文集成：理解项目使用的语言、框架、代码风格

现在，MiniCopilot 不仅能补全代码，还能根据自然语言描述直接生成代码。

下一讲，我们将实现代码审查与 Bug 检测——让 MiniCopilot 能自动发现代码中的问题。