Skills：AI Agent 基础设施化的全链路深度解析——从概念验证到生产落地的架构革命

2026年8月，技术圈发生了一件被大多数媒体忽略、但真正让工程师群体兴奋的事情：GitHub Trending 前 10 名里，有 8 席是 AI Agent 项目。

这背后揭示了一个深刻的技术趋势——AI Agent 正在经历一场从「玩具」到「基础设施」的范式跃迁。当 Skills、Radio、Memory 从概念词汇变成实际的代码库并登上 GitHub Trending，意味着整个行业已经形成了某种共识：AI Agent 的未来不在于单个模型的强大，而在于多个 Agent 之间如何通信、如何调用工具、如何共享记忆。

本文将深入拆解这场架构革命的三个核心层次：Skills 层（标准化技能）、Radio 层（多 Agent 通信）、Memory 层（持久化记忆），以及它们如何共同构成了 AI Agent 时代的「水电煤」。

一、从单 Agent 到多 Agent 协作：为什么我们需要基础设施

1.1 单 Agent 的瓶颈：一个模型解决不了所有问题

2023-2024 年，我们见证了单 Agent 的狂飙突进。Claude Code、Cursor、GitHub Copilot 这样的产品本质上都是一个 Agent：给定一个 prompt，Agent 自主规划并执行任务。

但单 Agent 有一个天然的天花板——全能性悖论（Versatile Trap）。没有一个模型既能成为顶级的代码审计员，又能在法律合同中找出风险条款，同时还能做数据可视化。

更深层的问题在于：单 Agent 的上下文窗口是有限的。 当任务复杂度超过模型的推理深度时，Agent 会出现「遗忘」——早先确定的架构决策在中途被推翻。

1.2 多 Agent 协作：组织架构的数字化映射

解决上述问题的思路很简单：不要让一个 Agent 做所有事，让不同的 Agent 做不同的事。

这与软件工程中的微服务架构如出一辙。单体应用之所以演进出微服务，是因为复杂业务需要不同专业能力的独立服务各自演进。多 Agent 系统同理：

• 规划 Agent 负责理解需求、拆解任务
• 代码 Agent 负责具体的实现
• 测试 Agent 负责质量保障
• 审查 Agent 负责安全审计

但问题来了：当多个 Agent 需要协作时，它们之间如何通信？如何共享上下文？如何保证数据一致性？

这就是 Skills、Radio、Memory 三个基础设施层要解决的问题。

1.3 2026年8月的标志性事件：Trendy 的结构性变化

8月8日的 GitHub Trending 是一个历史性的快照：

8个项目中，Skills 层占 5 席，Radio 层占 2 席，Memory 层占 1 席。这不是某个公司的单点突破，而是整个行业在用脚投票，选择了相同的架构方向。

二、Skills 层：让 AI Agent 学会「标准化技能」

2.1 为什么需要标准化技能？

在 Skills 出现之前，每个 AI Agent 项目都是这样的：

"你是一个Python专家，熟悉Django框架，做过3年Web开发，
会用Git，会写单元测试，了解CI/CD，了解Docker部署......"

这段 prompt 很长、很模糊、很难维护。每次换模型或换 Agent 实现，这段 prompt 都需要重新调优。而且这些「技能描述」无法复用——你在 Claude Code 里写的技能描述，换到 Codex 就得重写一遍。

Skills 的核心理念是：将技能定义从 prompt 字符串中抽离出来，变成可独立管理、可版本控制、可共享的标准化文件。

2.2 Skills 的技术标准

Skills 的标准格式是一个包含 .skills.js 文件和 SKILL.md 文件的目录：

my-skill/
├── .skills.js          # 技能元数据
├── SKILL.md            # 技能定义（Markdown格式）
├── scripts/            # 辅助脚本（可选）
├── templates/          # 代码模板（可选）
└── README.md           # 使用说明（可选）

.skills.js 元数据文件的结构如下：

// .skills.js
module.exports = {
  name: 'python-tdd',
  version: '1.0.0',
  description: 'Test-Driven Development workflow for Python projects',
  triggers: [
    'write tests', 'tdd', 'test driven', 'pytest'
  ],
  tools: ['shell', 'file_system'],
  icon: '🐍',
  author: 'Your Name',
  license: 'MIT',
  dependencies: [],
  matchers: {
    filePatterns: ['**/*.py', '**/test_*.py', '**/*_test.py'],
    keywords: ['test', 'pytest', 'unittest', 'tdd']
  }
};

SKILL.md 技能定义文件则包含了完整的技能指导：

# Python TDD Skill

## 角色定义
你是一个严格遵循 TDD 原则的 Python 开发者。

## 核心工作流
遵循 Red-Green-Refactor 循环：
1. 写一个失败的测试（Red）
2. 写最小代码让测试通过（Green）
3. 重构代码（Refactor）

## 工具使用规范
- 使用 `pytest` 作为测试框架
- 使用 `coverage` 测量测试覆盖率
- 使用 `black` 格式化代码
- 使用 `ruff` 进行代码检查

## 代码示例

### 示例：实现一个加法函数

**步骤1：写失败的测试**
```python
# tests/test_calculator.py
import pytest
from calculator import add

def test_add_two_positive_numbers():
    assert add(2, 3) == 5

def test_add_negative_numbers():
    assert add(-1, -1) == -2

步骤2：实现最小代码

# calculator.py
def add(a, b):
    return a + b

步骤3：运行测试验证

pytest tests/test_calculator.py -v

常见陷阱

陷阱1：跳过 TDD 直接写实现

错误做法：先写实现，再补测试
正确做法：始终先写测试，再写实现

陷阱2：测试覆盖不足

每个公共函数必须至少有一个测试。
每个边界条件必须单独测试。

陷阱3：测试之间的依赖

测试必须是完全独立的。
不要在测试之间共享可变状态。

### 2.3 Skills 的工作原理

Skills 的生命周期分为三个阶段：

#### 阶段1：发现（Discovery）

Agent 启动时，会扫描所有已安装的 Skills，只加载每个 Skill 的名称和描述：

```javascript
// Agent 启动时的发现逻辑
const discoveredSkills = await skillManager.discover();
/*
结果：
[
  { name: 'python-tdd', description: 'Test-Driven Development workflow', version: '1.0.0' },
  { name: 'rust-ownership', description: 'Rust memory safety patterns', version: '2.1.0' },
  { name: 'docker-compose', description: 'Multi-container orchestration', version: '1.0.0' }
]
*/

这个发现阶段决定了 Agent 何时激活某个 Skill——当用户任务与 Skill 的描述关键词匹配时，该 Skill 被标记为候选。

阶段2：激活（Activation）

当 Agent 判断某个 Skill 与当前任务相关时，将该 Skill 的完整 SKILL.md 内容注入到上下文：

// Skill 激活逻辑
async function activateSkill(agent, task) {
  const relevantSkills = await skillMatcher.findRelevant(task);

  for (const skill of relevantSkills) {
    const skillContent = await skillLoader.load(skill.name);
    agent.context.inject({
      role: 'system',
      content: skillContent.fullMarkdown
    });

    if (skill.tools) {
      agent.registerTools(skill.tools);
    }
  }
}

阶段3：执行（Execution）

Agent 按照 Skill 的指令执行任务。Skill 可以包含：

• 代码模板：直接可用的代码框架
• 工具集：预定义的命令和 API 调用
• 流程规范：标准化的操作步骤
• 最佳实践：经过验证的设计模式

2.4 Skills 的生态版图

截至 2026 年 8 月，Skills 层已经形成了清晰的生态分层：

第一层：官方/顶级项目

• mattpocock/skills：Remotion 维护的 Agent Skills 最佳实践库，8月5日更新，包含 60+ 标准技能
• addyosmani/agent-skills：Addy Osmani 出品的开发者工具技能集
• google/skills：Google 官方推出的技能规范

第二层：框架集成

• obra/superpowers：面向 AI Coding Agent 的超级技能框架
• Codex/skillforge：Codex 平台的技能工厂
• superpowers-cli：命令行工具，用于管理 superpowers

第三层：领域专用

• remotion-dev/skills：视频/动画制作领域技能
• openclaw：AI Agent 运行时技能规范

这种分层与 npm 生态非常相似——顶层是通用规范，中间层是框架工具，底层是垂直领域实现。

三、Radio 层：让多个 AI Agent 互相「对话」

3.1 为什么 Agent 之间需要 Radio？

多 Agent 协作最大的挑战不是「做什么」，而是「谁来做什么」以及「结果怎么传递」。

当一个任务被分解为多个子任务后：

主 Agent：分析需求
    ↓
    ├→ 子Agent1：写前端代码
    ├→ 子Agent2：写后端API
    └→ 子Agent3：写数据库Schema
    ↓
主 Agent：整合结果

在这个过程中，我们需要解决：

1. 通信协议：Agent 之间用什么格式传递消息？
2. 状态同步：如何保证所有 Agent 看到一致的共享状态？
3. 错误处理：某个子 Agent 失败后，系统如何反应？
4. 并发控制：多个 Agent 同时访问同一资源时如何协调？

3.2 Radio 的设计哲学

Radio 的命名本身就暗示了它的设计哲学：像对讲机一样，让不同 Agent 在同一频道上通信。

传统的微服务通信依赖 HTTP/REST 或 gRPC，这要求服务之间必须有明确的 API 契约。在 AI Agent 场景中，这种刚性契约过于笨重——Agent 的能力是动态的，prompt 是可以随时修改的。

Radio 采用了一种更灵活的模式：发布-订阅（Pub/Sub）+ 广播（Broadcast）。

// Radio 的核心通信模型
class RadioChannel {
  constructor(channelName) {
    this.channelName = channelName;
    this.subscribers = new Map(); // agentId -> handler
    this.messageQueue = [];
  }

  async broadcast(sender, message) {
    const envelope = {
      from: sender.agentId,
      timestamp: Date.now(),
      payload: message,
      channel: this.channelName
    };

    for (const [agentId, handler] of this.subscribers) {
      if (agentId !== sender.agentId) {
        await handler(envelope);
      }
    }
  }

  subscribe(agent, handler) {
    this.subscribers.set(agent.agentId, handler);
  }

  unsubscribe(agentId) {
    this.subscribers.delete(agentId);
  }
}

3.3 实战：用 Radio 构建多 Agent 协作系统

让我们通过一个实际例子来理解 Radio 的工作方式：构建一个「代码审查 Agent 团队」。

# radio_demo.py - 多 Agent 协作示例

from dataclasses import dataclass
from typing import Callable
from enum import Enum
import asyncio
import json


class MessageType(Enum):
    TASK_ASSIGNED = "task_assigned"
    RESULT_READY = "result_ready"
    ERROR = "error"


@dataclass
class AgentMessage:
    id: str
    sender: str
    message_type: MessageType
    payload: dict
    timestamp: float
    channel: str = "default"


class Radio:
    """多 Agent 通信基础设施"""

    def __init__(self):
        self.channels: dict[str, list[tuple[str, Callable]]] = {}

    def subscribe(self, agent_id: str, channel: str, handler: Callable):
        if channel not in self.channels:
            self.channels[channel] = []
        self.channels[channel].append((agent_id, handler))

    async def broadcast(self, sender_id: str, channel: str, message: AgentMessage):
        if channel not in self.channels:
            return

        tasks = []
        for agent_id, handler in self.channels[channel]:
            if agent_id != sender_id:
                tasks.append(asyncio.create_task(handler(message)))

        if tasks:
            await asyncio.gather(*tasks, return_exceptions=True)


class CodeReviewTeam:
    """代码审查团队 - 演示 Radio 多 Agent 协作"""

    def __init__(self):
        self.radio = Radio()
        self.agents = {}

    def register_agent(self, agent_id: str, role: str, specialty: list[str]):
        self.agents[agent_id] = {
            "role": role,
            "specialty": specialty,
            "status": "idle"
        }
        self.radio.subscribe(agent_id, "team", self._create_handler(agent_id))

    def _create_handler(self, agent_id: str) -> Callable:
        async def handler(message: AgentMessage):
            if message.message_type == MessageType.TASK_ASSIGNED:
                await self._handle_task(agent_id, message)
        return handler

    async def start_review(self, code: str, language: str):
        task_msg = AgentMessage(
            id=f"task_{asyncio.get_event_loop().time()}",
            sender="coordinator",
            message_type=MessageType.TASK_ASSIGNED,
            payload={
                "code": code,
                "language": language,
                "request_id": f"review_{asyncio.get_event_loop().time()}"
            },
            timestamp=asyncio.get_event_loop().time()
        )

        if language in ["python", "javascript", "typescript"]:
            primary_agent = "security_expert"
        elif language in ["rust", "c", "cpp"]:
            primary_agent = "memory_safety_expert"
        else:
            primary_agent = "general_expert"

        self.agents[primary_agent]["status"] = "working"
        await self.radio.broadcast("coordinator", "team", task_msg)
        return {"task_id": task_msg.payload["request_id"], "status": "dispatched"}

    async def _handle_task(self, agent_id: str, message: AgentMessage):
        agent = self.agents[agent_id]
        agent["status"] = "working"

        review_result = await self._do_review(agent_id, message.payload)

        result_msg = AgentMessage(
            id=f"result_{asyncio.get_event_loop().time()}",
            sender=agent_id,
            message_type=MessageType.RESULT_READY,
            payload={
                "request_id": message.payload["request_id"],
                "agent": agent_id,
                "result": review_result
            },
            timestamp=asyncio.get_event_loop().time()
        )

        await self.radio.broadcast(agent_id, "team", result_msg)
        agent["status"] = "idle"

    async def _do_review(self, agent_id: str, payload: dict) -> dict:
        return {
            "agent": agent_id,
            "issues_found": 3,
            "severity": ["high", "medium", "low"],
            "recommendations": [
                "建议添加输入验证",
                "考虑使用更安全的加密库",
                "添加错误处理"
            ]
        }


# 使用示例
async def main():
    team = CodeReviewTeam()

    team.register_agent("security_expert", "安全专家",
                       ["security", "cryptography", "authentication"])
    team.register_agent("performance_expert", "性能专家",
                       ["performance", "scalability", "caching"])
    team.register_agent("readability_expert", "可读性专家",
                       ["code_style", "documentation", "naming"])

    code_sample = """
def authenticate(username, password):
    query = f"SELECT * FROM users WHERE name='{username}'"
    return db.execute(query)
    """

    result = await team.start_review(code_sample, "python")
    print(json.dumps(result, indent=2, ensure_ascii=False))


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

3.4 Prime Agent 的 Radio 架构解析

PrimeIntellect-ai/prime-agent 是目前最热门的 Radio 层实现，它的核心架构如下：

┌─────────────────────────────────────────────────────────┐
│                    Prime Agent                          │
├─────────────────────────────────────────────────────────┤
│  ┌──────────┐  ┌──────────┐  ┌──────────┐            │
│  │ Planning │  │ Executor │  │ Memory   │            │
│  │  Agent   │  │  Agent   │  │  Agent   │            │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘            │
│       │              │              │                   │
│       └──────────────┼──────────────┘                   │
│                      │                                   │
│              ┌───────┴───────┐                          │
│              │  Radio Hub   │  ← 消息路由中枢           │
│              └───────────────┘                          │
│                      │                                   │
│       ┌──────────────┼──────────────┐                   │
│       │              │              │                   │
│  ┌────┴────┐   ┌────┴────┐   ┌────┴────┐              │
│  │ Session │   │ Tool    │   │ Context │              │
│  │ Manager │   │ Registry│   │ Buffer  │              │
│  └─────────┘   └─────────┘   └─────────┘              │
└─────────────────────────────────────────────────────────┘

Prime Agent 的 Radio Hub 解决了三个核心问题：

1. 会话树与会话隔离

// Prime Agent 的会话管理
class SessionTree {
  constructor(rootSessionId) {
    this.root = {
      id: rootSessionId, parent: null, children: [],
      agents: [], sharedState: {}
    };
    this.current = this.root;
  }

  fork(agentId, reason) {
    const branch = {
      id: `fork_${Date.now()}`,
      parent: this.current.id,
      children: [],
      agents: [agentId],
      sharedState: { ...this.current.sharedState },
      forkReason: reason
    };
    this.current.children.push(branch.id);
    this.sessions[branch.id] = branch;
    return branch;
  }

  merge(branchId) {
    const branch = this.sessions[branchId];
    Object.assign(this.current.sharedState, branch.sharedState);
    branch.status = 'merged';
  }
}

2. 共享内存与状态同步

// 跨 Agent 共享状态
class SharedState {
  constructor() {
    this.state = new Map();
    this.listeners = new Map();
  }

  set(key, value, agentId) {
    const oldValue = this.state.get(key);
    this.state.set(key, {
      value,
      lastModified: Date.now(),
      lastModifiedBy: agentId
    });
    this._notify(key, oldValue, value, agentId);
  }

  get(key) {
    return this.state.get(key)?.value;
  }

  watch(key, agentId, callback) {
    if (!this.listeners.has(key)) {
      this.listeners.set(key, []);
    }
    this.listeners.get(key).push({ agentId, callback });
  }

  _notify(key, oldValue, newValue, changedBy) {
    const callbacks = this.listeners.get(key) || [];
    for (const { agentId, callback } of callbacks) {
      if (agentId !== changedBy) {
        callback(newValue, oldValue);
      }
    }
  }
}

3. 工具调用的幂等性保证

// 防止重复执行
class ToolRegistry {
  constructor() {
    this.tools = new Map();
    this.callHistory = new Map();
  }

  async execute(toolName, args, agentId) {
    const callId = this._generateCallId(toolName, args);

    if (this.callHistory.has(callId)) {
      const cached = this.callHistory.get(callId);
      if (Date.now() - cached.timestamp < 30000) {
        return cached.result;
      }
    }

    const tool = this.tools.get(toolName);
    const result = await tool.execute(args);

    this.callHistory.set(callId, {
      result,
      timestamp: Date.now(),
      executedBy: agentId
    });

    return result;
  }

  _generateCallId(toolName, args) {
    return `${toolName}:${JSON.stringify(args)}`;
  }
}

3.5 Cloudflare Computer：Radio 的边缘计算版本

cloudflare/computer 是另一个值得关注的 Radio 层实现，特点是将 Agent 运行在边缘节点，让通信延迟降到最低。

// Cloudflare Computer 的边缘 Agent 通信
// 使用 Durable Objects 做状态同步

export class AgentRouter {
  async fetch(request) {
    const url = new URL(request.url);

    if (url.pathname === "/dispatch") {
      const body = await request.json();
      return this.dispatchToEdge(body);
    }

    if (url.pathname === "/sync") {
      const body = await request.json();
      return this.syncState(body);
    }
  }

  async dispatchToEdge(task) {
    const nearestEdge = await this.getNearestEdge();
    const edges = await this.getRedundantEdges(nearestEdge, 3);

    const results = await Promise.allSettled(
      edges.map(edge => this.sendToEdge(edge, task))
    );

    return this.majorityVote(results);
  }

  async syncState(update) {
    const do = this.env.AGENT_STATE.get(
      this.env.AGENT_STATE.idFromName(update.sessionId)
    );
    return await do.patch(update);
  }
}

四、Memory 层：让 AI Agent 拥有「持久记忆」

4.1 为什么 Agent 需要 Memory？

当前大多数 AI Agent 都是「金鱼式遗忘」——每次对话都是从零开始。这意味着：

• Agent 无法记住之前的架构决策
• 无法从错误中学习
• 无法积累项目特有的领域知识
• 无法维护跨会话的上下文

Memory 层的目标就是解决这个问题——让 AI Agent 拥有持久化的、可检索的记忆能力。

4.2 Memory 的三层架构

成熟的 Agent Memory 系统通常包含三层：

┌─────────────────────────────────────────┐
│           Semantic Memory               │  ← 长期知识：项目文档、技术栈说明
├─────────────────────────────────────────┤
│           Episodic Memory               │  ← 经验记录：之前的任务执行记录、结果
├─────────────────────────────────────────┤
│           Working Memory                │  ← 当前上下文：当前会话的状态
└─────────────────────────────────────────┘

语义记忆（Semantic Memory）：存储结构化的知识

# semantic_memory.py - 语义记忆存储
from dataclasses import dataclass
from typing import Optional
import json
import hashlib


@dataclass
class KnowledgeEntry:
    key: str
    value: str
    category: str
    tags: list[str]
    embedding: list[float]
    created_at: float
    updated_at: float
    version: int


class SemanticMemory:
    """语义记忆：存储结构化知识"""

    def __init__(self, vector_store, storage):
        self.vector_store = vector_store  # 向量数据库
        self.storage = storage            # KV 存储
        self.version_counter = {}

    async def store(self, key: str, value: str,
                    category: str, tags: list[str]) -> str:
        embedding = await self._embed(value)

        entry = KnowledgeEntry(
            key=key, value=value, category=category, tags=tags,
            embedding=embedding,
            created_at=self._now(),
            updated_at=self._now(),
            version=self._next_version(key)
        )

        storage_key = f"semantic:{key}"
        await self.storage.set(storage_key, json.dumps(self._entry_to_dict(entry)))
        await self.vector_store.upsert(
            id=hashlib.md5(key.encode()).hexdigest(),
            vector=embedding,
            metadata={"key": key, "category": category}
        )
        return key

    async def retrieve(self, query: str, top_k: int = 5,
                       category: Optional[str] = None) -> list[KnowledgeEntry]:
        query_embedding = await self._embed(query)
        results = await self.vector_store.search(
            vector=query_embedding,
            top_k=top_k,
            filter={"category": category} if category else None
        )
        entries = []
        for r in results:
            key = r.metadata["key"]
            storage_key = f"semantic:{key}"
            data = await self.storage.get(storage_key)
            if data:
                entries.append(self._dict_to_entry(json.loads(data)))
        return entries

    async def _embed(self, text: str) -> list[float]:
        import hashlib
        h = hashlib.sha256(text.encode()).digest()
        return [b / 255.0 for b in h[:256]]

    def _now(self) -> float:
        import time
        return time.time()

    def _next_version(self, key: str) -> int:
        v = self.version_counter.get(key, 0) + 1
        self.version_counter[key] = v
        return v

情景记忆（Episodic Memory）：存储执行历史

# episodic_memory.py - 情景记忆存储

@dataclass
class Episode:
    id: str
    task_description: str
    actions: list[dict]
    outcome: str  # success/failure/partial
    artifacts: dict
    timestamp: float
    session_id: str
    tags: list[str]


class EpisodicMemory:
    """情景记忆：存储任务执行历史"""

    def __init__(self, storage):
        self.storage = storage
        self.index = {}  # 标签索引

    async def record(self, episode: Episode):
        key = f"episode:{episode.id}"
        await self.storage.set(key, json.dumps(self._episode_to_dict(episode)))

        for tag in episode.tags:
            if tag not in self.index:
                self.index[tag] = []
            self.index[tag].append(episode.id)

        session_key = f"session:{episode.session_id}"
        session_episodes = await self.storage.get(session_key) or "[]"
        episodes = json.loads(session_episodes)
        episodes.append(episode.id)
        await self.storage.set(session_key, json.dumps(episodes))

    async def recall_similar(self, task: str, limit: int = 10) -> list[Episode]:
        task_tags = self._extract_tags(task)
        candidate_ids = set()
        for tag in task_tags:
            if tag in self.index:
                candidate_ids.update(self.index[tag])

        episodes = []
        for ep_id in list(candidate_ids)[:limit]:
            key = f"episode:{ep_id}"
            data = await self.storage.get(key)
            if data:
                episodes.append(self._dict_to_episode(json.loads(data)))
        return episodes

    def _extract_tags(self, text: str) -> list[str]:
        words = text.lower().split()
        return [w for w in words if len(w) > 3]

工作记忆（Working Memory）：维护当前会话状态

# working_memory.py - 工作记忆

class WorkingMemory:
    """工作记忆：当前会话的实时状态"""

    def __init__(self, session_id: str):
        self.session_id = session_id
        self.context: dict = {}
        self.stack: list[dict] = []
        self.buffer: list[str] = []
        self.locks: dict[str, str] = {}

    def push_context(self, context: dict):
        self.stack.append(self.context)
        self.context = {**self.context, **context}

    def pop_context(self) -> dict:
        if self.stack:
            self.context = self.stack.pop()
        return self.context

    def get(self, key: str, default=None):
        return self.context.get(key, default)

    def set(self, key: str, value):
        self.context[key] = value

    def lock(self, resource: str, agent_id: str) -> bool:
        if resource in self.locks:
            return False
        self.locks[resource] = agent_id
        return True

    def unlock(self, resource: str, agent_id: str) -> bool:
        if self.locks.get(resource) == agent_id:
            del self.locks[resource]
            return True
        return False

    def snapshot(self) -> dict:
        return {
            "session_id": self.session_id,
            "context": self.context,
            "stack_depth": len(self.stack),
            "buffer_size": len(self.buffer),
            "locks": self.locks
        }

4.3 superpowers-cli 的 Memory 实践

superpowers-cli 是 Memory 层的另一个重要实现，它的特点是将 Memory 与 Skills 深度整合：

// superpowers-cli 的 Memory 与 Skills 整合

class SuperpowersAgent {
  constructor(config) {
    this.memory = new AgentMemory(config.storage);
    this.skills = new SkillManager(config.skillsDir);
    this.radio = new RadioHub();
  }

  async processTask(task) {
    // 1. 从 Memory 中检索相关历史经验
    const similarEpisodes = await this.memory.recall(task.description);

    // 2. 找到适用的 Skills
    const applicableSkills = await this.skills.match(task);

    // 3. 构建增强的 prompt
    const enhancedPrompt = this.buildPrompt(task, {
      memories: similarEpisodes,
      skills: applicableSkills
    });

    // 4. 执行任务
    const result = await this.execute(enhancedPrompt);

    // 5. 将结果存入 Memory
    await this.memory.record({
      task: task.description,
      actions: result.steps,
      outcome: result.success ? 'success' : 'failure',
      lessons: this.extractLessons(result)
    });

    return result;
  }

  buildPrompt(task, context) {
    let prompt = task.originalPrompt;

    if (context.memories.length > 0) {
      prompt += "\n\n## 相关经验\n";
      for (const m of context.memories) {
        prompt += `- ${m.task}: ${m.outcome}\n`;
      }
    }

    for (const skill of context.skills) {
      prompt += `\n\n---\n${skill.fullContent}\n---\n`;
    }

    return prompt;
  }
}

五、Skills + Radio + Memory：完整的多 Agent 协作系统

5.1 架构整合

当 Skills、Radio、Memory 三层组合在一起时，我们得到的是一个完整的多 Agent 协作平台：

┌────────────────────────────────────────────────────────────────┐
│                     Multi-Agent Platform                        │
├────────────────────────────────────────────────────────────────┤
│   ┌──────────────┐  ┌──────────────┐  ┌──────────────┐       │
│   │   Skills     │  │    Radio     │  │    Memory    │       │
│   │   Layer      │  │    Layer     │  │    Layer     │       │
│   │  技能定义    │  │  消息路由    │  │  长期记忆    │       │
│   │  工具注册    │  │  状态同步    │  │  经验积累    │       │
│   │  流程规范    │  │  并发控制    │  │  上下文检索  │       │
│   └──────┬───────┘  └──────┬───────┘  └──────┬───────┘       │
│          └─────────────────┼─────────────────┘                │
│                            │                                  │
│                   ┌────────┴────────┐                        │
│                   │  Orchestrator   │  ← 任务编排中枢          │
│                   │  - 任务分解     │                        │
│                   │  - Agent 分配   │                        │
│                   │  - 结果聚合     │                        │
│                   └────────┬────────┘                        │
│          ┌────────────────┼────────────────┐                │
│   ┌──────┴──────┐  ┌──────┴──────┐  ┌──────┴──────┐         │
│   │  Code Agent │  │ Test Agent │  │ Review Agent│         │
│   │  (Skills:   │  │  (Skills:  │  │  (Skills:   │         │
│   │   python,   │  │   pytest,  │  │   security, │         │
│   │   fastapi)  │  │   coverage)│  │   best-prac)│         │
│   └─────────────┘  └────────────┘  └────────────┘         │
└────────────────────────────────────────────────────────────────┘

5.2 端到端实战：构建一个 AI 代码审查团队

# complete_code_review_team.py - 完整的多 Agent 协作系统

import asyncio
import json
from dataclasses import dataclass, field
from typing import Optional
from enum import Enum


class AgentRole(Enum):
    PLANNER = "planner"
    CODER = "coder"
    TESTER = "tester"
    REVIEWER = "reviewer"
    DEPLOYER = "deployer"


@dataclass
class Task:
    id: str
    description: str
    language: str
    code: str
    context: dict = field(default_factory=dict)


@dataclass
class Skill:
    name: str
    role: AgentRole
    description: str
    tools: list[str]
    instructions: str
    version: str = "1.0.0"


# ==================== Skills Layer ====================

class SkillsRegistry:
    def __init__(self):
        self.skills: dict[str, Skill] = {}
        self._register_default_skills()

    def _register_default_skills(self):
        self.skills["python-architecture"] = Skill(
            name="python-architecture",
            role=AgentRole.CODER,
            description="Python 项目架构设计规范",
            tools=["file_system", "git", "docker"],
            instructions="""
## Python 架构设计规范

### 目录结构

project/
├── src/
│ ├── init.py
│ ├── main.py
│ ├── config/settings.py
│ ├── models/init.py
│ ├── services/init.py
│ └── api/routes.py
├── tests/
│ ├── init.py
│ ├── conftest.py
│ ├── unit/
│ └── integration/
├── pyproject.toml
└── README.md

### 依赖注入规范
```python
# src/services/dependency_injection.py
from typing import TypeVar, Callable

T = TypeVar('T')

class Container:
    def __init__(self):
        self._services = {}
        self._singletons = {}

    def register(self, interface: type[T], factory: Callable[..., T],
                 singleton: bool = False):
        self._services[interface] = {
            'factory': factory,
            'singleton': singleton
        }

    def resolve(self, interface: type[T]) -> T:
        config = self._services.get(interface)
        if not config:
            raise KeyError(f"Service {interface} not registered")

        if config['singleton']:
            if interface not in self._singletons:
                self._singletons[interface] = config['factory']()
            return self._singletons[interface]

        return config['factory']()

"""
)

    self.skills["security-review"] = Skill(
        name="security-review",
        role=AgentRole.REVIEWER,
        description="安全审查技能库",
        tools=["code_scanner", "dependency_check"],
        instructions="""

安全审查规范

OWASP Top 10 检查清单

A01:2021 - Broken Access Control

• 所有 API 端点都有权限检查
• 用户只能访问自己的资源
• CORS 配置正确
• 文件路径没有路径穿越漏洞

A02:2021 - Cryptographic Failures

• 敏感数据不存储在代码中
• 使用强加密算法（AES-256, RSA-2048+）
• 密钥轮换机制已实现

A03:2021 - Injection

• 参数化查询防 SQL 注入
• 输入验证防 XSS
• 命令执行最小权限原则

代码示例

安全的 SQL 查询

# ✅ 正确：参数化查询
cursor.execute(
    "SELECT * FROM users WHERE id = %s",
    (user_id,)
)

# ❌ 错误：字符串拼接
cursor.execute(
    f"SELECT * FROM users WHERE id = {user_id}"
)

安全的文件操作

import os

def safe_read_file(user_provided_path: str, base_dir: str):
    real_path = os.path.realpath(user_provided_path)
    real_base = os.path.realpath(base_dir)

    if not real_path.startswith(real_base):
        raise PermissionError("Path traversal detected")

    with open(real_path) as f:
        return f.read()

"""
)

    self.skills["pytest-quality"] = Skill(
        name="pytest-quality",
        role=AgentRole.TESTER,
        description="Pytest 测试质量规范",
        tools=["pytest", "coverage", "tox"],
        instructions="""

Pytest 测试规范

测试文件结构

tests/
├── conftest.py           # 共享 fixtures
├── unit/
│   ├── __init__.py
│   ├── test_models.py
│   └── test_services.py
└── integration/
    ├── __init__.py
    ├── test_api.py
    └── test_database.py

Fixture 规范

# conftest.py
import pytest
from app import create_app
from app.database import init_db, drop_db

@pytest.fixture(scope="session")
def app():
    app = create_app(testing=True)
    with app.app_context():
        init_db()
    yield app
    drop_db()

@pytest.fixture
def client(app):
    return app.test_client()

@pytest.fixture
def auth_headers():
    return {
        "Authorization": "Bearer test_token_12345"
    }

参数化测试

@pytest.mark.parametrize("input,expected", [
    ("2 + 2", 4),
    ("3 * 3", 9),
    ("10 - 5", 5),
    ("8 / 2", 4),
])
def test_calculator_basic(input, expected):
    result = eval(input)
    assert result == expected

异步测试

import pytest
import pytest_asyncio

@pytest_asyncio.fixture
async def async_client():
    app = create_app()
    async with app.test_client() as client:
        yield client

@pytest.mark.asyncio
async def test_async_api(async_client):
    response = await async_client.get("/api/users/1")
    assert response.status_code == 200
    data = await response.get_json()
    assert data["id"] == 1

"""
)

def get_skills_for_role(self, role: AgentRole) -> list[Skill]:
    return [s for s in self.skills.values() if s.role == role]

def match_skills(self, task: Task) -> list[Skill]:
    matched = []
    for skill in self.skills.values():
        if any(keyword in task.description.lower()
               for keyword in skill.name.split("-")):
            matched.append(skill)
    return matched

==================== Memory Layer ====================

class AgentMemory:
def init(self):
self.semantic = {}
self.episodes = {}
self.working = {}

async def store_episode(self, task_id: str, actions: list[dict], outcome: str):
    self.episodes[task_id] = {
        "actions": actions,
        "outcome": outcome,
        "timestamp": asyncio.get_event_loop().time()
    }

async def recall_similar(self, query: str) -> list[dict]:
    results = []
    for ep_id, ep in self.episodes.items():
        if any(query.lower() in str(a) for a in ep["actions"]):
            results.append(ep)
    return results[:5]

def set_working(self, key: str, value):
    self.working[key] = value

def get_working(self, key: str, default=None):
    return self.working.get(key, default)

==================== Radio Layer ====================

class RadioHub:
def init(self):
self.channels: dict[str, list[asyncio.Queue]] = {}
self.subscribers: dict[str, str] = {}

async def subscribe(self, agent_id: str, channel: str) -> asyncio.Queue:
    if channel not in self.channels:
        self.channels[channel] = []
    queue = asyncio.Queue()
    self.channels[channel].append(queue)
    self.subscribers[id(queue)] = agent_id
    return queue

async def broadcast(self, sender_id: str, channel: str, message: dict):
    if channel not in self.channels:
        return
    for queue in self.channels[channel]:
        if self.subscribers.get(id(queue)) != sender_id:
            await queue.put({
                "from": sender_id,
                "message": message,
                "timestamp": asyncio.get_event_loop().time()
            })

==================== Agent Implementation ====================

class BaseAgent:
def init(self, agent_id: str, role: AgentRole,
skills: SkillsRegistry, memory: AgentMemory,
radio: RadioHub):
self.agent_id = agent_id
self.role = role
self.skills = skills
self.memory = memory
self.radio = radio
self.message_queue: Optional[asyncio.Queue] = None
self.active_skills: list[Skill] = []

async def initialize(self):
    self.message_queue = await self.radio.subscribe(self.agent_id, "team")
    self.active_skills = self.skills.get_skills_for_role(self.role)

def apply_skills(self, context: str) -> str:
    if not self.active_skills:
        return context
    enhanced = context
    for skill in self.active_skills:
        enhanced += f"\n\n### {skill.name} 规范\n{skill.instructions}"
    return enhanced

class CodeReviewAgent(BaseAgent):
async def execute_task(self, task: Task) -> dict:
self.apply_skills(task.description)
issues = []

    if "sql" in task.code.lower() and "'" in task.code:
        if "+" in task.code or "f\"" in task.code:
            issues.append({
                "type": "sql_injection",
                "severity": "high",
                "message": "检测到潜在的 SQL 注入风险，使用参数化查询"
            })

    if any(secret in task.code.lower()
           for secret in ["password", "api_key", "secret", "token"]):
        if any(s in task.code for s in ['=', ': ']):
            issues.append({
                "type": "hardcoded_secret",
                "severity": "critical",
                "message": "检测到硬编码的敏感信息"
            })

    if "except" not in task.code and "try" in task.code:
        issues.append({
            "type": "bare_except",
            "severity": "medium",
            "message": "使用裸 except 可能捕获意外异常"
        })

    return {
        "agent_id": self.agent_id,
        "issues_found": len(issues),
        "issues": issues,
        "summary": f"审查完成，发现 {len(issues)} 个问题"
    }

class TestingAgent(BaseAgent):
async def execute_task(self, task: Task) -> dict:
self.apply_skills(task.description)
test_code = f"""
import pytest

测试 {task.language} 代码

class TestCodeQuality:
def test_basic_functionality(self):
# TODO: 基于 {task.description} 生成测试
pass

def test_edge_cases(self):
    # 边界条件测试
    pass

"""
return {
"agent_id": self.agent_id,
"test_code": test_code,
"coverage_expected": "80%+",
"summary": "测试代码生成完成"
}

==================== Orchestrator ====================

class ReviewOrchestrator:
def init(self):
self.skills = SkillsRegistry()
self.memory = AgentMemory()
self.radio = RadioHub()
self.review_agent = CodeReviewAgent(
"reviewer_1", self.skills, self.memory, self.radio)
self.testing_agent = TestingAgent(
"tester_1", self.skills, self.memory, self.radio)

async def initialize(self):
    await self.review_agent.initialize()
    await self.testing_agent.initialize()

async def run_full_review(self, task: Task) -> dict:
    results = {
        "task_id": task.id,
        "reviews": [],
        "tests": None,
        "overall_status": "pending"
    }

    review_task = asyncio.create_task(self.review_agent.execute_task(task))
    test_task = asyncio.create_task(self.testing_agent.execute_task(task))

    review_result, test_result = await asyncio.gather(
        review_task, test_task, return_exceptions=True)

    if not isinstance(review_result, Exception):
        results["reviews"].append(review_result)
    if not isinstance(test_result, Exception):
        results["tests"] = test_result

    if results["reviews"]:
        total_issues = sum(r.get("issues_found", 0) for r in results["reviews"])
        critical_issues = sum(
            1 for r in results["reviews"]
            for i in r.get("issues", [])
            if i.get("severity") == "critical"
        )

        if critical_issues > 0:
            results["overall_status"] = "rejected"
        elif total_issues > 5:
            results["overall_status"] = "needs_fix"
        else:
            results["overall_status"] = "approved"
    else:
        results["overall_status"] = "error"

    await self.memory.store_episode(
        task.id, actions=[results], outcome=results["overall_status"])

    return results

async def main():
orchestrator = ReviewOrchestrator()
await orchestrator.initialize()

task = Task(
    id="task_001",
    description="审查用户认证模块的安全性",
    language="python",
    code="""

def authenticate_user(username, password):
query = f"SELECT * FROM users WHERE name='{username}'"
result = db.execute(query)
if result and result[0].password == password:
return True
return False
"""
)

results = await orchestrator.run_full_review(task)
print("\n" + "="*60)
print("代码审查报告")
print("="*60)
print(json.dumps(results, indent=2, ensure_ascii=False))

if name == "main":
asyncio.run(main())

---

## 六、性能优化与踩坑清单（20条）

### 6.1 Skills 性能优化

**Skills 元数据缓存**：当 Skills 数量超过 100 个时，启动扫描会很慢。建议使用 Redis/Memory 缓存元数据，TTL 设置为 1 小时。

**Skills 冲突检测**：当多个 Skills 同时激活时，它们的指令可能相互冲突。使用冲突图（conflict_graph）检测并按优先级排序：

```javascript
// Skills 优先级与冲突检测
class SkillActivator {
  _build_conflict_graph() {
    return {
      "python-tdd": ["python-chaos"],
      "security-strict": ["security-lax"],
      "fast-prototype": ["production-ready"]
    };
  }

  async activate(task, max_skills = 3) {
    const candidates = await this.registry.match(task);
    const activated = [];
    for (const skill of candidates) {
      if (activated.length >= max_skills) break;
      const has_conflict = activated.some(
        active => this.conflict_graph[active.name]?.includes(skill.name)
      );
      if (!has_conflict) activated.push(skill);
    }
    this.active_skills = activated;
    return activated;
  }
}

6.2 Radio 性能优化

消息批处理：当多个 Agent 高频广播时，消息队列可能堆积。使用 100ms 批处理窗口合并消息，减少 RPC 开销。

Agent 失联检测：Agent 崩溃时，任务会卡死。使用心跳机制（5秒间隔、30秒超时）检测失联 Agent 并重新分配任务：

class AgentHealthMonitor:
    def __init__(self, radio: Radio):
        self.radio = radio
        self.agent_heartbeats: dict[str, float] = {}
        self.heartbeat_interval = 5.0
        self.timeout = 30.0

    async def start_heartbeat(self, agent_id: str):
        while True:
            self.agent_heartbeats[agent_id] = asyncio.get_event_loop().time()
            await self.radio.broadcast(agent_id, "heartbeat", {
                "type": "heartbeat", "agent_id": agent_id
            })
            await asyncio.sleep(self.heartbeat_interval)

    async def check_health(self) -> dict[str, bool]:
        now = asyncio.get_event_loop().time()
        health = {}
        for agent_id, last in self.agent_heartbeats.items():
            is_alive = (now - last) < self.timeout
            health[agent_id] = is_alive
            if not is_alive:
                await self._handle_agent_down(agent_id)
        return health

6.3 Memory 性能优化

向量检索延迟：Memory 存储大量条目后，向量检索变慢。使用分类索引和标签索引预过滤候选集，减少向量搜索范围。

记忆过期：Agent 长期运行后，记忆会积累大量陈旧数据。使用 TTL + 版本号机制自动清理过期记忆。

6.4 完整踩坑清单（20条）

七、总结与展望

7.1 三层架构的价值

Skills、Radio、Memory 三层基础设施的出现，标志着 AI Agent 开发进入了一个新的阶段：

Skills 层解决的问题是「Agent 知道怎么做」——通过标准化的技能定义，将领域知识从 prompt 中抽离出来，变成可复用、可版本控制、可共享的资产。

Radio 层解决的问题是「多个 Agent 能协作」——通过发布-订阅的消息模式，让不同 Agent 在同一频道上通信，而不需要知道彼此的实现细节。

Memory 层解决的问题是「Agent 能记住过去」——通过三层记忆架构，让 Agent 能够积累经验、学习教训，而不必每次都从零开始。

7.2 生态发展趋势

从 2026 年 8 月的 GitHub Trending 数据来看：

1. Skills 标准化运动：SkillHub 平台已经收录了 142.5K+ Skills，覆盖 29 个分类，类似于当年 npm 之于 JavaScript 的生态建设。

2. Radio 协议收敛：虽然目前有多个 Radio 实现（Prime Agent、Cloudflare Computer 等），但它们的核心理念正在收敛——发布-订阅、状态同步、并发控制。

3. 垂直领域分化：通用 Skills 框架之上，正在生长出大量垂直领域的技能集——Web 开发、数据科学、安全审计等。

7.3 结语

2026 年的 AI Agent 基础设施化，就像 2010 年的云计算、2015 年的容器化、2020 年的微服务——它是一个必然的技术演进方向，只是来得比大多数人预期的更快。

当 Skills 变成 npm，当 Radio 变成 HTTP，当 Memory 变成数据库——AI Agent 的开发就会像今天的 Web 开发一样：不再需要从零构建每一个功能，而是站在基础设施的肩膀上，快速构建真正有价值的产品。

这场革命已经开始了。你，准备好了吗？

参考项目：

• mattpocock/skills - https://github.com/remotion-dev/skills
• PrimeIntellect-ai/prime-agent - https://github.com/PrimeIntellect-ai/prime-agent
• cloudflare/computer - https://github.com/cloudflare/computer
• obra/superpowers - https://github.com/obra/superpowers
• superpowers-cli - https://github.com/obra/superpowers-cli
• SkillHub - https://www.skillhub.club