课程概述

本课程将深入学习Cursor AI的高级功能，包括自定义提示词、代码模板创建、工作流自动化等，帮助你进一步提升开发效率，实现个性化的AI辅助开发体验。

学习目标

• 掌握自定义提示词的创建和使用方法
• 学会创建和管理代码模板
• 理解工作流自动化的实现原理
• 掌握高级功能配置和优化技巧

课程大纲

1. 自定义提示词

1.1 提示词基础概念

• 提示词定义：指导AI生成代码的指令和描述
• 提示词结构：包含任务描述、约束条件、示例等
• 提示词类型：功能型、优化型、重构型、测试型等
• 提示词效果：影响AI生成代码的质量和准确性

1.2 提示词设计原则

1. 明确性：清晰描述需求和目标
2. 具体性：提供具体的输入输出要求
3. 约束性：明确代码的限制和规范
4. 示例性：提供参考示例和代码风格

1.3 提示词模板系统

// 提示词配置文件
{
    "prompts": {
        "web_api": {
            "name": "Web API开发",
            "description": "生成RESTful API接口",
            "template": "请创建一个{framework}的{method}接口，路径为{path}，功能是{function}。要求：\n1. 包含完整的错误处理\n2. 添加输入验证\n3. 返回标准JSON格式\n4. 包含API文档注释\n\n示例输入：{input_example}\n期望输出：{output_example}",
            "variables": {
                "framework": "Flask",
                "method": "POST",
                "path": "/api/users",
                "function": "创建新用户",
                "input_example": "{\"username\": \"testuser\", \"email\": \"test@example.com\"}",
                "output_example": "{\"success\": true, \"user_id\": 1, \"message\": \"用户创建成功\"}"
            },
            "tags": ["api", "web", "backend"],
            "usage_count": 0
        },
        "database_model": {
            "name": "数据库模型",
            "description": "生成数据库模型类",
            "template": "请创建一个{orm}的{model_name}模型类，包含以下字段：{fields}。要求：\n1. 使用适当的字段类型\n2. 添加字段验证\n3. 包含关系定义\n4. 添加模型方法\n\n数据库类型：{db_type}",
            "variables": {
                "orm": "SQLAlchemy",
                "model_name": "User",
                "fields": "id, username, email, password, created_at, updated_at",
                "db_type": "MySQL"
            },
            "tags": ["database", "model", "orm"],
            "usage_count": 0
        },
        "test_case": {
            "name": "测试用例生成",
            "description": "为代码生成测试用例",
            "template": "请为以下{language}代码生成完整的测试用例：\n\n{code}\n\n要求：\n1. 使用{test_framework}框架\n2. 测试正常情况和异常情况\n3. 包含边界条件测试\n4. 使用Mock对象模拟依赖\n5. 达到{coverage}%的测试覆盖率",
            "variables": {
                "language": "Python",
                "test_framework": "pytest",
                "coverage": "90"
            },
            "tags": ["testing", "unit_test", "quality"],
            "usage_count": 0
        }
    }
}

1.4 提示词管理系统

# 提示词管理类
import json
import os
from typing import Dict, List, Optional
from datetime import datetime

class PromptManager:
    """提示词管理类"""
   
    def __init__(self, config_file: str = "prompts_config.json"):
        self.config_file = config_file
        self.prompts = self._load_prompts()
   
    def _load_prompts(self) -> Dict:
        """加载提示词配置"""
        if os.path.exists(self.config_file):
            with open(self.config_file, 'r', encoding='utf-8') as f:
                return json.load(f)
        return {"prompts": {}}
   
    def _save_prompts(self):
        """保存提示词配置"""
        with open(self.config_file, 'w', encoding='utf-8') as f:
            json.dump(self.prompts, f, ensure_ascii=False, indent=2)
   
    def create_prompt(self, prompt_id: str, name: str, description: str,
                     template: str, variables: Dict, tags: List[str]) -> bool:
        """创建新的提示词"""
        if prompt_id in self.prompts["prompts"]:
            return False
       
        self.prompts["prompts"][prompt_id] = {
            "name": name,
            "description": description,
            "template": template,
            "variables": variables,
            "tags": tags,
            "usage_count": 0,
            "created_at": str(datetime.now())
        }
       
        self._save_prompts()
        return True
   
    def get_prompt(self, prompt_id: str) -> Optional[Dict]:
        """获取提示词"""
        return self.prompts["prompts"].get(prompt_id)
   
    def render_prompt(self, prompt_id: str, **kwargs) -> Optional[str]:
        """渲染提示词模板"""
        prompt = self.get_prompt(prompt_id)
        if not prompt:
            return None
       
        try:
            rendered = prompt["template"]
            for key, value in kwargs.items():
                if key in prompt["variables"]:
                    rendered = rendered.replace(f"{{{key}}}", str(value))
           
            # 更新使用次数
            prompt["usage_count"] += 1
            self._save_prompts()
           
            return rendered
        except Exception as e:
            print(f"渲染提示词失败: {e}")
            return None
   
    def search_prompts(self, query: str, tags: List[str] = None) -> List[Dict]:
        """搜索提示词"""
        results = []
        query_lower = query.lower()
       
        for prompt_id, prompt in self.prompts["prompts"].items():
            # 文本搜索
            if (query_lower in prompt["name"].lower() or
                query_lower in prompt["description"].lower()):
                results.append({"id": prompt_id, **prompt})
                continue
            
            # 标签搜索
            if tags and any(tag in prompt["tags"] for tag in tags):
                results.append({"id": prompt_id, **prompt})
       
        return results
   
    def update_prompt(self, prompt_id: str, **kwargs) -> bool:
        """更新提示词"""
        if prompt_id not in self.prompts["prompts"]:
            return False
       
        prompt = self.prompts["prompts"][prompt_id]
        for key, value in kwargs.items():
            if key in prompt:
                prompt[key] = value
       
        prompt["updated_at"] = str(datetime.now())
        self._save_prompts()
        return True
   
    def delete_prompt(self, prompt_id: str) -> bool:
        """删除提示词"""
        if prompt_id in self.prompts["prompts"]:
            del self.prompts["prompts"][prompt_id]
            self._save_prompts()
            return True
        return False

1.5 提示词使用示例

# 使用自定义提示词生成代码
def generate_web_api():
    """使用Web API提示词生成代码"""
    prompt_manager = PromptManager()
   
    # 渲染提示词
    prompt = prompt_manager.render_prompt(
        "web_api",
        framework="Flask",
        method="POST",
        path="/api/products",
        function="创建新产品",
        input_example='{"name": "iPhone 15", "price": 5999, "category": "electronics"}',
        output_example='{"success": true, "product_id": 123, "message": "产品创建成功"}'
    )
   
    if prompt:
        print("生成的提示词:")
        print(prompt)
       
        # 这里可以调用AI API生成代码
        # generated_code = ai_generate_code(prompt)
        # return generated_code
   
    return None

# 使用数据库模型提示词
def generate_database_model():
    """使用数据库模型提示词生成代码"""
    prompt_manager = PromptManager()
   
    prompt = prompt_manager.render_prompt(
        "database_model",
        orm="SQLAlchemy",
        model_name="Product",
        fields="id, name, description, price, category, stock, created_at, updated_at",
        db_type="MySQL"
    )
   
    if prompt:
        print("生成的提示词:")
        print(prompt)
   
    return None

2. 代码模板创建

2.1 模板系统架构

• 模板引擎：基于Jinja2的模板渲染系统
• 模板存储：本地文件系统和云端同步
• 模板分类：按项目类型、语言、功能分类
• 模板版本：支持模板版本管理和更新

2.2 模板创建工具

# 代码模板创建工具
from jinja2 import Template, Environment, FileSystemLoader
import os
import re
from typing import Optional, List

class CodeTemplateManager:
    """代码模板管理类"""
   
    def __init__(self, template_dir: str = "templates"):
        self.template_dir = template_dir
        self.env = Environment(loader=FileSystemLoader(template_dir))
        self.env.trim_blocks = True
        self.env.lstrip_blocks = True
       
        # 注册自定义过滤器
        self.env.filters['snake_case'] = self._to_snake_case
        self.env.filters['camel_case'] = self._to_camel_case
        self.env.filters['pluralize'] = self._pluralize
   
    def _to_snake_case(self, text: str) -> str:
        """转换为蛇形命名"""
        return re.sub(r'(?<!^)(?=[A-Z])', '_', text).lower()
   
    def _to_camel_case(self, text: str) -> str:
        """转换为驼峰命名"""
        return ''.join(word.capitalize() for word in text.split('_'))
   
    def _pluralize(self, text: str) -> str:
        """复数化处理"""
        if text.endswith('y'):
            return text[:-1] + 'ies'
        elif text.endswith(('s', 'x', 'z', 'ch', 'sh')):
            return text + 'es'
        else:
            return text + 's'
   
    def create_template(self, template_name: str, template_content: str,
                       template_type: str = "general") -> bool:
        """创建新的代码模板"""
        template_path = os.path.join(self.template_dir, template_type, f"{template_name}.jinja2")
       
        # 确保目录存在
        os.makedirs(os.path.dirname(template_path), exist_ok=True)
       
        try:
            with open(template_path, 'w', encoding='utf-8') as f:
                f.write(template_content)
            return True
        except Exception as e:
            print(f"创建模板失败: {e}")
            return False
   
    def render_template(self, template_name: str, template_type: str = "general",
                       **context) -> Optional[str]:
        """渲染代码模板"""
        try:
            template_path = f"{template_type}/{template_name}.jinja2"
            template = self.env.get_template(template_path)
            return template.render(**context)
        except Exception as e:
            print(f"渲染模板失败: {e}")
            return None
   
    def list_templates(self, template_type: str = None) -> List[str]:
        """列出可用模板"""
        templates = []
       
        if template_type:
            type_dir = os.path.join(self.template_dir, template_type)
            if os.path.exists(type_dir):
                for file in os.listdir(type_dir):
                    if file.endswith('.jinja2'):
                        templates.append(file[:-7])  # 移除.jinja2扩展名
        else:
            for root, dirs, files in os.walk(self.template_dir):
                for file in files:
                    if file.endswith('.jinja2'):
                        rel_path = os.path.relpath(os.path.join(root, file), self.template_dir)
                        templates.append(rel_path[:-7])
       
        return templates

2.3 常用代码模板

{# Flask应用模板 #}
{# templates/flask/app.jinja2 #}
from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
from flask_cors import CORS
import os

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = os.environ.get('DATABASE_URL', 'sqlite:///app.db')
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY', 'dev-secret-key')

db = SQLAlchemy(app)
CORS(app)

# 模型定义
class {{ model_name }}(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    {% for field in fields %}
    {{ field.name }} = db.Column(db.{{ field.type }}{% if field.length %}({{ field.length }}){% endif %}{% if field.nullable == false %}, nullable=False{% endif %})
    {% endfor %}
    created_at = db.Column(db.DateTime, default=db.func.current_timestamp())
    updated_at = db.Column(db.DateTime, default=db.func.current_timestamp(), onupdate=db.func.current_timestamp())

# 路由定义
@app.route('/api/{{ model_name|pluralize }}', methods=['GET'])
def get_{{ model_name|pluralize }}():
    """获取所有{{ model_name }}"""
    try:
        {{ model_name|pluralize }} = {{ model_name }}.query.all()
        return jsonify({
            'success': True,
            'data': [{{ model_name|snake_case }}.to_dict() for {{ model_name|snake_case }} in {{ model_name|pluralize }}]
        }), 200
    except Exception as e:
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

@app.route('/api/{{ model_name|pluralize }}/<int:{{ model_name|snake_case }}_id>', methods=['GET'])
def get_{{ model_name }}({{ model_name|snake_case }}_id):
    """获取单个{{ model_name }}"""
    try:
        {{ model_name|snake_case }} = {{ model_name }}.query.get_or_404({{ model_name|snake_case }}_id)
        return jsonify({
            'success': True,
            'data': {{ model_name|snake_case }}.to_dict()
        }), 200
    except Exception as e:
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

@app.route('/api/{{ model_name|pluralize }}', methods=['POST'])
def create_{{ model_name }}():
    """创建新{{ model_name }}"""
    try:
        data = request.get_json()
        new_{{ model_name|snake_case }} = {{ model_name }}(**data)
        db.session.add(new_{{ model_name|snake_case }})
        db.session.commit()
       
        return jsonify({
            'success': True,
            'message': '{{ model_name }}创建成功',
            'data': new_{{ model_name|snake_case }}.to_dict()
        }), 201
    except Exception as e:
        db.session.rollback()
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

if __name__ == '__main__':
    with app.app_context():
        db.create_all()
    app.run(debug=True)

{# 数据库模型模板 #}
{# templates/sqlalchemy/model.jinja2 #}
from sqlalchemy import Column, Integer, String, DateTime, Boolean, Text, Float, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from datetime import datetime

Base = declarative_base()

class {{ class_name }}(Base):
    """{{ description }}"""
    __tablename__ = '{{ table_name }}'
   
    # 主键字段
    id = Column(Integer, primary_key=True, autoincrement=True)
   
    # 业务字段
    {% for field in fields %}
    {{ field.name }} = Column({{ field.type }}{% if field.length %}({{ field.length }}){% endif %}{% if field.nullable == false %}, nullable=False{% endif %}{% if field.default %}, default={{ field.default }}{% endif %}{% if field.comment %}, comment='{{ field.comment }}'{% endif %})
    {% endfor %}
   
    # 时间字段
    created_at = Column(DateTime, default=func.now(), comment='创建时间')
    updated_at = Column(DateTime, default=func.now(), onupdate=func.now(), comment='更新时间')
   
    # 关系定义
    {% for relation in relations %}
    {{ relation.name }} = relationship('{{ relation.target_class }}', back_populates='{{ relation.back_populates }}')
    {% endfor %}
   
    def __repr__(self):
        return f'<{{ class_name }}(id={self.id})>'
   
    def to_dict(self):
        """转换为字典"""
        return {
            'id': self.id,
            {% for field in fields %}
            '{{ field.name }}': self.{{ field.name }},
            {% endfor %}
            'created_at': self.created_at.isoformat() if self.created_at else None,
            'updated_at': self.updated_at.isoformat() if self.updated_at else None
        }
   
    @classmethod
    def from_dict(cls, data):
        """从字典创建实例"""
        # 过滤掉非字段属性
        field_names = [field.name for field in cls.__table__.columns]
        filtered_data = {k: v for k, v in data.items() if k in field_names}
        return cls(**filtered_data)

2.4 模板使用示例

# 使用模板生成Flask应用
def generate_flask_app():
    """生成Flask应用代码"""
    template_manager = CodeTemplateManager()
   
    # 定义模型字段
    fields = [
        {"name": "name", "type": "String", "length": 100, "nullable": False},
        {"name": "description", "type": "Text", "nullable": True},
        {"name": "price", "type": "Float", "nullable": False},
        {"name": "category", "type": "String", "length": 50, "nullable": False}
    ]
   
    # 渲染模板
    context = {
        "model_name": "Product",
        "fields": fields
    }
   
    generated_code = template_manager.render_template("app", "flask", **context)
   
    if generated_code:
        # 保存生成的代码
        with open("generated_app.py", "w", encoding="utf-8") as f:
            f.write(generated_code)
        print("Flask应用代码生成成功！")
        return generated_code
   
    return None

# 使用模板生成数据库模型
def generate_database_model():
    """生成数据库模型代码"""
    template_manager = CodeTemplateManager()
   
    # 定义字段和关系
    fields = [
        {"name": "username", "type": "String", "length": 80, "nullable": False, "comment": "用户名"},
        {"name": "email", "type": "String", "length": 120, "nullable": False, "comment": "邮箱"},
        {"name": "password_hash", "type": "String", "length": 255, "nullable": False, "comment": "密码哈希"},
        {"name": "is_active", "type": "Boolean", "default": True, "comment": "是否激活"}
    ]
   
    relations = [
        {"name": "orders", "target_class": "Order", "back_populates": "user"}
    ]
   
    # 渲染模板
    context = {
        "class_name": "User",
        "description": "用户模型",
        "table_name": "users",
        "fields": fields,
        "relations": relations
    }
   
    generated_code = template_manager.render_template("model", "sqlalchemy", **context)
   
    if generated_code:
        # 保存生成的代码
        with open("generated_user_model.py", "w", encoding="utf-8") as f:
            f.write(generated_code)
        print("用户模型代码生成成功！")
        return generated_code
   
    return None

3. 工作流自动化

3.1 自动化工作流概念

• 工作流定义：一系列自动化任务的组合
• 触发器：启动工作流的条件（文件变更、时间、手动等）
• 任务链：按顺序执行的任务序列
• 条件分支：根据条件选择不同的执行路径

3.2 工作流引擎

# 工作流自动化引擎
import asyncio
import logging
from typing import Dict, List, Callable, Any
from datetime import datetime
import json

class WorkflowEngine:
    """工作流自动化引擎"""
   
    def __init__(self):
        self.workflows = {}
        self.tasks = {}
        self.logger = logging.getLogger(__name__)
   
    def register_task(self, task_name: str, task_func: Callable):
        """注册任务函数"""
        self.tasks[task_name] = task_func
        self.logger.info(f"任务 {task_name} 注册成功")
   
    def create_workflow(self, workflow_id: str, name: str, description: str = ""):
        """创建工作流"""
        self.workflows[workflow_id] = {
            "id": workflow_id,
            "name": name,
            "description": description,
            "steps": [],
            "triggers": [],
            "created_at": datetime.now(),
            "enabled": True
        }
        self.logger.info(f"工作流 {workflow_id} 创建成功")
        return self.workflows[workflow_id]
   
    def add_workflow_step(self, workflow_id: str, step_name: str, task_name: str,
                         parameters: Dict = None, conditions: Dict = None):
        """添加工作流步骤"""
        if workflow_id not in self.workflows:
            raise ValueError(f"工作流 {workflow_id} 不存在")
       
        step = {
            "name": step_name,
            "task": task_name,
            "parameters": parameters or {},
            "conditions": conditions or {},
            "order": len(self.workflows[workflow_id]["steps"])
        }
       
        self.workflows[workflow_id]["steps"].append(step)
        self.logger.info(f"步骤 {step_name} 添加到工作流 {workflow_id}")
   
    def add_workflow_trigger(self, workflow_id: str, trigger_type: str,
                           trigger_config: Dict):
        """添加工作流触发器"""
        if workflow_id not in self.workflows:
            raise ValueError(f"工作流 {workflow_id} 不存在")
       
        trigger = {
            "type": trigger_type,
            "config": trigger_config,
            "enabled": True
        }
       
        self.workflows[workflow_id]["triggers"].append(trigger)
        self.logger.info(f"触发器 {trigger_type} 添加到工作流 {workflow_id}")
   
    async def execute_workflow(self, workflow_id: str, context: Dict = None):
        """执行工作流"""
        if workflow_id not in self.workflows:
            raise ValueError(f"工作流 {workflow_id} 不存在")
       
        workflow = self.workflows[workflow_id]
        if not workflow["enabled"]:
            self.logger.warning(f"工作流 {workflow_id} 已禁用")
            return False
       
        self.logger.info(f"开始执行工作流 {workflow_id}")
       
        context = context or {}
        context["workflow_id"] = workflow_id
        context["start_time"] = datetime.now()
       
        try:
            for step in workflow["steps"]:
                # 检查条件
                if not self._check_conditions(step["conditions"], context):
                    self.logger.info(f"步骤 {step['name']} 条件不满足，跳过")
                    continue
               
                # 执行任务
                await self._execute_step(step, context)
               
        except Exception as e:
            self.logger.error(f"工作流 {workflow_id} 执行失败: {e}")
            return False
       
        self.logger.info(f"工作流 {workflow_id} 执行完成")
        return True
   
    def _check_conditions(self, conditions: Dict, context: Dict) -> bool:
        """检查步骤执行条件"""
        for key, value in conditions.items():
            if key not in context or context[key] != value:
                return False
        return True
   
    async def _execute_step(self, step: Dict, context: Dict):
        """执行单个步骤"""
        task_name = step["task"]
        if task_name not in self.tasks:
            raise ValueError(f"任务 {task_name} 未注册")
       
        self.logger.info(f"执行步骤 {step['name']}: {task_name}")
       
        try:
            task_func = self.tasks[task_name]
            parameters = step["parameters"]
           
            # 支持同步和异步任务
            if asyncio.iscoroutinefunction(task_func):
                result = await task_func(**parameters, context=context)
            else:
                result = task_func(**parameters, context=context)
           
            context[f"step_{step['name']}_result"] = result
            self.logger.info(f"步骤 {step['name']} 执行成功")
           
        except Exception as e:
            self.logger.error(f"步骤 {step['name']} 执行失败: {e}")
            raise

3.3 常用自动化任务

# 常用自动化任务定义
import subprocess
import os
from typing import Dict, Any

class AutomationTasks:
    """自动化任务集合"""
   
    @staticmethod
    def run_tests(context: Dict = None) -> Dict[str, Any]:
        """运行测试任务"""
        try:
            result = subprocess.run(
                ["python", "-m", "pytest", "--cov=.", "--cov-report=html"],
                capture_output=True,
                text=True,
                cwd=context.get("project_path", ".")
            )
           
            return {
                "success": result.returncode == 0,
                "output": result.stdout,
                "error": result.stderr,
                "return_code": result.returncode
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "return_code": -1
            }
   
    @staticmethod
    def check_code_style(context: Dict = None) -> Dict[str, Any]:
        """检查代码风格任务"""
        try:
            result = subprocess.run(
                ["flake8", "."],
                capture_output=True,
                text=True,
                cwd=context.get("project_path", ".")
            )
           
            return {
                "success": result.returncode == 0,
                "output": result.stdout,
                "error": result.stderr,
                "return_code": result.returncode,
                "issues_count": len(result.stdout.splitlines()) if result.stdout else 0
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "return_code": -1
            }
   
    @staticmethod
    def build_project(context: Dict = None) -> Dict[str, Any]:
        """构建项目任务"""
        try:
            # 清理构建目录
            build_dir = context.get("build_dir", "build")
            if os.path.exists(build_dir):
                import shutil
                shutil.rmtree(build_dir)
           
            # 运行构建命令
            build_command = context.get("build_command", ["python", "setup.py", "build"])
            result = subprocess.run(
                build_command,
                capture_output=True,
                text=True,
                cwd=context.get("project_path", ".")
            )
           
            return {
                "success": result.returncode == 0,
                "output": result.stdout,
                "error": result.stderr,
                "return_code": result.returncode,
                "build_dir": build_dir
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "return_code": -1
            }
   
    @staticmethod
    def deploy_project(context: Dict = None) -> Dict[str, Any]:
        """部署项目任务"""
        try:
            deploy_script = context.get("deploy_script", "deploy.sh")
            deploy_env = context.get("deploy_env", "production")
           
            # 设置环境变量
            env = os.environ.copy()
            env["DEPLOY_ENV"] = deploy_env
           
            result = subprocess.run(
                [f"./{deploy_script}"],
                capture_output=True,
                text=True,
                cwd=context.get("project_path", "."),
                env=env
            )
           
            return {
                "success": result.returncode == 0,
                "output": result.stdout,
                "error": result.stderr,
                "return_code": result.returncode,
                "deploy_env": deploy_env
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "return_code": -1
            }

3.4 工作流配置示例

# 配置自动化工作流
def setup_development_workflow():
    """设置开发工作流"""
    engine = WorkflowEngine()
   
    # 注册任务
    engine.register_task("run_tests", AutomationTasks.run_tests)
    engine.register_task("check_code_style", AutomationTasks.check_code_style)
    engine.register_task("build_project", AutomationTasks.build_project)
    engine.register_task("deploy_project", AutomationTasks.deploy_project)
   
    # 创建代码质量检查工作流
    quality_workflow = engine.create_workflow(
        "code_quality_check",
        "代码质量检查",
        "自动检查代码质量和运行测试"
    )
   
    # 添加工作流步骤
    engine.add_workflow_step(
        "code_quality_check",
        "style_check",
        "check_code_style",
        {"project_path": "."}
    )
   
    engine.add_workflow_step(
        "code_quality_check",
        "run_tests",
        "run_tests",
        {"project_path": "."},
        {"step_style_check_result.success": True}  # 只有代码风格检查通过才运行测试
    )
   
    # 添加文件变更触发器
    engine.add_workflow_trigger(
        "code_quality_check",
        "file_change",
        {
            "watch_paths": ["./src", "./tests"],
            "file_extensions": [".py", ".js", ".ts"],
            "debounce_time": 5  # 5秒防抖
        }
    )
   
    # 创建部署工作流
    deploy_workflow = engine.create_workflow(
        "auto_deploy",
        "自动部署",
        "代码合并到主分支后自动部署"
    )
   
    # 添加部署步骤
    engine.add_workflow_step(
        "auto_deploy",
        "build",
        "build_project",
        {"project_path": ".", "build_dir": "dist"}
    )
   
    engine.add_workflow_step(
        "auto_deploy",
        "deploy",
        "deploy_project",
        {"project_path": ".", "deploy_env": "staging"},
        {"step_build_result.success": True}  # 只有构建成功才部署
    )
   
    # 添加Git推送触发器
    engine.add_workflow_trigger(
        "auto_deploy",
        "git_push",
        {
            "branch": "main",
            "auto_merge": True
        }
    )
   
    return engine

# 使用工作流
async def main():
    """主函数"""
    engine = setup_development_workflow()
   
    # 手动触发代码质量检查
    await engine.execute_workflow("code_quality_check", {
        "project_path": ".",
        "trigger_type": "manual"
    })
   
    # 手动触发部署
    await engine.execute_workflow("auto_deploy", {
        "project_path": ".",
        "deploy_env": "production"
    })

if __name__ == "__main__":
    asyncio.run(main())

高级技巧

1. 提示词优化

• A/B测试：对比不同提示词的效果
• 迭代优化：根据反馈不断改进提示词
• 上下文增强：添加更多上下文信息
• 多轮对话：使用多轮对话提升效果

2. 模板系统扩展

• 模板继承：支持模板间的继承关系
• 条件渲染：根据条件动态渲染内容
• 宏定义：定义可复用的代码片段
• 模板验证：验证模板的正确性

3. 工作流监控

• 执行日志：记录工作流执行过程
• 性能监控：监控任务执行时间
• 错误处理：优雅处理执行错误
• 状态管理：管理工作流执行状态

提示词效果不理想的解决方案

当你发现AI生成的代码或内容不符合预期时，可以尝试以下方法优化提示词效果：

1. 优化提示词结构
   • 明确描述任务目标，避免歧义。
   • 拆分复杂需求为多个简单指令，逐步引导AI。
2. 增加上下文信息
   • 提供更多背景说明、输入输出示例或相关约束条件。
   • 指定代码风格、技术栈或业务场景。
3. 添加示例和约束
   • 给出期望的输入输出示例，帮助AI理解需求。
   • 明确列出必须遵守的规则或限制。
4. 采用迭代优化
   • 根据AI生成结果不断调整和细化提示词。
   • 记录有效的提示词模板，积累经验。
5. A/B测试不同提示词
   • 同时尝试多种表达方式，比较生成效果，选择最佳方案。
6. 利用多轮对话
   • 通过多轮交互补充细节，逐步完善生成内容。
7. 参考社区和官方文档
   • 借鉴优秀的提示词案例，学习行业最佳实践。

小贴士：遇到复杂需求时，可以先让AI分步生成，再手动整合结果，提升整体质量。

下节课预告

下一节课将学习Cursor AI的实战应用，通过实际项目开发演示，让你真正掌握AI辅助编程的技能。