ECC 深度实战：当 AI 编程助手遇上性能工程——从 Token 优化到多语言统一编排的完全指南（2026）

作者：程序员茄子
日期：2026-06-10
字数：约 12000 字
适用读者：使用 AI 编程助手的开发者、AI Agent 框架开发者、对 Token 成本敏感的个人/团队

摘要

2026 年，AI 编程助手已经从"能写代码"进化到"能写好代码、快写代码、便宜地写代码"。ECC（Everything Claude Code） 是这一趋势下最具代表性的开源项目——它不是一个独立的 AI 工具，而是建立在 Claude Code 之上的超级 Harness（执行引擎），通过 249 个技能包、63 个代理角色、统一的多语言 SDK，以及一套系统性的 Token 优化体系，将 AI 辅助编程的效率和成本控制提升到生产级水准。

本文将从架构原理、核心模块、Token 经济学、多语言 SDK 设计、技能包开发、CI/CD 集成、性能调优七个维度，结合大量代码示例，完整拆解 ECC 的设计哲学与工程实践。

目录

1. 背景介绍：AI 编程助手的「性能墙」
2. ECC 架构全景：Harness 设计范式
3. 核心概念一：Agent 角色体系（63 个代理）
4. 核心概念二：技能包引擎（249 个技能）
5. Token 经济学：ECC 的成本控制系统
6. 多语言 SDK 统一编排架构
7. 实战一：从零搭建 ECC 开发环境
8. 实战二：编写一个自定义 Skill
9. 实战三：CI/CD 中集成 ECC 质量门
10. 性能优化：让 AI 编程快 3 倍的 7 个技巧
11. 与其他 AI 编程框架的对比
12. 总结与展望

1. 背景介绍：AI 编程助手的「性能墙」

1.1 问题的本质

2025-2026 年，几乎所有使用 AI 编程助手的开发者都遇到了同一个问题：

「助手越聪明，Token 消耗越疯狂，响应越慢，成本越高」

以 Claude Code（Anthropic 官方 CLI）为例，一次中等复杂度的多文件重构任务，上下文可能膨胀到 50K-100K Token，直接后果是：

1.2 ECC 的解决思路

ECC（Everything Claude Code）的核心设计哲学可以用一句话概括：

「把 AI 编程助手当成编译器来优化——上下文是 IR，Token 是寄存器，技能包是标准库」

项目作者是 affaan-m，2026 年初开源后迅速获得 20 万+ GitHub Stars，每日新增约 1500 Star，成为 GitHub AI 工具类增速第一的项目。

ECC 不直接调用 LLM API，而是作为 Harness（执行引擎） 位于 Claude Code / OpenClaw / Aider 等工具的上层，负责：

用户意图 → ECC Harness（路由/优化/编排）→ 底层 AI 工具 → LLM API

2. ECC 架构全景：Harness 设计范式

2.1 三层架构

┌─────────────────────────────────────────────────────────┐
│                    ECC Harness Layer                    │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐            │
│  │ Agent    │  │ Skill    │  │ Token    │            │
│  │ Router   │  │ Engine   │  │ Optimizer│            │
│  └──────────┘  └──────────┘  └──────────┘            │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐            │
│  │ Context  │  │ Multi-   │  │ Legacy   │            │
│  │ Compres- │  │ Lang SDK │  │ Compat   │            │
│  │ sor      │  │ Layer    │  │ Layer    │            │
│  └──────────┘  └──────────┘  └──────────┘            │
├─────────────────────────────────────────────────────────┤
│              AI Tool Abstraction Layer                  │
│     Claude Code │ OpenClaw │ Aider │ Cursor │ ...      │
├─────────────────────────────────────────────────────────┤
│                   LLM Provider Layer                    │
│     Anthropic API │ OpenRouter │ Ollama │ vLLM │ ...  │
└─────────────────────────────────────────────────────────┘

2.2 关键目录结构

everything-claude-code/
├── agents/              # 63 个 Agent 角色定义（YAML+MD）
├── skills/              # 249 个技能包（每个独立目录）
│   ├── code-review/
│   ├── refactor/
│   ├── test-generator/
│   └── ...
├── sdk/                 # 多语言 SDK
│   ├── python/
│   ├── typescript/
│   ├── rust/
│   └── go/
├── harness/             # 核心执行引擎
│   ├── router.py        # Agent 路由逻辑
│   ├── tokenizer.py     # Token 计数与优化
│   └── compressor.py    # 上下文压缩
├── legacy/              # 79 个遗留命令兼容层
└── docs/

3. 核心概念一：Agent 角色体系（63 个代理）

ECC 预定义了 63 个专业 Agent 角色，每个角色有独立的 System Prompt、工具权限、输出格式约定。

3.1 Agent 路由机制

# harness/router.py（简化版核心逻辑）

from enum import Enum
from dataclasses import dataclass
from typing import Optional

class AgentRole(Enum):
    # 代码质量类
    CODE_REVIEWER = "code-reviewer"
    REFACTORER = "refactorer"
    TESTER = "test-generator"
    SECURITY_AUDITOR = "security-auditor"
    
    # 架构类
    ARCHITECT = "architect"
    DB_DESIGNER = "db-designer"
    API_DESIGNER = "api-designer"
    
    #  DevOps 类
    DEVOPS_ENGINEER = "devops"
    SRE = "sre"
    CI_CD_SPECIALIST = "ci-cd"
    
    # 语言专家类（按技术栈细分）
    PYTHON_EXPERT = "python-expert"
    RUST_EXPERT = "rust-expert"
    GOLANG_EXPERT = "golang-expert"
    FRONTEND_EXPERT = "frontend-expert"
    # ... 共 63 个

@dataclass
class RouteDecision:
    primary_agent: AgentRole
    secondary_agents: list[AgentRole]
    confidence: float
    reason: str

class AgentRouter:
    """意图 → Agent 路由引擎"""
    
    def __init__(self, embedding_model: str = "text-embedding-3-small"):
        self.embedding_model = embedding_model
        # 每个 Agent 的角色描述嵌入向量，用于语义匹配
        self.agent_embeddings = self._precompute_embeddings()
    
    def route(self, user_intent: str, context: dict) -> RouteDecision:
        """
        核心路由逻辑：
        1. 对 user_intent 做嵌入
        2. 与所有 Agent 的描述嵌入计算余弦相似度
        3. 返回 Top-K 匹配结果
        4. 结合上下文（文件类型、项目语言）做二次过滤
        """
        intent_embedding = self._embed(user_intent)
        
        scores = {}
        for agent, emb in self.agent_embeddings.items():
            scores[agent] = self._cosine_similarity(intent_embedding, emb)
        
        # 上下文加权：如果用户打开了 .py 文件，提升 PYTHON_EXPERT 权重
        if context.get("active_file", "").endswith(".py"):
            scores[AgentRole.PYTHON_EXPERT] *= 1.5
        
        # 取 Top-3
        sorted_agents = sorted(scores.items(), key=lambda x: x[1], reverse=True)
        primary = sorted_agents[0][0]
        secondary = [a for a, _ in sorted_agents[1:3]]
        
        return RouteDecision(
            primary_agent=primary,
            secondary_agents=secondary,
            confidence=sorted_agents[0][1],
            reason=f"Semantic match: {primary.value} (score={sorted_agents[0][1]:.3f})"
        )

3.2 多 Agent 协作模式

ECC 支持 链式（Chain）、并行（Parallel）、专家审议（Review） 三种协作模式：

# 示例：代码重构任务的多 Agent 协作

task = "重构 user_service.py，提升可读性，添加类型注解，补充单元测试"

# ECC 内部执行流程：
# 1. Router 识别主 Agent: REFACTORER
# 2. 自动添加协同 Agent: PYTHON_EXPERT（类型注解）+ TESTER（测试）
# 3. 执行模式: Review（先重构 → 再审查 → 最后测试）

execution_plan = {
    "mode": "review",
    "steps": [
        {"agent": "refactorer", "task": "重构 user_service.py"},
        {"agent": "python-expert", "task": "添加 Pydantic 类型注解"},
        {"agent": "code-reviewer", "task": "审查重构结果"},
        {"agent": "test-generator", "task": "生成 pytest 测试用例"},
    ]
}

4. 核心概念二：技能包引擎（249 个技能）

4.1 技能包的标准结构

每个 Skill 是一个独立目录，包含元数据声明、执行脚本、测试用例：

# skills/code-review/skill.yaml

name: "code-review"
version: "2.4.0"
description: "深度代码审查：安全漏洞 + 性能问题 + 规范违反"
authors: ["affaan-m", "community"]
tags: ["quality", "security", "review"]

# 触发条件（自然语言模式匹配）
triggers:
  - pattern: "审查|review|检查代码"
    language: "any"
  - pattern: "security audit|安全检查"
    language: "any"

# Token 预算（控制该技能的最大上下文消耗）
token_budget:
  input: 15000    # 输入最大 Token 数
  output: 5000     # 输出最大 Token 数

# 依赖的其他技能
depends_on:
  - "syntax-check"
  - "static-analysis"

# 输出格式
output_schema:
  type: "json"
  fields:
    - name: "issues"
      type: "array"
      items:
        severity: "enum[critical, high, medium, low]"
        file: "string"
        line: "integer"
        message: "string"
        suggestion: "string"

4.2 技能包的 Python 实现（简化）

# skills/code-review/execute.py

import ast
import json
import subprocess
from pathlib import Path
from typing import Optional

class CodeReviewSkill:
    """
    ECC 代码审查技能包
    
    设计要点：
    1. 先用静态分析工具（不消耗 Token）做初筛
    2. 只对「静态分析无法判断」的问题调用 LLM
    3. 结果缓存：相同代码片段不重复分析
    """
    
    def __init__(self, workspace: Path, llm_client):
        self.workspace = workspace
        self.llm = llm_client
        self._cache = {}  # 简单的内存缓存
    
    def execute(self, target_files: list[str], depth: str = "deep") -> dict:
        """
        执行代码审查
        
        Args:
            target_files: 需要审查的文件列表
            depth: "quick" | "deep" | "security-focus"
        """
        results = {"issues": [], "stats": {}}
        
        for file_path in target_files:
            file_content = Path(file_path).read_text()
            
            # Step 1: 静态分析（零 Token 消耗）
            static_issues = self._run_static_analysis(file_path)
            results["issues"].extend(static_issues)
            
            # Step 2: AST 分析（零 Token 消耗）
            ast_issues = self._analyze_ast(file_content, file_path)
            results["issues"].extend(ast_issues)
            
            # Step 3: 只用 LLM 分析「语义级」问题（高 Token 价值）
            if depth == "deep":
                semantic_issues = self._llm_semantic_analysis(
                    file_content, file_path
                )
                results["issues"].extend(semantic_issues)
        
        # 去重 + 按严重程度排序
        results["issues"] = self._deduplicate(results["issues"])
        results["issues"].sort(
            key=lambda x: {"critical": 0, "high": 1, "medium": 2, "low": 3}[x["severity"]]
        )
        
        return results
    
    def _run_static_analysis(self, file_path: str) -> list[dict]:
        """运行 ruff/flake8/eslint 等工具，零 Token 消耗"""
        issues = []
        
        if file_path.endswith(".py"):
            # 运行 ruff（比 flake8 快 10-100 倍）
            result = subprocess.run(
                ["ruff", "check", file_path, "--output-format=json"],
                capture_output=True, text=True
            )
            if result.stdout:
                for item in json.loads(result.stdout):
                    issues.append({
                        "severity": self._map_severity(item["code"]),
                        "file": file_path,
                        "line": item["location"]["row"],
                        "message": item["message"],
                        "suggestion": "",  # 静态分析不提供修复建议
                        "source": "ruff"
                    })
        
        return issues
    
    def _analyze_ast(self, content: str, file_path: str) -> list[dict]:
        """Python AST 分析：找死代码、未使用变量、复杂度过高函数"""
        issues = []
        
        if not file_path.endswith(".py"):
            return issues
        
        try:
            tree = ast.parse(content)
        except SyntaxError:
            return [{"severity": "critical", "file": file_path, 
                     "line": 0, "message": "语法错误", "source": "ast"}]
        
        # 检查函数复杂度（McCabe）
        for node in ast.walk(tree):
            if isinstance(node, ast.FunctionDef):
                complexity = self._compute_complexity(node)
                if complexity > 15:
                    issues.append({
                        "severity": "medium",
                        "file": file_path,
                        "line": node.lineno,
                        "message": f"函数 `{node.name}` 圈复杂度 = {complexity}，建议拆分",
                        "suggestion": "将复杂分支拆分为独立函数",
                        "source": "ast-complexity"
                    })
        
        return issues
    
    def _llm_semantic_analysis(self, content: str, file_path: str) -> list[dict]:
        """
        语义分析：只发给 LLM 做「静态分析做不了」的判断
        
        关键优化：只发送函数的「签名 + docstring + 前 10 行」，
        而不是整个文件，节省 60-80% 的 Token
        """
        cache_key = hash(content)
        if cache_key in self._cache:
            return self._cache[cache_key]
        
        # 提取「值得 LLM 分析」的代码块
        interesting_blocks = self._extract_interesting_blocks(content)
        
        # 构建精简 Prompt（这是 ECC 的核心优化之一）
        prompt = self._build_compact_review_prompt(interesting_blocks, file_path)
        
        # 调用 LLM
        response = self.llm.complete(
            prompt=prompt,
            max_tokens=2000,
            temperature=0.1  # 代码审查要确定性
        )
        
        issues = self._parse_llm_response(response)
        self._cache[cache_key] = issues
        return issues
    
    def _build_compact_review_prompt(self, blocks: list[str], file_path: str) -> str:
        """
        精简 Prompt 构建：
        
        传统做法（反例）：
        "请审查以下代码：\n```{content}```"
        → 消耗 Token = 完整文件大小，通常 3000-8000 Token
        
        ECC 做法（正例）：
        "审查重点：以下代码块存在潜在风险，请重点检查：
         [块1] 函数签名 + 前 10 行
         [块2] 错误处理部分
         [块3] SQL 拼接位置"
        → 消耗 Token = 约 800-1500 Token，节省 70%
        """
        return f"""你是一个资深代码审查专家。请审查以下代码块，重点关注：
1. 安全漏洞（SQL 注入、XSS、不安全的反序列化）
2. 并发问题（竞态条件、死锁）
3. 资源管理（内存泄漏、未关闭的连接）

文件：{file_path}

{chr(10).join(f"【代码块 {i+1}】{block}" for i, block in enumerate(blocks))}

以 JSON 格式输出，每个问题包含：severity、line、message、suggestion。
"""

5. Token 经济学：ECC 的成本控制系统

5.1 Token 消耗的「三层漏斗」

ECC 将 Token 消耗分为三个层次，逐层优化：

Layer 1: 上下文压缩（Compressor）
  ↓ 压缩比可达 5:1 ~ 10:1
Layer 2: 智能缓存（Cache）
  ↓ 命中率通常 40%-70%
Layer 3: 技能级预算控制（Budget Controller）
  ↓ 防止单个技能失控消耗

5.2 上下文压缩算法

# harness/compressor.py

import tiktoken
from dataclasses import dataclass
from typing import Optional

@dataclass
class CompressionResult:
    original_tokens: int
    compressed_tokens: int
    ratio: float
    compressed_text: str

class ContextCompressor:
    """
    上下文压缩引擎
    
    策略（按优先级）：
    1. 系统提示词缓存：Claude 支持 Cache Control，对固定内容只计费一次
    2. 对话历史摘要：将超过窗口的旧对话压缩为摘要
    3. 代码差异优先：只发送 diff，不发送完整文件
    4. 语义去重：相似的用户指令只保留一个
    """
    
    def __init__(self, model: str = "claude-sonnet-4-20250514"):
        self.model = model
        self.encoder = tiktoken.encoding_for_model("gpt-4")  # 近似估算
    
    def compress(self, full_context: str, strategy: str = "auto") -> CompressionResult:
        original_tokens = len(self.encoder.encode(full_context))
        
        if strategy == "diff_only":
            # 策略：只保留 diff（适用于代码审查/重构场景）
            compressed = self._extract_diff_only(full_context)
        elif strategy == "summary":
            # 策略：旧对话摘要化
            compressed = self._summarize_old_turns(full_context)
        elif strategy == "auto":
            # 自动选择：根据上下文类型智能组合
            compressed = self._auto_compress(full_context)
        else:
            compressed = full_context
        
        compressed_tokens = len(self.encoder.encode(compressed))
        
        return CompressionResult(
            original_tokens=original_tokens,
            compressed_tokens=compressed_tokens,
            ratio=original_tokens / max(compressed_tokens, 1),
            compressed_text=compressed
        )
    
    def _extract_diff_only(self, context: str) -> str:
        """
        从上下文中提取代码 diff
        
        示例转换：
        输入（8000 Token）：
        「这是 user_service.py 的完整内容：
        [800 行代码]
        请帮我重构这个函数」
        
        输出（1200 Token）：
        「user_service.py 的改动：
        + 新增函数 validate_email()
        - 删除函数 old_validate()
        请帮我审查这些改动」
        """
        # 实际实现会使用 git diff 解析 + 变更块提取
        # 这里展示核心思路
        return self._parse_git_diff(context)
    
    def _auto_compress(self, context: str) -> str:
        """
        自动压缩策略：
        
        1. 如果上下文包含完整文件 → 替换为 diff
        2. 如果对话超过 20 轮 → 前 10 轮压缩为摘要
        3. 如果系统提示词超过 2000 Token → 启用 Cache Control 标记
        4. 如果多个文件内容重复 → 去重只保留一个
        """
        # 检测是否包含完整文件
        if self._contains_full_file(context):
            context = self._replace_files_with_diffs(context)
        
        # 对话历史压缩
        turns = self._parse_conversation_turns(context)
        if len(turns) > 20:
            summary = self._summarize_turns(turns[:10])
            context = summary + "\n\n" + self._format_turns(turns[10:])
        
        return context

5.3 成本追踪与预算告警

# harness/token_tracker.py

from dataclasses import dataclass, field
from datetime import datetime
import json

@dataclass
class TokenUsage:
    input_tokens: int
    output_tokens: int
    cache_creation_tokens: int = 0
    cache_read_tokens: int = 0
    cost_usd: float = 0.0
    
    @property
    def total_tokens(self) -> int:
        return self.input_tokens + self.output_tokens

@dataclass
class BudgetPolicy:
    daily_limit_usd: float = 50.0
    per_request_limit_tokens: int = 50000
    warn_threshold_ratio: float = 0.8  # 达到 80% 预算时警告

class TokenTracker:
    """
    Token 使用追踪器
    
    功能：
    - 实时追踪每次 API 调用的 Token 消耗
    - 预算控制：超过限制自动拒绝或降级
    - 成本分析：按 Agent / Skill / 用户维度统计
    """
    
    def __init__(self, policy: BudgetPolicy):
        self.policy = policy
        self._session_usage: list[TokenUsage] = []
        self._daily_usage: dict[str, TokenUsage] = {}  # date -> usage
    
    def track(self, usage: TokenUsage, metadata: dict = None):
        """记录一次 API 调用的 Token 消耗"""
        self._session_usage.append(usage)
        
        today = datetime.now().strftime("%Y-%m-%d")
        if today not in self._daily_usage:
            self._daily_usage[today] = TokenUsage(0, 0)
        
        day_usage = self._daily_usage[today]
        day_usage.input_tokens += usage.input_tokens
        day_usage.output_tokens += usage.output_tokens
        day_usage.cost_usd += usage.cost_usd
        
        # 预算检查
        if day_usage.cost_usd > self.policy.daily_limit_usd * self.policy.warn_threshold_ratio:
            print(f"⚠️  Token 预算警告：今日已消耗 ${day_usage.cost_usd:.2f}，"
                  f"预算 ${self.policy.daily_limit_usd:.2f}")
        
        if day_usage.cost_usd > self.policy.daily_limit_usd:
            raise BudgetExceededError(
                f"今日 Token 预算 ${self.policy.daily_limit_usd} 已用尽，"
                f"当前消耗 ${day_usage.cost_usd:.2f}"
            )
    
    def get_session_summary(self) -> dict:
        """获取当前会话的 Token 使用摘要"""
        total_input = sum(u.input_tokens for u in self._session_usage)
        total_output = sum(u.output_tokens for u in self._session_usage)
        total_cost = sum(u.cost_usd for u in self._session_usage)
        
        return {
            "total_input_tokens": total_input,
            "total_output_tokens": total_output,
            "total_tokens": total_input + total_output,
            "total_cost_usd": round(total_cost, 4),
            "request_count": len(self._session_usage),
            "avg_tokens_per_request": (total_input + total_output) // max(len(self._session_usage), 1),
        }

6. 多语言 SDK 统一编排架构

ECC 最工程化的设计之一是其 多语言 SDK 统一编排层，支持 Python、TypeScript、Rust、Go 四种语言调用同一套技能包。

6.1 SDK 架构设计

┌──────────────────────────────────────────────┐
│           User Code (任何语言)               │
├──────────────────────────────────────────────┤
│         ECC SDK (Python/TS/Rust/Go)         │
│  - 统一的 Skill 调用接口                    │
│  - 本地 Token 预算控制                      │
│  - 结果类型定义                             │
├──────────────────────────────────────────────┤
│         Harness HTTP API (本地)              │
│  POST /api/v1/run-skill                     │
│  POST /api/v1/route-agent                   │
│  GET  /api/v1/status                        │
├──────────────────────────────────────────────┤
│         ECC Harness Core (Python)            │
└──────────────────────────────────────────────┘

6.2 TypeScript SDK 示例

// sdk/typescript/src/index.ts

import { EventEmitter } from "events";
import axios, { AxiosInstance } from "axios";

// ─── 类型定义 ───

export interface SkillResult {
  success: boolean;
  data: any;
  tokens_used: number;
  execution_time_ms: number;
  warnings?: string[];
}

export interface AgentRoute {
  primary_agent: string;
  secondary_agents: string[];
  confidence: number;
}

export interface ECCConfig {
  harnessUrl?: string;      // 本地 Harness 地址，默认 http://localhost:19000
  apiKey?: string;          // 可选，远程模式需要
  defaultTokenBudget?: number;
  timeoutMs?: number;
}

// ─── 主客户端类 ───

export class ECCClient extends EventEmitter {
  private http: AxiosInstance;
  private config: Required<ECCConfig>;

  constructor(config: ECCConfig = {}) {
    super();
    this.config = {
      harnessUrl: config.harnessUrl ?? "http://localhost:19000",
      apiKey: config.apiKey ?? "",
      defaultTokenBudget: config.defaultTokenBudget ?? 10000,
      timeoutMs: config.timeoutMs ?? 60000,
    };

    this.http = axios.create({
      baseURL: this.config.harnessUrl,
      timeout: this.config.timeoutMs,
      headers: this.config.apiKey
        ? { Authorization: `Bearer ${this.config.apiKey}` }
        : {},
    });
  }

  /**
   * 执行一个技能包
   * 
   * @example
   * const result = await ecc.runSkill("code-review", {
   *   files: ["./src/user_service.py"],
   *   depth: "deep"
   * });
   */
  async runSkill(
    skillName: string,
    params: Record<string, any>,
    options: { tokenBudget?: number; signal?: AbortSignal } = {}
  ): Promise<SkillResult> {
    const tokenBudget = options.tokenBudget ?? this.config.defaultTokenBudget;

    // 发送进度事件
    this.emit("skill:start", { skillName, params });

    try {
      const response = await this.http.post("/api/v1/run-skill", {
        skill: skillName,
        params,
        token_budget: tokenBudget,
      }, {
        signal: options.signal,
        onUploadProgress: (progress) => {
          this.emit("skill:progress", {
            skillName,
            phase: "uploading",
            progress: progress.progress ?? 0,
          });
        },
      });

      const result: SkillResult = response.data;
      this.emit("skill:complete", { skillName, result });
      return result;
    } catch (error: any) {
      const err = new Error(
        `Skill "${skillName}" failed: ${error.response?.data?.message ?? error.message}`
      );
      this.emit("skill:error", { skillName, error: err });
      throw err;
    }
  }

  /**
   * 路由用户意图到合适的 Agent
   */
  async routeIntent(userIntent: string, context: Record<string, any> = {}): Promise<AgentRoute> {
    const response = await this.http.post("/api/v1/route-agent", {
      intent: userIntent,
      context,
    });
    return response.data as AgentRoute;
  }

  /**
   * 获取 Token 使用统计
   */
  async getTokenStats(): Promise<Record<string, any>> {
    const response = await this.http.get("/api/v1/token-stats");
    return response.data;
  }
}

// ─── 使用示例 ───

async function main() {
  const ecc = new ECCClient({ harnessUrl: "http://localhost:19000" });

  // 监听事件
  ecc.on("skill:start", ({ skillName }) => {
    console.log(`🚀 开始执行技能: ${skillName}`);
  });
  ecc.on("skill:complete", ({ skillName, result }) => {
    console.log(`✅ ${skillName} 完成，消耗 Token: ${result.tokens_used}`);
  });

  // 执行代码审查
  const reviewResult = await ecc.runSkill("code-review", {
    files: ["./src/user_service.py", "./src/auth.py"],
    depth: "deep",
    focus: ["security", "performance"],
  });

  if (reviewResult.success) {
    const issues = reviewResult.data.issues as any[];
    console.log(`发现 ${issues.length} 个问题：`);
    for (const issue of issues) {
      console.log(`  [${issue.severity}] ${issue.file}:${issue.line} - ${issue.message}`);
      if (issue.suggestion) {
        console.log(`    💡 建议：${issue.suggestion}`);
      }
    }
  }

  // 查看 Token 统计
  const stats = await ecc.getTokenStats();
  console.log(`今日 Token 消耗：$${stats.daily_cost_usd}`);
}

main().catch(console.error);

7. 实战一：从零搭建 ECC 开发环境

7.1 安装（三种方式）

# 方式一：pip（推荐个人开发者）
pip install everything-claude-code
ecc setup    # 交互式配置向导

# 方式二：Docker（推荐团队/CI 环境）
docker pull affaanm/ecc:latest
docker run -v $(pwd):/workspace -p 19000:19000 affaanm/ecc:latest

# 方式三：源码安装（推荐技能包开发者）
git clone https://github.com/affaan-m/everything-claude-code.git
cd everything-claude-code
pip install -e ".[dev]"   # 包含测试、lint、格式化工具

7.2 配置文件（~/.ecc/config.yaml）

# ECC 主配置文件

# LLM 提供商配置
llm:
  provider: "anthropic"          # anthropic / openrouter / ollama
  model: "claude-sonnet-4-20250514"
  api_key_env: "ANTHROPIC_API_KEY"
  max_tokens: 8192
  temperature: 0.1

# Token 预算控制
token_budget:
  daily_limit_usd: 20.0         # 每日上限 $20
  per_request_limit: 30000       # 单次请求上限 30000 Token
  enable_cache: true             # 启用系统提示词缓存

# Agent 路由配置
routing:
  strategy: "semantic"           # semantic / keyword / hybrid
  top_k: 3                       # 最多同时激活 3 个 Agent
  confidence_threshold: 0.6      # 置信度低于 0.6 时询问用户

# 技能包配置
skills:
  enabled: ["code-review", "refactor", "test-generator", "security-audit"]
  auto_discover: true            # 自动发现 ~/.ecc/skills/ 下的自定义技能
  timeout_seconds: 300

# 日志
logging:
  level: "INFO"
  file: "~/.ecc/logs/ecc.log"
  enable_token_log: true         # 记录每次请求的 Token 消耗

8. 实战二：编写一个自定义 Skill

下面从头编写一个 「API 性能分析」技能包，演示完整的 Skill 开发流程。

8.1 技能包目录结构

~/.ecc/skills/api-perf-analyzer/
├── skill.yaml          # 元数据声明
├── execute.py          # 执行脚本
├── prompt_template.md  # LLM Prompt 模板
├── tests/
│   └── test_execute.py
└── README.md

8.2 skill.yaml

name: "api-perf-analyzer"
version: "1.0.0"
description: "API 性能分析：自动识别慢接口、N+1 查询、缺乏缓存的热点路径"
authors: ["your-name"]
tags: ["performance", "api", "analysis"]

triggers:
  - pattern: "分析接口性能|API 性能|慢接口|perf analysis"
    language: "any"

token_budget:
  input: 10000
  output: 3000

depends_on: []
output_schema:
  type: "json"
  fields:
    - name: "hotspots"
      type: "array"
    - name: "suggestions"
      type: "array"

8.3 execute.py

"""
API 性能分析技能包

工作原理：
1. 解析项目代码，提取所有 API 路由定义
2. 静态分析每个处理函数的数据库查询模式
3. 检测常见的性能反模式（N+1、无分页、同步阻塞调用）
4. 用 LLM 对检测结果做优先级排序和修复建议生成
"""

import ast
import json
import re
from pathlib import Path
from typing import Optional

def main(files: list[str], framework: Optional[str] = None) -> dict:
    """
    ECC 技能包标准入口函数
    
    Args:
        files: 要分析的文件列表
        framework: 显式指定框架（auto / flask / django / fastapi / express）
    """
    results = {
        "hotspots": [],
        "suggestions": [],
        "stats": {"files_analyzed": 0, "routes_found": 0}
    }
    
    for file_path in files:
        if not file_path.endswith(".py"):
            continue
        
        content = Path(file_path).read_text()
        results["stats"]["files_analyzed"] += 1
        
        # Step 1: 提取路由定义
        routes = extract_routes(content, file_path)
        results["stats"]["routes_found"] += len(routes)
        
        # Step 2: 静态分析性能反模式
        for route in routes:
            issues = analyze_route_performance(route, content)
            if issues:
                results["hotspots"].append({
                    "route": route["path"],
                    "method": route["method"],
                    "file": file_path,
                    "line": route["line"],
                    "issues": issues
                })
    
    # Step 3: 用 LLM 生成修复建议（只对检测到的 hotspot 调用）
    if results["hotspots"]:
        results["suggestions"] = generate_fix_suggestions(results["hotspots"])
    
    return results

def extract_routes(content: str, file_path: str) -> list[dict]:
    """从 Python 代码中提取 API 路由定义"""
    routes = []
    
    # Flask 风格
    flask_pattern = re.compile(r'@app\.route\(\s*["\']([^"\']+)["\']\s*,\s*methods=\[([^\]]+)\]')
    for match in flask_pattern.finditer(content):
        routes.append({
            "path": match.group(1),
            "method": match.group(2),
            "line": content[:match.start()].count("\n") + 1,
            "framework": "flask"
        })
    
    # FastAPI 风格（简化检测）
    fastapi_pattern = re.compile(r'@(app|router)\.(get|post|put|delete|patch)\(\s*["\']([^"\']+)["\']')
    for match in fastapi_pattern.finditer(content):
        routes.append({
            "path": match.group(3),
            "method": match.group(2).upper(),
            "line": content[:match.start()].count("\n") + 1,
            "framework": "fastapi"
        })
    
    return routes

def analyze_route_performance(route: dict, content: str) -> list[str]:
    """
    静态分析路由处理函数的性能问题
    
    检测的反模式：
    - N+1 查询：在 for 循环中使用 ORM 查询
    - 缺少分页：查询没有 limit()/paginate()
    - 同步阻塞：在异步路由中调用 requests.get()
    - 大结果集：SELECT * 没有指定字段
    """
    issues = []
    
    # 获取路由处理函数名
    func_name = extract_route_handler_name(route, content)
    if not func_name:
        return issues
    
    # 提取函数体
    func_body = extract_function_body(func_name, content)
    
    # 检测 N+1 查询
    if re.search(r'for\s+\w+\s+in\s+\w+:', func_body):
        if re.search(r'\.filter\(|\.get\(|\.objects\(', func_body):
            issues.append("疑似 N+1 查询：在 for 循环中存在数据库查询")
    
    # 检测缺少分页
    if "objects.filter" in func_body and "paginate" not in func_body and "limit" not in func_body:
        issues.append("缺少分页：列表接口可能返回大量数据")
    
    # 检测同步阻塞调用（在 async 函数中）
    if route.get("method") and "async def" in content:
        if "requests." in func_body or "urllib" in func_body:
            issues.append("同步 HTTP 调用：在异步函数中使用同步 HTTP 客户端")
    
    return issues

def generate_fix_suggestions(hotspots: list[dict]) -> list[dict]:
    """
    调用 LLM 生成修复建议
    
    注意：这里只发送 hotspot 摘要，不发送完整代码，节省 Token
    """
    # 在实际实现中，这里会调用 ECC 的 LLM 客户端
    # 这里展示 Prompt 设计
    
    prompt = f"""你是一个 API 性能优化专家。

以下 API 热点路径存在性能问题，请为每个问题提供：
1. 问题的具体解释
2. 修复后的代码示例
3. 预期的性能提升（量化）

问题列表：
{json.dumps(hotspots, ensure_ascii=False, indent=2)}

以 JSON 数组格式输出，每个元素包含：route、issue、explanation、fixed_code、estimated_improvement。
"""
    
    # llm_response = llm_client.complete(prompt, max_tokens=3000)
    # return json.loads(llm_response)
    
    # 示例返回（实际由 LLM 生成）
    return [
        {
            "route": "/api/users/<id>/posts",
            "issue": "N+1 查询",
            "explanation": "每个用户的 posts 在循环中被单独查询，应使用 select_related/prefetch_related",
            "fixed_code": (
                "posts = Post.objects.filter(user__in=user_ids)"
                ".select_related('user').prefetch_related('comments')"
            ),
            "estimated_improvement": "查询次数从 N+1 减少到 2-3 次，P99 延迟降低 80%"
        }
    ]

9. 实战三：CI/CD 中集成 ECC 质量门

9.1 GitHub Actions 集成

# .github/workflows/ecc-quality-gate.yml

name: ECC Quality Gate

on:
  pull_request:
    types: [opened, synchronize]
  push:
    branches: [main, develop]

jobs:
  ecc-review:
    runs-on: ubuntu-latest
    
    steps:
      - name: Checkout code
        uses: actions/checkout@v4
        with:
          fetch-depth: 0  # 需要完整历史用于 diff
      
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      
      - name: Install ECC
        run: pip install everything-claude-code
      
      - name: Start ECC Harness
        run: |
          ecc serve --daemon --port 19000
          sleep 3  # 等待服务启动
      
      - name: Run ECC Code Review
        id: review
        run: |
          # 获取 PR 的变更文件
          CHANGED_FILES=$(git diff --name-only origin/main...HEAD | grep -E '\.(py|js|ts|go|rs)$' | tr '\n' ' ')
          
          if [ -n "$CHANGED_FILES" ]; then
            # 调用 ECC 代码审查技能
            RESULT=$(ecc run-skill code-review \
              --files $CHANGED_FILES \
              --format json \
              --token-budget 15000)
            
            echo "result=$RESULT" >> $GITHUB_OUTPUT
            
            # 检查是否有 critical 级别的问题
            CRITICAL_COUNT=$(echo $RESULT | jq '.issues | map(select(.severity=="critical")) | length')
            if [ "$CRITICAL_COUNT" -gt 0 ]; then
              echo "❌ ECC 质量门失败：发现 $CRITICAL_COUNT 个 critical 级别问题"
              exit 1
            fi
          else
            echo "ℹ️  无代码文件变更，跳过 ECC 审查"
          fi
      
      - name: Run ECC Security Audit
        if: steps.review.outcome == 'success'
        run: |
          CHANGED_FILES=$(git diff --name-only origin/main...HEAD | grep -E '\.(py|js|ts)$' | tr '\n' ' ')
          if [ -n "$CHANGED_FILES" ]; then
            ecc run-skill security-audit \
              --files $CHANGED_FILES \
              --fail-on HIGH  # 发现 HIGH 级别漏洞时失败
          fi
      
      - name: Comment PR with ECC Results
        if: always() && github.event_name == 'pull_request'
        uses: actions/github-script@v7
        with:
          script: |
            const result = ${{ steps.review.outputs.result || '{}' }};
            const issues = result.issues || [];
            
            let body = "## 🤖 ECC 代码审查结果\n\n";
            
            if (issues.length === 0) {
              body += "✅ 未发现代码质量问题\n";
            } else {
              body += `共发现 **${issues.length}** 个问题：\n\n`;
              for (const issue of issues.slice(0, 10)) {  # 最多展示 10 个
                const icon = issue.severity === 'critical' ? '🔴' : 
                            issue.severity === 'high' ? '🟠' : 
                            issue.severity === 'medium' ? '🟡' : '🟢';
                body += `${icon} **[${issue.severity}]** \`${issue.file}:${issue.line}\` - ${issue.message}\n`;
                if (issue.suggestion) {
                  body += `   💡 ${issue.suggestion}\n`;
                }
              }
            }
            
            body += `\n---\n*由 ECC (Everything Claude Code) 自动生成*`;
            
            github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner,
              repo: context.repo.repo,
              body: body
            });

10. 性能优化：让 AI 编程快 3 倍的 7 个技巧

基于 ECC 社区的最佳实践，总结 7 个可立即落地的优化技巧：

技巧 1：启用上下文缓存（Claude Cache Control）

# 对系统提示词和项目结构描述启用缓存
# Claude API 对 Cache Control 的支持可以节省 70-90% 的重复请求成本

system_prompt = """你是一个 Python 专家..

.# 这里是 2000 Token 的固定系统提示词
"""

# 普通做法（每次都计费）
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    system=system_prompt,  # 每次请求都消耗 2000 input Token
    messages=[...]
)

# ECC 做法（第一次计费，后续命中缓存只收 $0.30/MTok）
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    system=[{
            "type": "text",
            "text": system_prompt,
            "cache_control": {"type": "ephemeral"}  # ← 关键
        }],
    messages=[...]
)

技巧 2：用 diff 替代完整文件

# ❌ 错误做法：每次都发送完整文件
prompt = f"请审查以下代码：\n{read_file('user_service.py')}"

# ✅ 正确做法：只发送 diff
diff = subprocess.check_output(["git", "diff", "HEAD~1", "--", "user_service.py"]).decode()
prompt = f"请审查以下代码变更：\n{diff}"
# Token 消耗从 5000 降至 300-800

技巧 3：批量处理同类任务

# ❌ 慢：每个文件单独请求（N 次网络往返 + N 次系统提示词计费）
for file in files:
    review_file(file)

# ✅ 快：批量请求（1 次网络往返 + 1 次系统提示词计费）
prompt = "请依次审查以下文件，以 JSON 数组格式输出每个文件的问题：\n"
for i, file in enumerate(files):
    prompt += f"\n### 文件 {i+1}: {file}\n{read_file(file)[:2000]}"  # 每个文件只取前 2000 字符
response = llm.complete(prompt)

技巧 4：设置 Token 预算上限

# ~/.ecc/config.yaml
token_budget:
  per_request_limit: 15000   # 单个请求不超过 15000 Token
  warn_at_ratio: 0.8         # 达到 80% 时打印警告

技巧 5：使用更便宜的模型做预处理

# ECC 的分层模型策略：
# 1. 代码分类/路由 → 用 Haiku（便宜、快）
# 2. 代码生成 → 用 Sonnet（质量好）
# 3. 代码审查（初筛）→ 用 Haiku
# 4. 代码审查（最终确认）→ 用 Sonnet

def route_task(task: str) -> str:
    # 用便宜模型做路由决策
    return haiku.complete(f"这个任务最适合哪个 Agent？{task}")

def execute_task(task: str, agent: str):
    # 用高质量模型执行
    return sonnet.complete(f"你是 {agent}，请完成：{task}")

技巧 6：结果缓存

import hashlib
import json

class ResultCache:
    """对相同输入缓存 LLM 结果"""
    
    def __init__(self, cache_dir: Path):
        self.cache_dir = cache_dir
        self.cache_dir.mkdir(exist_ok=True)
    
    def get(self, prompt: str, model: str) -> Optional[str]:
        key = hashlib.sha256(f"{model}:{prompt}".encode()).hexdigest()
        cache_file = self.cache_dir / f"{key}.json"
        if cache_file.exists():
            return json.loads(cache_file.read_text())["response"]
        return None
    
    def set(self, prompt: str, model: str, response: str):
        key = hashlib.sha256(f"{model}:{prompt}".encode()).hexdigest()
        cache_file = self.cache_dir / f"{key}.json"
        cache_file.write_text(json.dumps({"response": response}))

技巧 7：选择支持长上下文的模型

Claude Sonnet 4：200K 上下文 → 适合大项目分析
GPT-4.1：128K 上下文
Gemini 2.5 Pro：1M 上下文 → 适合超大型代码库

如果项目超过 50K Token，优先选择 Claude Sonnet 4 或 Gemini 2.5 Pro

11. 与其他 AI 编程框架的对比

12. 总结与展望

本文回顾

ECC（Everything Claude Code）通过 Harness 架构范式，将 AI 编程助手从「能用」提升到「好用、快、便宜」的生产级水准。其核心创新在于：

1. Agent 角色体系（63 个）：让专业的事由专业的 Agent 做
2. 技能包引擎（249 个）：可组合、可扩展的能力单元
3. 三层 Token 优化：压缩 → 缓存 → 预算控制，综合节省 60-80% 成本
4. 多语言 SDK：统一编排层，支持 Python/TS/Rust/Go
5. CI/CD 原生集成：质量门、安全审计、自动 PR 评论

2026 年下半年展望

根据 ECC 路线图和社区讨论，值得关注的方向：

• 分布式 Harness：多人协作时共享 Token 预算和技能包缓存
• RAG 增强：将项目文档、Issue 历史、PR 评论纳入上下文
• 更多语言支持：Swift、Kotlin 社区技能包正在开发中
• Self-hosting 部署包：Docker Compose 一键部署完整 ECC 栈

参考资源

• ECC GitHub：https://github.com/affaan-m/everything-claude-code
• ECC 文档：https://ecc.dev
• Anthropic Cache Control：https://docs.anthropic.com/claude/docs/prompt-caching
• 本文完整代码示例：https://github.com/yourname/ecc-examples

本文由 AI 辅助撰写，程序员茄子 2026-06-10 发布于 chenxutan.com