多Agent复杂系统实战：从Skill机制到公众号接入的完整架构解析

最近这半年，我的技术重心明显从"单次调用大模型做点好玩的功能"挪到了"怎么让AI真正接手一套复杂系统"。3月30日那天，我做了一次印象很深的实验：把一套多Agent复杂系统接到公众号上，让用户在对话框里跟它自然对话，系统内部自动拆任务、调技能、维护上下文、异步回推结果。整个折腾下来，不光跑通了，还让我把"AI操作系统"这个抽象概念落地成了能摸得着的架构。

这篇文章就把当天的完整过程拆开讲：为什么AI做复杂系统必须借操作系统那套思路？Skill机制怎么设计才能像乐高一样可插拔？公众号这个入口接入时有哪些容易翻车的细节？跟着我一步步走一遍，你也能搭出一套"小而完整"的多Agent复杂系统。

1. 为什么AI做复杂系统绕不开"操作系统"这个比喻？

先说说我做这个实验的动机。之前我做过不少AI小工具：给大模型包装几个函数，让它自己选工具、调工具、返回结果。单个Agent接两三个外部工具，其实问题不大。但当你手里的Agent数量超过五个、Skill超过十几个、还要处理多轮对话和多用户并发时，整个系统会迅速失控。

失控的典型表现是：A业务线的Agent在B用户的问题上返回了C用户的订单数据；某个Skill超时导致整条请求链卡死；新加一个技能要改路由代码，改完又怕影响之前的功能。这些问题单靠模型聪明是解决不了的，它的本质是系统设计问题。

我把这个问题跟操作系统的发展史对照了一下，发现AI复杂系统正在重复计算机操作系统当年走过的路。最开始我们写代码就类似DOS时代——一个程序跑到底，单任务、单进程，出了问题整个机器死机。后来有了多任务系统，进程可以并发跑，但进程之间可能互相干扰，于是引入了进程隔离、内存保护、调度器。再到后来的现代操作系统，文件系统、设备驱动、权限模型、中断机制分层清清楚楚。

现在的AI复杂系统正好卡在从"单任务"往"多进程管理"过渡的阶段。我们需要的不是又一个调大模型的框架，而是一套更像操作系统的承载层：谁先跑、谁后跑、资源不够怎么办、一个技能崩溃会不会拖垮主流程、不同用户的对话数据会不会串——这些都需要系统级的管理机制。

我整理了一张AI系统与操作系统的概念对照表，这张表后期成了我设计架构的核心依据：

这套对照不是说非得写个真正的AI操作系统内核，而是提醒我们：复杂AI系统的核心难点从来不在模型本身，而在"协调"。3月30日的实验，本质上就是围绕这张表做一次最小的工程验证。

2. 从单体对话框到多Agent协奏：任务拆解与分层思路

设计系统之前，先把目标定清楚。我想做的是一个能跑在公众号里的AI助手：用户在对话框里提问或下达任务，系统判断意图、拆解子任务，然后调度不同的Agent和Skill去完成，最后把结果整理成用户友好的回复。这个目标看起来不复杂，但要支撑它的是背后一整套分层架构。

我用了三个核心层级：能力层、调度层、状态层。能力层是所有Skill的集合——信息查询、订单处理、文本总结、异步通知，每个能力都是独立封装的一个技能。调度层负责意图识别和任务路由：给定用户输入，先判断属于哪个业务域，然后决定调用哪个Skill，要不要拆分成多个子步骤，多个子步骤之间有没有依赖关系。状态层负责所有会话上下文、用户画像、任务进度的存取，保证多轮对话不会丢上下文。

这个分层的理由是：把"模型能力"和"业务逻辑"解耦。调度层只要能读懂Skill的输入输出声明，就能在新增技能时做到零代码改动。这条规则听起来简单，但真正落地时需要把接口约定做到非常严格，我在第3节里详细说。

整个请求流转链路是这样的：用户在公众号对话框发消息，平台把消息推送给我部署的后端服务；服务先做签名校验，然后从事件内容中取出消息文本和用户编号；接着交给调度层——先做一轮轻量级的意图分类，再用一个路由决策模块判断当前用户处于什么会话状态（是新问题还是对上一话题的追问）；最后把任务分发给具体的Skill，Skill执行完返回结构化结果，再经过一层回复生成，拼装成自然语言返回给用户。

我当时花了很大精力在对话状态机上，原因很现实：公众号场景里用户不会告诉你"我要开始新话题了"，他可能直接追问一句"那价格呢"。如果没有状态机的上下文管理，这句话根本没法回答。状态层里我保存了当前活跃节点、上次回复涉及的业务实体、提问历史摘要三样东西。下次消息进来时，调度层会先加载这段状态，再决定居然追问还是开启新任务。

2.1 为什么把公众号当作系统的第一入口

很多朋友问我，为什么偏偏选公众号来做这个复杂系统的前端入口。我自己的判断是：公众号平台天然提供了一套"用户触达 + 消息信道 + 长期会话"的基础能力，我不需要自己开发App、不需要管理用户系统、不需要处理推送通道的长连接保活。只要做好消息服务和业务逻辑，就能验证整套复杂AI系统的架构。

另一个原因是公众号交互模式非常适合异步任务。复杂系统里很多Skill跑不完同步请求——比如我要调用一个外部数据处理服务，可能要等十几秒。公众号允许系统先回复一个"任务已收到、正在处理"，然后在后台跑完流程后再把结果推给用户。这种异步交互跟"操作系统里的中断"很类似，用户体验也自然。

把公众号接进来之后，我给了它一个身份定义：它既是一个交互终端，也是一个特殊的Skill。从系统视角看，公众号不过是"用户消息的接收器和结果分发器"，它跟别的Skill一样，都有输入（用户消息）、输出（回复内容）、状态（会话上下文）三个接口。这个抽象很有用——后面如果要再加其他终端（邮件、IM、语音助手），只需要实现同一套接口协议，系统主体完全不用动。

3. Skill机制的核心设计：让能力像乐高一样可插拔

从3月30日这次实验里我最大的收获，是Skill机制必须做成"接口先行"。大多数人在集成AI工具时习惯于给大模型塞函数定义，函数返回个字符串就完事了。但在复杂系统里，这种松散的约定会导致灾难：调度层根本不知道这次调用到底成功没有、要不要重试、结果里的数据字段是什么类型，稍一复杂就出问题。

我给每个Skill定义了一份标准清单：元信息、输入schema、输出schema、运行模式、权限声明、错误处理策略。输入输出schema解决"调度层怎么理解你"的问题；运行模式声明这个Skill是同步执行还是异步执行、超时阈值是多少；权限声明解决安全分级；错误处理策略决定失败时是重试、降级还是直接告知用户。

json复制{
  "skill_name": "order_refund_skill",
  "version": "2.1.0",
  "description": "处理用户退款申请的复杂业务流程",
  "input_schema": {
    "type": "object",
    "required": ["order_id", "reason_code"],
    "properties": {
      "order_id": {"type": "string", "pattern": "^ORD[0-9]{8}$"},
      "reason_code": {"type": "integer", "enum": [1, 2, 3, 4]},
      "memo": {"type": "string", "max_length": 200}
    }
  },
  "output_schema": {
    "type": "object",
    "required": ["code", "refund_status", "estimated_date"],
    "properties": {
      "code": {"type": "integer", "enum": [0, 1001, 1002]},
      "refund_status": {"type": "string", "enum": ["accepted", "rejected", "pending"]},
      "estimated_date": {"type": "string", "format": "date"}
    }
  },
  "runtime": {
    "mode": "async",
    "timeout_seconds": 15,
    "retry_policy": "once",
    "permission_level": "user_verified"
  },
  "error_handling": {
    "on_timeout": "notify_user_async",
    "on_validation_error": "return_code_1001"
  }
}

这套规则一旦定下来，调度层就变得极其简单——不感知每个Skill内部怎么做，只读取manifest、校验输入、调用执行器、校验输出、决定是否重试。这种设计就是我在第1节说的"系统调用"思想：上层只依赖接口协议，不依赖具体实现。

3.1 Skill注册表与热加载

Skill多了之后，手工路由等于自杀。我在系统里加了一个注册表：每个Skill上线时把manifest注册进去，系统启动时扫描Skill目录自动发现技能，运行过程中也能动态加载新技能。注册表保存了技能名、版本、输入输出schema、运行模式、权限等信息。

调度层匹配Skill时分为两步。第一步是意图分类：用户输入先被一个轻量模型或规则引擎转成语义意图标签，比如"退款申请""价格查询""进度追问"。第二步是技能匹配：意图标签加上当前会话状态，一起丢给注册表，找出可用的Skill候选列表。如果有多个候选，再按权限等级、优先级、历史成功率排序，选出最优解。

这一步让我体会到注册表的巨大价值：Agent系统要变得可演进，绝不能把技能列表写死在路由代码里。每次新增一个技能，创业者只需在目录下放一个manifest文件和对应的执行模块，系统重启或热加载后自动发现。真正做到了"加技能不改主流程"。

3.2 Skill执行器的通用运行框架

每个Skill内部可以完全是自己的业务代码，但外层执行器必须统一。我写了一个通用执行器基类，它做五件事：读取manifest、校验输入参数、调用业务函数、校验输出结果、提交日志与耗时数据。业务函数可以用任何语言和框架写，只要按约定的输入输出格式返回即可。

执行器的好处是让所有Skill共享一套监控能力。每个Skill每次运行的数据都会落日志：耗时多少、输入输出体积、超时没有、重试没有。这些可观测数据对于长期维护复杂系统极其重要——哪天某个技能变慢了，我能从监控面板上一眼看出是外部服务变慢了还是输入体量变大了。

4. 公众号Skill的完整落地过程：从接口校验到多轮追问

现在说说3月30日那天真正动手实现公众号Skill的过程。公众号接入的第一件事是配置服务器地址和token，这个环节的核心是签名校验——平台在验证我的服务器时，会把一个加密签名、时间戳、随机串和token一起发过来，我在本地按同样的算法拼接计算，比对结果一致才响应成功。这一步不是走形式，它保证后续所有消息都真的来自公众号平台，而不是伪造请求。

这里我建议首次接的时候一定要先把验签函数单独跑通。我就吃过这个亏，一开始把验签逻辑写在了主流程里，结果平台回调一直失败，日志里全是"Invalid signature"，排查半天才发现是我在拼接HTML实体时把参数顺序弄错了。

4.1 消息处理的主链路

平台推消息给我的服务时，payload是XML结构。我做了个消息解析层，把XML里的消息类型、用户标识、内容、事件类型抽出来，统一转成内部事件对象。然后事件进入调度层之前，会先做一次消息去重检查——这个细节非常重要，平台在异常情况下会重发消息，处理不好用户同一个指令会被执行两三遍。我维护了一个基于消息标识的Redis去重表，只处理第一次到达的消息。

主链路的简化版伪代码大概长这样：

python复制def handle_platform_message(xml_body):
    msg = parse_message(xml_body)
    if dedup_check(msg.msg_id):
        return "duplicate ignored"

    session = load_session(msg.user_id)
    intent = intent_classify(msg.content, session.recent_context)
    skill = skill_registry.match(intent, session.state)

    if skill.runtime.mode == "sync":
        result = skill.execute(msg.content, session)
        reply = compose_reply(result)
        return reply
    else:
        task_id = task_queue.enqueue(skill, msg.content, session)
        return "任务已收到，正在处理中，稍后推送结果"

这个链路看起来简单，但每行背后都有细节。比如加载session，不是只拉一条记录，而是要把用户最近几轮的意图摘要、活跃业务实体、待确认参数都恢复出来。再比如任务入队之后，如何异步回推结果——这里走的是自主发送消息通道，用户在会话存续期内能收到系统主动下发的任务完成通知。

4.2 多轮追问的上下文恢复

公众号场景里最麻烦的是多轮追问。用户第一轮说"帮我看下订单"，系统回复"请提供订单号"。用户第二轮真的发来订单号，这句话本身没有意图关键词，如果系统不记得上一轮在等什么，就会当作新消息重新分类，对话直接断掉。

我的解决办法是维护一个"待填充槽位"的状态机制。当Skill输出结果里带有缺参标记时，系统不是简单回复用户，而是记录当前Skill名称、等待的槽位key、尚未执行的完整请求参数包。下一轮用户消息进来时，调度层先检查session里是否有挂起的槽位，如果有就直接把消息内容填入对应槽位，然后重新触发技能执行。

这个机制让我想到了当年用爬虫框架时填表单的感觉——每一轮对话都是往同一个表单里填字段，填齐了就提交。用户不会意识到自己在填表，因为回复模板设计成了自然的追问语气，比如"还需要告诉我退款原因哦"。状态层保存的是"未完成的表单"，这是整个多轮对话逻辑里最核心的数据结构。

4.3 公众号响应的超时约束

复杂系统接入公众号一定要意识到一个现实约束：同步响应窗口很短。如果我在服务器收到消息后超过几秒没有返回，平台会认为服务异常，并且启动重试。这意味着任何超过响应时限的复杂任务都不应该在请求处理线程里同步执行。

我的策略是把所有耗时操作异步化：收到消息后，先快速校验、入队，立刻返回一个"任务受理中"的提示；后台线程池执行Skill，执行完成后再主动推送结果。这个方案在实际使用中很稳——用户感觉是"我说了个任务，几秒后系统自己回了结果"，体验比干等十几秒好得多。

对异步推送也有一个需要注意的边界：如果用户长期不跟公众号互动，部分主动服务受限，因此任务推送需要在会话有效期内尽快执行。实测下来，"异步任务+主动回传"的方式适合大多数业务型Skill；如果某些技能必须在窗口内完成，那就只能尽量用轻量模型、精简流程，把耗时压到窗口以内。

5. 实测中的三个大坑：上下文串味、重复执行与任务超时

这一节分享我认为最有价值的踩坑经验。3月30日的测试中，我先后踩了三个大坑，每个坑都几乎让我推倒重来。

坑一：上下文串味。 我把所有用户的对话历史放进了一个全局列表变量里调试，结果用户A登录测试的时候，系统回复里出现了用户B的订单信息。排查发现，session按用户区分这件事我在设计里做了，但调试时偷懒绕过了它。修复方案很简单：所有session对象key必须带上用户标识与会话标识的双重维度，绝对不能全局共享一个可变列表。这件事给我留下的教训是，即使一次临时调试，也不能绕过正式架构，否则会掩盖真实问题。

坑二：重复执行。 平台的重试机制加上我自己的异步队列重试策略叠加后，出现了一个诡异的现象：用户退单申请被执行了三次。第一次是平台重发消息，第二次是我自己的重试策略触发，第三次是队列消费完后我手动重新入队测试。到最后审计日志里三条执行记录，用户那边收到了三次结果推送。修复的核心是幂等设计：每个任务生成全局唯一的请求ID，执行层记录每个请求ID的处理状态，重复请求一律直接返回已处理结果。现在我的系统里任何Skill要具备"重复调用不会产生副作用"的能力。

坑三：任务超时后的状态不明确。 异步任务执行过程中，如果某个Skill调用的外部服务卡了超过一分钟，系统里任务状态还挂在"处理中"，用户问了一次进度，系统也只回复"正在处理"。后来我加了两个机制：任务队列配备超时监控，执行超过阈值的任务自动标记为"失败或降级"，同时触发重试或人工介入；对用户侧增加主动的进度回传，状态变化时推送一条简短通知。这就像操作系统里的看门狗，防止某个进程无声无息地卡死整个系统。

这三个坑其实指向同一个问题：复杂系统里最难的不是让单个组件工作，而是让组件在异常情况下还能保持整体稳定。稳定性不是上线后才考虑的问题，而要在第一个版本就把幂等、隔离、看门狗这些机制设计进去。

6. 进阶改造：事件驱动和任务队列带来的"AI操作系统感"

把基础版本跑通之后，我开始思考：怎么让这套系统更像一个真正可承载复杂业务的操作系统。3月30日晚上的加班改造，核心动作是引入事件总线和任务队列。

事件总线解决的是系统的解耦问题。之前各个模块之间直接互相调用，比如"意图分类服务"直接调用"路由服务"，路由直接调用Skill。一旦链路变长，任何一个环节改动都会波及上下游。有了事件总线后，所有模块只发布和订阅事件，互相之间没有直接依赖。典型的业务事件包括：消息到达、意图分类完成、任务开始执行、任务执行成功、任务失败、超时触发。每个事件都携带上下文快照，订阅方只处理自己关心的事件类型。

下表是我在实际实现中使用的事件与响应关系的对照：

任务队列负责的是削峰填谷和优先级管理。公众号消息的到达节奏很不均匀，早上可能一分钟一条，晚上活动期间可能一秒几十条。如果每个消息都立刻起线程跑到外部服务，很容易把后端依赖打爆。我的做法是：所有异步任务进入优先级队列，用户主动需求的任务优先级最高，普通后台任务排后；队列消费端有并发上限和限流策略，避免瞬时压力冲垮外部系统。

6.1 Skill的运行时隔离与权限分级

进阶改造里另一个重要模块是权限分级。公众号入口面对的是普通用户，但Skill里有一部分可能涉及敏感操作。我把Skill分成三个等级：普通技能（任何用户可触发）、用户验证技能（需要身份信息验证）、内部技能（只允许特定内部请求触发）。调度层在做技能匹配时，会先检查当前用户的会话属性是否满足权限要求，不满足的直接拒绝并提示。

这个设计对应到操作系统就是"用户态和内核态的分离"。如果不做权限隔离，一旦用户发现可以通过某种提示词让系统误触高级技能，整个系统就处于裸奔状态。我还加了行为审计：所有技能的调用记录都落日志，包含触发用户、Skill名称、输入输出的脱敏版本、触发时间，方便事后追溯。

6.2 这个架构后续还能怎么扩展

当我把事件总线、任务队列、Skill注册表、状态存储这套东西搭完，最直观的感受是：系统已经从"给我一个模型、我帮你接几个工具"进化到了"给你一套承载复杂业务的基础设施"。再往后扩展方向也很清晰：多个Agent之间可以互相调用对方的Skill，形成Agent间协作；Skill可以按租户维度隔离，做成多租户SaaS；长期记忆可以跟外部向量库打通，让每个用户拥有持久化的个人上下文。

这次3月30日实验做完之后，我对"AI操作系统"的理解从概念落到了一张架构图、一套代码和一堆踩坑经验。如果你也想做类似方向，我的建议是先别去研究太抽象的理论，就拿一个熟悉业务场景（比如公众号助手）搭一个最小系统，把Skill注册表、session隔离、异步任务、幂等去重这四个基础能力实现出来。这四个能力一旦跑稳，你再往上扩展任何复杂功能都会非常有底气。

最后分享一个我在反复调试中养成的习惯：每次加新功能之前，先回答三个问题——它有没有标准的输入输出契约？它失败时系统怎么感知？用户重复触发它会不会产生副作用？把这三个问题想清楚，复杂AI系统不会太难。