Solon TeamAgent 协作协议:不改成员,切换流水线与主管制

2026-07-21 38 预计阅读时间: 1 分钟
来源: oschina.net AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间:9 分钟

单个智能体解决的是“谁擅长做什么”,多智能体团队还必须回答三个更棘手的问题:下一步轮到谁、谁有权退回结果、满足什么条件才能停止。Solon AI 的 TeamAgent 将这些调度规则抽象为一等公民 TeamProtocol,使团队成员与协作方式分离。同一批 Agent 可以采用 SEQUENTIAL 组成线性流水线,也可以切换到 HIERARCHICAL,由主管分派、审核和决定是否继续。

协议不是调用顺序,而是团队控制面

把多个 Agent 依次写进一段业务代码并不难:

researcher -> writer -> reviewer

问题出现在任务失败、审核不通过或上下文膨胀之后。如果这些规则散落在业务逻辑里,增加一次返工就要修改多个 Agent,切换组织模式更接近重写工作流。

TeamProtocol 的价值在于把以下决策收敛到协议层:

  • 发言权转移:当前 Agent 完成后,下一位参与者是谁。
  • 任务路由:任务固定向后传递,还是由主管按状态动态分派。
  • 结果审核:审核意见只是附注,还是能触发返工。
  • 停止判定:达到目标、超过轮次、超时或失败时如何退出。
  • 状态传递:哪些上下文进入下一轮,哪些中间信息需要裁剪。

因此,协议应依赖稳定的成员能力接口,而不应知道每个 Agent 内部如何调用模型、检索知识或访问数据库。成员负责产出,协议负责组织。

SEQUENTIAL:确定性强的线性流水线

SEQUENTIAL 适合步骤明确、依赖关系稳定的任务。例如,先检索资料,再生成草稿,最后检查格式。它的优势是执行轨迹直观、延迟容易估算,也便于定位是哪一步引入了错误。

线性协议的边界同样明显:前序步骤一旦产出低质量结果,错误会继续向后传播;审核者即使发现问题,如果协议没有返工通道,也只能给出失败结论。生产环境不能只定义成员顺序,还要明确:

  • 某一步失败后是终止、跳过还是重试;
  • 重试是否使用相同输入与模型;
  • 审核不通过时能否回到指定成员;
  • 整条流水线的时间、轮次和 Token 预算。

对于高吞吐、低歧义任务,线性协议通常是更可控的起点。

HIERARCHICAL:主管掌握路由与停机权

HIERARCHICAL 不只是“在成员列表前增加一个 Manager”。主管需要读取团队状态,决定把任务交给谁,并根据审核结果选择结束、返工或换人。这种模式适合需要动态拆解、质量门禁或多轮修订的工作。

层级协议提供了更强的适应能力,但成本也更高:主管本身会消耗模型调用;错误的路由可能形成循环;如果停止条件模糊,团队会在“继续优化”中不断扩大上下文。主管提示词至少应约束以下输出:

你是团队主管。每轮只能执行一种动作:
1. delegate:选择一名成员并描述具体任务;
2. revise:根据审核意见要求指定成员修改;
3. finish:仅当验收条件全部满足时结束;
4. abort:超过预算或无法完成时停止。

禁止把任务委派给不存在的成员。
最多允许 3 轮返工。
输出必须包含 action、assignee、reason。

这类结构化决策比“请管理好团队”更容易校验、记录和恢复。

可以这样实践:用同一组成员切换协议

下面是一个可直接运行的教学模型,用来展示“成员不变,只替换协议”的设计。它不是 Solon TeamAgent 的真实 API;接入项目时,应将 SequentialProtocolHierarchicalProtocol 替换为实际的 TeamProtocol 实现,把三个处理函数替换为真实 Agent。

创建 team_demo.py

import os
from dataclasses import dataclass, field


@dataclass
class State:
    topic: str
    facts: list[str] = field(default_factory=list)
    draft: str = ""
    review: str = ""
    rounds: int = 0


def researcher(s: State) -> None:
    s.facts = [f"核心主题:{s.topic}", "协议负责路由、审核与停止"]


def writer(s: State) -> None:
    s.draft = ";".join(s.facts)
    if s.review:
        s.draft += ";已根据审核意见补充停止条件"


def reviewer(s: State) -> bool:
    passed = "停止条件" in s.draft
    s.review = "通过" if passed else "缺少明确的停止条件"
    return passed


class SequentialProtocol:
    def run(self, s: State) -> State:
        researcher(s)
        writer(s)
        reviewer(s)
        return s


class HierarchicalProtocol:
    def run(self, s: State) -> State:
        researcher(s)
        writer(s)
        while s.rounds < 2:
            s.rounds += 1
            if reviewer(s):
                break
            writer(s)  # 主管根据审核意见将任务退回给 writer
        return s


protocols = {
    "sequential": SequentialProtocol(),
    "hierarchical": HierarchicalProtocol(),
}

name = os.getenv("PROTOCOL", "sequential").lower()
if name not in protocols:
    raise SystemExit(f"unknown protocol: {name}")

result = protocols[name].run(State("TeamAgent 协作协议"))
print(f"protocol={name}")
print(f"rounds={result.rounds}")
print(f"review={result.review}")
print(f"draft={result.draft}")

分别运行两种模式:

PROTOCOL=sequential python team_demo.py
PROTOCOL=hierarchical python team_demo.py

在这个模型中,researcherwriterreviewer 没有因协议切换而改变。差异集中在控制流:线性模式执行一次即退出;层级模式允许主管依据审核结果发起返工,并受到最大轮次限制。

实际接入时还可以把协议选择外置为配置,但不要允许未审核的请求参数任意切换高成本协议:

team:
  protocol: HIERARCHICAL
  limits:
    max_rounds: 3
    timeout_seconds: 90
    max_model_calls: 12
  completion:
    require_reviewer_approval: true

上线前检查协议,而不只是检查提示词

团队式 Agent 的故障往往发生在 Agent 之间。准备采用 TeamAgent 时,可以用以下清单审视协议边界:

  • 每种状态是否都有明确的下一位执行者;
  • 审核失败是否对应确定的返工目标;
  • 是否同时设置最大轮次、超时和模型调用预算;
  • 协议切换后,成员输入输出契约是否保持一致;
  • 是否记录每次路由选择、审核结果和停止原因;
  • 进程重启后,团队状态能否恢复或安全终止;
  • 对外部写操作是否设置幂等键、权限检查和人工确认。

选择协议时,不必默认层级越复杂越好。步骤稳定、成本敏感的任务优先使用 SEQUENTIAL;只有当动态分派、返工和质量门禁确实能提高结果时,再采用 HIERARCHICAL。真正可维护的团队系统,不是拥有最多 Agent,而是能用清晰协议约束谁行动、谁负责,以及何时停止。


相关推荐