多Agent工具调度：依赖DAG与全局并发上限

题意与假设

这道练习题保留了面经中值得训练的核心约束，并将需要澄清的条件单独列出。现场拿到变体时，先和面试官确认这些条件。

• agents映射每个agent的本地task ID到依赖ID集合和无参callable；不跨agent依赖。训练使用共享thread pool，而非规定一个agent所有工具必须跑在同一物理线程。
• 依赖失败则后代skipped，其他独立任务继续；callable抛Exception标failed。任务图在开始前固定，循环/缺失依赖在任何副作用前拒绝。
• 全局limit限制同时执行的工具数，线程主要适合I/O型工具；Python CPU密集任务可另用进程/远端执行。示例无超时、运行中强制取消或持久化恢复。

解题路线

1. 先对所有图做Kahn拓扑检查，再执行。运行时remaining计数归零的任务进入ready，只有全部依赖status=success才能submit；失败后仍推进依赖计数，让后代被明确标skipped而非永久等待。
2. 调度线程统一修改状态，worker只调用工具并返回结果。running future数不超limit；wait FIRST_COMPLETED推进依赖，不让worker持有slot去等待还未提交的子任务，避免线程池死锁。
3. 从单agent到多agent只需把key拓展成(agent,task)，所有ready共享pool，因此limit是全局而不是每agent各一个limit。每agent公平需轮转ready队列/配额，当前FIFO不承诺公平延迟。
4. 动态增加tool call、重试和cancel需要显式任务状态机/attempt ID。Python future.cancel不能终止已运行线程；外部shell任务应有进程组和协作取消协议，取消后仍要回收日志和资源。

手动走一遍

a依赖b，b失败：b failed、a skipped；另一个agent的c无依赖仍成功。limit2，两个agent各有独立工具，用threading.Barrier证明确实同时运行；所有任务依赖开始顺序、上限和失败传播经904项真实线程核验。

Python 参考实现

下面的代码按上文假设实现，适合本地运行并追加自己的测试用例。

代码区可横向滚动；使用键盘时，先聚焦代码区再按方向键查看。

from concurrent.futures import ThreadPoolExecutor,wait,FIRST_COMPLETED
from collections import deque

def schedule(agents,limit):
    # agents[agent][task] = (local_dependency_ids, zero-argument callable)
    if not isinstance(limit,int) or limit<1:raise ValueError('positive worker limit')
    tasks={};children={};remaining={}
    for agent,specs in agents.items():
        for task,(deps,fn) in specs.items():
            key=(agent,task);deps=set(deps)
            if not callable(fn) or any(d not in specs for d in deps):raise ValueError('task/dependency')
            tasks[key]=(set((agent,d) for d in deps),fn);remaining[key]=len(deps);children[key]=[]
    for key,(deps,_) in tasks.items():
        for dep in deps:children[dep].append(key)
    # Validate the entire graph before executing any side effect.
    counts=remaining.copy();queue=deque(k for k,v in counts.items() if not v);visited=0
    while queue:
        key=queue.popleft();visited+=1
        for child in children[key]:
            counts[child]-=1
            if not counts[child]:queue.append(child)
    if visited!=len(tasks):raise ValueError('dependency cycle')
    ready=deque(k for k,v in remaining.items() if not v);status={};results={};running={}
    def finish(key,state,value=None):
        status[key]=state;results[key]=value
        for child in children[key]:
            remaining[child]-=1
            if not remaining[child]:ready.append(child)
    with ThreadPoolExecutor(max_workers=limit) as pool:
        while ready or running:
            while ready and len(running)<limit:
                key=ready.popleft();deps,fn=tasks[key]
                if any(status[d]!='success' for d in deps):finish(key,'skipped');continue
                running[pool.submit(fn)]=key
            if not running:continue
            done,_=wait(running,return_when=FIRST_COMPLETED)
            for future in done:
                key=running.pop(future)
                try:finish(key,'success',future.result())
                except Exception as error:finish(key,'failed',error)
    return status,results

复杂度

V任务E依赖，预检与调度元数据O(V+E)，空间O(V+E+limit)。实际工期取决于任务时间和DAG，至少max(关键路径,总工作/limit)，FIFO不承诺最优makespan。

易错点与追问

容易失分的边界

• 在worker里等待同pool其他future会耗尽slot。
• 每agent各创建limit线程会把全局并发放大到agent数×limit。
• 任务执行之后才发现环会留下无法撤销的副作用。

继续深挖

• 动态任务依赖如何禁止环并让重试不重复外部副作用？
• 不同agent优先级与公平配额怎样避免一个大图占满ready队列？