照着 Meta 的 Thin Orchestrator 造一个「薄编排器」：状态机路由、专家子智能体与不可变 JSON 交接

把胖编排器的失败归因为「提示词没写好」，会导向错误的优化方向。Meta 的复盘给出了更准确的描述：当编排器同时承担「理解领域语义」和「管理工作流状态」两件事时，它会交叉污染概念。他们试过另一条路——为每个任务手工精调系统提示词，成功率能到七到八成，但那要求工程师为每个具体任务单独调提示词；去掉这种人工陪跑，成功率掉到一到两成。这说明问题不在提示词工程，而在架构：你要求一个上下文同时承载三套领域知识和六阶段状态机，它在边界情况上必然失控。另一处佐证是，多智能体编排在批量任务上的准确率显著高于单体智能体，原因正是任务隔离避免了上下文互相干扰。

# 胖编排器：一个提示词里塞全部（示意，勿抄）
FAT_PROMPT = """
你是数据平台专家。以下是三个仓库的完整 schema 定义……
以下是 Python / C++ / SQL 三套语法与代码生成规则……
以下是六个流水线阶段的依赖顺序与校验要求……
以下是十类数据源的填充模式与错误恢复流程……
请完成用户的列接入请求。
"""

薄编排器的职责边界要写得像合同一样硬。它只拥有三件事：一是工作流定序，即根据当前状态决定下一个跑谁；二是可靠性检查，包括超时、重试与错误边界；三是状态管理，把上一个子智能体的产出传给下一个。它不做的事同样要写死：不解释 schema、不生成配置、不校验领域逻辑——它甚至不需要理解「数据列」是什么、六个阶段之间是什么关系。这条边界一旦被突破（比如让编排器顺手「看一眼校验结果对不对」），它就开始重新积累领域知识，几轮迭代后又会变回胖子。

class ThinOrchestrator:
    """只做三件事：定序、可靠性检查、状态管理。不做任何领域判断。"""

    def __init__(self, agents, templates, timeouts=30, max_retries=3):
        self.agents = agents          # 名字 -> 专家子智能体
        self.templates = templates    # 工作流模板：阶段顺序的状态机
        self.timeouts = timeouts
        self.max_retries = max_retries

    def run(self, template_name, initial_state):
        template = self.templates[template_name]
        state = dict(initial_state)
        for stage in template.stages:
            result = self.call_with_reliability(stage, state)
            state = template.apply(stage, state, result)   # 状态快照写在这里
            if state.get("error"):
                return self.agents["error_handler"].run(state)
        return state

值得注意的是，这个编排器对每一个请求都走固定的七步协议：读取目标子智能体的说明文件、检查目标是否已存在以避免重复、向用户收集必需字段、自动推断能推断的信息（仓库类型、已有配置等）、做一次可靠性闸门检查（有无告警、是否需要评审人）、把完整上下文交给子智能体、最后处理结果并确认。这七步里只有第六步涉及大模型推理，而且发生在上下文隔离的子智能体内部；其余六步都是确定性代码。

路由是薄编排器的心脏，也是最容易写胖的地方。正确写法是：路由只看工作流状态里的标志位，不检查任何内容。比如下面这段，它判断「schema 是否已被提出」「校验是否通过」，然后决定下一个是谁；它完全不去看 schema 长什么样、校验结果写了什么，而是信任专家子智能体把标志位设对。这样做的好处是路由逻辑与领域语义彻底解耦——你改了某个子系统的字段规则，路由代码一行都不用动。

def route_next_agent(self, workflow_state):
    if workflow_state.stage == "schema_inference":
        if workflow_state.schema_proposed:
            return self.agents["validation"]
        return self.agents["schema_inference"]

    if workflow_state.stage == "config_generation":
        lang = workflow_state.target_language          # 只是取值，不理解含义
        return self.agents["config_gen_" + lang]

    if workflow_state.stage == "deployment":
        if workflow_state.validation_passed:
            return self.agents["deployment"]
        return self.agents["error_handler"]

    raise UnknownStageError(workflow_state.stage)

这里的取舍要提前接受：薄编排器会用延迟换可靠性。胖编排器可能一步同时生成并校验配置，薄编排器要分别调用生成器、校验器，再回来确认结果，多出几次模型往返。Meta 的实践里这个代价是可接受的，因为数据列接入是批处理而非面向用户的接口；如果你的场景是强实时的用户请求，就要重新权衡，可能只在链路最长的少数环节上做拆分。

十四个专家子智能体各管一块：schema 管理知道配置注册表，填充知道十类数据源各自的代码生成方式，校验知道构建系统与质量检查。关键在于每个专家只加载自己那部分知识源（通常是各自几千 token 的领域文件），并且拿到彼此隔离的工具集。于是 schema 推断只看到源数据与已有列定义，Python 配置生成只看到目标 schema 与 Python 语法规则——没有任何一个智能体需要全貌。正是这种「看得少」换来了单任务 token 消耗的大幅下降，也避免了跨子系统的概念污染。

# 子智能体声明：知识源 + 工具集，两者都按域收窄
SPECIALISTS = {
  "schema_inference": {
      "knowledge": ["docs/config_registry.md"],       # 只加载这几份
      "tools":     ["read_source_table", "list_columns"],
      "emits":     {"schema_proposed": "boolean"},
  },
  "config_gen_python": {
      "knowledge": ["docs/python_syntax.md", "docs/target_schema.md"],
      "tools":     ["render_python_config", "lint_python"],
      "emits":     {"config_path": "string"},
  },
  "validation": {
      "knowledge": ["docs/build_system.md", "docs/quality_gates.md"],
      "tools":     ["build_check", "type_check"],
      "emits":     {"validation_passed": "boolean"},
  },
}

每个子智能体遵循统一结构，这点是规模化的前提：声明运行时要加载哪些知识源、可用的工具集、以及它会产出哪些结构化字段。有了这份声明，编排器就能自动做两件事——在调用前按声明装配上下文，在返回后按声明校验产出。新增一个专家不需要改编排器，只需要在注册表里加一项；这也是为什么十四个专家能被二十多条工作流复用来复去。

子智能体之间不传递无界的对话历史，只传递结构化的 JSON 工件：包含输出状态与执行元数据，并且带签名。编排器在唤醒下一个子智能体之前，先拿这份工件去校验 schema 不变量；校验不过就走错误分支。这套「不可变交接」有两个直接收益：一是故障定位变得便宜——出了问题，你看是哪个专家交出了不合 schema 的工件，而不是去猜编排器当时为什么做了那个路由决定；二是重试能做到局部化——某个子智能体编译异常或超时，只重试它自己，不必重新评估整条工作流历史。

import hashlib, json, jsonschema

def handoff(agent_name, payload):
    artifact = {
        "agent": agent_name,
        "payload": payload,
        "prev_hash": state.last_hash,                     # 串成哈希链
    }
    artifact["hash"] = hashlib.sha256(
        json.dumps(artifact, sort_keys=True).encode()).hexdigest()
    jsonschema.validate(payload, SCHEMAS[agent_name])     # 不变量校验
    return artifact                                        # 校验不过会抛异常

可靠性检查也要做成「不理解语义也能执行」的通用件：每次调用带三十秒超时，失败最多重试三次且间隔指数退避，单个智能体出错不会拖垮编排器，每完成一步就持久化状态快照。这四条对所有专家一视同仁，编排器不需要知道「schema 推断」是什么意思，也能给它加超时。这里有个细节值得照抄：错误隔离意味着故障被限制在子智能体边界内，而不是演变成编排器整体崩溃——这正是胖编排器的典型死法。

这套架构里最省钱的洞察是：把「要发生什么」和「每步怎么做」分开。编排器用工作流模板定义阶段顺序——也就是一个约束大模型决策空间的状态机。列接入是「schema → 校验 → 填充 → 测试」，事件接入是「schema → 填充 → 校验 → 测试 → 事件处理」，流水线迁移是「评估 → 规划 → 填充 → 校验 → 监控」。三条模板复用同一批十四个专家，只是顺序不同。所以当业务提出「我们还要支持流水线迁移」时，团队没有写任何新的智能体代码，只写了一条新模板，就拿到了完整的生产能力。

# 模板 = 阶段顺序的状态机；专家池是共享的
TEMPLATES = {
  "column_onboarding": ["schema", "validation", "population", "testing"],
  "event_onboarding":  ["schema", "population", "validation", "testing", "event_processing"],
  "pipeline_migration":["assessment", "planning", "population", "validation", "monitoring"],
}

落地顺序上给一条务实建议：不要一上来就拆十四个专家。先挑一条你已经在跑的、领域知识明显混在一起的工作流，把它拆成三到四个专家加一个薄路由，跑两周看故障定位是不是变快了、token 是不是降了；确认有效再横向扩。判断该不该用这个模式的标准也很清楚：它适合多领域、且可靠性与可维护性比延迟更重要的工作流；反过来，如果任务是单一领域、短链路，或者延迟是硬指标，胖一点的单体反而更合适。