LLMCompiler模式：用编译器思维重构AI技能编排

问题背景

我们团队在构建电商智能客服Agent时遇到了一个典型困境：用户一句"帮我查一下蓝色M码T恤有没有货，顺便看看有没有满减优惠"，背后需要调用商品查询、库存检查、价格计算、优惠信息获取等多个技能。

最初的实现是典型的hardcode编排——用if-else和顺序await串起整个流程。这种方式在简单场景下工作正常，但当SKU种类超过5000、优惠规则超过200条时，平均响应时间从2秒飙升至8秒。更头疼的是，当某个技能调用超时或返回异常时，整个链条的容错处理需要写大量的try-catch，业务逻辑和异常处理混在一起，维护成本极高。

为什么这个决策这么难做

我们一开始以为问题出在API调用性能上，于是花了两周时间优化HTTP连接池、压缩返回数据。但上线后发现延迟只降低了10%，根本原因根本没摸到。

实际上，这些技能调用之间存在大量可以并行的部分。比如"库存检查"和"优惠信息获取"完全不互相依赖，却因为hardcode的顺序编写被强制串行执行。我们后来又尝试用Promise.all并行化，但遇到了更棘手的问题：某些技能依赖其他技能的输出——价格计算必须等商品查询返回原价，优惠判断需要先知道库存状态。盲目并行导致大量数据竞争和undefined错误。

我们意识到，这不是简单的并行优化问题，而是任务编排缺乏对依赖关系的显式建模能力。

根因：缺乏编译器思维的任务编排

让我们从编译器设计的角度重新审视这个问题。如果我们把AI技能编排看作一个"任务编译器"，它应该包含三个核心阶段：

# LLMCompiler核心架构伪代码

class TaskCompiler:
    """任务编译器：将自然语言指令编译为可执行的任务图"""
    
    def compile(self, user_intent: str) -> TaskGraph:
        # 阶段1：前端解析 - 将自然语言解析为任务序列
        task_sequence = self.parser.parse(user_intent)
        
        # 阶段2：中间表示优化 - 分析依赖关系，构建DAG
        dag = self.optimizer.build_dag(task_sequence)
        
        # 阶段3：代码生成 - 调度器根据DAG执行任务
        results = self.executor.execute(dag)
        
        return results

class TaskGraph:
    """任务图：表达任务间的数据依赖关系"""
    
    def __init__(self):
        self.nodes: List[Task] = []  # 任务节点
        self.edges: List[Dependency] = []  # 依赖边
    
    def add_node(self, task: Task):
        self.nodes.append(task)
    
    def add_edge(self, from_task: str, to_task: str, data_key: str):
        """建立数据依赖：to_task需要from_task的data_key输出"""
        self.edges.append({
            "from": from_task,
            "to": to_task,
            "data_key": data_key
        })
    
    def get_execution_plan(self) -> List[List[str]]:
        """拓扑排序生成可并行执行的批次"""
        return topological_layers(self.edges)

关键在于依赖分析器如何工作。看看实际的调度逻辑：

class TaskScheduler:
    """任务调度器：根据DAG拓扑序动态决定执行策略"""
    
    def execute(self, dag: TaskGraph) -> Dict:
        results = {}
        execution_plan = dag.get_execution_plan()
        
        for layer in execution_plan:
            # 同一层的任务可以并行执行
            layer_tasks = []
            for task_id in layer:
                task = dag.get_task(task_id)
                # 检查依赖是否满足
                deps_ready = all(
                    dep in results for dep in task.dependencies
                )
                if deps_ready:
                    layer_tasks.append(task)
            
            # 并行执行这一层的任务
            layer_results = await asyncio.gather(
                *[self.exec_task(t, results) for t in layer_tasks]
            )
            
            # 收集结果供下一层使用
            for i, task_id in enumerate(layer):
                results[task_id] = layer_results[i]
        
        return results

以开头的电商场景为例，编译器会分析出：商品查询可以立即执行，库存检查和优惠信息获取都依赖商品ID可以并行，价格计算依赖商品原价和优惠信息。这样整个执行从8秒的串行变为约3秒的并行（最大耗时路径决定总延迟）。

可移植的原则

1. 如果你在设计多技能协作的Agent系统，先用DAG建模所有技能间的数据依赖，而不是直接写执行顺序。
2. 如果你在处理串行任务性能瓶颈，先用拓扑排序分析哪些任务可以并行，不要凭直觉并行——依赖分析错误比串行更危险。
3. 如果你在构建任务编排层，保持前端解析（理解意图）和后端执行（调度任务）的解耦，这样你可以在不改变解析逻辑的情况下优化执行策略。
4. 如果你在实现容错机制，不要在每个技能调用里写try-catch，而是让调度器统一处理超时和异常，根据DAG重新规划后续执行路径。

结尾

LLMCompiler模式的核心价值在于将"如何组织任务执行"这个问题从业务代码中抽离出来，让调度器根据依赖关系自动做出最优决策。如果你正在被hardcode的技能编排所困扰，建议先从分析现有流程的依赖关系开始，画出任务图，这往往是重构的第一步。