Agent 开发入门｜最小版的智能体Supervisor

大家好，我是想风！

如果上篇你还有点懵，那么看完 这篇你应该能理解至少一半了！

上一篇 Agent 开发入门｜手写一个 Claude Code，机制很多循环就一个 摊开了 20 集的地图。

这一篇动手：用其中三集——s01 循环、s06 子智能体、s15 团队——从零写一个最小 Supervisor。

Supervisor 就是主管——自己不干活，只派活和验收。88 行，装一个 SDK （软件开发工具包）就能跑。

目录：

1. Supervisor 是什么：主管不干活，只派活和验收

2. 动手：88 行写个最小 Supervisor

3. 跑起来

4. 常见坑

5. 往上加什么

一、Supervisor 是什么：主管不干活，只派活和验收

先补一句 S01 就讲了一个循环：

问模型 → 要工具就执行、把结果塞回去、再问 → 不要工具，退出。

本文的主管也是这样，每个手下也是，用的是同一个函数。

先说结论：

Supervisor 就是一个只有「派活」这一个工具的智能体。

它碰不到文件，跑不了命令。

能做的只有三件事：拆任务、派给手下、看汇报决定下一步。

这个形状，系列里有两集已经讲到一半：

所以本文的 Supervisor = s06 的派活方式 + s15 的 Lead 角色，砍掉并行和收件箱。

s06 里有个比方：子智能体就像开一个新终端去追一件事，追完关掉，结论写进笔记。

主管拿到的只有那条笔记，不是整个终端的滚动记录。

回传整个对话列表，等于没隔离。

手下的每一条命令输出都会塞进主管的上下文，派不了几次活就装不下了。

二、动手：88 行写个最小 Supervisor

要什么：

• Python 3.10 以上（最新版 anthropic SDK 要求 3.10）

• 一个 Anthropic API Key，在 console.anthropic.com 申请

• 一个空目录（worker 会在里面真的写文件、跑命令）

下面的命令按 Mac 写，一行一行贴进终端：

python3 --version                     # 看版本，要 3.10 以上
mkdir sv-demo && cd sv-demo           # 空目录，worker 只在这里折腾
python3 -m venv .venv                 # 建一个独立环境，装的包不弄乱系统
source .venv/bin/activate             # 进环境，提示符前面会多出 (.venv)
pip install anthropic                 # 官方 SDK
export ANTHROPIC_API_KEY=sk-ant-xxx   # 换成你的 key
echo $ANTHROPIC_API_KEY               # 能打印出 key，说明设上了

.venv 是隐藏目录，ls 看不到，不影响 worker 干活。

export 只对当前这个终端窗口有效。

关掉窗口再开，要重新 cd sv-demo、source .venv/bin/activate、export 一遍。

走中转或者国内兼容服务的，再加两行：

export ANTHROPIC_BASE_URL=对方给的地址   # SDK 会自己读
export MODEL_ID=对方的模型名            # 代码默认是 claude-opus-5，对方不一定认

地址和模型名去哪找：learn-claude-code 仓库的 .env.example 里列了 GLM、Kimi、DeepSeek、MiniMax 几家。

兼容服务在工具调用的细节上可能和官方有出入。想要最稳，用官方的。

然后建文件，推荐用 VS Code：新建一个文件，存到 sv-demo 目录，文件名 supervisor.py。

用 Mac 自带的「文本编辑」也行，先按 ⇧⌘T 转成纯文本再存，不然存出来是富文本，Python 读不了。

代码如下，整段粘进去（需修改密钥 deepseek为例）：

import os, subprocess
from anthropic import Anthropic

#client = Anthropic()  # 自动读环境变量 ANTHROPIC_API_KEY
client = Anthropic(base_url="https://api.deepseek.com/anthropic", auth_token="这里你填写你自己的API KEY")

MODEL = os.getenv("MODEL_ID", "deepseek-flash") #deepseek-flash 模型 如果你是其他平台，需换上其他平台对应模型

# ① 手下有哪些人：名字 → 岗位说明（就是 worker 的 system prompt）
WORKERS = {
    "coder": "你是程序员。用 bash 写代码、跑代码。做完用一两句话汇报：改了哪些文件、跑出什么结果。",
    "tester": "你是测试。用 bash 运行和检查代码，只报告通过还是失败、失败在哪。不要改代码。",
}

# ② 两个工具：worker 用 bash 干活，主管用 delegate 派活
BASH = {
    "name": "bash",
    "description": "在当前目录执行一条 shell 命令，返回输出",
    "input_schema": {"type": "object",
                     "properties": {"command": {"type": "string"}},
                     "required": ["command"]},
}
DELEGATE = {
    "name": "delegate",
    "description": "把一个子任务派给某个 worker，返回它的汇报",
    "input_schema": {"type": "object",
                     "properties": {"worker": {"type": "string", "enum": list(WORKERS)},
                                    "task": {"type": "string"}},
                     "required": ["worker", "task"]},
}

DANGER = ["rm -rf", "sudo", "shutdown", "mkfs"]


def run_bash(command):  # 权限门（s03）+ 超时
    if any(d in command for d in DANGER):
        return "拒绝执行：命中危险命令"
    try:
        r = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=60)
    except subprocess.TimeoutExpired:
        return "超时：60 秒没跑完"
    return (r.stdout + r.stderr)[-5000:] or "(没有输出)"


# ③ 循环（s01）：主管和 worker 共用这一个
def agent_loop(system, messages, tools, handlers, max_turns):
    for _ in range(max_turns):
        resp = client.messages.create(model=MODEL, max_tokens=16000, system=system,
                                      messages=messages, tools=tools)
        messages.append({"role": "assistant", "content": resp.content})
        if resp.stop_reason != "tool_use":  # 不再要工具，说明干完了
            return "".join(b.text for b in resp.content if b.type == "text")
        results = []
        for block in resp.content:
            if block.type != "tool_use":
                continue
            try:
                output, err = handlers[block.name](**block.input), False
            except Exception as e:  # 工具报错也要回给模型，不能让循环崩
                output, err = f"工具出错：{e}", True
            results.append({"type": "tool_result", "tool_use_id": block.id,
                            "content": output, "is_error": err})
        messages.append({"role": "user", "content": results})
    return "达到轮数上限，没做完"


# ④ 派活（s06）：全新 messages，只回结论；worker 没有 delegate，不能再往下派
def run_worker(worker, task):
    print(f"\n[派活] {worker}: {task}")
    result = agent_loop(WORKERS[worker], [{"role": "user", "content": task}],
                        [BASH], {"bash": run_bash}, max_turns=20)
    result = result or "(worker 没留话)"
    print(f"[汇报] {worker}: {result}")
    return result


# ⑤ 主管：只有 delegate 一个工具，自己碰不到文件
SUPERVISOR = """你是主管，手下有两个 worker：coder（写代码）、tester（测代码）。
你自己不碰文件，只能用 delegate 派活，一次派一件。
拿到汇报要验收：tester 说没通过，就把失败原因转给 coder 重做。
全部通过后，用三句话向用户交差。"""


def supervise(goal):
    return agent_loop(SUPERVISOR, [{"role": "user", "content": goal}],
                      [DELEGATE], {"delegate": run_worker}, max_turns=10)


if __name__ == "__main__":
    print("\n[交差]", supervise(input("给主管的目标 > ")))

一共五块，对着系列看：

① WORKERS

手下就是一段岗位说明。加一个人，加一行。

② 两个工具

对应 s02。工具 = 一份说明书（给模型看）+ 一个函数（给代码跑）。

handlers 这个字典，就是把工具名和真正要跑的函数对应起来。

③ agent_loop

就是 s01 那个循环，多了一个 max_turns 上限。

主管和 worker 都调它，只是传进去的岗位说明、工具不一样。

④ run_worker

s06 的三个要点都在这：

• 每次派活，都是全新的 messages

• 只把最后一句文字结论回给主管

• worker 的工具里没有 delegate，不会一层套一层往下派

⑤ SUPERVISOR

主管的工具只有 delegate。

不给它 bash，它就只能派活。

整份代码最关键的，是 run_worker 里那个 [BASH]——它决定了谁能干活、谁只能管人。

三、跑起来

先确认三件事：人在 sv-demo 目录里，提示符前有 (.venv)，ls 能看到 supervisor.py。

python supervisor.py   # 在 sv-demo 目录里跑

中途想停，按 Ctrl + C。

提示符出来后，给一个有「写」有「测」的目标：

给主管的目标 > 写一个 fizzbuzz.py，打印 1 到 15，再让 tester 验证输出对不对

fizzbuzz.py 脚本解释：

FizzBuzz 就是一个口头报数游戏，现在则是计算机编程领域最经典的初级面试编程题之一

游戏与题目规则

从 1 开始依次往下数数（通常数到 100）：

• 当数字是 3 的倍数时，不读数字，改读 "Fizz"；

• 当数字是 5 的倍数时，不读数字，改读 "Buzz"；

• 当数字同时是 3 和 5 的倍数（即 15 的倍数）时，读 "FizzBuzz"；

• 如果以上条件都不满足，就直接读出这个数字本身。

输出示例

从 1 数到 15 的部分结果如下： 1, 2, Fizz, 4, Buzz, Fizz, 7, 8, Fizz, Buzz, 11, Fizz, 13, 14, FizzBuzz

屏幕上会交替出现两种行：

• [派活] coder: ……——主管把活拆好，派了出去

• [汇报] coder: ……——worker 干完回的那一两句话

最后一行 [交差]，是主管给你的总结。

每次跑，拆法和措辞都不一样，这是正常的。

要看的是流程：先派 coder，再派 tester，tester 说没过就再派回 coder。

跑完 ls 一下，fizzbuzz.py 会真的在目录里。

想看「验收打回」这一步，就故意给个容易写错的要求。比如「15 的倍数打印 FizzBuzz，但 30 要打印 Thirty」。

本人运行文件输出效果如下：

一次运行，背后发生了什么

对着代码走一遍：

1. 你输入目标，supervise() 把它当成主管对话的第一句

2. 主管问模型。模型手里只有 delegate，就回一个派活请求：派给 coder，附上一段任务说明

3. 代码执行 run_worker，屏幕打出 [派活]

4. coder 拿到一段全新的对话，里面只有那句任务说明。它在自己的循环里调 bash：写文件、跑文件、看输出

5. coder 不再要 bash，最后那段文字就是汇报。屏幕打出 [汇报]，这段话塞回主管的对话

6. 主管再问模型。这次多半派给 tester，流程同上

7. tester 说没过，主管把失败原因写进新的 task，再派给 coder

8. 主管不再调 delegate，最后那段文字就是 [交差]

主管从头到尾只看到三样东西：你的目标、它自己派出去的 task、worker 的汇报。

worker 跑过的每条命令、每段输出，主管一条都看不到。

跑通之后，先做这三件事

第一件：看 worker 到底敲了哪些命令。

在 run_bash 函数的第一行加一句：

    print("   $", command)   # 先把命令打出来，被拦的也打

再跑一次，[派活] 和 [汇报] 中间会多出 worker 敲过的每条命令。

第二件：加一个手下。

在 WORKERS 里加一行：

    "reviewer": "你是代码审查。用 bash 读代码，指出可读性和边界情况的问题。不要改代码。",

delegate 能派的 worker 列表（代码里那个 enum），是从 WORKERS 自动生成的，加人不用改它。

但 SUPERVISOR 第一句写着「手下有两个 worker」，要一起改成三个，写上 reviewer 是干嘛的。不改，主管多半想不起来派给它。

第三件：换个目标再跑。

比如「写一个 word_count.py，统计一段英文里每个词出现几次，再让 tester 验证」。

看主管怎么拆：先写还是先定测试，tester 打回几次。

四、常见坑

坑一、worker 不记得上一次。

每次派活都是全新的 messages，这是隔离的代价。

tester 打回后，主管如果只说「再改改」，coder 根本不知道改什么、改哪个文件。

所以 SUPERVISOR 里那句「把失败原因转给 coder」不能省。每次派活写的那段说明，要写成一个新人也看得懂的样子：文件名、哪里错、期望什么。

坑二、worker 回了一句空话。

worker 最后一轮只调了工具、没留文字，主管就收到一个空字符串，接着瞎猜。

代码里用 "(worker 没留话)" 兜了底，岗位说明里也要求「做完用一两句话汇报」。

两头都要有。

坑三、不设轮数上限。

模型偶尔会绕圈：改一下，跑一下，再改一下。

没有上限，它会一直跑，一直扣钱。

主管 10 轮、worker 20 轮，按任务大小调。s06 的教学代码给子智能体设的是 30 轮。

坑四、给主管也配了 bash。

图省事给主管加个 bash，它多半就自己下场写代码了——派活要写说明、等汇报，不如自己动手快。

Supervisor 之所以是 Supervisor，靠的是手里没有干活的工具，不是 prompt 里写一句「你不要干活」。

坑五、DANGER 黑名单挡不住所有危险命令。

rm -rf 拦住了，rm -r -f 就没拦住。

字符串黑名单只是演示 s03 那道门长什么样，不是安全方案。

一定在空目录里跑。要放进真项目，至少上容器，或者看 s18 的工作树隔离。

坑六、钱是按调用次数一次次算的。

主管每派一次活，worker 那边就是一整个循环。每一轮都是一次完整的 API 调用，对话越长，每次带的上下文越多。

最坏情况：主管 10 轮 × worker 20 轮，一个目标能跑出上百次调用。

练手可以换便宜的模型：export MODEL_ID=claude-sonnet-5。

报错速查

跑不起来，先对一下这张表：

五、往上加什么

这个最小版故意砍了很多。想往上长，系列里每一集都是一块现成的砖：

• 多个 worker 同时干、干完不散 → s15 智能体团队

• 队友之间要约定格式 → s16 团队协议

• 让 worker 自己去任务板领活 → s12 任务系统 + s17 自治的智能体

• 某个活要跑很久，主管不想干等 → s13 后台任务

• 几个 worker 改同一个仓库会打架 → s18 工作树隔离

• API 偶尔报错、限流 → s11 错误恢复

• 主管的对话越来越长 → s08 上下文压缩

尽管机制很多，但不管加哪块，改的都是循环外面的东西。

最后

88 行，该有的骨头都在：一个循环、两个工具、一道权限门、一层上下文隔离。

先跑通这个，再去看 s15 那 786 行，就知道多出来的每一段在补什么了。

系列地图见上一篇：Agent 开发入门｜手写一个 Claude Code，机制很多循环就一个。

#agent

参考来源：learn-claude-code 中文版 https://www.shoptofly.com/lcc/zh/ （s01、s06、s15 三集的行数、工具数和机制描述出自这里）

我是想风@xaiwind，一个关注出海与 AI、AIGC 的创作者。

本文基于AI辅助与个人实践融合创作，如果文章对你有帮助，欢迎点赞、收藏、关注，

本人文章持续维护，欢迎联系或者评论区见，每天进步一点，我们终会到达目的地！