工作流与多智能体：路由、并行和评审

Yui和Kai在五条工作流路径间协作，展示链式、路由、并行、编排与评审AI 生成配图

A6 的 Agent 循环让模型自己决定下一步。但很多任务的步骤其实是已知的：先分类再处理、先写提纲再写正文、写完再检查。这时与其让模型自由发挥，不如由你的代码把几次模型调用串起来——这就是工作流。它比 Agent 更好预测、更便宜，出了问题也更容易定位。

下面五种模式都基于同一个小工具函数：调一次模型，可选地要求 JSON Schema，并记录每一步。

python

import json, logging, os, time
from openai import OpenAI

client = OpenAI(base_url="https://hivegpt.cn/v1", api_key=os.environ["HIVEGPT_API_KEY"])
log = logging.getLogger("workflow")

def ask(system, user, schema=None, step=""):
    kwargs = {"response_format": {"type": "json_schema", "json_schema": schema}} if schema else {}
    t0 = time.time()
    resp = client.chat.completions.create(
        model="gpt-5.5",
        messages=[{"role": "system", "content": system}, {"role": "user", "content": user}],
        **kwargs,
    )
    text = resp.choices[0].message.content
    log.info("step=%s tokens=%s ms=%d out=%.200s", step, resp.usage.total_tokens, (time.time() - t0) * 1000, text)
    return json.loads(text) if schema else text

1. 链式调用：固定步骤，中间设关卡

适合：任务能拆成固定的几步，每一步都比整件事简单。比如「写提纲 → 写正文 → 压缩长度」「抽取信息 → 翻译 → 排版」。

关键在于步骤之间的检查：上一步不合格就停下或重做，不要把错误一路传下去。

python

def write_article(topic):
    outline = ask("为主题列 3–5 个小标题，每行一个，不要别的文字。", topic, step="outline")
    heads = [line for line in outline.splitlines() if line.strip()]
    if not 3 <= len(heads) <= 5:                      # 关卡：提纲不合格，后面就不用写了
        raise ValueError(f"提纲不合格：{outline}")
    draft = ask("按提纲写一篇 600 字以内的介绍。", f"主题：{topic}\n提纲：\n{outline}", step="draft")
    if len(draft) > 900:                              # 能用代码判断的，就别再问模型
        draft = ask("把下面的文字压缩到 600 字以内，信息不要丢。", draft, step="shorten")
    return draft

2. 路由：先分类，再交给专门的处理

适合：输入分成明显不同的几类，每类需要不同的提示词、工具或模型。比如客服消息分成查订单、退款、开发票、投诉；简单问题交给便宜的模型，复杂问题交给强一点的模型。

把所有情况塞进一个大提示词，规则会互相干扰；先分类，每个处理函数只管一件事，提示词短、好测、好改。

python

HANDLERS = {
    "order_status": handle_order,     # 查物流：调订单接口
    "refund": handle_refund,          # 退款：走退款流程
    "invoice": handle_invoice,
    "complaint": handle_human,        # 投诉：直接转人工
}

def route(message):
    r = ask(ROUTER_SYSTEM, message, ROUTER_SCHEMA, step="route")
    if r["needs_human"]:
        return handle_human(message, r)
    handler = HANDLERS.get(r["intent"], handle_human)   # 分发由代码做；没见过的意图走兜底
    return handler(message, r)

模型只负责分类和抽字段，真正调用哪个函数由你的代码决定。分类结果用 enum 限定，代码里就不会出现拼错的意图名。

3. 并行：互不依赖的子任务同时做

适合两种情况：

分工：同一份输入从几个角度分别处理，比如代码评审分成安全、性能、可读性三个检查，每个提示词只关注一项，比一个提示词全包更仔细。
投票：同一个问题问几次取多数，用在判断类任务上（是否违规、是否是广告），减少偶然误判。

python

import asyncio
from openai import AsyncOpenAI

aclient = AsyncOpenAI(base_url="https://hivegpt.cn/v1", api_key=os.environ["HIVEGPT_API_KEY"])
limit = asyncio.Semaphore(4)                    # 控制并发，别撞上 RPM 限制

async def aask(system, user):
    async with limit:
        resp = await aclient.chat.completions.create(
            model="gpt-5.5",
            messages=[{"role": "system", "content": system}, {"role": "user", "content": user}],
        )
        return resp.choices[0].message.content

CHECKS = {
    "安全": "只检查安全问题：注入、越权、密钥写在代码里。没有问题就回答「无」。",
    "性能": "只检查性能问题：循环里查数据库、重复计算、没有分页。没有问题就回答「无」。",
    "可读性": "只检查命名和结构。没有问题就回答「无」。",
}

async def review_code(code):
    results = await asyncio.gather(*(aask(p, code) for p in CHECKS.values()))
    return dict(zip(CHECKS, results))

async def is_ad(text, n=3):
    votes = await asyncio.gather(*(aask("这条评论是不是广告？只回答「是」或「否」。", text) for _ in range(n)))
    return sum(v.strip().startswith("是") for v in votes) * 2 > n

并行省的是时间，不省钱：三个检查就是三次调用的费用。

4. 编排者–执行者：先规划，再分派

适合：子任务事先不知道，要看输入才能决定。比如「给这个需求写技术方案」，有的需求要拆成接口、数据表、前端三部分，有的只需要一部分。和并行的区别是：子任务清单由模型生成，而不是写死在代码里。

python

PLAN_SCHEMA = {
    "name": "plan", "strict": True,
    "schema": {
        "type": "object",
        "properties": {"tasks": {"type": "array", "items": {
            "type": "object",
            "properties": {"title": {"type": "string"}, "instruction": {"type": "string"}},
            "required": ["title", "instruction"], "additionalProperties": False}}},
        "required": ["tasks"], "additionalProperties": False,
    },
}

async def orchestrate(request):
    plan = ask("把需求拆成 2–5 个能独立完成的子任务。", request, PLAN_SCHEMA, step="plan")
    tasks = plan["tasks"][:5]                         # 数量在代码里再限一次
    outputs = await asyncio.gather(*(
        aask("完成分配给你的子任务，只输出结果。", f"总需求：{request}\n你的子任务：{t['instruction']}")
        for t in tasks
    ))
    merged = "\n\n".join(f"## {t['title']}\n{o}" for t, o in zip(tasks, outputs))
    return ask("把各部分整合成一份完整的文档，去掉重复，统一术语。", merged, step="merge")

执行者看不到彼此的结果，所以末尾的合并步骤要负责去重和统一口径。

5. 评审–优化：写稿、按标准挑错、修改

适合：有明确的评价标准，而且「指出问题」比「一次写对」容易。比如营销文案要满足字数、必含信息、禁用词；翻译要符合术语表。

python

REVIEW_SCHEMA = {
    "name": "review", "strict": True,
    "schema": {
        "type": "object",
        "properties": {"passed": {"type": "boolean"}, "problems": {"type": "array", "items": {"type": "string"}}},
        "required": ["passed", "problems"], "additionalProperties": False,
    },
}
CRITERIA = "1. 写明价格和活动日期；2. 语气友好，不催促；3. 不用「最」「第一」「顶级」等绝对化用语。"

def write_with_review(brief, max_rounds=3):
    draft = ask("根据要点写一段活动文案。", brief, step="draft")
    for i in range(max_rounds):
        if len(draft) > 200:                          # 字数这类规则用代码查，又准又省
            problems = [f"超过 200 字（现在 {len(draft)} 字）"]
        else:
            review = ask(f"按下面的标准逐条检查文案，只列出不符合的地方：\n{CRITERIA}", draft, REVIEW_SCHEMA, step=f"review{i}")
            if review["passed"]:
                return draft
            problems = review["problems"]
        draft = ask("根据问题改写文案，只输出新文案。", f"原文：\n{draft}\n\n问题：\n" + "\n".join(problems), step=f"revise{i}")
    log.warning("评审 %d 轮仍未通过，交给人工", max_rounds)
    return draft

两个要点：一定要有轮数上限，否则可能一直改不过；标准要写具体，「写得更好一点」这种标准，评审永远能挑出毛病。

「多智能体」是什么

所谓多智能体，通常就是上面这些模式，只是每个角色有自己的系统提示词和工具：路由员只会分类，退款专员只能调退款接口，评审员只读不写。拆开的好处是每个角色的提示词短、权限小、可以单独测试；代价也很实在：

调用更多：每多一个角色、每多一轮评审，就多一次费用。
更慢：串行的步骤延迟会叠加。
更难排查：结果不对时，要弄清是哪一步、哪个角色出的错。

所以顺序应该是：一次调用能解决就用一次；不够再加一步链式检查；确实需要时才引入路由、并行或评审；需要临场决定步骤时才用 Agent。另外两条习惯能省很多排查时间：

步骤之间传结构化数据：用 JSON Schema 输出，下一步拿字段，而不是从上一段自然语言里再猜一遍。
每一步都记日志：步骤名、输入摘要、输出、token、耗时，并带上同一个请求 ID，出问题时能把整条链路串起来。

动手：客服消息路由

下面是一个路由器：系统提示词说明了客服应用的五种意图和对应的处理函数，模型返回分类、紧急程度、是否转人工和抽出的字段。换几条消息试试，比如「发票抬头开错了怎么改」「我登录不上了，验证码一直收不到」。

▶ 动手试试

系统提示词（这次请求一起发送，点开查看）

你是网店客服系统的路由器，只做分类和抽取，不回答用户。意图和处理函数：order_status 查订单和物流 → order_lookup；refund 退货退款 → refund_flow；invoice 发票 → invoice_service；account 登录、账号、密码 → account_support；complaint 投诉或其他无法归类 → human_agent。用户情绪激烈、明确要投诉、或涉及金额纠纷时 needs_human 为 true。urgency：影响使用或用户很着急为 high，一般问题为 normal，咨询类为 low。字段没提到就填 null。reason 用一句中文说明判断依据。

登录后运行登录 HiveGPT 后每天有免费运行次数

示例输出（之前运行的结果）

{
  "intent": "refund",
  "handler": "refund_flow",
  "urgency": "high",
  "needs_human": true,
  "fields": { "order_id": "A20261003118", "product": "蓝牙耳机" },
  "reason": "用户要求退货退款，问题重复出现且表示要投诉，情绪激烈，需要人工跟进。"
}

拿到这个结果后，代码按 needs_human 和 handler 分发；order_id 直接交给退款流程去查库，而不必再让模型读一遍原话。

小结

步骤已知就用工作流：链式加关卡、路由分类分发、并行分工或投票、编排者动态拆任务、评审按标准改到通过为止（设轮数上限）。
多智能体就是给每个角色单独的提示词和工具；每加一个角色，调用次数、延迟和排查难度都会上去。
先用够用的简单流程；步骤间传 JSON，能用代码检查的不交给模型，每一步都记日志。

下一课：工作流和 Agent 会读网页、文档和工具结果，这些内容里可能藏着指令。「安全护栏」 讲怎么防提示注入、越权和不安全的输出。

工作流与多智能体：路由、并行和评审

1. 链式调用：固定步骤，中间设关卡 ​

2. 路由：先分类，再交给专门的处理 ​

3. 并行：互不依赖的子任务同时做 ​

4. 编排者–执行者：先规划，再分派 ​

5. 评审–优化：写稿、按标准挑错、修改 ​

「多智能体」是什么 ​

动手：客服消息路由 ​

小结 ​

1. 链式调用：固定步骤，中间设关卡

2. 路由：先分类，再交给专门的处理

3. 并行：互不依赖的子任务同时做

4. 编排者–执行者：先规划，再分派

5. 评审–优化：写稿、按标准挑错、修改

「多智能体」是什么

动手：客服消息路由

小结