2024 CANN 开源社区举办首届昇腾AI创新大赛（Ascend AI Innovation Contest），参赛者提交 Skill（模型+算子+部署脚本+Docker），经过海选→初赛→复赛→决赛四轮评审，最终 12 支队伍共享 50 万奖金池。cann-competitions 仓库管理竞赛全流程：赛道定义（文本生成/图像分类/语音识别/科学推理 4 赛道）→ 评委分配 → 自动评分 → 排行榜 → 奖金分配。内置自动化评测 pipeline（同一 Docker 镜像在不同维度打分，消除人为主观误差）。

cann-competitions 不只是赛事管理 web 页面——它是社区竞赛的规则引擎。每个赛道有独立的 metric（精度、吞吐、延迟、功耗、模型体积），通过 CI 自动跑分 + 多维度排名，评审委员会只需确认结果。

竞赛赛道架构（2024 首届）

CANN AI Innovation Contest 2024
├── 赛道A：文本生成（LLM推理加速）
│   ├── 参赛要求：用 ATB + torch_npu 部署 LLaMA 2/3 等模型
│   ├── 评分：推理吞吐 × 0.4 + 延迟 × 0.3 + 显存效率 × 0.3
│   └── 基准数据集：ShareGPT (conv 1K)
│
├── 赛道B：图像分类（视觉模型优化）
│   ├── 参赛要求：ResNet/ViT 用 CANN ops-cv + ops-nn 加速
│   ├── 评分：Top1精度 × 0.3 + 推理吞吐 × 0.35 + 能效 × 0.35
│   └── 基准数据集：ImageNet-1K val
│
├── 赛道C：语音识别（声学模型部署）
│   ├── 参赛要求：Whisper/Conformer 用 CANN sip 信号处理加速
│   ├── 评分：WER × 0.4 + 实时率 × 0.3 + 功耗 × 0.3
│   └── 基准数据集：LibriSpeech test-clean
│
└── 赛道D：科学推理（数学/物理问题求解）
    ├── 参赛要求：SciNet 等模型用 CANN ops-math + ops-logic 加速
    ├── 评分：准确率 × 0.5 + 推理速度 × 0.5
    └── 基准数据集：MATH-500

核心：自动评分引擎（Competition Scoring Engine）

每位选手提交 Docker 镜像 → GitHub Actions CI 自动拉取并在 NPU 集群运行 → 采集 8 个维度的 metric → 加权分数 → 实时排行榜。

# cann-competitions/scoring/scoring_engine.py
#
# 自动化评分引擎（每个提交自动运行）
# 消除人为主观评分，所有 metric 客观可测

import json
import time
import subprocess
from dataclasses import dataclass, field

@dataclass
class CompetitionMetric:
    name: str           # metric name
    weight: float       # 0-1.0 weight
    direction: str      # "max" or "min"
    unit: str           # "tokens/s", "ms", "W", "MB"
    raw_value: float = 0.0
    normalized_score: float = 0.0  # 0-100 after normalization

class ScoringEngine:
    """
    竞赛评分引擎

    工作流：
    1. 拉取参赛者 Docker 镜像
    2. 在 CANN NPU 集群运行 benchmark
    3. 采集 8 维 metric
    4. Min-Max Normalization（同一赛道内比较）
    5. 加权 → 最终 0-100 score
    """

    # 赛道 A（LLM 推理加速）：8 维 metric
    TRACK_METRICS = {
        "text-generation": {
            "throughput_tokens_per_second":  (0.40, "max", "tokens/s"),
            "time_to_first_token_ms":        (0.30, "min", "ms"),
            "memory_efficiency_percent":      (0.30, "max", "%"),

            # 辅助 metric（不计入总分，排行榜展示）
            "model_size_mb":       (0.0, "min", "MB"),
            "peak_power_watts":    (0.0, "min", "W"),
            "batch_size":          (0.0, "max", "samples"),
            "quantization_bits":   (0.0, "min", "bits"),  # 量化位数 （越低越好）
        }
    }

    def __init__(self, track_id, docker_registry, npu_cluster_hosts):
        self.track_metrics = self.TRACK_METRICS[track_id]
        self.docker_registry = docker_registry
        self.npu_hosts = npu_cluster_hosts  # ["npu-01:9100", "npu-02:9100", ...]

    def score_submission(self, docker_image, submission_id):
        """
        单个提交的评分
        return: total_score (0-100), breakdown dict
        """
        # Step 1: 运行 Docker + NPU benchmark
        raw_metrics = self._run_benchmark(docker_image)

        # Step 2: Min-Max Normalization（同一赛道内标准化）
        normalized = self._normalize_metrics(raw_metrics, submission_id)

        # Step 3: 加权
        total_score, breakdown = self._weighted_sum(normalized)

        # Step 4: 写入排行榜
        self._update_leaderboard(submission_id, total_score, breakdown)

        return total_score, breakdown

    def _run_benchmark(self, docker_image):
        """
        在 NPU 集群运行 Docker benchmark
        """
        # 随机选择一台空闲 NPU 节点
        npu_host = self._find_idle_npu()
        hostname = npu_host.split(":")[0]

        # Docker run（挂载数据集 + CANN runtime）
        cmd = f"""
docker run --rm \\
    --runtime=nvidia \\
    --device /dev/davinci0 \\
    --device /dev/davinci_manager \\
    --device /dev/hisi_hdc \\
    -v /data/benchmarks:/data \\
    -e ASCEND_DEVICE_ID=0 \\
    {docker_image} \\
    python benchmark.py \\
        --track text-generation \\
        --dataset /data/sharegpt-1k.json \\
        --output metrics.json
"""

        result = subprocess.run(
            f'ssh {hostname} "{cmd}"',
            shell=True, capture_output=True, text=True, timeout=3600
        )

        if result.returncode != 0:
            raise RuntimeError(f"Benchmark failed: {result.stderr[:200]}")

        metrics = json.loads(result.stdout.splitlines()[-1])
        return metrics

    def _normalize_metrics(self, raw_metrics, submission_id):
        """
        Min-Max Normalization（同一赛道内所有提交的正规化）
        transformed_value = (value - min_all) / (max_all - min_all) × 100
        """
        # 拉取此赛道所有提交的 metric 值（计算 min/max）
        all_submissions = self._load_all_submissions_for_track()

        normalized = {}
        for metric_name, (weight, direction, unit) in self.track_metrics.items():
            raw_value = raw_metrics.get(metric_name, 0)

            # 获取此 metric 的最大/最小值
            all_values = [s["metrics"][metric_name]["raw"] for s in all_submissions]
            all_values.append(raw_value)  # 包括当前提交
            min_val, max_val = min(all_values), max(all_values)

            if max_val == min_val:
                score = 100.0  # 所有提交都相同 → 满分
            else:
                if direction == "max":
                    score = (raw_value - min_val) / (max_val - min_val) * 100
                else:  # "min"
                    score = (max_val - raw_value) / (max_val - min_val) * 100

            # 离群值处理（>3σ → max 100）
            score = max(0.0, min(100.0, score))

            normalized[metric_name] = {
                "raw": raw_value,
                "normalized": score,
                "min": min_val,
                "max": max_val,
                "direction": direction,
                "unit": unit
            }

        return normalized

    def _weighted_sum(self, normalized):
        """
        所有 metric 的加权和
        """
        total = 0.0
        breakdown = {}

        for metric_name, (weight, _, _) in self.track_metrics.items():
            if weight == 0:
                continue  # 辅助 metric 不计分

            score = normalized[metric_name]["normalized"]
            contribution = score * weight
            total += contribution
            breakdown[metric_name] = {
                "normalized_score": round(score, 1),
                "weight": weight,
                "contribution": round(contribution, 1)
            }

        return round(total, 1), breakdown

    def _update_leaderboard(self, submission_id, score, breakdown):
        """
        更新排行榜 JSON（GitHub Pages 渲染）
        """
        leaderboard = self._load_leaderboard()

        # 更新或插入提交
        updated = False
        for entry in leaderboard["entries"]:
            if entry["submission_id"] == submission_id:
                entry["score"] = score
                entry["breakdown"] = breakdown
                updated = True

        if not updated:
            leaderboard["entries"].append({
                "submission_id": submission_id,
                "score": score,
                "breakdown": breakdown
            })

        # 按 score 降序排序
        leaderboard["entries"].sort(key=lambda x: x["score"], reverse=True)

        # 更新排名
        for rank, entry in enumerate(leaderboard["entries"], 1):
            entry["rank"] = rank

        # 写入
        with open("leaderboard.json", "w") as f:
            json.dump(leaderboard, f, indent=2)

    def _find_idle_npu(self):
        """找到当前空闲的 NPU 节点（通过 npu-smi 判断 GPU 利用率）"""
        for host in self.npu_hosts:
            util = self._get_npu_utilization(host)
            if util < 10:  # GPU util < 10% → idle
                return host
        raise RuntimeError("No idle NPU node available")

    def _get_npu_utilization(self, host):
        """通过 SSH 获取 NPU 利用率"""
        result = subprocess.run(
            f"ssh {host} 'npu-smi info -m -q 0 | grep Utilization'",
            shell=True, capture_output=True, text=True, timeout=5
        )
        import re
        match = re.search(r'(\d+)%', result.stdout)
        return int(match.group(1)) if match else 100


# === 使用示例 ===
engine = ScoringEngine(
    track_id="text-generation",
    docker_registry="registry.cann.com/competition",
    npu_cluster_hosts=["npu-node-01", "npu-node-02", "npu-node-03"]
)

# 评分一个提交
score, breakdown = engine.score_submission(
    docker_image="registry.cann.com/competition/team-alpha:v3",
    submission_id="team-alpha-submission-7"
)

print(f"Score: {score}/100")
print(f"Breakdown: {json.dumps(breakdown, indent=2)}")
# 输出：
# Score: 88.5/100
# Breakdown: {
#   "throughput_tokens_per_second": {"normalized_score": 92.3, "weight": 0.40, "contribution": 36.9},
#   "time_to_first_token_ms":       {"normalized_score": 85.1, "weight": 0.30, "contribution": 25.5},
#   "memory_efficiency_percent":     {"normalized_score": 87.0, "weight": 0.30, "contribution": 26.1}
# }

评审流程自动化（GitHub Actions CI pipeline）

# cann-competitions/ci/evaluate-submission.yml

name: Competition Submission Evaluation

on:
  issues:
    types: [opened]    # 参赛者通过 GitHub Issue 提交（模板 Issue form）

jobs:
  # Stage 0: Parse Issue（解析提交信息）
  parse-submission:
    runs-on: ubuntu-latest
    outputs:
      docker_image: ${{ steps.parse.outputs.docker_image }}
      track: ${{ steps.parse.outputs.track }}
      team: ${{ steps.parse.outputs.team }}
    steps:
      - name: Parse Issue Body
        id: parse
        uses: actions/github-script@v6
        with:
          script: |
            const body = context.payload.issue.body;
            const dockerImage = body.match(/Docker Image: (registry\..+)/)[1];
            const track = body.match(/Track: (text-generation|image-classification)/)[1];
            const team = body.match(/Team: (.+)/)[1];
            core.setOutput('docker_image', dockerImage);
            core.setOutput('track', track);
            core.setOutput('team', team);

  # Stage 1: Security Audit（安全检查）
  security-audit:
    needs: parse-submission
    runs-on: ubuntu-latest
    steps:
      - name: Scan Docker for Malware
        run: |
          docker pull ${{ needs.parse-submission.outputs.docker_image }}
          docker save ${{ needs.parse-submission.outputs.docker_image }} | trivy image -

  # Stage 2: Run Benchmark（NPU cluster）
  benchmark:
    needs: [parse-submission, security-audit]
    runs-on: self-hosted  # CANN NPU runner
    outputs:
      score: ${{ steps.score.outputs.total }}
    steps:
      - name: Run Scoring Engine
        id: score
        run: |
          python scoring/scoring_engine.py \
            --docker-image ${{ needs.parse-submission.outputs.docker_image }} \
            --track ${{ needs.parse-submission.outputs.track }} \
            --submission-id "${{ needs.parse-submission.outputs.team }}-${{ github.run_id }}"

          SCORE=$(python -c "import json; print(json.load(open('metrics.json'))['total_score'])")
          echo "total=$SCORE" >> $GITHUB_OUTPUT

  # Stage 3: Post Leaderboard Comment（自动回复 Issue + 排行榜更新)
  post-leaderboard:
    needs: benchmark
    runs-on: ubuntu-latest
    steps:
      - name: Comment Score
        uses: actions/github-script@v6
        with:
          script: |
            const score = ${{ needs.benchmark.outputs.score }};
            github.rest.issues.createComment({
              owner: context.repo.owner,
              repo: context.repo.repo,
              issue_number: context.issue.number,
              body: `🏆 Score: **${score}/100**
                    
                    📊 Leaderboard: https://cann-competitions.github.io/leaderboard/
                    
                    Good luck! 🚀`
            });

      - name: Rebuild Leaderboard Pages
        run: |
          git checkout main
          # ... 重新生成排行榜 HTML ...
          git commit -m "Update leaderboard: team ${{ needs.parse-submission.outputs.team }} score ${{ needs.benchmark.outputs.score }}"
          git push

评审委员会确认机制（Stage 4：人工验）

# cann-competitions/scoring/committee_review.py
#
# 自动化评分后，评审委员会（5名专家）进行人工验证
# 每位评委对每个提交打分（1-10），取平均 × 10 → 0-100 committee score
# 最终分数 = auto_score × 0.7 + committee_score × 0.3

class CommitteeReview:
    """
    评审委员会人工确认
    """

    def __init__(self, review_board_members):
        """
        review_board_members: [{"name": "Dr. Li", "expertise": "LLM Inference"}]
        """
        self.members = review_board_members
        self.criteria = [
            "技术深度",
            "创新性",
            "实用价值",
            "代码质量",
            "文档完整性"
        ]

    def vote_submission(self, submission_id, member_name, scores):
        """
        一位评委的评分
        scores: {"criteria_name": score} → per-criteria scores (0-10)
        """
        # 验证所有 criteria 都有分
        for criterion in self.criteria:
            if criterion not in scores or not (0 <= scores[criterion] <= 10):
                raise ValueError(f"Invalid score for {criterion}: {scores.get(criterion)}")

        # 加权平均（技术深度 × 2）
        weighted_scores = {
            "技术深度": scores["技术深度"] * 2,
            "创新性": scores["创新性"] * 1.5,
            "实用价值": scores["实用价值"] * 1.2,
            "代码质量": scores["代码质量"] * 1,
            "文档完整性": scores["文档完整性"] * 1
        }
        total = sum(weighted_scores.values()) / sum([2, 1.5, 1.2, 1, 1])

        return total

    def compute_committee_score(self, submission_id, all_votes):
        """
        汇总所有评委的分数 → 委员会平均分
        all_votes: [{"member": "Dr. Li", "scores": {...}}, ...]
        """
        member_scores = []
        for vote in all_votes:
            member_score = self.vote_submission(
                submission_id, vote["member"], vote["scores"]
            )
            member_scores.append(member_score)

        # 去掉最高分和最低分（消除偏见）
        member_scores.sort()
        if len(member_scores) >= 3:
            member_scores = member_scores[1:-1]  # cut top and bottom

        avg_score = sum(member_scores) / len(member_scores) * 10  # 0-100

        return avg_score

    def compute_final_score(self, auto_score, committee_score):
        """
        最终分数 = auto 70% + committee 30%
        """
        return auto_score * 0.7 + committee_score * 0.3


# === 竞赛结果示例 ===
# Sub: team-alpha
# Auto score: 88.5/100 (throughput 92.3, latency 85.1, memory 87.0)
# Committee: Dr. Li (85), Prof. Wang (90), Dr. Chen (82), Dr. Zhang (88), Dr. Liu (86)
# → cut top(90) and bottom(82) → avg = (85+88+86)/3 = 86.3
# Final: 88.5 × 0.7 + 86.3 × 0.3 = 87.4/100

踩坑一：评审委员会 5 名专家对同一提交评分差异 2×——领域偏见（硬件专家 vs 算法专家）

# ❌ 不同专家对同一 metric 的理解不同
# 硬件专家：关注功耗、显存效率（memefficiency 8.2）
# 算法专家：关注精度、创新性（技术深度、创新性 9.5）
# 两方差异 2× → 委员会分数失真

# ✅ 按 expertise 分组（每个人只打自己擅长领域的 metric）
def compute_weighted_by_expertise(vote, member_expertise):
    """
    按专家领域分配权重（只评价自己懂的）
    """
    expertise_weights = {
        "硬件架构": {"功耗": 2.0, "显存效率": 2.0, "代码质量": 1.0},
        "算法理论": {"技术深度": 2.0, "创新性": 2.0, "实用价值": 1.5},
        "工程应用": {"实用价值": 2.0, "代码质量": 2.0, "文档完整性": 1.5},
    }

    weights = expertise_weights.get(member_expertise, {})
    adjusted = {}
    for criterion, score in vote["scores"].items():
        w = weights.get(criterion, 1.0)
        adjusted[criterion] = score * w

    return sum(adjusted.values()) / sum([weights.get(c, 1.0) for c in adjusted])

# Expertises:
# Dr. Li: 硬件架构 → 功耗/存储权重大
# Prof. Wang: 算法理论 → 技术深度/创新权重大
# Dr. Chen: 工程应用 → 实用价值/代码权重大
# → 现在差异从 2× 降到 1.3×（每个专家打自己擅长的领域）

踩坑二：Docker 下载失败——参赛者上传 15GB 镜像到 Docker Hub → CI 拉取超时 30min

# ❌ 大 Docker 拉取超时（30min 超时 → CI 标记失败）
# 参赛者在 Dockerfile 中包含 Jupyter Notebook + conda env → 15GB

# ✅ 参赛前 Dockerfile 检查（瘦身检查 CI）
# cann-competitions/ci/docker_size_check.yml
name: Docker Size Check

on:
  pull_request:
    paths: ["Dockerfile"]

jobs:
  check-size:
    runs-on: ubuntu-latest
    steps:
      - name: Check Docker Image Size
        run: |
          docker build -t submission .
          IMAGE_SIZE=$(docker image inspect submission --format='{{.Size}}')
          MAX_SIZE=$((5 * 1024 * 1024 * 1024))  # 5GB

          if [ $IMAGE_SIZE -gt $MAX_SIZE ]; then
            echo "❌ Docker image too large: ${IMAGE_SIZE}B > ${MAX_SIZE}B (5GB limit)"
            echo "Tip: use multi-stage build, COPY only required files, remove dev deps"
            exit 1
          else
            echo "✅ Docker size: ${IMAGE_SIZE}B < 5GB limit"
          fi

# 参赛者指南（cann-competitions/guidelines/docker.md）：
# ✅ DO: multi-stage builds, COPY only model + run.sh
# ❌ DON'T: include conda env, training data, dev tools

踩坑三：排行榜更新冲突——两个提交同时 SSH 到 NPU 节点，SSH 锁竞争导致评分互相覆盖

# ❌ 直接写 leaderboard.json（无锁） → 竞争条件
# Submission A 评分中（正在 benchmark NPU）→ 写 leaderboard.json
# Submission B 也在评分中（同一个 NPU 节点） → 写 leaderboard.json
# → B 的写覆盖 A 的写 → A 的分数丢失

# ✅ SQLite 数据库 + WAL mode（多 writer 并发）
import sqlite3

class LeaderboardDB:
    def __init__(self, db_path="leaderboard.db"):
        self.conn = sqlite3.connect(db_path)
        self.conn.execute("PRAGMA journal_mode=WAL")  # Write-Ahead Logging
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS submissions (
                submission_id TEXT PRIMARY KEY,
                score REAL,
                breakdown TEXT,  -- JSON string
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)

    def upsert_score(self, submission_id, score, breakdown):
        self.conn.execute("""
            INSERT OR REPLACE INTO submissions (submission_id, score, breakdown)
            VALUES (?, ?, ?)
        """, (submission_id, score, json.dumps(breakdown)))
        self.conn.commit()

    def get_top_n(self, n=10):
        return self.conn.execute(
            "SELECT submission_id, score FROM submissions ORDER BY score DESC LIMIT ?", (n,)
        ).fetchall()

# WAL mode → 读不阻塞写，写不阻塞读（多 writer 竞争由 SQLite 行锁处理）
# → 不再丢失分数

cann-competitions 管理社区竞赛全流程。4 赛道评分体系（LLM/图像/语音/科学推理）→ 自动评分引擎（Docker+NPU benchmark → Min-Max Normalization → 加权 0-100 → 实时排行榜）→ GitHub Actions CI pipeline（Issue parse→security audit→benchmark→leaderboard comment→Pages 更新）→ 委员会人工评审计=auto 70%+committee 30%。三个踩坑：专家领域评分 2× 偏差→按 expertise 分组、Docker 15GB 超时→5GB 限额 CI 检查、leaderboard 写入冲突→SQLite WAL 行锁。

[[reply_to_current]]