29.3 代码生成模块

29.3.1 代码生成概述

代码生成模块是编程 Agent 的核心能力之一，它能够根据自然语言描述生成高质量的代码。代码生成涉及需求理解、架构设计、代码实现等多个环节。

代码生成流程

用户需求 ↓ 需求分析与理解 ↓ 架构设计 ↓ 代码实现 ↓ 代码验证 ↓ 优化与改进 ↓ 最终代码

29.3.2 需求分析

需求提取器

 
    class RequirementExtractor:
        """需求提取器"""
 
        def __init__(self, llm_client: LLMClient):
            self.llm_client = llm_client
 
        async def extract(self, user_request: str) -> Requirement:
            """提取需求"""
            prompt = f"""
            分析用户需求，提取关键信息：
 
            用户需求：{user_request}
 
            请提取以下信息：
            1. 功能需求（需要实现什么功能）
            2. 技术栈（使用的编程语言、框架等）
            3. 约束条件（性能、安全、兼容性等）
            4. 输入输出（预期的输入和输出）
            5. 特殊要求（代码风格、注释要求等）
 
            以 JSON 格式返回结果。
            """
 
            response = await self.llm_client.complete(prompt)
            return self._parse_requirement(response)
 
        def _parse_requirement(self, response: str) -> Requirement:
            """解析需求"""
            try:
                data = json.loads(response)
                return Requirement(
                    functional_requirements=data.get('functional_requirements', []),
                    tech_stack=data.get('tech_stack', {}),
                    constraints=data.get('constraints', {}),
                    inputs=data.get('inputs', []),
                    outputs=data.get('outputs', []),
                    special_requirements=data.get('special_requirements', {})
                )
            except json.JSONDecodeError:
                raise ValueError("Invalid requirement format")
 
    ```### 需求验证器
 
```python
    class RequirementValidator:
    """需求验证器"""
    def validate(self, requirement: Requirement) -> ValidationResult:
    """验证需求"""
    issues = []
    # 检查功能需求
    if not requirement.functional_requirements:
    issues.append("No functional requirements specified")
    # 检查技术栈
    if not requirement.tech_stack:
    issues.append("No tech stack specified")
    # 检查约束条件
    if 'performance' in requirement.constraints:
    perf = requirement.constraints['performance']
    if not isinstance(perf, dict) or 'max_time' not in perf:
    issues.append("Invalid performance constraint")
    return ValidationResult(
    valid=len(issues) == 0,
    issues=issues
    )

29.3.3 架构设计

架构设计器

 
class ArchitectureDesigner:
    """架构设计器"""
 
    def __init__(self, llm_client: LLMClient):
        self.llm_client = llm_client
        self.design_patterns = self._load_design_patterns()
 
    async def design(self, requirement: Requirement) -> Architecture:
        """设计架构"""
        prompt = f"""
        根据需求设计软件架构：
 
        功能需求：{requirement.functional_requirements}
        技术栈：{requirement.tech_stack}
        约束条件：{requirement.constraints}
 
        请设计：
        1. 系统架构（模块划分、层次结构）
        2. 类设计（类、接口、继承关系）
        3. 数据结构（数据模型、存储方案）
        4. 接口设计（API、函数签名）
        5. 设计模式（适用的设计模式）
 
        以 JSON 格式返回架构设计。
        """
 
        response = await self.llm_client.complete(prompt)
        return self._parse_architecture(response)
 
    def _parse_architecture(self, response: str) -> Architecture:
        """解析架构"""
        try:
            data = json.loads(response)
            return Architecture(
                system_architecture=data.get('system_architecture', {}),
                class_design=data.get('class_design', []),
                data_structures=data.get('data_structures', []),
                interfaces=data.get('interfaces', []),
                design_patterns=data.get('design_patterns', [])
            )
        except json.JSONDecodeError:
            raise ValueError("Invalid architecture format")
 
    def _load_design_patterns(self) -> Dict[str, DesignPattern]:
        """加载设计模式"""
        return {
            'singleton': DesignPattern(
                name='Singleton',
                description='确保一个类只有一个实例',
               适用场景='需要全局唯一访问点'
            ),
            'factory': DesignPattern(
                name='Factory',
                description='创建对象的接口',
                适用场景='需要灵活创建对象'
            ),
            'observer': DesignPattern(
                name='Observer',
                description='定义对象间的一对多依赖',
                适用场景='需要事件通知机制'
            )
        }
 
```### 架构评估器
 
```python
class ArchitectureEvaluator:

"""架构评估器"""

    def evaluate(self, architecture: Architecture,
    requirement: Requirement) -> EvaluationResult:

"""评估架构""" scores = {}

评估模块化

scores['modularity'] = self._evaluate_modularity(architecture)

评估可扩展性

scores['extensibility'] = self._evaluate_extensibility(architecture)

评估性能

scores['performance'] = self._evaluate_performance( architecture, requirement )

评估可维护性

scores['maintainability'] = self._evaluate_maintainability(architecture)

计算总分

total_score = sum(scores.values()) / len(scores)

    return EvaluationResult(
    total_score=total_score,
    scores=scores,
    recommendations=self._generate_recommendations(scores)
    )
    def _evaluate_modularity(self, architecture: Architecture) -> float:

"""评估模块化"""

检查模块划分

modules = architecture.system_architecture.get('modules', [])

    if not modules:
    return 0.0

模块越多，模块化程度越高

score = min(len(modules) / 10.0, 1.0)

    return score
    def _evaluate_extensibility(self, architecture: Architecture) -> float:

"""评估可扩展性"""

检查设计模式使用

patterns = architecture.design_patterns

    if not patterns:
    return 0.5

使用设计模式提高可扩展性

score = 0.5 + min(len(patterns) / 5.0, 0.5)

    return score
    def _evaluate_performance(self, architecture: Architecture,
    requirement: Requirement) -> float:

"""评估性能"""

检查性能约束

constraints = requirement.constraints.get('performance', {})

    if not constraints:
    return 0.8  # 默认分数

评估架构是否满足性能要求

score = 0.8 # 基础分数

检查缓存策略

    if 'caching' in architecture.system_architecture:
    score += 0.1

检查并发处理

    if 'concurrency' in architecture.system_architecture:
    score += 0.1
    return min(score, 1.0)
    def _evaluate_maintainability(self, architecture: Architecture) -> float:

"""评估可维护性"""

检查类设计

    classes = architecture.class_design
    if not classes:
    return 0.5

评估类的复杂度

avg_methods = sum( len(c.get('methods', [])) for c in classes ) / len(classes)

方法数量适中，可维护性高

    if 5 <= avg_methods <= 15:
    score = 1.0
    elif avg_methods < 5:
    score = 0.8
    else:
    score = 0.6
    return score
    def _generate_recommendations(self,
    scores: Dict[str, float]) -> List[str]:

"""生成建议""" recommendations = []

    if scores['modularity'] < 0.7:
    recommendations.append(

"建议增加模块划分，提高模块化程度" )

    if scores['extensibility'] < 0.7:
    recommendations.append(

"建议使用更多设计模式，提高可扩展性" )

    if scores['maintainability'] < 0.7:
    recommendations.append(

"建议简化类设计，降低复杂度" )

    return recommendations

## 29.3.4 代码实现

### 代码生成器

    ```python

```python
    class CodeGenerator:

"""代码生成器"""

        def __init__(self, llm_client: LLMClient):
            self.llm_client = llm_client
            self.code_templates = self._load_code_templates()
 
        async def generate(self, architecture: Architecture,
                          requirement: Requirement) -> GeneratedCode:

"""生成代码"""

生成类代码

            class_codes = []
            for class_design in architecture.class_design:
                code = await self._generate_class_code(
                    class_design,
                    requirement
                )
                class_codes.append(code)

生成接口代码

interface_codes = []

            for interface in architecture.interfaces:
                code = await self._generate_interface_code(
                    interface,
                    requirement
                )
                interface_codes.append(code)

生成主程序代码

main_code = await self._generate_main_code( architecture, requirement )

组合所有代码

full_code = self._combine_codes(

                class_codes,
                interface_codes,
                main_code
            )
 
            return GeneratedCode(
                full_code=full_code,
                class_codes=class_codes,
                interface_codes=interface_codes,
                main_code=main_code
            )
 
        async def _generate_class_code(self, class_design: Dict,
                                      requirement: Requirement) -> str:

"""生成类代码""" prompt = f""" 根据类设计生成代码：

类名：{class_design.get('name')} 方法：{class_design.get('methods', [])} 属性：{class_design.get('attributes', [])} 父类：{class_design.get('parent', 'None')} 编程语言：{requirement.tech_stack.get('language', 'Python')}

请生成完整的类代码，包括：

  1. 类定义
  2. 所有方法的实现
  3. 必要的注释
  4. 错误处理 """
            return await self.llm_client.complete(prompt)
 
        async def _generate_interface_code(self, interface: Dict,
                                           requirement: Requirement) -> str:

"""生成接口代码""" prompt = f""" 根据接口设计生成代码：

接口名：{interface.get('name')} 方法：{interface.get('methods', [])} 编程语言：{requirement.tech_stack.get('language', 'Python')}

请生成完整的接口代码。 """

            return await self.llm_client.complete(prompt)
 
        async def _generate_main_code(self, architecture: Architecture,
                                      requirement: Requirement) -> str:

"""生成主程序代码""" prompt = f""" 根据架构和需求生成主程序代码：

功能需求：{requirement.functional_requirements} 类：{[c.get('name') for c in architecture.class_design]} 接口：{[i.get('name') for i in architecture.interfaces]} 编程语言：{requirement.tech_stack.get('language', 'Python')}

请生成主程序代码，包括：

  1. 初始化代码
  2. 主要业务逻辑
  3. 示例用法 """
            return await self.llm_client.complete(prompt)
 
        def _combine_codes(self, class_codes: List[str],
                          interface_codes: List[str],
                          main_code: str) -> str:

"""组合代码""" combined = []

添加导入

            combined.append("# Generated Code")
            combined.append("")

添加接口

            if interface_codes:
                combined.append("# Interfaces")
                for code in interface_codes:
                    combined.append(code)
                    combined.append("")

添加类

            if class_codes:
                combined.append("# Classes")
                for code in class_codes:
                    combined.append(code)
                    combined.append("")

添加主程序

            combined.append("# Main Program")
            combined.append(main_code)
 
            return "\n".join(combined)
 
 
class CodeOptimizer:
"""代码优化器"""
def __init__(self, llm_client: LLMClient):
self.llm_client = llm_client
async def optimize(self, code: str,
requirement: Requirement) -> OptimizedCode:
"""优化代码"""
# 分析代码问题
issues = await self._analyze_issues(code)
# 生成优化建议
suggestions = await self._generate_suggestions(
code,
issues,
requirement
)
# 应用优化
optimized_code = await self._apply_optimizations(
code,
suggestions
)
return OptimizedCode(
original_code=code,
optimized_code=optimized_code,
issues=issues,
suggestions=suggestions
)
async def _analyze_issues(self, code: str) -> List[CodeIssue]:
"""分析代码问题"""
prompt = f"""
分析以下代码的问题：
{code}
请识别：
1. 性能问题
2. 安全问题
3. 代码风格问题
4. 潜在的 bug
5. 可维护性问题
以 JSON 格式返回问题列表。
"""
response = await self.llm_client.complete(prompt)
return self._parse_issues(response)
async def _generate_suggestions(self, code: str,
issues: List[CodeIssue],
requirement: Requirement) -> List[Suggestion]:
"""生成优化建议"""
prompt = f"""
基于代码问题生成优化建议：
代码：{code}
问题：{issues}
约束条件：{requirement.constraints}
请生成具体的优化建议，包括：
1. 问题描述
2. 优化方案
3. 预期效果
以 JSON 格式返回建议列表。
"""
response = await self.llm_client.complete(prompt)
return self._parse_suggestions(response)
async def _apply_optimizations(self, code: str,
suggestions: List[Suggestion]) -> str:
"""应用优化"""
optimized_code = code
for suggestion in suggestions:
if suggestion.applicable:
optimized_code = await self._apply_suggestion(
optimized_code,
suggestion
)
return optimized_code
async def _apply_suggestion(self, code: str,
suggestion: Suggestion) -> str:
"""应用单个建议"""
prompt = f"""
应用以下优化建议到代码：
原始代码：{code}
优化建议：{suggestion.description}
优化方案：{suggestion.solution}
请返回优化后的代码。
"""
return await self.llm_client.complete(prompt)

29.3.5 代码验证

代码验证器

 
class CodeValidator:
    """代码验证器"""
 
    def __init__(self, tool_manager: ToolManager):
        self.tool_manager = tool_manager
 
    async def validate(self, code: str,
                      requirement: Requirement) -> ValidationResult:
        """验证代码"""
 
        results = []
 
        # 语法检查
 
        syntax_result = await self._check_syntax(code, requirement)
        results.append(syntax_result)
 
        # 类型检查
 
        type_result = await self._check_types(code, requirement)
        results.append(type_result)
 
        # 逻辑检查
 
        logic_result = await self._check_logic(code, requirement)
        results.append(logic_result)
 
        # 性能检查
 
        performance_result = await self._check_performance(
            code,
            requirement
        )
        results.append(performance_result)
 
        # 综合结果
 
        all_passed = all(r.passed for r in results)
 
        return ValidationResult(
            passed=all_passed,
            results=results,
            issues=self._collect_issues(results)
        )
 
    async def _check_syntax(self, code: str,
                           requirement: Requirement) -> CheckResult:
        """检查语法"""
        language = requirement.tech_stack.get('language', 'python')
 
        try:
            if language == 'python':
                result = await self._check_python_syntax(code)
            else:
                result = CheckResult(
                    check_type='syntax',
                    passed=True,
                    message=f"Syntax check for {language} not implemented"
                )
 
            return result
 
        except Exception as e:
            return CheckResult(
                check_type='syntax',
                passed=False,
                message=f"Syntax error: {str(e)}"
            )
 
    async def _check_python_syntax(self, code: str) -> CheckResult:
        """检查 Python 语法"""
        try:
            compile(code, '<string>', 'exec')
            return CheckResult(
                check_type='syntax',
                passed=True,
                message="Syntax is valid"
            )
        except SyntaxError as e:
            return CheckResult(
                check_type='syntax',
                passed=False,
                message=f"Syntax error at line {e.lineno}: {e.msg}"
            )
 
    async def _check_types(self, code: str,
                          requirement: Requirement) -> CheckResult:
        """检查类型"""
 
        # 使用类型检查工具
 
        tool = self.tool_manager.get_tool('type_checker')
 
        if not tool:
            return CheckResult(
                check_type='type',
                passed=True,
                message="Type checker not available"
            )
 
        try:
            result = await tool.execute({'code': code})
 
            if result.success:
                return CheckResult(
                    check_type='type',
                    passed=True,
                    message="Type check passed"
                )
            else:
                return CheckResult(
                    check_type='type',
                    passed=False,
                    message=f"Type check failed: {result.error}"
                )
        except Exception as e:
            return CheckResult(
                check_type='type',
                passed=False,
                message=f"Type check error: {str(e)}"
            )
 
    async def _check_logic(self, code: str,
                          requirement: Requirement) -> CheckResult:
        """检查逻辑"""
 
        # 分析代码逻辑
 
        issues = []
 
        # 检查空指针
 
        if 'None' in code and 'if' not in code:
            issues.append("Potential None reference without check")
 
        # 检查资源泄漏
 
        if 'open(' in code and 'close(' not in code:
            issues.append("Potential resource leak (file not closed)")
 
        if issues:
            return CheckResult(
                check_type='logic',
                passed=False,
                message=f"Logic issues: {', '.join(issues)}"
            )
        else:
            return CheckResult(
                check_type='logic',
                passed=True,
                message="Logic check passed"
            )
 
    async def _check_performance(self, code: str,
                                requirement: Requirement) -> CheckResult:
        """检查性能"""
        issues = []
 
        # 检查嵌套循环
 
        if code.count('for ') > 2:
            issues.append("Deep nested loops may cause performance issues")
 
        # 检查大列表操作
 
        if 'list(' in code and 'range(' in code:
            issues.append("Consider using generator expressions for large ranges")
 
        if issues:
            return CheckResult(
                check_type='performance',
                passed=False,
                message=f"Performance issues: {', '.join(issues)}"
            )
        else:
            return CheckResult(
                check_type='performance',
                passed=True,
                message="Performance check passed"
            )
 
    def _collect_issues(self,
                       results: List[CheckResult]) -> List[str]:
        """收集所有问题"""
        issues = []
 
        for result in results:
            if not result.passed:
                issues.append(result.message)
 
        return issues

通过实现这些组件，我们可以构建一个完整的代码生成模块，能够从需求分析到代码验证的全流程自动化。