Subterranean Agent — 将工作流 SOP 编译进 LLM 权重

Skill-Agentic-Workflow-Compilation · 16-智能体工程

causalexperimentoptimizationmulti_agentvisual_generation广告与投放客服与VOCMAS与智能体工程风控与合规视觉内容生成WF-B 广告优化WF-C 客服分诊WF-G Listing内容优化

年化 ROI$150/天 → 约 $0.5–1.2/天（按 128–462× 节省比例）

业务优先级⭐⭐⭐⭐⭐

业务视角

适用角色CTO / 技术负责人 · 产品经理 · 数据工程师

适用平台跨境运营 AI Agent 工程落地 · Amazon SP API + LLM 集成 · 多平台数据采集 Agent

什么情况下用想把 AI 集成到业务系统，但 LLM 稳定性差、幻觉问题、成本控制都是挑战；Agent 任务失败了不知道哪步出了问题

成功是什么样的AI Agent 在生产环境稳定运行，失败可追踪，成本可控，复杂任务完成率 >85%

业务痛点

LLM 返回结果不稳定不可靠AI 幻觉导致业务决策错误Agent 任务失败了不知道哪步出问题AI 调用成本控制不住

1. 解决的问题

每日需要上架数百个 SKU，每个 SKU 经过「标题优化→图片描述生成→合规检查→关键词填写」4 步 SOP，当前用 LangGraph 编排，frontier 模型成本约 $0.15/SKU × 1000 SKU = $150/天

2. 核心算法逻辑

传统 LangGraph/CrewAI 等框架在运行时通过 Orchestrator 逐步调用 LLM 完成多步 SOP：每一步都需要独立的 API 调用、上下文窗口填充、token 计费。Subterranean Agent 的核心洞察是：当 SOP 固定时，Orchestrator 的编排逻辑可以从"运行时解释"变为"编译时参数化"——把整个多步工作流的决策逻辑直接烧录进单个模型权重。

3. 业务应用场景

业务问题：每日需要上架数百个 SKU，每个 SKU 经过「标题优化→图片描述生成→合规检查→关键词填写」4 步 SOP，当前用 LangGraph 编排，frontier 模型成本约 $0.15/SKU × 1000 SKU = $150/天。

数据要求： - 历史 SOP 执行轨迹（输入 SKU 信息 → 各步骤中间结果 → 最终 Listing） - 标准化 SOP 定义文件（每步 Prompt 模板 + 输出 Schema） - 至少 500 条高质量执行示例用于 SFT 训练数据

预期产出： - 编译后单次 API 调用完成全部 4 步 SOP - 输出结构化 JSON 包含：optimized_title / image_description / compliance_flags / keywords - 响应时间从 4 次串行调用（约 12s）降至 1 次推理（约 3s）

4. 输入数据要求

请查看原始代码模板获取输入规格。

5. 输出结果

请查看原始代码模板获取输出规格。

6. 业务价值 / ROI

需要搭建 SFT + GRPO 微调基础设施（A100 或等效 GPU × 8）
需要构建 SOP 执行轨迹数据集（每个 SOP 至少 500–2000 条样本）
编译后模型需要评估框架验证质量保留率（≥87%）
重编译 CI/CD 管线需要工程化（30–50 分钟/次）

7. 代码模板

代码块数量：1 · 路径：未检测到

"""
Subterranean Agent — 工作流编译范式模拟
论文: Compiling Agentic Workflows into LLM Weights (arXiv:2605.22502)
模拟将固定 SOP 从「运行时编排」转为「编译时参数化」的核心范式
"""

from __future__ import annotations
import time
import json
from dataclasses import dataclass, field
from typing import Any

# ─── 数据类定义 ────────────────────────────────────────────────────────────────

@dataclass
class WorkflowStep:
    """单个 SOP 步骤的元数据"""
    step_name: str
    prompt_template: str
    expected_output_schema: dict[str, str]
    estimated_input_tokens: int = 500
    estimated_output_tokens: int = 200

    def format_prompt(self, context: dict[str, Any]) -> str:
        """将上下文变量填充进 prompt 模板"""
        try:
            return self.prompt_template.format(**context)
        except KeyError as e:
            raise ValueError(f"步骤 [{self.step_name}] 缺少上下文变量: {e}") from e


@dataclass
class SOPWorkflow:
    """固定 SOP 工作流定义（多步骤序列）"""
    workflow_name: str
    description: str
    steps: list[WorkflowStep] = field(default_factory=list)

    def add_step(self, step: WorkflowStep) -> "SOPWorkflow":
        self.steps.append(step)
        return self

    @property
    def total_steps(self) -> int:
        return len(self.steps)

    @property
    def total_estimated_tokens(self) -> int:
        """运行时编排模式：每步独立调用的总 token 消耗（含上下文重复填充）"""
        # 每步都需要重新填充完整上下文（SOP说明 + 历史步骤输出 + 当前步骤 prompt）
        CONTEXT_OVERHEAD_PER_STEP = 800  # Orchestrator 系统 prompt + CoT 模板
        HISTORY_ACCUMULATION = 150       # 每步会累积前序输出
        total = 0
        for i, step in enumerate(self.steps):
            context_size = (
                step.estimated_input_tokens
                + CONTEXT_OVERHEAD_PER_STEP
                + i * HISTORY_ACCUMULATION    # 历史步骤输出的累积
            )
            total += context_size + step.estimated_output_tokens

8. 论文来源

2605.22502