CodexQA

行业与实践工具与框架

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

CodexQA 团队阅读约 5 分钟

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

本文目录

多智能体

提示:

如果需要子智能体委派、持久记忆和结构化规划,先看 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)

觉得有用,转给同事

微信扫码

用微信扫一扫,在手机上打开后即可转发。

用 RSS 订阅

提交勘误