从超级个体到超级团队

摘要

过去两年，大模型让"超级个体"成为现实：一个人借助 AI 可以在一天内完成过去一周的工作。但组织很快发现一个尴尬的事实——个体效率提升了 10 倍，团队效率却只提升了 1.2 倍。原因在于：个人与 AI 的交互是私有的、临时的、不可复用的，这些能力无法沉淀为组织资产。

本文从工程视角出发，讨论"AI 操作系统"这一概念：如何把个人与 AI 协作产生的提示词、上下文、工具链、决策经验，抽象为可版本化、可编排、可审计的组织级资产，从而完成从超级个体到超级团队的生产力跃迁。文章包含完整的代码示例，涵盖资产抽象层、上下文仓库、Agent 编排、权限治理与度量体系。

---

引言：为什么超级个体没有带来超级团队

先看一个真实场景。

某研发团队有 8 名工程师，每个人都熟练使用 AI 编码助手。个人视角下：

• 写单元测试：从 40 分钟缩短到 5 分钟
• 排查线上日志：从 30 分钟缩短到 3 分钟
• 写技术方案初稿：从 2 小时缩短到 20 分钟

看起来很美好。但团队层面的数据却很平淡：

指标	引入 AI 前	引入 AI 后	提升
人均编码产出	100	180	1.8x
团队交付周期	10 天	8.5 天	1.18x
需求返工率	25%	23%	1.09x
新人上手时间	30 天	28 天	1.07x

个体 1.8x，团队 1.18x，这就是著名的"AI 生产力漏斗"。漏斗的每一层都在漏掉价值：

个人能力 → 团队能力 → 组织资产
  100%  →   40%   →    10%

漏掉的部分去哪了？答案是：它们停留在个人的聊天窗口里。

• 张三调教了 3 周的 Prompt，只存在于他的浏览器历史里
• 李四整理的上下文规则，新人根本不知道
• 王五踩过的坑，下一个人还会再踩一遍

要跨越这个漏斗，我们需要一个操作系统层，把个人 AI 能力结构化、资产化、可编排化。这就是本文要讨论的核心。

---

一、重新定义问题：AI 能力的四种资产形态

在动手写代码之前，先明确我们要沉淀什么。个人与 AI 协作时，产生的价值可以拆成四类：

1.1 提示资产（Prompt Asset）

可复用的指令模板，包括角色设定、任务描述、输出格式约束。

1.2 上下文资产（Context Asset）

让 AI 理解业务的"背景知识"：代码规范、领域术语、架构约束、历史决策。

1.3 工具资产（Tool Asset）

AI 可以调用的函数、API、脚本，把"能说"变成"能做"。

1.4 经验资产（Experience Asset）

决策模式与踩坑记录：什么场景用什么方案，为什么不用另一种方案。

这四类资产对应四个技术层次，我们逐一实现。

---

二、资产抽象层：用代码定义 Prompt 与上下文

第一步是把散落在聊天窗口里的东西，变成可版本控制的结构化文件。

2.1 定义资产的数据模型

# assets/models.py
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from datetime import datetime
import hashlib
import json


@dataclass
class PromptAsset:
    """提示资产：可复用的指令模板"""
    asset_id: str
    name: str
    template: str                       # 含 {variable} 占位符
    variables: List[str] = field(default_factory=list)
    tags: List[str] = field(default_factory=list)
    owner: str = "unknown"
    version: str = "1.0.0"
    created_at: str = field(default_factory=lambda: datetime.now().isoformat())

    def render(self, **kwargs) -> str:
        """渲染模板，缺失变量会明确报错，避免 AI 收到残缺指令"""
        missing = [v for v in self.variables if v not in kwargs]
        if missing:
            raise ValueError(f"缺少变量: {missing}")
        return self.template.format(**kwargs)

    def fingerprint(self) -> str:
        """内容指纹，用于去重与变更检测"""
        raw = json.dumps({
            "template": self.template,
            "variables": sorted(self.variables),
        }, sort_keys=True)
        return hashlib.sha256(raw.encode()).hexdigest()[:12]


@dataclass
class ContextAsset:
    """上下文资产：业务背景知识"""
    asset_id: str
    name: str
    content: str
    scope: str                          # 作用域：global / team / project
    priority: int = 0                   # 优先级，数值越大越靠前
    tags: List[str] = field(default_factory=list)

2.2 用 YAML 存储资产，纳入 Git 管理

资产不应该藏在数据库里，而应该像代码一样被 review、被 diff、被回滚。

# assets/prompts/code_review.yaml
asset_id: prompt.code_review
name: 代码评审助手
version: 1.2.0
owner: platform-team
tags: [review, quality]
variables: [language, diff, guidelines]
template: |
  你是一名资深 {language} 工程师，请对以下代码变更进行评审。

  ## 评审准则
  {guidelines}

  ## 待评审变更

{diff}

## 输出要求
1. 按严重程度分级：Blocker / Major / Minor
2. 每条问题给出具体行号与修复建议
3. 若变更整体无问题，明确回复 "LGTM"

# assets/contexts/team_conventions.yaml
asset_id: context.team_conventions
name: 团队工程规范
scope: team
priority: 100
tags: [convention, backend]
content: |
  - 所有对外接口必须做参数校验，使用 pydantic
  - 数据库访问必须通过 repository 层，禁止在 service 直接写 SQL
  - 日志使用结构化 JSON，禁止 print
  - 异常必须携带 trace_id，便于链路追踪

2.3 资产仓库：加载、检索与组合

# assets/registry.py
import os
import yaml
from typing import List, Dict
from .models import PromptAsset, ContextAsset


class AssetRegistry:
    """资产注册中心：统一加载、检索、组合资产"""

    def __init__(self, root_dir: str):
        self.root_dir = root_dir
        self.prompts: Dict[str, PromptAsset] = {}
        self.contexts: Dict[str, ContextAsset] = {}
        self._load_all()

    def _load_all(self):
        for sub, loader in (("prompts", self._load_prompt),
                            ("contexts", self._load_context)):
            path = os.path.join(self.root_dir, sub)
            if not os.path.isdir(path):
                continue
            for fname in os.listdir(path):
                if fname.endswith((".yaml", ".yml")):
                    loader(os.path.join(path, fname))

    def _load_prompt(self, path: str):
        with open(path, encoding="utf-8") as f:
            data = yaml.safe_load(f)
        asset = PromptAsset(**data)
        self.prompts[asset.asset_id] = asset

    def _load_context(self, path: str):
        with open(path, encoding="utf-8") as f:
            data = yaml.safe_load(f)
        asset = ContextAsset(**data)
        self.contexts[asset.asset_id] = asset

    def search(self, keyword: str, kind: str = "prompt") -> List:
        """按关键词检索资产，供 AI 或人工发现可复用能力"""
        pool = self.prompts if kind == "prompt" else self.contexts
        results = []
        for asset in pool.values():
            blob = f"{asset.name} {getattr(asset, 'tags', [])}"
            if keyword.lower() in blob.lower():
                results.append(asset)
        return results

    def compose_context(self, tags: List[str], max_chars: int = 4000) -> str:
        """按标签组合上下文，按优先级排序并做长度裁剪"""
        selected = [c for c in self.contexts.values()
                    if set(c.tags) & set(tags)]
        selected.sort(key=lambda c: c.priority, reverse=True)

        buf, total = [], 0
        for ctx in selected:
            block = f"### {ctx.name}\n{ctx.content}\n"
            if total + len(block) > max_chars:
                break
            buf.append(block)
            total += len(block)
        return "\n".join(buf)

关键设计点：

1. 资产即代码：YAML + Git，天然支持 review、回滚、审计
2. 显式变量：render 时校验变量，避免残缺指令导致 AI 输出漂移
3. 可组合上下文：按标签动态拼装，而不是把整个知识库塞给模型

---

三、上下文仓库：从"喂资料"到"精准检索"

把上下文全量塞进 Prompt 是新手做法，成本高且会稀释注意力。组织级方案是检索增强。

3.1 一个轻量的向量检索实现

# context/retriever.py
import numpy as np
from typing import List, Tuple


class SimpleRetriever:
    """
    轻量上下文检索器。
    生产环境可替换为向量数据库（如 pgvector / milvus），
    这里用 numpy 演示核心逻辑，便于理解。
    """

    def __init__(self, embed_fn):
        self.embed_fn = embed_fn          # 传入 embedding 函数
        self.docs: List[str] = []
        self.meta: List[dict] = []
        self.matrix: np.ndarray = None

    def add(self, texts: List[str], metas: List[dict]):
        vecs = np.array([self.embed_fn(t) for t in texts], dtype=np.float32)
        # L2 归一化，之后内积即余弦相似度
        vecs /= (np.linalg.norm(vecs, axis=1, keepdims=True) + 1e-9)
        self.docs.extend(texts)
        self.meta.extend(metas)
        self.matrix = vecs if self.matrix is None else np.vstack([self.matrix, vecs])

    def query(self, text: str, top_k: int = 3,
              threshold: float = 0.3) -> List[Tuple[str, dict, float]]:
        if self.matrix is None:
            return []
        q = np.array(self.embed_fn(text), dtype=np.float32)
        q /= (np.linalg.norm(q) + 1e-9)
        scores = self.matrix @ q
        idx = np.argsort(-scores)[:top_k]
        return [(self.docs[i], self.meta[i], float(scores[i]))
                for i in idx if scores[i] >= threshold]

3.2 上下文注入的完整流程

# context/pipeline.py
from assets.registry import AssetRegistry
from context.retriever import SimpleRetriever


class ContextPipeline:
    """把资产 + 检索结果组装成最终 Prompt"""

    def __init__(self, registry: AssetRegistry, retriever: SimpleRetriever):
        self.registry = registry
        self.retriever = retriever

    def build(self, prompt_id: str, user_query: str,
              tags: List[str], **variables) -> str:
        prompt = self.registry.prompts[prompt_id]

        # 1. 静态上下文（团队规范等）
        static_ctx = self.registry.compose_context(tags)

        # 2. 动态检索（历史决策、相似案例）
        hits = self.retriever.query(user_query, top_k=3)
        dynamic_ctx = "\n".join(
            f"- [{m.get('source', 'unknown')}] {d}" for d, m, _ in hits
        )

        # 3. 合并注入
        variables.setdefault("guidelines", static_ctx)
        variables["context"] = dynamic_ctx
        return prompt.render(**variables)

收益：同一个 Prompt 资产，在不同项目、不同问题下自动注入不同上下文，个人经验被"检索"出来复用。

---

四、Agent 编排：把个人工作流变成团队工作流

超级个体的秘密往往是一套隐性的工作流：先查日志、再定位代码、再改、再测、再写复盘。把它显式化，就变成了团队可用的 Agent。

4.1 定义可编排的 Agent 与工具

# agent/core.py
from typing import Callable, Dict, List, Any
import json


class Tool:
    """工具资产：AI 可调用的原子能力"""
    def __init__(self, name: str, description: str,
                 func: Callable, schema: Dict):
        self.name = name
        self.description = description
        self.func = func
        self.schema = schema          # JSON Schema，供模型理解参数

    def invoke(self, **kwargs) -> Any:
        return self.func(**kwargs)


class Agent:
    """可复用 Agent：Prompt + 工具集 + 执行循环"""

    def __init__(self, name: str, llm_client, system_prompt: str,
                 tools: List[Tool], max_steps: int = 8):
        self.name = name
        self.llm = llm_client
        self.system_prompt = system_prompt
        self.tools = {t.name: t for t in tools}
        self.max_steps = max_steps

    def _tool_specs(self) -> List[Dict]:
        return [{"name": t.name, "description": t.description,
                 "parameters": t.schema} for t in self.tools.values()]

    def run(self, user_input: str, trace: List[Dict] = None) -> str:
        """执行循环：思考 → 调用工具 → 观察 → 再思考"""
        trace = trace if trace is not None else []
        messages = [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", "content": user_input},
        ]

        for step in range(self.max_steps):
            resp = self.llm.chat(messages, tools=self._tool_specs())
            trace.append({"step": step, "type": "assistant",
                          "content": resp.get("content", "")})

            # 没有工具调用，说明任务完成
            if not resp.get("tool_calls"):
                return resp["content"]

            for call in resp["tool_calls"]:
                tool = self.tools.get(call["name"])
                if tool is None:
                    result = f"未知工具: {call['name']}"
                else:
                    try:
                        result = tool.invoke(**call["arguments"])
                    except Exception as e:
                        result = f"工具执行失败: {e}"

                trace.append({"step": step, "type": "tool",
                              "name": call["name"], "result": str(result)[:500]})
                messages.append({"role": "assistant",
                                 "content": None, "tool_calls": [call]})
                messages.append({"role": "tool", "name": call["name"],
                                 "content": json.dumps(result, ensure_ascii=False,
                                                       default=str)})

        return "达到最大步数，任务未完成"

4.2 注册团队工具集

# agent/tools_registry.py
import subprocess
from agent.core import Tool


def search_code(keyword: str, path: str = ".") -> str:
    """在代码库中搜索关键词"""
    out = subprocess.run(
        ["grep", "-rn", "--include=*.py", keyword, path],
        capture_output=True, text=True, timeout=10
    )
    return out.stdout[:2000] or "未找到匹配"


def query_logs(service: str, minutes: int = 30, level: str = "ERROR") -> str:
    """查询服务日志（示例：对接真实日志系统）"""
    # 实际项目替换为 ES / Loki 查询
    return f"[{service}] 最近 {minutes} 分钟 {level} 日志：\n<示例日志数据>"


def run_tests(target: str) -> str:
    """运行指定测试"""
    out = subprocess.run(
        ["pytest", target, "-q", "--no-header"],
        capture_output=True, text=True, timeout=120
    )
    return (out.stdout + out.stderr)[-2000:]


TEAM_TOOLS = [
    Tool("search_code", "在代码库中按关键词搜索代码",
         search_code,
         {"type": "object",
          "properties": {"keyword": {"type": "string"},
                         "path": {"type": "string"}},
          "required": ["keyword"]}),
    Tool("query_logs", "查询线上服务日志",
         query_logs,
         {"type": "object",
          "properties": {"service": {"type": "string"},
                         "minutes": {"type": "integer"},
                         "level": {"type": "string"}},
          "required": ["service"]}),
    Tool("run_tests", "运行指定路径的测试",
         run_tests,
         {"type": "object",
          "properties": {"target": {"type": "string"}},
          "required": ["target"]}),
]

4.3 组装一个"线上问题排查" Agent

# agent/incident_agent.py
from agent.core import Agent
from agent.tools_registry import TEAM_TOOLS

INCIDENT_PROMPT = """你是一名 SRE，负责排查线上故障。

工作流程：
1. 先用