多智能体协作的秩序：任务系统、状态机与工程实践

1. 多智能体系统最缺的不是聪明的模型而是一套秩序先说个直观感受。现在单智能体聊天、单链路工具调用已经不算什么新鲜事了真正容易翻车的是把多个智能体放到一起干活。你会发现每个智能体单独都挺聪明可一旦让它们自由沟通场面很快就会失控A 在等 B 的结论B 又觉得 C 给的信息不完整C 反过来质疑 A 的假设整场对话陷入一种“礼貌但毫无产出”的循环。跑了几轮之后你甚至不知道系统到底推进了哪个目标、哪个步骤卡住了、谁该为此负责。OpenHarness 这个名字很容易让人误解成“又一个大模型框架”但我更愿意把它理解为多智能体协作与任务系统的连接层。它的核心不是某个模型而是一整套把智能体组织起来的基础设施你定义角色、划分任务、规定输入输出格式、跟踪每一步状态、处理失败重试、保留审计痕迹。换句话说它给一群本来各说各话的智能体安装了一套团队协作的秩序。这篇内容适合谁如果你正准备把多个智能体放进真实业务而不是停留在“两个 Agent 互相聊天好酷”的演示阶段如果你已经跑通了最简单的 ReAct 循环但发现在多步骤、多角色、需要持久化任务状态时无从下手如果你需要给多智能体系统加监控、加错误恢复、加并发调度那这篇文章应该能帮你把整体思路捋顺。我会从架构选型、任务模型、上下文传递、实际编码、调参、排障这几个维度展开全程以我自己的落地经验为线索来聊。2. 协作架构的设计让任务而不是模型成为主角2.1 三种主流协作范式多智能体怎么协作业界目前基本跑不出三种范式。第一种是中心化编排也叫 orchestrator-worker。一个调度者负责任务拆分、分配、汇总执行者干具体活。团队里只要有一个灵魂项目经理它拆任务、分活、收结果其他成员按指令执行。这种模式最稳定也最好观测但瓶颈集中在调度器本身如果调度器的规划能力不足整个系统都会拖后腿。第二种是对等自由协作所有智能体地位平等它们直接彼此对话、讨论、投票、交付结果。说白了就是会议室里没人主持大家自由发言。好处是适合头脑风暴、方案评审这类任务能产生一些中心化编排想不到的差异化视角。坏处也很明显容易发散、容易重复讨论、难以收敛而且整个过程的 token 消耗像流水一样哗哗地走。第三种是分层递归分解。顶层智能体只负责把复杂目标拆成子目标子目标再派给下一级智能体或子协调器逐层深入最后结果逐层汇总回来。这种模式很适合全局规划比如制定一个大型项目的交付计划或者生成一份覆盖多个领域的综合报告。代价是架构复杂度成倍上涨每一层都要考虑状态衔接和上下文传递调度开销明显高于前两种。没有绝对的好坏只有适不适合。你追求的如果是可交付、可审计、可运维那中心化编排几乎是最稳妥的起点如果只是探索创意的早期原型对等协作可能带来惊喜。OpenHarness 这类强调“任务系统”的框架本质上就是默认选择偏中心化的路线因为任务本身需要一个明确的责任人、明确的状态和明确的完成定义而自由协作很难给出这些。2.2 为什么我会坚持“以任务为中心”而不是“以对话为中心”很多人在做多智能体 demo 的时候习惯把智能体之间的关系建模成“对话”A 说一句话B 回一句话A 再回一句…… 这种建模方式在真实系统里是灾难。最简单的例子你让分析智能体产出一份季度营收数据汇总让审核智能体检查这份汇总让写作智能体据此生成报告。如果采用对话式建模分析智能体的“所有思考过程”都会留在聊天记录里审核智能体看到的是一堆对话而不是一份结构化的“待审核文档”写作智能体想要调用最初的数据源还得从对话历史里反推。最要命的是任何一个智能体回复里带了千分位逗号或者小数点格式不统一后面的链路全都会出错。所以我做多智能体系统时会刻意引入一个概念任务就是协作的基本单位对话框只是智能体工作时的“草稿纸”严格定义的输入输出结构才是交接棒。OpenHarness 那一套“任务系统”之所以在协作里这么关键就是因为它扮演了一个类似“看板 交接协议”的角色。看板上每一张卡片代表一个任务卡片写着谁负责、当前什么状态、依赖谁、结果放哪交接协议则规定了卡片里的数据结构长什么样。智能体不需要理解整个项目它只需要看懂自己这张卡片做好、提交然后拿下一张。这样一来协作就从“两个人面对面聊不清楚”变成了“流水线上每个工位各干各的活”效率和可控性完全不是一个量级。2.3 任务系统是团队协作的“契约层”有一点经常被忽略任务系统不只是队列它其实是智能体之间的契约层。契约里写死了输入字段、输出格式、验收标准。智能体的自由度被限制在这个契约范围内这一点恰恰是它稳定运行的前提。打个比方你让一个实习生去整理会议纪要如果他自由发挥可能写成小说体、诗歌体也可能漏掉关键决定。你给他一张表单表头是“会议时间/参会人/讨论议题/最终结论/待办事项”他按表填结果就稳定了。多智能体系统里的任务结构就是那张表单。我在设计任务对象时最重视的从来不是队列性能而是输入输出 schema 是否清晰、状态流转是否明确。后面第三节我会详细拆任务模型的字段设计这里先建立起“任务即契约”的概念。3. 任务系统的核心模型与状态机3.1 一个最基本的任务对象长什么样我见过不少半路出家做智能体框架的开发者任务对象就两个字段一个 prompt一个 response。这种模型应付单轮调用没问题一旦进入多智能体协作立刻寸步难行。你不清楚这个任务是哪个父任务拆出来的不知道它有没有依赖别的任务不知道它重试了几次更不知道它现在卡在哪个环节。我自己项目里长期使用的任务模型大致长这样from enum import Enum from dataclasses import dataclass, field from typing import Optional, List, Dict, Any class TaskState(str, Enum): PENDING pending # 已创建等待依赖完成 READY ready # 依赖已满足可被调度 RUNNING running # 执行中 COMPLETED completed # 成功结束 FAILED failed # 失败且不再重试 TIMEOUT timeout # 超时 BLOCKED blocked # 阻塞等待外部输入 CANCELLED cancelled # 被取消 dataclass class Task: task_id: str task_type: str payload: Dict[str, Any] state: TaskState TaskState.PENDING assignee_agent: Optional[str] None parent_task_id: Optional[str] None dependencies: List[str] field(default_factorylist) result: Optional[Dict[str, Any]] None error_info: Optional[Dict[str, Any]] None retry_count: int 0 max_retries: int 3 timeout_seconds: int 120 created_at: str updated_at: str attempt_token: str 每个字段的存在都有它的理由我挑几个容易忽略的说。parent_task_id用来建立任务树。一个大任务被拆成几十个子任务执行到一半失败了你得能顺着树往上追溯定位是哪个子任务导致整条链路受阻。没有这个字段排查故障时只能靠猜。dependencies是调度核心。它表示这个任务要等哪些任务先完成。严格来说我只让它存依赖任务的 ID不直接存结果结果通过任务 ID 去存储层取。这样依赖关系更加清晰也方便做 DAG 调度。assignee_agent表示这个任务派给哪个智能体是后面 Agent Registry 的键值。我不喜欢把智能体写死在业务代码里而是通过任务类型和智能体能力列表动态匹配这样加一个新人或者替换某个模型不需要改主流程。attempt_token是幂等标记。多智能体系统里任务重试是家常便饭但重试绝不能导致外部副作用重复执行。这个 token 会随任务下发执行器每次调用外部接口或者变更业务状态时都必须带着它做去重。后面排查问题那节我会展开聊。3.2 状态机设计与流转逻辑多智能体任务的状态机看起来简单但真做起来容易搞出一堆隐藏 bug。我最终沉淀下来的流转规则如下PENDING表示创建成功但还没达到执行条件。它的依赖任务列表不为空且至少有一个依赖没完成。调度器周期性扫描所有 PENDING 任务检查依赖状态全部完成就转移到READY。READY任务可以被 worker 拉取。这里有个细节多个 worker 并发拉取同一个任务必须靠数据库行的条件更新或者 Redis 原子指令来抢避免一个任务被两个智能体同时执行。用得比较多的手段是“UPDATE tasks SET stateRUNNING, assignee_agent? WHERE task_id? AND stateREADY”如果影响行数为 0说明被别人抢走了就跳过。RUNNING不是终点。执行过程中可能发生超时、模型输出解析失败、依赖的外部 API 挂了。超时我会单独标记成TIMEOUT和普通FAILED区分开因为超时后的重试策略通常更激进一些比如立刻重试一次而不是退避等待。如果失败且重试次数用尽才标记为FAILED。我见过有人为了图省事把超时也直接塞进FAILED结果排查问题时无法区分“模型回复慢导致超时”和“模型生成的结果业务校验不过导致失败”这是两种本质不同的失败原因混在一起会让告警失真。最终状态我设计为四种成功、失败、超时、取消。BLOCKED是一个中间态表示任务在执行过程中发现需要人工介入比如某个关键决策不能让模型自己拍板。系统把任务标记为 BLOCKED并持久化地等待一个外部回调把任务唤醒。3.3 任务分解从大目标到可执行步骤任务系统光有状态机还不够还得解决“一个任务到底怎么拆”。OpenHarness 场景下最典型的做法是引入一个或者多个规划智能体Planner Agent它专门负责分解目标。我记得第一次把复杂任务交给规划智能体时给出的子任务粒度完全是随机的。有些子任务拆得特别碎连“读取文件并输出文件大小”都要单独成一个任务结果调度开销比执行开销还大有些子任务又拆得太大一个子任务里混杂了检索、总结、写报告三个职责执行智能体在单次上下文里根本干不完输出质量明显下降。后来我总结出一个经验法则子任务的粒度应该控制在“单个智能体在一个上下文窗口内可以独立完成一个完整交付物”的级别。比如写行业分析报告不要拆成“做行业分析”这种大任务也不要拆到“查一家公司的官网”这种琐碎级别而是拆成“数据检索与汇总”“趋势分析与洞察”“报告初稿撰写”“事实核查与修订”四个环节。前一个环节的输出正好是后一个环节的输入每个环节都能对应一个明确的交付物。任务之间的依赖关系大部分情况下是一个有向无环图不是简单的一串列表。比如“起草商业计划书”依赖“市场调研报告”和“财务测算模型”两个任务同时完成然后才能进入下一步。所以我坚持用 DAG 来表达依赖关系而不是用一个线性队列。3.4 上下文管理协作里最难啃的骨头如果说任务模型是骨架那上下文管理就是血肉。多智能体系统跑着跑着效果变差十有八九是上下文出了问题。我处理上下文的原则是只把完成当前任务需要的那一小块上下文传给执行智能体绝不传递整个历史。比如一个报告生成任务PDF 源文件可能有几十页我不会把 PDF 全文塞进提示词而是先做检索把相关的几个段落和引用信息作为上下文传入再让模型基于这些材料写报告。中间产物也不会跟着任务一起传递而是落到一个独立的存储里后续任务通过引用 ID 去取。更具体一点我在传给下游任务的上下文中通常是“任务背景描述 上游任务的结构化结果 本次任务输出格式要求”三部分。上游任务的结构化结果只取它真正需要的字段不需要的全丢掉。这么做一方面能显著降低 token 开销另一方面也避免了模型面对海量信息时“迷失在长文本里”的通病。上下文传递还有一个容易忽略的点要给每个任务定义“一句话背景”。因为下游执行智能体可能并不具备全局视角它不知道这个任务在整个流程中扮演什么角色。给它一段简洁的背景描述能明显减少模型因为“不理解为什么做这件事”而产生的废话输出。这个背景可以由规划智能体在分解任务时自动写入也可以由人工模板定制。4. 实践落地构建 OpenHarness 任务系统的具体做法4.1 一个最小可用的模块切分如果让我给一个从零开始的团队提供最小的架构推荐我不会一上来就上一堆微服务和消息中间件。太久了我发现架构首先要做到的是“职责清晰”其次才是追求规模和性能。最小可用的模块是这样API/交互入口接收外部请求把一个业务目标转成一条根任务。规划器Planner把根任务拆成 DAG 形式的子任务集。调度器Scheduler周期性扫描任务把 READY 状态的任务分发给可用的执行 worker。执行器Agent Executor包装具体的智能体调用负责填充提示词、调用大模型、解析输出、校验 schema、上报状态。任务存储Task Store持久化任务状态和结果是系统的唯一事实来源。事件流/日志记录状态变更、错误信息、token 消耗等用于观测。模块之间不建议直接调用对方内部的数据库而是通过明确的任务状态迁移来联动。调度器只负责“把 READY 变成 RUNNING”执行器只负责“把 RUNNING 变成 COMPLETED/FAILED/TIMEOUT”规划器只负责“创建新的子任务”。边界清楚之后后面加监控、加人工审核、加并行扩容都会轻松很多。4.2 状态存储选型从内存到数据库再到事件溯源我见过非常多的原型项目任务状态直接放在一个 Python 字典里。demo 阶段没问题但真实使用中只要进程一重启所有任务状态全部丢失那些正在 RUNNING 的任务永远卡死这肯定不能接受。稍微认真一点的做法是用 Redis 存状态。读取快、过期策略灵活适合单机或者少量 worker 的场景。但 Redis 有一个天然的缺点很难做复杂的任务查询和条件更新尤其是“把所有依赖已完成且状态为 PENDING 的任务捞出来”这种关系型操作写起来很别扭。我最终选择的方案是 PostgreSQL 作为任务存储外加 Redis 做轻量级缓存和分布式锁。PostgreSQL 的处理方式是把任务表做成一张宽表把 state、dependencies、assignee_agent 等都放进去配合条件更新实现状态转移。这样查询、审计、扩展性都不错。如果你的系统规模已经大到单库扛不住可以考虑把任务表按 task_id hash 分片或者引入事件溯源机制每次任务状态变更都写一条事件记录当前状态由事件流聚合得出。这样调试和回放都很方便但是开发和维护成本高不少。我个人建议在没到“千万级任务量”之前老老实实用数据库宽表就好。4.3 核心代码调度器与执行器的骨架下面我给出一个非常精简但完整的调度器骨架帮助理解状态是怎么流转的。import asyncio from typing import Dict class Scheduler: def __init__(self, store, agent_registry, planner): self.store store self.agent_registry agent_registry self.planner planner self.running_tasks: Dict[str, asyncio.Task] {} self.max_concurrency 8 async def run_forever(self): while True: ready_tasks self.store.fetch_ready_tasks(limit10) for task in ready_tasks: # 条件更新抢任务 if not self.store.claim_task(task.task_id, worker_idscheduler-1): continue if len(self.running_tasks) self.max_concurrency: # 释放任务等下一轮 self.store.release_task(task.task_id, worker_idscheduler-1) continue task.state RUNNING fut asyncio.create_task(self._execute(task)) self.running_tasks[task.task_id] fut await asyncio.sleep(1) async def _execute(self, task): try: agent self.agent_registry.get_agent(task.assignee_agent) result await asyncio.wait_for( agent.run(task), timeouttask.timeout_seconds ) self.store.complete_task(task.task_id, result) except asyncio.TimeoutError: self.store.handle_timeout(task.task_id) except Exception as exc: self.store.handle_failure(task.task_id, error_messagestr(exc)) finally: self.running_tasks.pop(task.task_id, None)这里面有两个容易被忽略的点。第一fetch_ready_tasks要做成“只捞 READY 状态”不能把 PENDING 任务捞出来。因为 PENDING 的依赖可能还没完成提前执行必然拿不到上游结果。调度器要有一个单独的逻辑持续检查 PENDING 任务的依赖是否全部完成一旦满足就更新为 READY。第二claim_task那条条件更新很关键是分布式下防重复执行的唯一有效手段。如果直接拉了任务列表就开始执行两个调度器实例并发时就会重复执行同一个任务轻则浪费模型调用费重则导致业务数据错乱。执行器的骨架长这样class AgentExecutor: def __init__(self, model_client, prompt_builder, schema_validator): self.model_client model_client self.prompt_builder prompt_builder self.schema_validator schema_validator async def run(self, task: Task) - Dict: # 1. 构建任务上下文 context self.prompt_builder.build_context(task) prompt self.prompt_builder.build_prompt( roletask.assignee_agent, task_typetask.task_type, contextcontext, output_schemaOUTPUT_SCHEMAS[task.task_type] ) # 2. 调用模型 raw_output await self.model_client.complete( promptprompt, temperature0, response_format{type: json_object}, attempt_tokentask.attempt_token ) # 3. 校验输出格式 parsed self.schema_validator.parse_and_validate( raw_output, OUTPUT_SCHEMAS[task.task_type] ) if not parsed.is_valid: raise SchemaValidationError(parsed.error_message) return parsed.data执行器的核心思想是它不关心任务的业务含义只负责“把任务输进去把符合 schema 的结果输出来”。业务逻辑全部放在 task_type 对应的 prompt 模板和输出 schema 里这样新增一种任务类型不需要动执行器代码。4.4 关键参数怎么定多智能体任务系统跑起来后有几个参数是需要反复调的。我把它们整理成了一个速查表。参数建议初始值设置理由单任务超时时间60-120 秒太短容易误杀复杂任务太长会拖垮调度吞吐失败重试次数2-3 次超过 3 次还失败通常不是偶发问题继续重试只会烧钱模型 temperature0-0.2代码、数据提取、格式输出必须低温度保证确定性和稳定性单次上下文字数上限模型窗口的 60%-70%留出足够的输出空间避免模型因为“没地方写了”而截断最大并发执行数4-8取决于模型 API 的限流配额和业务对延迟的容忍度任务拆分粒度单个交付物/每个子任务拆太碎调度开销大拆太大单模型窗口装不下调参要结合具体场景没有万能答案。比如你用的是超长窗口模型上下文上限可以拉到 80% 甚至 90%如果你调用的外部 API 经常抖动重试次数可以增加到 5 次但要把重试退避时间拉长。总的原则是先求稳定再求速度。宁可让一个任务慢一点完成也不要因为并发过高导致集体超时。5. 常见问题与排查习惯实录5.1 智能体陷入“假推进”的循环我最初做多智能体协作时遇到最扎心的问题是任务状态一直显示 RUNNING但系统实际上没干任何有用的事。两个智能体在一个细枝末节上反复交换意见谁也没有往下推进。模型根本没有“我要完成任务”的意识它只是顺着对话惯性继续回应对方的话。破解这个问题的核心思路是把“讨论”和“交付”分开。如果任务本身就是“生成一份草案”那么执行模型输出的必须是“草案”而不是“对上一句话的回应”。如果任务需要多轮评审那就把每一轮评审也定义成独立任务比如“一稿评审”完成后必须产生“评审意见结果”评审不通过就进入“修订”任务通过就进入“终稿确认”任务。让状态机来强制推进而不是指望模型自觉。另一个有效手段是给每个任务设定最大回复轮次。无论双方讨论得多热闹达到轮次上限后强制进入决策环节必须输出一个明确结论。这个限制一般不会影响质量反而能逼模型收敛。5.2 上下文越来越长效果越来越差长上下文是最近两年被反复吹捧的能力但实际用起来根本不是百毒不灵。模型面对超长上下文时往往忽略中段信息只关注开头和结尾。在多智能体链路里这个毛病因为“上游结果逐级累积”而被无限放大。第三个智能体拿到的可能是前两个智能体的完整输出历史几千个字里真正有用的只有两行。解决手段就是前面提过的“按需传递”“不要传历史只传结果”。每个任务从上游任务的结果里精确取用需要的字段。如果下游需要的数据形态和上游输出不一致就通过一个轻量转换任务来做字段映射而不是指望下游智能体“理解上下文后自己提取”。我自己还会在提示词里专门写一句“你只依赖上下文中明确的引用数据和结构化字段不要依赖未出现的背景假设。”这句话看起来朴素却能显著减少模型胡编乱造的概率。5.3 模型输出格式不稳定JSON 偶尔爆炸做多智能体系统模型输出格式稳定性是决定系统可靠性的关键。绕不开的一个问题是模型偶尔会返回不合法 JSON或者字段类型不对、多一个逗号、少一个右括号。我的做法分两层。第一层尽量用模型服务商提供的 structured output 或者 function calling 能力从源头把合法格式的概率拉高。第二层在代码里实现一个专门的“输出修复器”负责把不合法的模型输出修复成合法 JSON。这个修复器不用太复杂找常见的修补模式即可补全缺失的右括号、把单引号替换成双引号、去掉末尾的逗号等等。如果修复器也修复不了就带着“上次 schema 校验失败的详细错误信息”让模型重试一次。大多数情况下模型看到具体的 schema 错误后会自己纠正。这里要注意重试时不要简单重复原 prompt要把错误信息拼进去并且语气明确地要求“严格按照给定 JSON schema 输出”。5.4 任务中途挂了怎么恢复分布式系统跑得越久各种状态越容易出现“僵尸任务”进程启动时看到一堆 RUNNING 但实际已经执行失败的任务或者 worker 崩了之后任务永久卡死。这种情况必须靠“恢复机制”兜底。我在执行器启动时会主动扫描那些状态为 RUNNING 但已经超过任务超时时间上限还未更新的任务统一将它们标记为 FAILED 或者重新入队。同时在任务状态更新时加上expects_version或updated_at条件避免旧的执行结果覆盖新状态。“幂等”是我在这里反复强调的另一个关键词。任务重试不可避免关键是不能因为重试造成外部系统重复操作。比如一个任务要求“生成订单并通知用户”重试时绝不能生成两张订单、发两次通知。我一般在任务里带一个业务侧唯一的幂等键外部系统根据幂等键决定是否创建新记录。这个习惯一旦养成处理重试逻辑会省非常多心。5.5 没有链路追踪出问题就是无头悬案最后一个坑更多是工程习惯问题。多智能体系统和传统微服务一样必须有可观测性体系。如果只打几个 print 日志任务出错时你根本不知道问题出在规划、调度还是执行环节。我在项目里会为每条根任务生成一个全局 trace_id并把它透传到所有子任务、所有智能体调用、所有外部 API 请求中。任务状态的每次变化都会记录一条事件包含谁触发的、从什么状态迁到什么状态、耗时多少、token 消耗多少。排障时直接按 trace_id 拉出整条链路的时间线一眼就能看出卡点和瓶颈在哪里。对小型项目哪怕刚开始不上完整链路追踪产品也至少要保证日志里能按 request_id 把所有子任务串联起来。这一个动作能帮你省掉八成的排查时间。6. 把多智能体系统做稳定靠的是工程纪律说句大实话把多个智能体组装到一起并不难真正难的是让这套系统长期稳定、可维护、出了问题能快速定位。我做了几个项目之后最大的体会是系统稳定性的上限不取决于最强的那个模型而取决于你的任务契约是否清晰、状态流转是否严谨、幂等是否兜底、可观测是否到位。这几个点全部做到位即使模型偶尔犯浑系统也能通过重试和人工干预把它纠回来。我早期犯过不少错比如让智能体自由交流太久结果把时间和额度全花在无意义的往返上比如任务结果不做 schema 校验下游拿到脏数据最后整条链路崩掉比如完全不考虑上下文裁剪跑半个小时后提示词已经膨胀到连模型自己都分不清重点。后来老老实实把任务系统当成一个分布式系统来做状态机、依赖 DAG、幂等键、超时重试、链路追踪全部配上系统才真正变得可信。如果你现在正准备搭多智能体系统我的建议很简单别急着让两个最聪明的模型互相对话先把任务模型定义清楚把状态机画明白把一个任务从创建到成功完成的整条链路跑通再逐步加协作复杂度。这条路上你会踩不少坑但只要你守住“任务即契约”这个基本原则大多数问题都能在早期被化解掉。