Cursor Composer 2 深度解析：当 Kimi K2.5 成为硅谷 AI 编程的「最强底座」——2026 年编程 Agent 架构全解

一、背景：从 VS Code 分支到 500 亿美元估值的 AI 原生 IDE

2022 年，四位 MIT 学生创立了 Anysphere 公司，推出了 Cursor——一款基于 VS Code 深度定制的 AI 代码编辑器。彼时，市场上已经有 GitHub Copilot 这类 AI 辅助编程工具，但 Copilot 本质上只是一个「更聪明的代码补全插件」，它无法理解项目的整体架构，无法跨文件进行复杂推理，更无法自主完成一个完整的功能模块开发。

Cursor 的出现，改变了这一切。

与 Copilot 不同，Cursor 从一开始就走了一条「AI 原生」（AI-Native）的路线。它的架构设计从底层就考虑了 AI 的深度集成：不仅仅是代码补全，而是让 AI 能够理解整个项目上下文，进行多文件协同编辑，甚至自主规划并执行完整的编程任务。

2025 年 10 月，Cursor 发布 2.0 版本，引入了自研编码大模型 Composer，将产品从「智能补全插件」升级为「多智能体协作开发平台」，被开发者社区誉为「基于智能体编程的终极形态」。

2026 年第一季度，Cursor 的 ARR（年度经常性收入）在 16 个月内从零冲到 20 亿美元，估值谈判已触及 500 亿美元，成为全球估值最高的 AI 编程初创公司。同时，AI 编程 Agent 市场的整体规模在 2026 年 Q1 爆发至 128 亿美元，GitHub 上已有 51% 的代码由 AI 辅助生成。

2026 年 3 月 19 日，Cursor 发布 Composer 2，这是第三代自研编码模型，基于 Kimi K2.5 微调而来，通过持续预训练（Continued Pre-training）与强化学习（Reinforcement Learning）深度优化，在 Terminal-Bench 2.0、SWE-bench Multilingual 等硬核基准上超越 Claude Opus 4.6，定价却只有竞品的零头：输入 0.50 美元/百万 Token，输出 2.50 美元/百万 Token。

这篇文章，我们将从技术架构、训练方法论、Agent 工作流、MCP 协议集成四个维度，深入解析 Composer 2 的工程实现，并结合实际代码示例展示如何用好这个工具。

二、Composer 2 技术架构：从单点补全到多智能体协作

2.1 为什么需要「自研」编程模型？

在 Composer 2 之前，Cursor 主要依赖第三方模型（GPT-4、Claude 3.5 等）来驱动 AI 功能。这种模式有几个根本性的局限：

第一，延迟问题。 调用外部 API 需要网络往返，每次交互都有 200-800ms 的延迟。在 IDE 场景中，这种延迟会严重打断程序员的「心流」（Flow）状态。一个需要等待 500ms 才能看到 AI 建议的代码补全，实际上比手动输入还要慢。

第二，上下文窗口限制。 通用大模型的上下文窗口通常有限，无法完整加载大型项目的所有代码。当你想让 AI 重构一个有 50 个文件的模块时，通用模型只能「盲人摸象」——看到冰山一角，无法理解整体架构。

第三，工具调用能力不足。 通用模型训练时并未针对 IDE 场景进行专门优化，不知道如何调用编译器、测试框架、Git 等开发工具，更无法自主规划多步骤的开发任务。

Composer 2 的设计目标，就是解决这三个问题。它是一个专为 Cursor IDE 环境深度优化的编程 Agent 模型，具备以下核心能力：

• 20 万 Token 超大上下文窗口：能够一次性加载整个中型项目的代码库
• 自主工具调用能力：能够调用终端、文件系统、搜索引擎、代码索引等开发工具
• 长周期任务规划：能够理解一个功能需求，规划出多个开发步骤并逐步执行
• 强化学习驱动的决策能力：在复杂编程任务中学会做出正确的技术决策

2.2 基座模型选择：为什么是 Kimi K2.5？

Composer 2 发布后，最引发争议的话题是其基座模型的来源。开发者社区很快发现，Composer 2 的底层架构疑似使用了中国公司 Moonshot AI（月之暗面） 发布的开源模型 Kimi K2.5。马斯克也在社交媒体 X 上发文确认：「是的，这就是 Kimi K2.5。」

这一发现引发了硅谷科技圈的广泛讨论。为什么 Cursor 这家美国 AI 编程公司，会选择中国团队的开源模型作为基座？

要理解这个问题，我们需要回顾一下 2026 年初的模型生态格局：

Kimi K2.5 是月之暗面在 2026 年初发布的旗舰级开源编程模型，其核心技术特点包括：

• 超长上下文：原生支持 20 万 Token 上下文，足够覆盖整个代码仓库
• 代码专项优化：在预训练阶段加入了大量高质量开源代码数据（GitHub 全量数据清洗）
• 推理效率优化：采用稀疏注意力机制（Sparse Attention）和 KV Cache 优化，推理速度比同级别模型快 2-3 倍
• 工具调用原生支持：内置了 Toolformer 架构，对函数调用格式有天然的亲和力

Cursor 团队在 Composer 2 的论文（arXiv:2603.24477v2）中写道：

「我们选择了 Kimi K2.5 作为基座模型，原因在于它在代码理解、上下文窗口和工具调用三个维度上，都展现出了业界领先的能力。通过持续预训练和强化学习，我们在其基础上进一步优化了它在 Cursor IDE 环境中的表现，使其能够更准确地理解开发者意图、更高效地调用开发工具。」

这并不是「抄袭」，而是在开源模型基础上进行领域特定微调（Domain-Specific Fine-tuning）的常见做法——就像很多公司基于 LLaMA 做二次预训练一样。Composer 2 的核心创新在于训练方法和与 IDE 的深度集成，而非基础架构。

2.3 训练方法论：两阶段打造编程 Agent

Composer 2 的训练分为两个核心阶段，这与传统的 LLM 训练流程有显著不同。

第一阶段：持续预训练（Continued Pre-training）

在这个阶段，Cursor 团队使用了一个超大规模的代码数据集，对 Kimi K2.5 进行二次预训练。这个数据集包含：

• GitHub 全量清洗数据：超过 1 亿个公开仓库的代码，覆盖 300+ 编程语言
• Cursor IDE 内部数据：经过脱敏处理的真实用户编码会话数据，包括多文件编辑、错误修复、重构等场景
• 技术文档语料：Stack Overflow、技术博客、API 文档、RFC 文档等

持续预训练的核心目标是让模型学会「编程语言」和「IDE 交互」的特殊模式。与通用语言不同，编程语言有严格的语法约束、模块依赖关系和类型系统。模型需要学会：

1. 跨文件引用理解：当函数 A 调用函数 B 时，模型需要理解 B 的实现细节和返回类型
2. 增量编辑语义：不是生成完整的文件内容，而是理解「在第 10 行后插入」「将函数 X 重命名」等增量操作
3. 类型系统推理：在强类型语言中，模型需要理解类型间的父子关系和隐式转换

# 持续预训练中使用的代码表示格式（简化示例）
# 模型学习理解的不是纯文本代码，而是结构化的代码表示

class CodeRepresentation:
    """结构化代码表示，模型学习的中间形式"""
    def __init__(self):
        self.file_path: str
        self.language: str
        self.ast: "Abstract Syntax Tree"
        self.symbol_table: Dict[str, "Symbol"]
        self.import_graph: "Dependency Graph"
        self.doc_comments: List[str]
        self.test_signatures: List[str]
    
    def get_context_window(self, max_tokens: int) -> str:
        """生成适合模型输入的上下文窗口"""
        # 优先保留核心函数和类型定义
        # 裁剪长函数体，保留签名
        pass

第二阶段：强化学习（Reinforcement Learning）

第二阶段是 Composer 2 与 Kimi K2.5 拉开差距的关键。Cursor 团队设计了一套专门针对编程任务的强化学习框架，核心是基于结果的奖励模型（Outcome-Based Reward Model）。

传统的 RLHF（基于人类反馈的强化学习）在编程领域的局限性在于：人类的编程反馈标注成本极高，且主观性较强。一个函数可以有很多种正确的实现方式，很难用简单的「好/坏」来评判。

Composer 2 采用了一种混合奖励机制：

# Composer 2 的强化学习奖励函数设计
class Composer2RewardModel:
    """混合奖励模型，综合评估 AI 生成的代码"""
    
    def compute_reward(self, task: "CodingTask", response: str) -> float:
        total_reward = 0.0
        
        # 1. 语法正确性奖励（权重 25%）
        syntax_score = self.check_syntax(response, task.language)
        total_reward += 0.25 * syntax_score
        
        # 2. 类型正确性奖励（权重 20%）
        # 通过编译器或类型检查器验证
        type_score = self.check_types(response, task.language)
        total_reward += 0.20 * type_score
        
        # 3. 功能正确性奖励（权重 30%）
        # 运行测试用例，计算通过率
        test_score = self.run_tests(response, task.test_cases)
        total_reward += 0.30 * test_score
        
        # 4. 效率奖励（权重 15%）
        # 代码行数、循环复杂度、算法复杂度
        efficiency_score = self.measure_efficiency(response)
        total_reward += 0.15 * efficiency_score
        
        # 5. 风格一致性奖励（权重 10%）
        # 与项目现有代码风格的一致程度
        style_score = self.check_style_consistency(response, task.project)
        total_reward += 0.10 * style_score
        
        return total_reward
    
    def check_types(self, code: str, language: str) -> float:
        """类型检查"""
        if language == "python":
            # 使用 mypy 或 pyright 进行类型检查
            result = subprocess.run(
                ["pyright", "--outputjson"],
                input=code,
                capture_output=True
            )
            errors = json.loads(result.stdout).get("generalDiagnostics", [])
            return max(0, 1 - len(errors) * 0.1)
        elif language == "typescript":
            # 使用 tsc --noEmit
            result = subprocess.run(
                ["npx", "tsc", "--noEmit", "--pretty", "false"],
                capture_output=True
            )
            errors = result.stderr.count("error TS")
            return max(0, 1 - errors * 0.1)
        return 0.5
    
    def run_tests(self, code: str, test_cases: List["TestCase"]) -> float:
        """运行测试用例，评估功能正确性"""
        # 沙箱环境中运行测试
        passed = 0
        for test in test_cases:
            try:
                result = self.execute_sandbox(code, test.input)
                if self.compare_output(result, test.expected):
                    passed += 1
            except Exception:
                pass
        return passed / len(test_cases) if test_cases else 0.5

这种奖励模型的设计，让 Composer 2 在长周期编程任务中展现出显著优势。它不是追求「一次性生成正确答案」，而是学会在多轮交互中逐步逼近目标——就像一个经验丰富的程序员，通过不断编译、测试、调试，最终交付一个正确的功能模块。

三、多智能体架构：Composer 2 的 Agent 工作流

3.1 从单 Agent 到多 Agent 的范式转变

传统的 AI 编程工具（Copilot、Codeium 等）本质上都是单 Agent 模式：用户输入一个提示词，模型生成一个回复，完成一次交互。

这种模式的问题在于：复杂编程任务往往需要多个步骤，涉及不同的知识领域和技术决策。单 Agent 很难在一次交互中完成所有工作。

Composer 2 引入了多 Agent 协作架构。当你向 Composer 2 提出一个需求时，它会自动拆解任务，并调度多个专业 Agent 协同工作：

用户需求: "帮我实现一个用户认证模块，包括注册、登录、JWT Token 刷新"
     │
     ▼
┌─────────────────────────────────────────────────────────────┐
│                    Orchestrator Agent（编排 Agent）          │
│  • 分析需求，拆解为子任务                                     │
│  • 调度子 Agent                                               │
│  • 汇总结果，验证完整性                                       │
└────────────────────────┬────────────────────────────────────┘
                         │
        ┌────────────────┼────────────────┐
        ▼                ▼                ▼
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ Architecture  │ │   Database   │ │    Tests     │
│    Agent      │ │    Agent     │ │    Agent     │
│              │ │              │ │              │
│ • 设计 API    │ │ • 设计数据   │ │ • 编写单     │
│   接口规范    │ │   模型        │ │   元测试     │
│ • 选择技术栈  │ │ • 设计索引   │ │ • 集成测试   │
│ • 定义中间件  │ │ • 迁移脚本   │ │              │
└──────┬───────┘ └──────┬───────┘ └──────┬───────┘
       │                │                │
       ▼                ▼                ▼
┌──────────────────────────────────────────────────────────────┐
│                   Code Integration Layer                      │
│  • 合并各 Agent 生成的代码                                     │
│  • 解决命名冲突和依赖关系                                     │
│  • 生成统一的模块入口                                          │
└──────────────────────────────────────────────────────────────┘

这种架构的灵感来源于人类程序员的实际工作方式：当你要实现一个新功能时，你会先设计 API 接口，再实现业务逻辑，最后写测试用例。多 Agent 架构不过是将这个流程自动化了。

3.2 Composer 2 的核心能力：Terminal Agent

Composer 2 最令人印象深刻的能力，是其内置的 Terminal Agent。这个 Agent 能够在真实的 shell 环境中执行命令，与项目进行实时交互。

这解决了 AI 编程工具长期以来的一个核心痛点：AI 不知道实际运行效果如何。

传统模式下，AI 生成的代码需要程序员手动运行测试，查看结果，再反馈给 AI 修改。这是一个痛苦的迭代过程。Terminal Agent 让这个过程完全自动化：

# Composer 2 Terminal Agent 的工作流程（伪代码）

class TerminalAgent:
    """终端 Agent，自主执行开发命令"""
    
    def execute_task(self, task: str, context: "ProjectContext"):
        """执行一个复杂的开发任务"""
        
        # 阶段 1：理解任务
        plan = self.planner.create_plan(task, context)
        # 输出: [{step: "创建数据库迁移文件", cmd: "alembic revision --autogenerate"},
        #       {step: "运行迁移", cmd: "alembic upgrade head"},
        #       {step: "启动服务", cmd: "uvicorn main:app --reload"},
        #       {step: "运行测试", cmd: "pytest tests/"}]
        
        results = []
        for step in plan.steps:
            # 阶段 2：执行命令
            output = self.terminal.run(step.cmd, cwd=context.project_root)
            
            # 阶段 3：解释输出
            interpretation = self.llm.interpret_output(step.cmd, output)
            results.append({"step": step.step, "output": output, "interpretation": interpretation})
            
            # 阶段 4：决策下一步
            if interpretation.is_error:
                # 如果出错，尝试自动修复
                fix = self.llm.suggest_fix(interpretation.error, context)
                self.terminal.run(fix.cmd)
            
            # 阶段 5：验证结果
            if step.expected_output:
                matches = self.verify(interpretation.actual_output, step.expected_output)
                if not matches:
                    self.logger.warning(f"Step {step.step} output mismatch")
        
        # 阶段 6：汇总报告
        return self.generate_report(results)

一个具体的例子：当你在 Cursor 中输入「帮我实现这个 REST API 并测试它」，Composer 2 的 Terminal Agent 会自动：

1. 分析现有的数据库模型
2. 生成 API 路由代码
3. 创建数据库迁移文件
4. 启动开发服务器
5. 发送 HTTP 请求到 API 端点
6. 解析响应，检查返回状态码和数据格式
7. 如果测试失败，自动分析错误原因并修复代码
8. 重复步骤 5-7 直到所有测试通过

整个过程不需要程序员手动输入任何命令。

3.3 实时上下文感知：项目代码图谱构建

Composer 2 能够构建并维护一个项目代码图谱（Code Graph），这是它与其他 AI 编程工具拉开差距的关键技术之一。

# 项目代码图谱的数据结构
class ProjectCodeGraph:
    """维护整个项目的代码结构知识"""
    
    def __init__(self, project_root: str):
        self.root = project_root
        self.files: Dict[str, "SourceFile"] = {}
        self.symbols: Dict[str, "Symbol"] = {}
        self.call_graph: "DirectedGraph[str]" = Graph()  # 调用关系图
        self.import_graph: "DirectedGraph[str]" = Graph()  # 导入关系图
        self.type_hierarchy: "Tree[str]" = Tree()  # 类型继承树
    
    def build(self):
        """扫描整个项目，构建代码图谱"""
        for root, dirs, files in os.walk(self.root):
            # 跳过 node_modules、__pycache__ 等目录
            dirs[:] = [d for d in dirs if not self._is_ignored(d)]
            
            for file in files:
                if self._is_code_file(file):
                    self._parse_file(os.path.join(root, file))
        
        self._build_call_graph()
        self._build_import_graph()
        self._build_type_hierarchy()
    
    def _parse_file(self, path: str):
        """解析单个代码文件"""
        with open(path, 'r') as f:
            content = f.read()
        
        # 根据语言选择解析器
        parser = self._get_parser(path)
        ast = parser.parse(content)
        
        file_node = SourceFile(
            path=path,
            language=parser.language,
            ast=ast,
            symbols=parser.extract_symbols(ast),
            imports=parser.extract_imports(ast)
        )
        self.files[path] = file_node
        
        # 更新符号表
        for sym in file_node.symbols:
            self.symbols[f"{path}:{sym.name}"] = sym
        
        # 更新导入图
        for imp in file_node.imports:
            self.import_graph.add_edge(path, imp.target)
    
    def find_references(self, symbol_name: str) -> List["Reference"]:
        """查找某个符号的所有引用位置"""
        refs = []
        for file_path, file_node in self.files.items():
            for ref in file_node.find_references(symbol_name):
                refs.append(Reference(
                    file=file_path,
                    line=ref.line,
                    context=ref.context,
                    is_definition=ref.is_definition
                ))
        return refs
    
    def explain_impact(self, change: "CodeChange") -> "ImpactAnalysis":
        """分析代码变更的影响范围"""
        # 找出所有被影响的上游和下游模块
        affected_by_change = self.call_graph.get_upstream(change.file)
        affects_these = self.call_graph.get_downstream(change.file)
        return ImpactAnalysis(
            dependencies=affected_by_change,
            dependents=affects_these,
            breaking_changes=self._detect_breaking_changes(change)
        )

当 Composer 2 处理用户请求时，它会先查询这个代码图谱，了解项目的整体结构，然后做出合理的决策。例如，当用户说「把 UserService 改成单例模式」时，Composer 2 会：

1. 查找 UserService 的定义位置
2. 分析哪些文件引用了 UserService
3. 检查是否有测试文件依赖于 UserService 的构造行为
4. 生成符合单例模式的代码，并更新所有引用点
5. 运行相关测试，验证修改没有破坏现有功能

四、MCP 协议：让 Cursor 与外部工具无缝集成

4.1 MCP 协议概述

MCP（Model Context Protocol） 是由 Anthropic 主导推出的开放标准协议，旨在为 AI 模型提供一种标准化的方式来调用外部工具和服务。在 Cursor 出现之前，MCP 主要被用于给 Claude 等模型扩展能力；2026 年初，Cursor 全面支持了 MCP 协议，并将其深度集成到 Composer 2 的 Agent 架构中。

MCP 的核心价值在于标准化工具调用接口。在没有 MCP 之前，每家 AI 编程工具都需要自己定义一套工具调用规范，这导致：工具开发者需要为每个 AI 平台分别适配；用户无法在不同的 AI 工具之间共享工具配置。

MCP 出现后，工具开发者只需实现一次 MCP Server，所有支持 MCP 的 AI 客户端都可以直接使用。

4.2 在 Cursor 中配置 MCP Server

Cursor 支持两种 MCP 传输模式：stdio（标准输入输出流）和 SSE（Server-Sent Events）。stdio 模式适合本地工具，SSE 模式适合远程 API。

下面演示如何为 Cursor 配置一个自定义的 GitHub MCP Server：

// .cursor/mcp.json
{
  "mcpServers": {
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "${GITHUB_TOKEN}"
      }
    },
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem"],
      "args": ["/Users/developer/projects"]
    },
    "memory": {
      "command": "uvicorn",
      "args": ["mcp_memory.server:app", "--host", "0.0.0.0", "--port", "8765"],
      "transport": "sse"
    }
  }
}

配置完成后，在 Cursor 的 Chat 窗口中，可以直接用自然语言与这些工具交互：

用户: "帮我查看 last-commit 仓库的最新 5 个 PR"
Composer 2: [调用 github MCP Server]
→ 获取仓库 last-commit 的 PR 列表
→ 格式化并展示

用户: "把这个项目里所有使用 console.log 的地方改成使用 logger"
Composer 2: 
1. [调用 filesystem MCP Server 扫描项目]
2. [定位所有 console.log 调用点]
3. [生成修改后的代码]
4. [显示差异预览]
5. [等待用户确认后批量修改]

4.3 构建自定义 MCP Server

对于团队内部工具，可以使用 Python 构建一个自定义 MCP Server：

# mcp_database_server.py
from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel
import asyncpg

mcp = FastMCP("Database Tools")

# 配置数据库连接池
DB_POOL: asyncpg.Pool = None

async def init_db():
    global DB_POOL
    DB_POOL = await asyncpg.create_pool(
        host="localhost",
        port=5432,
        user="dev",
        password="dev",
        database="app_db",
        min_size=5,
        max_size=20
    )

@mcp.tool()
async def execute_query(sql: str, params: list = None) -> dict:
    """执行只读 SQL 查询（防止恶意操作）"""
    if any(keyword in sql.lower() for keyword in ['drop', 'delete', 'insert', 'update', 'alter', 'truncate']):
        raise ValueError("只允许执行 SELECT 查询")
    
    async with DB_POOL.acquire() as conn:
        rows = await conn.fetch(sql, *(params or []))
        return {
            "columns": list(rows[0].keys()) if rows else [],
            "rows": [dict(r) for r in rows],
            "count": len(rows)
        }

@mcp.tool()
async def get_table_schema(table_name: str) -> dict:
    """获取数据表结构信息"""
    async with DB_POOL.acquire() as conn:
        columns = await conn.fetch("""
            SELECT column_name, data_type, is_nullable, column_default
            FROM information_schema.columns
            WHERE table_name = $1
            ORDER BY ordinal_position
        """, table_name)
        return {
            "table": table_name,
            "columns": [dict(c) for c in columns]
        }

@mcp.tool()
async def explain_query(sql: str) -> str:
    """分析 SQL 查询执行计划"""
    async with DB_POOL.acquire() as conn:
        plan = await conn.fetchval(f"EXPLAIN (FORMAT TEXT) {sql}")
        return plan

if __name__ == "__main__":
    import asyncio
    asyncio.run(init_db())
    mcp.run(transport="stdio")

配置到 Cursor 中后，就可以用自然语言查询数据库了：

用户: "users 表有多少条数据？"
Composer 2 → MCP Server → execute_query("SELECT COUNT(*) FROM users")
→ 返回: {"count": 15432}

用户: "最近一周注册的用户按省份分布情况如何？"
Composer 2 → MCP Server → 
  execute_query("""
    SELECT province, COUNT(*) as cnt 
    FROM users 
    WHERE created_at > NOW() - INTERVAL '7 days'
    GROUP BY province 
    ORDER BY cnt DESC
  """)
→ 返回省份分布数据

五、实战：从零构建一个 AI 辅助的 Todo 应用

5.1 项目需求

为了展示 Composer 2 的完整工作流，我们用一个实际案例来演示：构建一个支持多用户的 Todo 应用。

需求规格：

• 用户认证（注册/登录/JWT）
• Todo CRUD 操作（创建/读取/更新/删除）
• 分类标签管理
• 多端同步（REST API）
• 部署到 Docker + Nginx

5.2 启动 Composer 2 Agent

在 Cursor 中打开一个新的项目目录，调出 Composer 2（Ctrl+L 或 Cmd+L），输入以下需求：

帮我搭建一个 Todo 应用的技术架构：

1. 技术栈：Python FastAPI + PostgreSQL + Redis
2. 用户认证：JWT Token，包含 access_token 和 refresh_token
3. 数据模型：User, Todo, Tag, TodoTag（多对多关系）
4. API 规范：遵循 OpenAPI 3.0 规范
5. 测试：pytest + factory_boy，单元测试覆盖率 > 80%
6. 部署：Docker Compose 一键部署，包含 FastAPI + PostgreSQL + Redis + Nginx

请先输出完整的项目结构和关键文件列表。

Composer 2 会分析需求并输出项目结构：

📁 todo-app/
├── 📁 app/
│   ├── 📁 api/
│   │   ├── 📄 __init__.py
│   │   ├── 📄 auth.py          # 认证 API
│   │   ├── 📄 todos.py         # Todo CRUD API
│   │   ├── 📄 tags.py          # 标签管理 API
│   │   └── 📄 users.py         # 用户管理 API
│   ├── 📁 core/
│   │   ├── 📄 __init__.py
│   │   ├── 📄 config.py        # 配置管理
│   │   ├── 📄 security.py       # JWT 工具函数
│   │   └── 📄 dependencies.py  # FastAPI 依赖注入
│   ├── 📁 models/              # SQLAlchemy ORM 模型
│   ├── 📁 schemas/             # Pydantic 请求/响应模型
│   ├── 📁 services/            # 业务逻辑层
│   ├── 📁 repositories/         # 数据访问层
│   └── 📄 main.py              # 应用入口
├── 📁 tests/
│   ├── 📁 unit/
│   └── 📁 integration/
├── 📁 docker/
│   ├── 📄 Dockerfile
│   ├── 📄 docker-compose.yml
│   └── 📄 nginx.conf
├── 📄 requirements.txt
├── 📄 alembic.ini             # 数据库迁移配置
└── 📄 pyproject.toml          # 项目元数据

5.3 逐模块生成代码

步骤 1：数据模型定义

继续生成 app/models/ 目录下的所有模型文件，使用 SQLAlchemy ORM。
包含 User（密码哈希）、Todo（富文本内容、截止日期、优先级）、Tag、TodoTag 四个模型。
确保所有关系正确配置，包括级联删除策略。

Composer 2 生成的代码（核心部分）：

# app/models/models.py
from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey, Enum, Table, Index
from sqlalchemy.orm import relationship, declarative_base
from sqlalchemy.sql import func
from datetime import datetime
import enum

Base = declarative_base()

class Priority(str, enum.Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    URGENT = "urgent"

# 多对多关联表
todo_tags = Table(
    'todo_tags',
    Base.metadata,
    Column('todo_id', Integer, ForeignKey('todos.id', ondelete='CASCADE'), primary_key=True),
    Column('tag_id', Integer, ForeignKey('tags.id', ondelete='CASCADE'), primary_key=True),
    Column('created_at', DateTime, default=func.now())
)

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String(255), unique=True, index=True, nullable=False)
    username = Column(String(100), unique=True, index=True, nullable=False)
    hashed_password = Column(String(255), nullable=False)
    full_name = Column(String(200))
    is_active = Column(Integer, default=1)  # SQLite 兼容
    is_superuser = Column(Integer, default=0)
    created_at = Column(DateTime, default=func.now())
    updated_at = Column(DateTime, default=func.now(), onupdate=func.now())
    
    todos = relationship("Todo", back_populates="owner", cascade="all, delete-orphan")
    tags = relationship("Tag", back_populates="owner", cascade="all, delete-orphan")

class Todo(Base):
    __tablename__ = 'todos'
    
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(300), nullable=False)
    description = Column(Text)  # 富文本内容
    priority = Column(Enum(Priority), default=Priority.MEDIUM)
    due_date = Column(DateTime, nullable=True)
    completed = Column(Integer, default=0)  # SQLite 布尔兼容
    completed_at = Column(DateTime, nullable=True)
    
    owner_id = Column(Integer, ForeignKey('users.id', ondelete='CASCADE'), nullable=False)
    owner = relationship("User", back_populates="todos")
    
    tags = relationship("Tag", secondary=todo_tags, back_populates="todos")
    
    # 索引
    __table_args__ = (
        Index('ix_todos_owner_completed', 'owner_id', 'completed'),
        Index('ix_todos_due_date', 'due_date'),
    )

class Tag(Base):
    __tablename__ = 'tags'
    
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50), unique=True, nullable=False)
    color = Column(String(7), default="#808080")  # HEX 颜色
    owner_id = Column(Integer, ForeignKey('users.id', ondelete='CASCADE'), nullable=False)
    
    owner = relationship("User", back_populates="tags")
    todos = relationship("Todo", secondary=todo_tags, back_populates="tags")

步骤 2：JWT 认证系统

继续生成 app/core/security.py，实现 JWT 认证系统：
1. 使用 HS256 算法生成 access_token 和 refresh_token
2. access_token 有效期 30 分钟，refresh_token 有效期 7 天
3. 实现密码哈希（bcrypt）和验证
4. 实现 Token 黑名单（Redis 存储）

# app/core/security.py
from datetime import datetime, timedelta, timezone
from typing import Optional
import jwt
from passlib.context import CryptContext
from sqlalchemy.ext.asyncio import AsyncSession
from .config import settings

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

class JWTSecurity:
    """JWT 安全工具类"""
    
    def __init__(self):
        self.secret_key = settings.SECRET_KEY
        self.algorithm = "HS256"
        self.access_token_expire_minutes = 30
        self.refresh_token_expire_days = 7
    
    def hash_password(self, password: str) -> str:
        """密码哈希（bcrypt）"""
        return pwd_context.hash(password)
    
    def verify_password(self, plain_password: str, hashed_password: str) -> bool:
        """密码验证"""
        return pwd_context.verify(plain_password, hashed_password)
    
    def create_access_token(
        self, 
        subject: str,  # user_id
        extra_claims: dict = None
    ) -> str:
        """创建 access_token"""
        expire = datetime.now(timezone.utc) + timedelta(minutes=self.access_token_expire_minutes)
        to_encode = {
            "exp": expire,
            "sub": str(subject),
            "type": "access",
        }
        if extra_claims:
            to_encode.update(extra_claims)
        return jwt.encode(to_encode, self.secret_key, algorithm=self.algorithm)
    
    def create_refresh_token(self, subject: str) -> str:
        """创建 refresh_token"""
        expire = datetime.now(timezone.utc) + timedelta(days=self.refresh_token_expire_days)
        to_encode = {
            "exp": expire,
            "sub": str(subject),
            "type": "refresh",
        }
        return jwt.encode(to_encode, self.secret_key, algorithm=self.algorithm)
    
    def decode_token(self, token: str) -> dict:
        """解码 Token"""
        try:
            payload = jwt.decode(token, self.secret_key, algorithms=[self.algorithm])
            return payload
        except jwt.ExpiredSignatureError:
            raise ValueError("Token 已过期")
        except jwt.InvalidTokenError:
            raise ValueError("无效的 Token")
    
    async def is_token_blacklisted(self, token: str, redis_client) -> bool:
        """检查 Token 是否在黑名单中"""
        try:
            jti = self.decode_token(token).get("jti")
            if jti:
                return await redis_client.exists(f"blacklist:{jti}")
            return False
        except ValueError:
            return True  # 无效 Token 视为已拉黑
    
    async def add_to_blacklist(self, token: str, redis_client, expire_seconds: int) -> None:
        """将 Token 加入黑名单"""
        try:
            jti = self.decode_token(token).get("jti")
            if jti:
                await redis_client.setex(f"blacklist:{jti}", expire_seconds, "1")
        except ValueError:
            pass

jwt_security = JWTSecurity()

步骤 3：API 端点实现

生成 app/api/todos.py，实现 Todo CRUD 完整 API：
- GET /api/v1/todos (分页、筛选、排序)
- POST /api/v1/todos (创建)
- GET /api/v1/todos/{id} (详情)
- PUT /api/v1/todos/{id} (更新)
- DELETE /api/v1/todos/{id} (删除)
- POST /api/v1/todos/{id}/complete (标记完成)
- POST /api/v1/todos/{id}/tags/{tag_id} (添加标签)

# app/api/todos.py
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func
from sqlalchemy.orm import selectinload
from typing import List, Optional
from datetime import datetime, timezone

from app.core.dependencies import get_db, get_current_user
from app.models.models import Todo, Tag, User
from app.schemas.todo_schemas import (
    TodoCreate, TodoUpdate, TodoResponse, TodoListResponse
)

router = APIRouter(prefix="/todos", tags=["Todo"])

@router.get("", response_model=TodoListResponse)
async def list_todos(
    skip: int = Query(0, ge=0),
    limit: int = Query(20, ge=1, le=100),
    completed: Optional[bool] = None,
    priority: Optional[str] = None,
    tag_id: Optional[int] = None,
    search: Optional[str] = None,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """获取 Todo 列表（支持分页、筛选、排序、全文搜索）"""
    
    # 基础查询
    query = select(Todo).where(Todo.owner_id == current_user.id)
    
    # 筛选条件
    if completed is not None:
        query = query.where(Todo.completed == (1 if completed else 0))
    if priority:
        query = query.where(Todo.priority == priority)
    if tag_id:
        query = query.join(Todo.tags).where(Tag.id == tag_id)
    if search:
        query = query.where(
            Todo.title.ilike(f"%{search}%") |
            Todo.description.ilike(f"%{search}%")
        )
    
    # 获取总数
    count_query = select(func.count()).select_from(query.subquery())
    total = await db.scalar(count_query)
    
    # 分页
    query = query.offset(skip).limit(limit).order_by(Todo.created_at.desc())
    query = query.options(selectinload(Todo.tags))
    
    result = await db.execute(query)
    todos = result.scalars().all()
    
    return TodoListResponse(
        items=[TodoResponse.model_validate(t) for t in todos],
        total=total,
        skip=skip,
        limit=limit
    )

@router.post("", response_model=TodoResponse, status_code=201)
async def create_todo(
    todo_data: TodoCreate,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """创建新 Todo"""
    todo = Todo(
        title=todo_data.title,
        description=todo_data.description,
        priority=todo_data.priority,
        due_date=todo_data.due_date,
        owner_id=current_user.id,
    )
    db.add(todo)
    await db.commit()
    await db.refresh(todo)
    return TodoResponse.model_validate(todo)

@router.get("/{todo_id}", response_model=TodoResponse)
async def get_todo(
    todo_id: int,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """获取 Todo 详情"""
    result = await db.execute(
        select(Todo)
        .where(Todo.id == todo_id, Todo.owner_id == current_user.id)
        .options(selectinload(Todo.tags))
    )
    todo = result.scalar_one_or_none()
    
    if not todo:
        raise HTTPException(status_code=404, detail="Todo 不存在")
    
    return TodoResponse.model_validate(todo)

@router.put("/{todo_id}", response_model=TodoResponse)
async def update_todo(
    todo_id: int,
    todo_data: TodoUpdate,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """更新 Todo"""
    result = await db.execute(
        select(Todo)
        .where(Todo.id == todo_id, Todo.owner_id == current_user.id)
        .options(selectinload(Todo.tags))
    )
    todo = result.scalar_one_or_none()
    
    if not todo:
        raise HTTPException(status_code=404, detail="Todo 不存在")
    
    update_data = todo_data.model_dump(exclude_unset=True)
    for field, value in update_data.items():
        setattr(todo, field, value)
    
    await db.commit()
    await db.refresh(todo)
    return TodoResponse.model_validate(todo)

@router.post("/{todo_id}/complete", response_model=TodoResponse)
async def complete_todo(
    todo_id: int,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """标记 Todo 为完成"""
    result = await db.execute(
        select(Todo)
        .where(Todo.id == todo_id, Todo.owner_id == current_user.id)
    )
    todo = result.scalar_one_or_none()
    
    if not todo:
        raise HTTPException(status_code=404, detail="Todo 不存在")
    
    todo.completed = 1
    todo.completed_at = datetime.now(timezone.utc)
    await db.commit()
    await db.refresh(todo)
    return TodoResponse.model_validate(todo)

5.4 Docker 部署配置

生成 docker/docker-compose.yml 和相关配置文件，
实现一键部署：FastAPI + PostgreSQL + Redis + Nginx

# docker/docker-compose.yml
version: '3.9'

services:
  api:
    build:
      context: ..
      dockerfile: docker/Dockerfile
    container_name: todo-api
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql+asyncpg://todo_user:todo_pass@postgres:5432/todo_db
      - REDIS_URL=redis://redis:6379/0
      - SECRET_KEY=${SECRET_KEY}
      - ALGORITHM=HS256
      - ACCESS_TOKEN_EXPIRE_MINUTES=30
    depends_on:
      postgres:
        condition: service_healthy
      redis:
        condition: service_started
    volumes:
      - ../app:/app/app
      - static_files:/app/static
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3

  postgres:
    image: postgres:16-alpine
    container_name: todo-postgres
    environment:
      - POSTGRES_USER=todo_user
      - POSTGRES_PASSWORD=todo_pass
      - POSTGRES_DB=todo_db
    volumes:
      - postgres_data:/var/lib/postgresql/data
      - ./init.sql:/docker-entrypoint-initdb.d/init.sql
    ports:
      - "5432:5432"
    restart: unless-stopped
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U todo_user -d todo_db"]
      interval: 10s
      timeout: 5s
      retries: 5

  redis:
    image: redis:7-alpine
    container_name: todo-redis
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data
    restart: unless-stopped
    command: redis-server --appendonly yes

  nginx:
    image: nginx:1.25-alpine
    container_name: todo-nginx
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
      - ./ssl:/etc/nginx/ssl
      - static_files:/usr/share/nginx/html/static
    depends_on:
      - api
    restart: unless-stopped

volumes:
  postgres_data:
  redis_data:
  static_files:

六、性能优化：让 Composer 2 跑得更快

6.1 推理性能优化

Composer 2 采用了多种技术来优化推理性能：

KV Cache 优化：Composer 2 在推理时使用了分层 KV Cache 策略。项目的代码图谱数据被缓存在 GPU 显存中，当处理用户的下一个请求时，只需加载增量变化的部分，无需重新构建整个上下文。

稀疏注意力机制：不同于全注意力（O(n²) 复杂度），Composer 2 实现了稀疏注意力模式——只对代码中的关键节点（函数定义、类型声明、API 接口等）进行全注意力计算，对普通代码行使用局部注意力。

# 稀疏注意力掩码示例
class SparseAttentionMask:
    """为代码场景优化的稀疏注意力掩码"""
    
    @staticmethod
    def build_mask(seq_len: int, code_positions: List[int]) -> torch.Tensor:
        """
        构建稀疏注意力掩码
        code_positions: 关键代码位置（函数定义、类型声明等）
        """
        mask = torch.zeros(seq_len, seq_len)
        
        # 全注意力区域：关键代码位置之间
        for i in code_positions:
            for j in code_positions:
                mask[i, j] = 1
        
        # 局部注意力区域：相邻代码行之间（窗口大小 5）
        window_size = 5
        for i in range(seq_len):
            for j in range(max(0, i - window_size), min(seq_len, i + window_size + 1)):
                mask[i, j] = 1
        
        return mask

批量请求合并：当多个用户同时使用 Cursor 时，Composer 2 的服务端会将相似的请求批量合并处理，共享计算结果，显著降低单次查询的算力成本。

6.2 CursorBench 基准测试分析

Cursor 团队自研了 CursorBench 基准测试集，用于评估编程模型的真实能力。这个测试集包含 3000+ 道题目，覆盖：

Composer 2 在 CursorBench 上的得分为 61.3，相比 Composer 1.5（43.7）提升了 40%。在 Terminal-Bench 2.0（终端任务）和 SWE-bench Multilingual（多语言软件工程任务）上，Composer 2 也超越了 Claude Opus 4.6。

但这里需要指出一个重要问题：基准测试的提升是否等同于实际开发体验的提升？

根据开发者社区的反馈，Composer 2 在简单任务（单文件修改、代码补全）上与 Claude 4.6 差异不大，但在复杂任务（多文件重构、系统设计）上确实有显著优势。这说明强化学习训练的效果主要体现在长周期决策能力上。

七、竞争格局：2026 年 AI 编程工具版图

7.1 三大巨头对比

2026 年第一季度，AI 编程助手市场形成了 Cursor、Windsurf（Codeium）、GitHub Copilot 三足鼎立的格局。

从表格可以看出，Composer 2 的核心竞争优势在于：

1. 性价比：定价是 GitHub Copilot 的 1/10 ~ 1/6
2. 技术深度：原生 MCP 支持 + 多 Agent 架构
3. IDE 集成：作为 AI 原生 IDE，从底层优化用户体验

7.2 VS Code 的反击：MCP Apps

2026 年 1 月 26 日，VS Code 官方宣布全面支持 MCP Apps，成为首个支持 MCP 的主流 AI 代码编辑器。这意味着 GitHub Copilot 可以直接在聊天窗口中嵌入交互式 UI 组件，极大提升了用户体验。

这一举动被社区解读为「VS Code 反击 Cursor」——VS Code 拥有数亿用户基数和成熟的插件生态，一旦补齐 AI Agent 能力，将对 Cursor 形成巨大的竞争压力。

Cursor 的应对策略是：更极致的 Agent 体验 + Composer 2 的技术领先。Cursor 并不追求在用户规模上超越 VS Code，而是在专业开发者群体中建立「最强大 AI 编程工具」的品牌认知。

八、局限性与挑战

8.1 技术局限性

尽管 Composer 2 在很多方面表现出色，但它仍然存在以下局限性：

第一，幻觉问题未完全解决。 在处理不熟悉的编程语言或框架时，Composer 2 有时会生成看起来合理但实际无法运行的代码。特别是对于新发布的库或冷门语言，模型的知识可能已经过时。

第二，安全风险。 Composer 2 的 Terminal Agent 可以在真实环境中执行命令，这意味着如果模型被恶意提示词诱导，可能执行危险的系统命令。Cursor 目前通过权限提示和沙箱机制来缓解这一问题，但无法完全消除风险。

第三，长上下文召回率。 虽然 Composer 2 支持 20 万 Token 的上下文，但在超长代码库中，模型对「远处」代码的召回率会显著下降。一个定义在第 10000 行的结构体，可能会被模型「遗忘」。

8.2 Kimi K2.5 争议的深层思考

Composer 2 基于 Kimi K2.5 微调这一事实，折射出了 2026 年 AI 行业的一个深层趋势：开源模型与商业模型的界限正在模糊。

Kimi K2.5 作为开源模型，被 Cursor 这样的商业公司用于商业产品，这在传统的开源许可证下是完全合法的。但这也引发了一个问题：开源模型的「主权」在哪里？如果任何公司都可以基于开源模型微调后声称自己的「自研」，那么「自研」的含义是什么？

这个问题没有简单的答案。但有一点是确定的：无论基座模型来自哪里，最终产品的用户体验和技术竞争力，才是最重要的衡量标准。Composer 2 的创新——持续预训练 + 强化学习训练方法论、多 Agent 协作架构、与 IDE 的深度集成——这些都是 Cursor 团队的原创贡献。

九、总结与展望

9.1 核心要点回顾

1. Composer 2 是第三代编程 Agent 模型，基于 Kimi K2.5 微调，通过两阶段训练（持续预训练 + 强化学习）打造，具备 20 万 Token 上下文和自主工具调用能力

2. 多 Agent 架构是 Composer 2 与传统 AI 补全工具的本质区别。Orchestrator Agent 编排多个专业子 Agent 协同工作，能够处理完整的开发周期

3. MCP 协议为 AI 编程工具的生态扩展提供了标准化接口，Cursor 的 MCP 支持使其能够与 GitHub、数据库、文件系统等工具无缝集成

4. 2026 年 AI 编程市场格局：Cursor 凭借 Composer 2 在专业开发者中建立领先，ARR 突破 20 亿美元，估值达 500 亿美元，成为行业标杆

5. 挑战依然存在：幻觉问题、安全风险、超长上下文的召回率问题尚未完全解决

9.2 未来展望

展望未来 1-2 年，我们可以预期：

• 更强的自主性：Composer 3 可能会实现「完全自主开发」——用户只需描述产品需求，AI 就能独立完成从需求分析、架构设计、编码、测试到部署的全流程
• 多模态集成：将视觉模型与代码模型结合，实现 UI 设计稿直接转代码（Design-to-Code）
• 实时协作：类似 Figma 的多人实时协作编程体验，每个参与者都有 AI 辅助
• 垂直领域优化：为移动开发、游戏开发、数据科学等垂直领域推出专项优化版本

软件工程正在被重新定价。GitHub 上 51% 的代码已由 AI 生成，这一比例还在快速增长。Cursor Composer 2 的出现，不是终点，而是 AI 编程 Agent 时代的新起点。

作为程序员，我们的价值不在于「写代码的速度」，而在于理解问题、定义问题、设计系统、评估方案的能力。AI 可以帮我们更快地实现，但方向的判断，永远需要人来完成。

参考来源：

• Cursor Composer 2 发布公告（2026-03-19）
• arXiv:2603.24477v2 — Composer 2 论文
• 腾讯网《Cursor 推出 Composer 2 模型》（2026-03-20）
• 新浪网《2026年AI编程助手深度测评》（2026-03-29）
• 今日头条《AI编程Agent争夺战》（2026-04-05）
• 今日头条《Cursor Composer 2 的基座模型揭秘》（2026-04-09）
• 腾讯网《Cursor团队发布专业编程AI Composer 2》（2026-04-07）