最近在折腾一个 AI Agent 项目技能越加越多调度却越来越乱。原本只是想给助手加几个 Skills结果发现真正卡脖子的不是“能不能调用”而是“该先调谁、怎么并发、失败了怎么降级”。折腾一段时间之后我用 130 行核心代码做了一个叫DeepAgentForce的轻量调度内核专门解决 Agent Skills 的“丝滑”编排问题现在已经开源。这 130 行代码干的事情其实非常具体把技能声明成带依赖关系的节点自动拓扑排序能并行的技能绝不排队依赖没满足的技能先等着随时超时、随时降级全程状态可观测。如果你在做 Agent 应用或者准备给现有助手塞一堆工具函数这个实现思路能让你少踩很多坑。1. 为什么要给 Agent 做“Skills 调度”以及我踩了什么坑1.1 先聊清楚什么是 Agent SkillsAgent Skills 这个概念最近在圈子里被反复讨论。它跟普通 Function Calling 最大的区别是函数是原子操作而 Skill 是一个“能力包”。一个 Skill 可以包含提示词模板、子步骤、校验规则、甚至内部的一小段工作流。举个例子一个“酒店推荐”技能它内部可能要查库存、算价格、比对评分、生成推荐理由这些子逻辑不需要 Agent 主脑去理解。Agent 只需要知道“我注册了一个酒店推荐技能输入是预算和城市输出是推荐列表”然后就可以在合适的时机调用它。这种封装方式很像螺丝刀套装里的批头——Agent 是手柄Skills 是不同规格的批头任务来了换一个就行。好处是主脑不需要承载所有领域知识坏处是当你有十几个批头的时候选哪个、按什么顺序用、能不能同时用两个就成了新的问题。1.2 我最初串行调用的惨痛教训我最早实现技能编排的方式非常原始if-else 串联。用户问“帮我推荐一个适合周末短途旅行的海边城市”我就先调天气技能再调交通技能最后调住宿技能一个接一个排着队来。很快问题就暴露了。第一是顺序经常搞错。有一次我写反了依赖先跑去查住宿再查天气结果给用户推了一个周末有暴雨的城市。第二是慢每个技能调用都得花两三秒三个技能串起来就是七八秒用户早就没了耐心。第三是上下文爆炸每个技能执行的时候我把前面所有技能的完整输出都塞给大模型结果 Prompt 越来越长后面技能的输出质量明显变差。更难受的是维护性。技能一多if-else 就变成了蜘蛛网加一个新技能得想着放在哪个分支里删一个旧技能还得担心别的地方有没有引用。跟几个做 Agent 的朋友聊下来发现大家几乎都走了一遍这条路只是有人把锅甩给“模型太笨”有人开始怀疑架构。问题不在模型在我的调度策略。我需要一个机制让技能自己告诉我它们之间的依赖关系然后由调度器统一安排执行顺序。DeepAgentForce 就是从这个需求里长出来的。2. DeepAgentForce 的核心设计130 行怎么装下这么多功能2.1 用声明式描述技能而不是写一堆 if-elseDeepAgentForce 的第一条设计原则是Skill 自己交代元数据。我不用去读你的函数体你只需要在定义技能的时候声明“我依赖谁”“我的优先级是什么”“超时多久”“我输入什么”。我用一个装饰器搞定这件事skill( nameweather_query, dependencies[], priority20, timeout15, 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 调度器内部的状态流转为什么能做到“丝滑”有了图和入度剩下的就是一个清晰的状态机。每个技能节点在任何时刻都处于以下状态之一状态含义PENDING入度不为零等待前置技能完成READY入度为零已进入可执行队列等待 worker 领取RUNNING正在执行DONE正常完成产物已写入上下文FAILED执行异常或超时SKIPPED前置失败导致不再需要执行所谓“丝滑”拆开看就是三个能力能并行的不排队、能恢复的不放弃、能绕开的不死磕。能并行的不排队靠的是多个 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_workers3): 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)), timeoutmeta[timeout], ) self.ctx[name] result self._solve_dependencies(name, finishedTrue) except asyncio.TimeoutError: self._handle_failure(name, ftimeout 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, finishedTrue): 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计算预算上限无51sweather_query查询目的地天气无103shotel_recommender推荐酒店budget_calc, weather_query502sitinerary_builder生成行程weather_query404scopywriter生成种草文案hotel_recommender, itinerary_builder902s这个依赖结构里budget_calc和weather_query互不依赖应该同时执行hotel_recommender要等它们两个完成才能启动itinerary_builder只需要等weather_query所以它实际上和hotel_recommender是同一批次可以并行copywriter最后跑。技能定义代码大概是这样的风格skill(namebudget_calc, dependencies[], priority5, timeout3, description根据用户输入计算行程预算) async def budget_calc(person_count: int 2, **ctx): return {budget: 5000} skill(nameweather_query, dependencies[], priority10, timeout15, description查询目的地天气情况) async def weather_query(city: str 某南方沿海城市, **ctx): ... skill(namehotel_recommender, dependencies[budget_calc, weather_query], priority50, timeout8, 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 (priority5) [10:00:01] READY weather_query (priority10) [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 并发调度、以及不带超时控制的并发调度来跑。执行方式总耗时失败处理备注if-else 串行14.2s无一个挂了全挂每个技能 2-5s累计相加并发调度无超时8.7s无可能卡死两个无依赖技能并行但有风险DeepAgentForce8.1s每次任务自动超时含日志和状态机开销从数据看并发调度相比串行的收益是实打实的四五个技能的链路能省下几乎一半时间。更重要的是超时控制把最坏执行时间从“不可控”变成了“可控”这一点在生产环境的权重比提速还高。还有个容易被忽略的收益是 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(keylambda 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 常见问题速查表现象可能原因解决方案两个技能总是先执行一个同级 READY 未按 priority 排序入队前排序技能收到大量无用历史数据ctx 整体透传按 dependencies 筛选输入任务长时间 RUNNING技能内部网络请求无 IO 超时技能内单独加 wait_for模型输出非法 JSON模型返回自由文本正则抽取 白名单过滤 默认顺序回退技能 A 失败后整个链路中断没有降级策略配置 skip_on_failure 标记非关键技能最后再分享一个我实际用下来的心得。DeepAgentForce 这套 130 行的调度内核表面上是在做任务编排本质上是在划定“模型”和“机制”的边界。大模型负责看意图、定方向调度器负责盯节奏、管异常、保底线。把这两件事拆开之后Agent 的稳定性才真正可控。我自己后续会在流式输出和条件分支上继续迭代比如让某个技能按前序技能的不同结果走不同分支同时也想把执行瀑布图开源出来。如果你也遇到过技能调度乱成一团的问题可以先拿这套思路去改自己手上的小项目别急着上重框架130 行很多时候已经够用了。