最近在折腾一个 AI Agent 项目，技能越加越多，调度却越来越乱。原本只是想给助手加几个 Skills，结果发现真正卡脖子的不是“能不能调用”，而是“该先调谁、怎么并发、失败了怎么降级”。折腾一段时间之后，我用 130 行核心代码做了一个叫DeepAgentForce的轻量调度内核，专门解决 Agent Skills 的“丝滑”编排问题，现在已经开源。

这 130 行代码干的事情其实非常具体：把技能声明成带依赖关系的节点，自动拓扑排序，能并行的技能绝不排队，依赖没满足的技能先等着，随时超时、随时降级，全程状态可观测。如果你在做 Agent 应用或者准备给现有助手塞一堆工具函数，这个实现思路能让你少踩很多坑。

1. 为什么要给 Agent 做“Skills 调度”，以及我踩了什么坑

1.1 先聊清楚什么是 Agent Skills

Agent Skills 这个概念，最近在圈子里被反复讨论。它跟普通 Function Calling 最大的区别是：函数是原子操作，而 Skill 是一个“能力包”。一个 Skill 可以包含提示词模板、子步骤、校验规则、甚至内部的一小段工作流。

举个例子，一个“酒店推荐”技能，它内部可能要查库存、算价格、比对评分、生成推荐理由，这些子逻辑不需要 Agent 主脑去理解。Agent 只需要知道“我注册了一个酒店推荐技能，输入是预算和城市，输出是推荐列表”，然后就可以在合适的时机调用它。

这种封装方式很像螺丝刀套装里的批头——Agent 是手柄，Skills 是不同规格的批头，任务来了换一个就行。好处是主脑不需要承载所有领域知识，坏处是：当你有十几个批头的时候，选哪个、按什么顺序用、能不能同时用两个，就成了新的问题。

1.2 我最初串行调用的惨痛教训

我最早实现技能编排的方式非常原始：if-else 串联。用户问“帮我推荐一个适合周末短途旅行的海边城市”，我就先调天气技能，再调交通技能，最后调住宿技能，一个接一个，排着队来。

很快问题就暴露了。第一是顺序经常搞错。有一次我写反了依赖，先跑去查住宿，再查天气，结果给用户推了一个周末有暴雨的城市。第二是慢，每个技能调用都得花两三秒，三个技能串起来就是七八秒，用户早就没了耐心。第三是上下文爆炸，每个技能执行的时候，我把前面所有技能的完整输出都塞给大模型，结果 Prompt 越来越长，后面技能的输出质量明显变差。

更难受的是维护性。技能一多，if-else 就变成了蜘蛛网，加一个新技能得想着放在哪个分支里，删一个旧技能还得担心别的地方有没有引用。跟几个做 Agent 的朋友聊下来，发现大家几乎都走了一遍这条路，只是有人把锅甩给“模型太笨”，有人开始怀疑架构。

问题不在模型，在我的调度策略。我需要一个机制，让技能自己告诉我它们之间的依赖关系，然后由调度器统一安排执行顺序。DeepAgentForce 就是从这个需求里长出来的。

2. DeepAgentForce 的核心设计：130 行怎么装下这么多功能

2.1 用声明式描述技能，而不是写一堆 if-else

DeepAgentForce 的第一条设计原则是：Skill 自己交代元数据。我不用去读你的函数体，你只需要在定义技能的时候，声明“我依赖谁”“我的优先级是什么”“超时多久”“我输入什么”。

我用一个装饰器搞定这件事：

@skill( name="weather_query", dependencies=[], priority=20, timeout=15, description="查询指定未来一周的天气情况", ) async def weather_query(city: str, **ctx): ...

这里面的字段含义很明确：

• name：技能的唯一标识，其他技能通过这个名字声明依赖；
• dependencies：这个技能执行前必须完成的技能名列表；
• priority：同一批次可执行技能中的排序依据，数值越小越先跑；
• timeout：单个技能的最长执行时间，超时直接判定失败；
• description：给大模型规划器看的说明，也是自动路由的依据。

为什么选用声明式而不是把调用逻辑写死在调度器里？因为声明式数据可以被检查、被规划、被可视化。调度器不需要理解“天气查询”的业务逻辑，只需要知道它是一个没有依赖的节点、优先级是 20。这也让后面接入大模型做意图规划变得非常简单——模型读 description 就够了。

2.2 一张隐形的执行图：依赖怎么从“乱序”变成“有序”

技能之间的依赖关系，本质是一个有向无环图（DAG）。每个技能是一个节点，依赖关系是边。“酒店推荐”依赖“预算计算”和“天气查询”，那么图上就有两条边从后面两个节点指向它。

调度器拿到一个规划列表后，会先做两件事：建图、算入度。入度是“我还差多少个依赖技能没完成”的计数。入度为零，意味着所有前置技能都跑完了，这个节点立刻进入可执行队列。

这个逻辑我用一个很短的函数实现：

def build_graph(plan): nodes = set(plan) inc = {n: 0 for n in nodes} edges = {} for n in nodes: for dep in SKILL_REGISTRY[n].dependencies: if dep not in nodes: continue edges.setdefault(dep, []).append(n) inc[n] += 1 return inc, edges

建完图之后，调度器会用一个队列维护所有入度为零的节点，谁先完成，谁去给后置节点“解锁”。这跟餐厅后厨出餐是一个套路：蒸饭的时候同时切菜，切完菜才能下锅炒，炒完才能装盘，装盘不等备菜。后厨不会傻傻地先做完一道菜再做下一道，Agent Skills 调度也不应该。

如果你发现技能图里存在循环依赖，比如 A 等 B、B 等 A，拓扑排序会直接失败。我选择在初始化阶段就拒绝这类非法配置，宁可抛异常，也不跑到运行期才暴露问题。

2.3 调度器内部的状态流转，为什么能做到“丝滑”

有了图和入度，剩下的就是一个清晰的状态机。每个技能节点在任何时刻都处于以下状态之一：

所谓“丝滑”，拆开看就是三个能力：能并行的不排队、能恢复的不放弃、能绕开的不死磕。

能并行的不排队，靠的是多个 worker 从同一个队列里取任务；能恢复的不放弃，靠的是超时和重试；能绕开的不死磕，靠的是降级规则——某个非关键技能失败后，调度器可以按配置决定是中断整条链路，还是跳过它继续跑下游。

每次状态变化，调度器会广播一个事件。我在事件里带上技能名、状态、耗时和结果摘要，这样上层可以做实时日志、链路追踪、甚至画一个动态执行的仪表盘。这也是 DeepAgentForce 被我用在正式环境时最舒服的地方：问题定位从“猜”变成了“看”。

3. 从零实现一个可运行的最小版本

3.1 目录结构与技能注册，先搭一个清爽的骨架

DeepAgentForce 的目录结构非常简洁，核心就几个文件：

deepagentforce/ ├── engine.py # 调度引擎，核心约 130 行 ├── registry.py # 技能注册表与装饰器 ├── planner.py # 大模型规划器（可选） ├── demo.py # 示例入口 └── skills/ └── travel.py # 示例技能集

registry.py负责维护一个全局技能注册表，所有技能定义都往这个表里登记。它同时承担一个重要的校验工作：技能定义合法性检查。如果两个技能重名、依赖项引用了不存在的技能、或者出现循环依赖，注册阶段就能发现。

下面是注册表的核心实现：

SKILL_REGISTRY = {} def skill( name: str, dependencies: list = None, priority: int = 10, timeout: int = 30, description: str = "", ): def decorator(fn): SKILL_REGISTRY[name] = { "name": name, "fn": fn, "dependencies": dependencies or [], "priority": priority, "timeout": timeout, "description": description, } return fn return decorator

这种全局注册的写法可能被一些人诟病“不够优雅”，但在轻量框架里它非常实用：技能函数不需要被调度器显式引用，只要模块被 import 过，装饰器就会把它登记进去。我在实际项目里会在启动时统一导入技能包目录，确保所有技能都完成注册。

3.2 核心调度引擎逐段解析，130 行拆开看

真正干活的代码都在engine.py里。我把它压缩到大约 130 行，下面这段是为了方便讲解稍微展开过的版本，但逻辑完全一致。

import asyncio import time from collections import deque class SkillEngine: def __init__(self, max_workers=3): self.max_workers = max_workers self.ctx = {} async def run(self, plan: list): # 1. 建图、算入度 inc = {n: 0 for n in plan} edges = {} for node in plan: meta = SKILL_REGISTRY[node] for dep in meta["dependencies"]: if dep in plan: edges.setdefault(dep, []).append(node) inc[node] += 1 self.inc = inc self.edges = edges self.status = {n: "PENDING" for n in plan} # 2. 创建就绪队列，入度为零的先入队 ready_q = asyncio.Queue() for node in plan: if inc[node] == 0: ready_q.put_nowait(node) self.status[node] = "READY" workers = [ asyncio.create_task(self._worker(ready_q)) for _ in range(self.max_workers) ] await ready_q.join() for w in workers: w.cancel() return self.ctx async def _worker(self, ready_q): while True: name = await ready_q.get() try: await self._execute(name) finally: ready_q.task_done() async def _execute(self, name): meta = SKILL_REGISTRY[name] self.status[name] = "RUNNING" start = time.time() try: result = await asyncio.wait_for( meta["fn"](**self._build_input(name)), timeout=meta["timeout"], ) self.ctx[name] = result self._solve_dependencies(name, finished=True) except asyncio.TimeoutError: self._handle_failure(name, f"timeout after {meta['timeout']}s") except Exception as e: self._handle_failure(name, str(e)) finally: self.status[name] = "DONE" print(f"[skill] {name} -> {self.status[name]} ({time.time() - start:.2f}s)") def _build_input(self, name): # 从 ctx 里筛出技能声明需要的字段，不让历史数据全部灌进去 meta = SKILL_REGISTRY[name] return {k: self.ctx[k] for k in meta["dependencies"]} def _solve_dependencies(self, name, finished=True): if name not in self.edges: return for child in self.edges[name]: self.inc[child] -= 1 if self.inc[child] == 0: self.status[child] = "READY" asyncio.create_task(self._execute(child))

有几个细节值得单独拿出来说。

ready_q.join()是关键的阻塞点——它会等待队列中所有任务都调用过task_done()才返回，保证了run()不会提前结束。

_execute里我用asyncio.wait_for实现了超时控制，比手动计时更可靠。它会在超时时直接抛出TimeoutError，挂在技能函数上的网络请求也会被取消，避免了线程泄漏。

_solve_dependencies用asyncio.create_task去启动下游技能，而没有继续走队列，这样能减少一次出队入队的延迟，技能之间衔接更紧凑。我实测过微小但可感知的提速，在任务是轻量计算时尤其明显。

3.3 让大模型参与规划：最小接入方式

调度器本身不挑任务，它只需要一份“该执行哪些技能”的列表。这份列表可以由规则生成，也可以由大模型动态规划。我把后者做成了一个单独的planner.py，只在需要的时候引入。

核心思路很简单：把技能注册表里的name和description用 JSON 格式传给模型，让它输出一个技能执行序列数组。我给模型设计的提示词模板如下：

你是任务规划器。以下是可以调用的技能列表： [{name}, {description}]（每项以JSON呈现） 请根据用户请求，输出一个按执行顺序排列的技能名称数组。 只能使用列表中存在的技能名称，不要解释原因。输出必须是合法JSON数组。 用户请求：{user_request}

解析响应的部分我写得比较宽容：先用正则从模型输出里抓取第一对[...]，再从里面解析 JSON。如果解析失败，就退回到默认顺序。这段兜底逻辑很重要，因为模型输出经常带一些“思考过程”之类的多余文本，硬解析很容易翻车。

老实说，大模型规划没有想象中那么难接，难点反而在“模型返回了没注册的技能名”或者“漏掉了关键技能”。我的处理策略是：解析完后做一遍白名单过滤，不在注册表里的名字直接丢弃，保证调度器永远拿到合法输入。

4. 用“旅行规划”场景验证调度效果

4.1 一个标准的五技能旅行规划示例

为了验证 DeepAgentForce 不是玩具，我搭建了一个旅行规划 Demo。技能清单如下：

这个依赖结构里，budget_calc和weather_query互不依赖，应该同时执行；hotel_recommender要等它们两个完成才能启动；itinerary_builder只需要等weather_query，所以它实际上和hotel_recommender是同一批次，可以并行；copywriter最后跑。

技能定义代码大概是这样的风格：

@skill(name="budget_calc", dependencies=[], priority=5, timeout=3, description="根据用户输入计算行程预算") async def budget_calc(person_count: int = 2, **ctx): return {"budget": 5000} @skill(name="weather_query", dependencies=[], priority=10, timeout=15, description="查询目的地天气情况") async def weather_query(city: str = "某南方沿海城市", **ctx): ... @skill(name="hotel_recommender", dependencies=["budget_calc", "weather_query"], priority=50, timeout=8, description="根据预算和天气推荐酒店") async def hotel_recommender(**ctx): budget = ctx["budget_calc"]["budget"] weather = ctx["weather_query"]["forecast"] ...

你会发现我在技能函数里并没有写任何调度逻辑，只是通过**ctx去拿上游产物。依赖关系全部体现在装饰器参数里，代码可读性一下子高了很多。

4.2 实际运行日志长什么样

跑起来之后，日志输出是这样的（我做了精简，但格式是真实的）：

[10:00:01] READY budget_calc (priority=5) [10:00:01] READY weather_query (priority=10) [10:00:01] RUN budget_calc [10:00:01] RUN weather_query [10:00:03] DONE budget_calc (2.05s) [10:00:04] DONE weather_query (3.10s) [10:00:04] READY hotel_recommender [10:00:04] READY itinerary_builder [10:00:04] RUN hotel_recommender [10:00:04] RUN itinerary_builder [10:00:08] DONE itinerary_builder (4.02s) [10:00:09] DONE hotel_recommender (5.01s) [10:00:09] READY copywriter [10:00:09] RUN copywriter [10:00:11] DONE copywriter (2.11s)

这份日志非常有用。你能直接看到哪些技能被并行执行了，哪些在等依赖，哪些技能耗时长，哪些技能拖慢了整条链。我在正式环境里加了事件埋点，把同样的日志推到图表面板上，效果跟 CI 流水线视图类似，每次 Agent 跑完我都能看到一张执行瀑布图。

4.3 串行和并行的实测对比数据

我拿同一个任务做了三组对照，分别用最朴素的 if-else 串行、DeepAgentForce 并发调度、以及不带超时控制的并发调度来跑。

从数据看，并发调度相比串行的收益是实打实的，四五个技能的链路能省下几乎一半时间。更重要的是，超时控制把最坏执行时间从“不可控”变成了“可控”，这一点在生产环境的权重比提速还高。

还有个容易被忽略的收益是 token 花费。因为_build_input只挑依赖项产物喂给下游技能，而不是把历史记录全部塞进去，后面的技能看到的输入更精简，模型输出质量也稳定了。我统计过，改造后单个任务的 token 开销降了大概三成。

5. 常见问题与调试心得

5.1 技能互相抢占 / 优先级不生效

有朋友跑起来第一件事就是问我：为什么我设置了 priority，两个技能还是乱序执行？

原因是入队的时候我直接按技能列表的顺序 put，没有对同批次优先级做排序。要在 READY 阶段严格按 priority 执行，就得改用优先队列，或者入队前先把同批次节点按 priority 排序。我的做法是后者，因为更直观：

ready_items = [node for node in plan if inc[node] == 0] ready_items.sort(key=lambda n: SKILL_REGISTRY[n]["priority"]) for node in ready_items: ready_q.put_nowait(node)

注意priority只在“同批次可执行节点”之间比较，不代表它能跨越依赖链提前执行。如果你希望某个技能无论如何都先跑，从设计上就应该把它放进依赖链的最上游，而不是依赖优先级这个弱信号。

5.2 上下文爆炸问题

我最初版本里有一个偷懒的写法：把ctx整个传给每个技能。技能只要写**ctx就能拿到上游产物，写起来很爽，代价是上下文像雪球一样滚。等到第五个技能执行的时候，它收到的是前面四个技能的全部中间产物，包括一些完全用不上的日志、调试信息、甚至大模型生成的中间文案。

解决思路就是我在_build_input里做的依赖项白名单：只把dependencies里声明过的技能产物传给当前技能。对于真需要全局变量的场景，我额外留了一个shared字段，只有显式声明了才会注入。这个改动让我后续再也不用为“上下文长度超限”头疼。

5.3 脚本卡死 / 假死

卡死问题最典型的原因是技能内部发起了网络请求，但没设置 IO 超时。调度器层超时只能兜底异常场景，如果你每个技能都不设 timeout，而每次请求又恰好永远挂起，那个技能就会一直处于 RUNNING 状态。

我自己的排查套路是两步：第一步，看engine.py日志里有没有技能长时间停留在 RUNNING；第二步，在技能函数里给网络调用再包一层asyncio.wait_for，不要把调度器的超时当作最后防线。调度器的 timeout 是保险丝，不是刹车片。

5.4 大模型规划不稳定怎么办

模型把用户请求解析成技能列表时，偶尔会输出“思考过程+JSON”的混合文本，或者干脆返回一个空数组。我在 planner 里加了三次容错：正则提取 JSON 数组、JSON 解析失败后二次尝试、最终回退到默认技能顺序。

从这几次迭代看，最有效的还是把提示词写死成约束性语言：只允许输出 JSON、只允许使用白名单技能、不要解释。如果你接的模型支持 JSON Schema 强制约束，直接开启约束模式，这类问题基本绝迹。我用下来的经验是：与其期望模型稳定，不如把模型当不稳定组件，然后在外面加一层确定性的解析和过滤护栏。

5.5 常见问题速查表

最后再分享一个我实际用下来的心得。DeepAgentForce 这套 130 行的调度内核，表面上是在做任务编排，本质上是在划定“模型”和“机制”的边界。大模型负责看意图、定方向，调度器负责盯节奏、管异常、保底线。把这两件事拆开之后，Agent 的稳定性才真正可控。

我自己后续会在流式输出和条件分支上继续迭代，比如让某个技能按前序技能的不同结果走不同分支，同时也想把执行瀑布图开源出来。如果你也遇到过技能调度乱成一团的问题，可以先拿这套思路去改自己手上的小项目，别急着上重框架，130 行很多时候已经够用了。