第四章教程 · 把方向盘交给模型，然后装上刹车

第一部分

交权四步

同一个任务——「做一份研究计划，找到《银河系漫游指南》」——写四遍。每一遍你都比上一遍少管一点事。

第一步：一个 agent，三把工具，你全权指挥

对应清单 4.1

顺序写死在提示词里模型只是照做

这一版就是第三章的自然延伸：一个 agent 同时连三个 MCP server。注意提示词里那句「先…然后…最后…」——执行顺序是你用大白话写死的。

import asyncio
import os
from pathlib import Path

from agents import Agent, Runner, set_default_openai_api, set_tracing_disabled
from agents.mcp import MCPServerStdio, MCPServerStdioParams
from dotenv import load_dotenv

# 仓库原文没有这三行，是为了接第三方模型加的
load_dotenv()
set_default_openai_api("chat_completions")
set_tracing_disabled(True)

SANDBOX = os.path.dirname(os.path.abspath(__file__))
SCRIPT = Path(__file__).with_name("01_research_tools_mcp_server.py").resolve()


async def main():
    # 先把三个 server 都实例化出来……
    servers = [
        MCPServerStdio(
            name="Research Tools",
            params=MCPServerStdioParams(
                command="mcp",
                args=["run", str(SCRIPT)],
            ),
        ),
        MCPServerStdio(
            name="sequential-thinking",
            params={
                "command": "npx",
                "args": ["-y", "@modelcontextprotocol/server-sequential-thinking"],
            },
        ),
        MCPServerStdio(
            name="filesystem",
            params={
                "command": "npx",
                "args": ["-y", "@modelcontextprotocol/server-filesystem", SANDBOX],
            },
        ),
    ]

    instructions = """
你是一个研究助手，可以用工具来做研究和规划。
给你一个研究目标后，先用研究工具找到研究来源。
然后用 sequential thinking 工具来规划研究。
最后用文件系统工具把研究计划写成一个文本文件。
    """

    # ……然后一次性把它们全打开
    async with (
        servers[0] as research_srv,
        servers[1] as thinking_srv,
        servers[2] as fs_srv,
    ):
        agent = Agent(
            name="Assistant",
            instructions=instructions,
            mcp_servers=[research_srv, thinking_srv, fs_srv],
        )
        goal = """
做一份研究计划，用来找到《银河系漫游指南》这本书
"""
        print("运行中……", goal)
        result = await Runner.run(agent, goal)
        print(result.final_output)


if __name__ == "__main__":
    asyncio.run(main())

async with (a as x, b as y, c as z) 这个写法是 Python 3.10+ 的括号版多上下文管理器——三个子进程一起开、一起关。这也是为什么这本书要求 3.11+。

这一版的问题，正是后面三步的动机

一个 agent 拿着三个 server 的全部工具，意味着它的工具清单很长、提示词要交代所有步骤。工具一多，模型挑错的概率就上升；提示词一长，它就开始漏步骤。这就是为什么下一步要拆。

第二步：拆成三个 agent，但队还是你排的

对应清单 4.2 · 书里叫「智能体流」

顺序写死在 Python 里每个 agent 只管自己那段

三个 agent，各有各的提示词、各连各的 server。但关键在最后那三行—— 顺序是你用 Python 代码排的，一个的输出手动喂给下一个：

    # 先把三个 agent 都实例化出来……
    research_agent = Agent(
        name="Research Agent",          # ← 名字必须英文，中文会撞名
        instructions="""
你是一个研究助手。
你的职责是找到研究来源。
""",
    )
    thinking_agent = Agent(
        name="Thinking Agent",
        instructions="""
你是一个研究助手。
你的职责是规划研究。
""",
    )
    filesystem_agent = Agent(
        name="Filesystem Agent",
        instructions="""
你是一个研究助手。
你的职责是把研究计划写成一个文本文件。
""",
    )

    # …… server 部分和上一步完全一样，此处省略 ……

    async with (
        servers[0] as research_srv,
        servers[1] as thinking_srv,
        servers[2] as fs_srv,
    ):
        goal = """
做一份研究计划，用来找到《银河系漫游指南》这本书
"""
        print("运行中……", goal)
        # 这就是「流」：你亲手把上一个的输出喂给下一个
        research_agent.mcp_servers = [research_srv]
        result = await Runner.run(research_agent, goal)
        thinking_agent.mcp_servers = [thinking_srv]
        result = await Runner.run(thinking_agent, result.final_output)
        filesystem_agent.mcp_servers = [fs_srv]
        result = await Runner.run(filesystem_agent, result.final_output)
        print(result.final_output)

三点值得说：

提示词短了一大截。每个 agent 只需要知道自己那一件事，不用再背「先…然后…最后…」。工具清单也短了——一个 agent 只看得见一个 server。 这就是拆分的全部好处。
research_agent.mcp_servers = [...] 是建完之后再塞进去的。 因为 server 必须在 async with 里才活着，而 agent 在外面就建好了。这个「先建 agent、进了上下文再挂 server」的写法，本章反复出现。
此刻还没有任何「多智能体」的魔法。就是三次普通的 Runner.run，用 Python 变量传结果。你完全掌控流程，也完全没享受到自动化。

第三步：给它一个会随机翻脸的工具

对应清单 4.3 / 4.4 · 决策

你只给条件模型自己判断怎么办

这一步的道具是个小而坏的 server：它每次返回 0 到 3 个来源，随机的。

import random

from mcp.server.fastmcp import FastMCP

# 建一个 MCP 服务器
mcp = FastMCP("Research Tools")


@mcp.tool()
def get_research_sources() -> list[str]:
    """随机提供 0 到 3 个研究来源。"""
    search_sources = [
        "Wikipedia",
        "Google",
        "YouTube",
    ]
    num_sources = random.randint(0, 3)
    if num_sources == 0:
        return []
    return random.sample(search_sources, num_sources)

这个随机是故意的，别去修

和第二章那个「故意跑崩的 04」一样，这里的随机是教学设计。它逼着你面对一个真实问题：工具返回空列表时，agent 该怎么办？ 编几个来源出来？（幻觉）还是老实说找不到？

副作用是这一章的示例结果不可复现——同样的代码跑两次，行为可能完全不同。你多跑几次才能看全所有分支。这也正好解释了后面为什么需要护栏： 你已经没法预测模型会遇到什么了。

配套的 03_agent_to_agent_decisions.py 在提示词里加了「绝不要编造研究来源」这类约束，让 agent 自己决定拿到空列表时怎么处理。 注意此刻控制权已经过半了：你不再规定动作，只规定原则。

第四步：移交——你连队都不排了

对应清单 4.5 · 这一章的分水岭

你只声明谁能转给谁模型决定何时转、转给谁

对比第二步：那三行 Runner.run 没了，只剩一行—— 启动第一个 agent，然后剩下的它们自己商量。

    class ResearchSourcesModel(BaseModel):
        research_sources: List[str]
        """研究要用到的来源列表。"""

    research_agent = Agent(
        name="Research Agent",
        output_type=ResearchSourcesModel,
        instructions="""
你是一个研究助手。
你的职责是找到研究来源。
不要编造或虚构任何研究来源。
永远移交给 thinking agent。
""",
    )
    thinking_agent = Agent(
        name="Thinking Agent",
        instructions="""
你是一个研究规划助手。
你的职责是规划研究。
你会从 research agent 那里收到一个研究来源列表。
用 sequentialThinking 工具，基于这些来源做出研究计划。
永远移交给 filesystem agent。
""",
    )
    filesystem_agent = Agent(
        name="Filesystem Agent",
        instructions="""
你是一个文件系统助手。
你的职责是把输出写成一个文本文件。
绝不要编造或虚构任何输出。
""",
    )

    # …… server 部分同前，省略 ……

    async with (
        servers[0] as research_srv,
        servers[1] as thinking_srv,
        servers[2] as fs_srv,
    ):
        goal = """
做一份研究计划，用来找到《银河系漫游指南》这本书
"""
        # 声明结构：谁连哪个 server，谁能移交给谁
        research_agent.mcp_servers = [research_srv]
        research_agent.handoffs = [thinking_agent]
        thinking_agent.mcp_servers = [thinking_srv]
        thinking_agent.handoffs = [filesystem_agent]
        filesystem_agent.mcp_servers = [fs_srv]

        print("运行中……", goal)
        # 只启动第一个，后面的转交由模型自己完成
        result = await Runner.run(
            research_agent,
            goal,
            max_turns=25,
        )
        print(result.final_output)

移交到底是什么？说穿了很朴素：handoffs=[thinking_agent] 会让 SDK 给 research_agent 偷偷加一个工具，名字叫 transfer_to_thinking_agent。模型「移交」的动作，其实就是调用了这个工具。没有任何魔法——移交就是一个自动生成的工具调用。

想通这一点，前面那条警告就顺理成章了：工具名从智能体名字拼出来，所以中文名会拼出 transfer_to______，三个 agent 撞成一个。

max_turns=25 是什么，为什么突然需要它

因为你已经不知道要跑几轮了。前面手排队时，就是三次调用，数得清清楚楚。现在模型自己决定转几次、转给谁——万一它在两个 agent 之间反复横跳呢？ max_turns 就是那根保险丝，跑够 25 轮还没结束就抛异常。

这是护栏思想的第一次出现，比正式讲 guardrails 早了一整节。交权的代价开始显现了。

第二部分

看见它在干什么

交权之后最难受的是看不见。这两节讲怎么把流程画出来、怎么在移交发生时插一脚。

把 agent 流画成图

对应清单 4.6

05_visualizing_agent_flows.py 用 draw_graph() 把「谁能移交给谁、谁挂了哪些工具」画成一张关系图。核心就一句：

from agents.extensions.visualization import draw_graph

# …… 照常把 agent 和 handoffs 配好之后 ……

draw_graph(research_agent).view()
input("按回车继续……")

这个文件有三个反直觉的地方

一、它根本不运行 agent。整个文件从头到尾没有 Runner.run—— 它只是把静态结构画出来。别指望看到执行过程。

二、它会卡住。.view() 会调起系统的图片查看器弹窗，后面还跟着 input() 等你敲回车。在自动化脚本里跑它会永远挂着。

三、要装 graphviz。不是 pip 装的那个 Python 包，是系统级的 graphviz （brew install graphviz / apt install graphviz）。 requirements.txt 里的 openai-agents[viz] 只装了 Python 那半边。

在移交发生的那一刻插一脚

对应清单 4.7 · 文件名拼错了

前面的 handoffs=[thinking_agent] 是简写。完整写法能让你挂个回调，在移交真正发生时被叫醒：

from agents import handoff

async def research_handoff(ctx, input_data: ResearchSourcesModel):
    print(f"移交发生了，带过去的数据：{input_data}")

research_agent.handoffs = [
    handoff(
        agent=thinking_agent,
        on_handoff=research_handoff,          # 移交时调这个
        input_type=ResearchSourcesModel,      # 移交时必须带这个结构的数据
    )
]

input_type 是这里的重点：它强制模型在移交时交出一份结构化数据，而不是随便甩一段话过去。这等于给两个 agent 之间的接口加了类型—— 和第二章 output_type 的思路一模一样，只是用在了 agent 之间。

顺带一提这个文件名：06_agent_to_agent_monitroing_handoffs.py—— 是 monitroing，不是 monitoring。仓库里就这么拼的，别打错。

第三部分

兜住它

权交出去了，现在得防着点。护栏（guardrail）就是在输入进模型之前、输出离开模型之后，各架一道闸门。

输入护栏和输出护栏

对应清单 4.8

模型还是自己决定但出口有闸门了

护栏是两个装饰器加一个返回值。返回值里的 tripwire_triggered 是绊线开关——设成 True，整个运行立刻抛异常中止。

class ResearchOutputModel(BaseModel):
    """研究智能体的输出模型。"""

    research_plan: str
    """最终的研究计划文本。"""
    research_plan_file: str
    """研究计划文件的路径。"""


@input_guardrail
async def research_guardrail(
    ctx: RunContextWrapper[None], agent: Agent, input: str | list[TResponseInputItem]
) -> GuardrailFunctionOutput:
    forbidden_research = "《银河系漫游指南》"
    if forbidden_research in input:
        research_forbidden = True
    else:
        research_forbidden = False

    return GuardrailFunctionOutput(
        output_info=f"用户问的是：{input}",
        tripwire_triggered=research_forbidden,
    )


@output_guardrail
async def research_output_guardrail(
    ctx: RunContextWrapper, agent: Agent, output: ResearchOutputModel
) -> GuardrailFunctionOutput:
    if len(output.research_plan) < 100:
        insufficient_research = True
    else:
        insufficient_research = False

    return GuardrailFunctionOutput(
        # 仓库原文这里少了个 f 前缀，是个 bug —— 见下方说明
        output_info=f"研究计划长度：{len(output.research_plan)}",
        tripwire_triggered=insufficient_research,
    )

挂上去，然后用 try/except 接住绊线：

        agent = Agent(
            name="Assistant",
            instructions=instructions,
            mcp_servers=[research_srv, thinking_srv, fs_srv],
            output_type=ResearchOutputModel,
            input_guardrails=[research_guardrail],
            output_guardrails=[research_output_guardrail],
        )
        goal = """
做一份研究计划，用来找到《银河系漫游指南》这本书
"""
        try:
            print("运行中……", goal)
            result = await Runner.run(agent, goal)
            print(result.final_output)
        except InputGuardrailTripwireTriggered as input_tripped:
            print(f"""
输入护栏绊线触发：
{input_tripped.guardrail_result.output.output_info}
""")
        except OutputGuardrailTripwireTriggered as output_tripped:
            print(f"""
输出护栏绊线触发：
{output_tripped.guardrail_result.output.output_info}
""")
            print("Done")

仓库这里有个真 bug：少了个 f

仓库原文第 58 行是这么写的：

output_info="research plan length: {len(output.research_plan)}"

少了 f 前缀，所以它打印出来的是花括号本身，而不是真实长度。我验证过，输出就是字面量 research plan length: {len(output.research_plan)}。上面那版我补上了 f。这个 bug 不影响护栏工作 （绊线判断用的是另一个变量），只是让你在排查时看不到真实数字—— 而护栏的信息恰恰就是给排查用的。

翻译这一段时踩到的真问题

输入护栏干的事是 if forbidden_research in input—— 一个朴素的子串匹配。原文里禁的是英文书名，而我把 goal 翻成了中文《银河系漫游指南》，所以禁词也必须跟着翻，两边一字不差才拦得住。少改一处，护栏就静默失效——不报错，只是永远不触发。

这暴露了这类护栏的本质弱点：它拦的是字符串，不是意图。用户写「银河系漫游指南」（不带书名号）就绕过去了。书里 4.4.2 接着就用「拿另一个 agent 当护栏」来解决这个问题—— 让模型去判断意图，而不是靠 in。

剩下三个护栏例子

对应清单 4.9 / 4.10 / 4.11

这三个是同一套思路的加码，代码结构和上面高度重复，就不整段贴了，说清楚区别：

08_agent_guardrails.py（清单 4.9）——拿 agent 当护栏。 护栏函数里跑一个专门的小 agent 去判断「这个请求该不该拦」，而不是用 in 硬匹配。这解决了上面那个「拦字符串不拦意图」的毛病，代价是每次请求多烧一次模型调用。
09_agent_passoff_guardrails.py（清单 4.10）——给移交流程加护栏。 前面的护栏都挂在单个 agent 上；这个演示的是多个 agent 互相移交时，护栏该挂在哪。（仓库里还有个 09x_agent_handoff_guardrails.py，书里没提，是同一主题的变体。）
10_agent__guardrails_retry.py（清单 4.11）——绊线之后重试。 前面的护栏一触发就抛异常、整个流程结束。这个演示接住异常、改一改、再跑一次—— 毕竟真实系统里你不能一有问题就死给用户看。 注意文件名是双下划线 10_agent__guardrails_retry.py，书里写的是单下划线，照着敲会找不到文件。

番外

书里没有的第 12 个文件

11_agent_orchestration_tools.py 不在任何清单里，但它演示的模式在后面几章会反复出现，值得单独看一眼。

把 agent 包成工具：另一种交权方式

仓库有、书里没有 · 对应第 1 章说的「中心辐射式编排」

总控 agent 始终在位但它自己挑用谁

移交是接力棒——转出去就不回来了。这个文件演示的是另一种玩法： 把 agent 包进 @function_tool 里，当普通工具用。主 agent 调完之后控制权还在自己手上，可以接着调下一个。

@function_tool
async def research_agent(instructions: str) -> ResearchSourcesModel:
    """
    用研究智能体来查找研究来源。
    """
    agent = Agent(
        name="Research Agent",
        instructions="""
你是一个研究助手。
你的职责是找到研究来源。
绝不要编造或虚构任何研究来源。
""",
        output_type=ResearchSourcesModel,
        mcp_servers=[research_srv],
    )
    async with research_srv:
        result = await Runner.run(agent, instructions)
        return result.final_output


orchestration_agent = Agent(
    name="Orchestration Agent",
    instructions="""
你是一个研究规划与编排助手。
你的职责是规划研究、找到已有的研究成果并更新它。
用 research agent 来查找研究来源。
用 sequentialThinking 工具，基于这些来源做出研究计划。
用 filesystem agent 来查找已有研究并更新。
用 filesystem agent 把输出写成一个文本文件。
""",
    tools=[research_agent, filesystem_agent],   # ← agent 当工具用
)

移交 vs. agent 当工具，怎么选？一句话： 要不要回来。流水线式的、转出去就不管了——用移交。需要一个总控来回调度、汇总结果——用 agent 当工具。书里第 1 章讲的「装配线」和「中心辐射」，对应的正是这两种。

这个文件有个真实隐患

research_srv 是个模块级的全局单例，而 async with research_srv: 写在了工具函数里面。意味着模型每调一次这个工具，就对同一个 server 对象重新进出一次上下文。

而模型完全可能连着调它两次——这正是「agent 当工具」的卖点。 同一个实例反复 connect / cleanup，是个货真价实的坑。 真要这么写，server 的生命周期应该提到外面去管，像本章其他文件那样在 main() 里 async with 一次。

把方向盘交给模型，
然后装上刹车

动手之前