CodexQA

Industry & PracticeTools & Frameworks

Inspect 多智能体：工作流、工具调用和交接怎么选

CodexQA 团队5 min read

英国 AI Security Institute 的 Inspect 文档：多智能体可以是交接、显式工作流，或把智能体当工具。建议先用 react 基线，再比多智能体。说明 run 不回写求解器状态，以及交接过滤。

In this piece

多智能体

提示：

如果需要子智能体委派、持久记忆和结构化规划，先看 Deep Agent。它开箱提供这些能力，不必自己接线。

概述

用 Inspect 的 Agent 协议实现多智能体系统，有几种办法。多智能体（multi-agent）是多个智能体分工，而不是一个循环包办全部步骤。

  1. 可以给顶层主管智能体交接能力，交给各自擅长不同任务的子智能体。交接（handoff）是把对话交给另一个智能体，它能看见历史并往上追加消息。
  1. 可以做一个智能体工作流，按阶段显式调用各个智能体。
  1. 可以把智能体做成模型的普通工具调用。

下面分别举例。

方法

探索多智能体结构时要记住：它们常常并不比简单的 react() 智能体更好。ReAct 是「推理一步、调用工具、再看结果」的循环。因此建议按下面的方法开发智能体：

  1. 先用基线 react() 智能体，这样才能量出各种改动是否真的提高了成绩。
  1. 再优化环境（任务定义）、工具选择和提示，以及智能体的系统提示。
  1. 可选：试验多智能体设计，并拿它们和你先前优化过的更简单智能体比。

Anthropic 的文章 Building Effective Agents 和后续视频 How We Build Effective Agents 强调了这些点，适合用来建立开发方法上的直觉。

工作流

用交接和工具做多智能体，是尽量让模型自己规划和路由。有时更适合显式编排。例如，许多深度研究智能体把规划、搜索和写作做成明确步骤。

可以用 run() 按预定或动态的顺序显式调用智能体。假设已经为研究流水线的各阶段写好了智能体，可以这样组成一个研究智能体：

from inspect_ai.agent import Agent, AgentState, agent, run
from inspect_ai.model import ChatMessageSystem

from research_pipeline import (
    research_planner, research_searcher, research_writer
)

@agent
def researcher() -> Agent:

    async def execute(state: AgentState) -> AgentState:
        """Research assistant."""

        state.messages.append(
            ChatMessageSystem(content="You are an expert researcher.")
        )

        state = await run(research_planner(), state)
        state = await run(research_searcher(), state)
        state = await run(research_writer(), state)

        return state

工作流里不一定总像上面那样把整个状态传入并赋回。也可以做更窄的查询，再用结果决定下一步。还可以并行执行某些步骤。例如：

from asyncio import gather

plans = await gather(
    run(web_search_planner(), state),
    run(experiment_planner(), state)
)

run() 会复制输入，因此适合像上面这样并行（两次并行运行不会对 state 做共享的、互相冲突的修改）。

在求解器里运行智能体

上面的复制行为也意味着 run() 不会把智能体的对话写回它的输入。它返回一个新的 AgentState，由调用方决定怎么用。这在从求解器里调用 run() 时尤其重要：如果丢掉返回的状态，智能体的对话和输出不会出现在样本的消息里（不过仍能在样本记录里看到），读取 state.output 或 state.messages 的评分器也看不到智能体的工作。把需要的字段拷回 TaskState：

@solver
def research_solver() -> Solver:
    async def solve(state: TaskState, generate: Generate) -> TaskState:
        # per-sample setup
        ...
        # run the agent, then reflect its conversation and
        # output back into the task state
        agent_state = await run(researcher(), state.messages)
        state.messages = agent_state.messages
        state.output = agent_state.output
        return state
    return solve

如果不需要在智能体周围写每个样本自己的逻辑，用 as_solver() 把智能体转成会自动更新 TaskState 的求解器。

工具

可以把智能体做成普通工具调用。这时智能体只看到一个输入字符串，并返回它最后一条助手消息的输出。

例如，下面的主管智能体把 web_surfer 做成工具：

from inspect_ai.agent import as_tool, react
from inspect_ai.dataset import Sample
from inspect_ai.tool import web_search
from math_tools import addition

web_surfer = react(
    name="web_surfer",
    description="Web research assistant",
    prompt="You are a tenacious web researcher that is expert "
           + "at using a web browser to answer questions.",
    tools=[web_search()]   
)

supervisor = react(
    prompt="You are an agent that can answer addition " 
            + "problems and do web research.",
    tools=[addition(), as_tool(web_surfer)]
)

交接

交接让主管智能体把工作委托给其他智能体。交接和工具调用不同：被交接的智能体既能看见对话历史，也能往历史里追加消息。

交接会自动以带 transfer_to 前缀的工具调用呈现给模型（例如 transfer_to_web_surfer），并提示模型：它处在一个可以把工作委托给其他智能体的多智能体系统里。

用 handoff() 包住一个智能体来创建交接。这些智能体往往是带专门提示和工具集的简单 react() 智能体。例如，创建一个可以交接过去的 web_surfer()：

from inspect_ai.agent react
from inspect_ai.tool import web_search

web_surfer = react(
    name="web_surfer",
    description="Web research assistant",
    prompt="You are a tenacious web researcher that is expert "
           + "at using a web browser to answer questions.",
    tools=[web_search()]
)

注意：

调用 react() 创建 web_surfer 时传入了 name 和 description。把 react 智能体用于交接时，这两个参数是必需的，主管模型才知道它的名字和能力。

然后可以创建一个主管智能体，它既能用普通工具，也能交接给网页浏览智能体。这里主管是标准的 react() 智能体，其他监督方式也可以。

from inspect_ai.agent import handoff
from inspect_ai.dataset import Sample
from math_tools import addition

supervisor = react(
    prompt="You are an agent that can answer addition " 
            + "problems and do web research.",
    tools=[addition(), handoff(web_surfer)]
)

task = Task(
    dataset=[
        Sample(input="Please add 1+1 then tell me what " 
                     + "movies were popular in 2020")
    ],
    solver=supervisor,
    sandbox="docker",
)

supervisor 既能用常规的 addition() 工具，也能 handoff() 给 web_surfer。网页浏览智能体有自己的 react 循环；因为是被交接过去的，它既能看到完整消息历史，也能把自己的消息追加进去。

交接过滤器

默认发生交接时：

  1. 目标智能体看到全局消息历史（系统消息除外）。
  1. 交接产生的消息用 content_only() 过滤器处理。它去掉系统消息和推理痕迹，并把工具调用转成文本，以免父模型看到它不知道来源的内容（例如推理或工具调用）而混淆。

可以传入另一个内置交接过滤器，或自己写过滤器。例如，用内置的 remove_tools 输入过滤器，从呈现给智能体的历史里去掉全部工具调用（有时必须这样，否则智能体会搞不清有哪些工具可用）：

from inspect_ai.agent import remove_tools

handoff(web_surfer, input_filter=remove_tools)

也可以用内置的 last_message 输出过滤器，只把智能体历史的最后一条消息追加到全局对话：

from inspect_ai.agent import last_message

handoff(web_surfer, output_filter=last_message)

不限于内置过滤器。input_filter 或 output_filter 都可以传函数，例如：

async def my_filter(messages: list[ChatMessage]) -> list[ChatMessage]:
    # filter messages however you need to...
    return messages

handoff(web_surfer, output_filter=my_filter)

UK AI Security Institute，Multi Agent，文档页无单独发布日期（检索于 2026-10-06），https://inspect.aisi.org.uk/multi-agent.html ，MIT License（Copyright (c) 2024 UK AI Security Institute）

Found it useful? Pass it on

WeChat

Scan with WeChat to open it on your phone and forward it.

Subscribe via RSS

Submit a correction

Keep reading

← All pieces