再开一个循环，需要改原循环吗？
用户要求先检查 main.py 与 expected.txt，获准后运行并核对。思想篇把读取过程交给子任务，父流程只拿结论。落到 Python，最容易走偏的是复制一份循环，再给它加“子代理专用”分支。其实第一课的 run_loop 已经会创建新消息、保存工具反馈、判断停止原因；我们只需为它提供另一份问题、工具和预算。
正式代码在公开仓库 examples/s06-subagent。本课新增 task(task: str) 工具，函数调用链是：父循环分发 task → run_subagent 收窄工具并运行子循环 → task_result 把结果关联原请求ID。宿主可以另存完整子结果，父模型只收到短反馈。
下面三个完整函数按公开提交和符号提取。点击行号设置断点，再点“调试”，图会跟随已有代码块的唯一执行游标；切换输入会重置断点。child_tools 与 task_result 纯内存执行；run_subagent 的模型和文件接口使用可展开查看的内存替身，内部仍执行同源原循环，方便观察函数的边界。真实嵌套循环和文件行为在公开 CLI 及测试中验证，两种运行方式不能混为一谈。
child_tools：能力由宿主挑选
父工具表里已有 read_file、run_main 与 todo_write，新加的 task 也在其中。子任务只负责检查两个文件，因此它只需要 read_file。这里用固定白名单构造新注册表，既不依赖模型自觉，也不把“暂时别用其他工具”当作授权规则。
deepcopy 复制工具描述与参数字典，避免修改子工具定义时影响父注册表；函数对象仍指向同一个正式读取实现，目录和字节上限继续有效。缺少 read_file 就抛出异常，没有回退为父工具全集的路径。
真实执行轨迹回放；纯内存操作，不调用模型、文件或进程。
def child_tools(tools):
"""宿主白名单生成独立注册表；不继承运行、清单和再次委派能力。"""
selected = {}
for name in ('read_file',):
if name not in tools:
raise ValueError('缺少只读工具：' + name)
selected[name] = deepcopy(tools[name])
return selected点击行号设置断点，再点“调试”；也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。
from copy import deepcopy
import json
from types import SimpleNamespace
QUESTION="教学默认问题"
scenario='normal'
tools={name:{"parameters":{},"run":lambda:None} for name in ("read_file","run_main","task","todo_write")}
if scenario=="missing":tools.pop("read_file")
task="读取 main.py，说明源码推导；不要运行。"
if scenario=="invalid":task=" "
call={"id":"parent_task_1"}
outcome={"status":"answered","text":"源码推导42；尚未运行。","calls":3,"messages":[{"role":"user","content":"子任务内部记录"}]}
if scenario=="limit":outcome.update(status="call_limit",text="已达子循环预算，尚未得出结论。",calls=1)
if scenario=="long":outcome["text"]="长"*1201
# 明确替代模型和文件接口，未调用SDK或真实文件。
def ask(history,restricted):
index=sum(item["role"]=="assistant" for item in history)
if scenario=="forbidden" and index>0:
return {"stop_reason":"end_turn","content":[{"type":"text","text":history[-1]["content"][0]["content"]}]}
if index<2:
name="run_main" if scenario=="forbidden" else "read_file"
return {"stop_reason":"tool_use","content":[{"type":"tool_use","id":"child_"+str(index),"name":name,"input":{"path":"main.py" if index==0 else "expected.txt"}}]}
return {"stop_reason":"end_turn","content":[{"type":"text","text":"源码推导42；尚未运行。"}]}
def execute_hooked(call,restricted,approve,hooks):
name=call["name"]
text="print(6 * 7)" if call["input"].get("path")=="main.py" else "42"
return {"result":{"type":"tool_result","tool_use_id":call["id"],"content":text if name in restricted else "未执行：未注册工具。","is_error":name not in restricted},"warnings":[]}
_s04=SimpleNamespace(execute_hooked=execute_hooked,audit_event=lambda *args:None)
# 调用入口
try:
observed=child_tools(tools)
print("完成：",observed.get("status",observed.get("content","只读工具已选出")))
except ValueError as error:
observed={"error":str(error)}
print(str(error))- 宿主只复制read_file，子任务不能运行或再次委派。
- 新消息只含委派文本；按独立预算同步运行。
- 状态、短结论和调用次数按原ID返回；失败和截断标记错误。
在 selected[name] = deepcopy(tools[name]) 停住，selected 仍为空；单步之后才出现 read_file。切换“缺少只读工具”，会在复制之前失败。这个函数本身没有执行文件，也没有发起模型请求。
白名单要同时用于模型可见的工具定义和实际分发，不能只从提示里隐藏 run_main，却让执行器仍拿父工具表。下一段代码把同一份 restricted 传给两边，正是为了封住这个缺口。
run_subagent：把新任务接回原循环
输入只有委派文本、宿主工具、模型回调与预算，没有父 messages 参数。入口先校验任务非空、最多600字符，预算为1到8次请求，再创建只读注册表。随后调用旧 run_loop，由旧函数创建 [user: task] 作为新历史。
模型回调收到 restricted，执行回调也使用 restricted。子工具请求依然经过第四课 Hook 和第三课权限入口：只有 read_file 可读取，其他名称即使被模型硬写进响应，也会返回未执行。Hook 在这里记录子调用的 before/after，父 task 外面则另有一组父 Hook，两者没有混成同一条工具记录。
真实执行轨迹回放；保留同源子循环；模型与文件接口使用公开内存教学替身。
def run_subagent(task, tools, ask, max_calls=4, emit=print):
"""每次调用创建新历史；同步等待，复用原循环的停止与协议判断。"""
if type(task) is not str or not task.strip() or len(task) > 600:
raise ValueError('子任务必须是1到600字符的文本。')
if type(max_calls) is not int or not 1 <= max_calls <= 8:
raise ValueError('子循环预算必须是1到8次模型请求。')
restricted = child_tools(tools)
entries, warnings = [], []
def execute(call):
hooks = {'before': [lambda p: _s04.audit_event('before', p, entries)],
'after': [lambda p: _s04.audit_event('after', p, entries)]}
report = _s04.execute_hooked(call, restricted, lambda *_: False, hooks)
warnings.extend(report['warnings'])
return report['result']
outcome = run_loop(lambda history: ask(history, restricted), execute,
max_calls=max_calls, emit=emit, question=task.strip())
return outcome点击行号设置断点，再点“调试”；也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。
from copy import deepcopy
import json
from types import SimpleNamespace
QUESTION="教学默认问题"
scenario='answered'
tools={name:{"parameters":{},"run":lambda:None} for name in ("read_file","run_main","task","todo_write")}
if scenario=="missing":tools.pop("read_file")
task="读取 main.py，说明源码推导；不要运行。"
if scenario=="invalid":task=" "
call={"id":"parent_task_1"}
outcome={"status":"answered","text":"源码推导42；尚未运行。","calls":3,"messages":[{"role":"user","content":"子任务内部记录"}]}
if scenario=="limit":outcome.update(status="call_limit",text="已达子循环预算，尚未得出结论。",calls=1)
if scenario=="long":outcome["text"]="长"*1201
# 明确替代模型和文件接口，未调用SDK或真实文件。
def ask(history,restricted):
index=sum(item["role"]=="assistant" for item in history)
if scenario=="forbidden" and index>0:
return {"stop_reason":"end_turn","content":[{"type":"text","text":history[-1]["content"][0]["content"]}]}
if index<2:
name="run_main" if scenario=="forbidden" else "read_file"
return {"stop_reason":"tool_use","content":[{"type":"tool_use","id":"child_"+str(index),"name":name,"input":{"path":"main.py" if index==0 else "expected.txt"}}]}
return {"stop_reason":"end_turn","content":[{"type":"text","text":"源码推导42；尚未运行。"}]}
def execute_hooked(call,restricted,approve,hooks):
name=call["name"]
text="print(6 * 7)" if call["input"].get("path")=="main.py" else "42"
return {"result":{"type":"tool_result","tool_use_id":call["id"],"content":text if name in restricted else "未执行：未注册工具。","is_error":name not in restricted},"warnings":[]}
_s04=SimpleNamespace(execute_hooked=execute_hooked,audit_event=lambda *args:None)
def child_tools(tools):
"""宿主白名单生成独立注册表；不继承运行、清单和再次委派能力。"""
selected = {}
for name in ('read_file',):
if name not in tools:
raise ValueError('缺少只读工具：' + name)
selected[name] = deepcopy(tools[name])
return selected
def run_loop(ask, execute, max_calls=4, emit=print, question=QUESTION):
"""ask 读完整记录，execute 读工具请求；返回状态和证据，避免把停下当作成功。"""
if max_calls < 1:
raise ValueError("max_calls 必须至少为 1")
messages = [{"role": "user", "content": question}]
def finish(status, text, calls):
return {"status": status, "text": text, "calls": calls, "messages": messages}
for number in range(1, max_calls + 1):
emit(f"模型调用 {number}")
try:
response = ask(messages)
except Exception as error:
# API 异常时交回控制权；不自动重试，也不伪造工具结果。
return finish("model_error", f"模型请求失败：{type(error).__name__}", number)
blocks = response.get("content")
reason = response.get("stop_reason")
if not isinstance(blocks, list) or any(
not isinstance(block, dict) or block.get("type") not in ("text", "tool_use")
for block in blocks
):
return finish("protocol_error", "响应不符合本课支持的格式。", number)
messages.append({"role": "assistant", "content": blocks})
calls = [block for block in blocks if block["type"] == "tool_use"]
text = "\n".join(block.get("text", "") for block in blocks if block["type"] == "text")
if reason == "end_turn" and not calls and text.strip():
return finish("answered", text, number)
if reason not in ("end_turn", "tool_use"):
return finish("model_stopped", f"模型停止：{reason}；需要检查响应。", number)
ids = [call.get("id") for call in calls]
if reason != "tool_use" or not calls or any(
not isinstance(call_id, str) or not call_id for call_id in ids
) or len(set(ids)) != len(ids):
return finish("protocol_error", "停止原因与工具请求不一致。", number)
# 上面已经保存完整 assistant 响应；每个结果必须关联原调用 ID。
results = []
for call in calls:
emit(f"处理工具请求 {call.get('name')}")
result = execute(call)
results.append(result)
emit("工具结果：" + result["content"].splitlines()[-1])
messages.append({"role": "user", "content": results})
emit("保存工具结果，供下一次调用读取")
return finish("call_limit", f"已达 {max_calls} 次模型调用上限；保留记录，交给用户。", max_calls)
# 调用入口
try:
observed=run_subagent(task,tools,ask,max_calls=1 if scenario=='limit' else 3,emit=lambda _:None)
print("完成：",observed.get("status",observed.get("content","只读工具已选出")))
except ValueError as error:
observed={"error":str(error)}
print(str(error))- 宿主只复制read_file，子任务不能运行或再次委派。
- 新消息只含委派文本；按独立预算同步运行。
- 状态、短结论和调用次数按原ID返回；失败和截断标记错误。
在 outcome = run_loop(...) 设置断点：停住时子循环尚未执行，outcome 还不存在；下一步回来后才有停止状态。“预算耗尽”返回 call_limit，“试图运行”收到未注册工具的反馈，随后替身返回一段说明。后者的 answered 只是“给了回答”，不表示执行过程序。
浏览器绑定保留第一课原循环，仅将模型与文件接口替换为内存函数；它没有真实调用 SDK 或读取磁盘。展开准备代码可以查看同源循环和替身。浏览器与正式 CLI 都走同样的过程：读取源码、读取预期、生成结论，正常需要3次子模型判断；预算1时完成第一次读取，随后返回 call_limit。预算限制与协议错误等分支都由原循环处理。
这里同步等待，没有创建线程、子进程或后台任务。函数内的 entries/warnings 是本次子工具观察的局部数据，当前只把 outcome 交给调用者，未提供独立审计持久化接口。父宿主自定义的 Hook 也不会自动继承；接入项目策略时，应显式提供子任务必需的检查，不能假定新上下文天然继承全部限制。
task_result：返回结论，同时返回它为什么停
run_subagent 返回的 outcome 含 status、text、calls 与完整子 messages。宿主可以保存它排查问题，但写回父历史时，只需要停止状态、短结论与调用次数。这个封装函数没有调用模型做二次摘要，text 直接来自子循环的最终回答或停止说明。
先看 failed = status != 'answered'。call_limit、model_error 和协议错误都会使结果标记为失败。随后检查长度：即使状态是 answered，超过1200字符仍明确截断并标记错误，避免父模型把不完整文本当成完整结论。
真实执行轨迹回放；纯内存操作，不调用模型、文件或进程。
def task_result(call, outcome):
"""父循环只获得关联ID、停止状态和有界结论；子历史保留在宿主。"""
status = outcome['status']
text = outcome['text']
failed = status != 'answered'
if len(text) > 1200:
text = text[:1200] + '（结论过长，已截断；请缩小子任务）'
failed = True
payload = {'status': status, 'summary': text, 'calls': outcome['calls']}
return {'type': 'tool_result', 'tool_use_id': call['id'],
'content': json.dumps(payload, ensure_ascii=False), 'is_error': failed}点击行号设置断点，再点“调试”；也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。
from copy import deepcopy
import json
from types import SimpleNamespace
QUESTION="教学默认问题"
scenario='answered'
tools={name:{"parameters":{},"run":lambda:None} for name in ("read_file","run_main","task","todo_write")}
if scenario=="missing":tools.pop("read_file")
task="读取 main.py，说明源码推导；不要运行。"
if scenario=="invalid":task=" "
call={"id":"parent_task_1"}
outcome={"status":"answered","text":"源码推导42；尚未运行。","calls":3,"messages":[{"role":"user","content":"子任务内部记录"}]}
if scenario=="limit":outcome.update(status="call_limit",text="已达子循环预算，尚未得出结论。",calls=1)
if scenario=="long":outcome["text"]="长"*1201
# 明确替代模型和文件接口，未调用SDK或真实文件。
def ask(history,restricted):
index=sum(item["role"]=="assistant" for item in history)
if scenario=="forbidden" and index>0:
return {"stop_reason":"end_turn","content":[{"type":"text","text":history[-1]["content"][0]["content"]}]}
if index<2:
name="run_main" if scenario=="forbidden" else "read_file"
return {"stop_reason":"tool_use","content":[{"type":"tool_use","id":"child_"+str(index),"name":name,"input":{"path":"main.py" if index==0 else "expected.txt"}}]}
return {"stop_reason":"end_turn","content":[{"type":"text","text":"源码推导42；尚未运行。"}]}
def execute_hooked(call,restricted,approve,hooks):
name=call["name"]
text="print(6 * 7)" if call["input"].get("path")=="main.py" else "42"
return {"result":{"type":"tool_result","tool_use_id":call["id"],"content":text if name in restricted else "未执行：未注册工具。","is_error":name not in restricted},"warnings":[]}
_s04=SimpleNamespace(execute_hooked=execute_hooked,audit_event=lambda *args:None)
# 调用入口
try:
observed=task_result(call,outcome)
print("完成：",observed.get("status",observed.get("content","只读工具已选出")))
except ValueError as error:
observed={"error":str(error)}
print(str(error))- 宿主只复制read_file，子任务不能运行或再次委派。
- 新消息只含委派文本；按独立预算同步运行。
- 状态、短结论和调用次数按原ID返回；失败和截断标记错误。
选择“未完成”，在返回行查看 failed 已为 True；原 call 的ID仍保留。选择“结论过长”，会额外经过截断和标错两行。最终 content 是 JSON 文本，仅含 status/summary/calls，子 messages 不在其中。
保留原ID才能让父循环正确配对那次 task 请求。result 的 is_error 表示工具任务未正常交付；payload.status 则保留子循环的具体停止原因。两个字段都保留，是因为“调用结束”与“结论可用”有两层判断：超长结论会保留 answered，同时以 is_error=True 提醒父流程缩小问题。
task 怎样接进现有分发？
注册表里的 task 接收一个字符串参数，闭包捕获当前 workspace、子模型回调与子预算。执行器先按第二课规则检查工具名和参数，再调用该闭包，将完整 outcome 交给 task_result；无效委派捕获为关联原ID的错误反馈。
父循环工具仍经过前后 Hook；todo_write 继续按第五课更新内存计划，read_file/run_main 继续按第三课决定权限。task 不直接运行程序，子注册表又不含 run_main，因此不会把运行许可藏进委派文本里。
tools = build_tools(workspace, state, ask_child, child_calls=4, traces=traces)
outcome = run_loop(
lambda history: ask_parent(prepare_messages(history, state)),
lambda call: execute_with_subagent(call, tools, approve, hooks)["result"],
max_calls=4, question=QUESTION,
)这段装配里的 prepare_messages 仍在每次父请求上附当前计划，traces 由宿主保留完整子 outcome。默认脚本聚焦委派，没有重复前课的四次计划更新；在线模型仍可使用清单工具。子工具定义用第二课字符串 schema，父工具定义则保留第五课 todo_write 的数组 schema，加上 task 的字符串参数。
父模型收到子结论后重新读 expected.txt，再申请运行。读取原预期让实际比较有直接依据；子结论中“尚未运行”的限定也不会被偷偷删去。若用户拒绝，得到的是未执行反馈；允许时，才运行 main.py 并比较完整输出正文与预期。
在公开仓库跑一次
Python3.10+ 标准库可运行离线路径：
python3 examples/s06-subagent/user_code/practice.py --decision deny
python3 examples/s06-subagent/user_code/practice.py --decision allow
python3 examples/s06-subagent/user_code/practice.py --decision allow --child-calls 1
python3 examples/s06-subagent/user_code/verify_practice.py
python3 examples/s06-subagent/user_code/verify_fixtures.py默认模型是固定响应替身，只识别示例中的 print(6 * 7)；它不是通用源码分析器。文件读取和获准后的进程真实执行。正常路径父4次、子3次模型判断，最后输出42并核对；拒绝路径报告未执行。子预算1时反馈未完成，示例父模型直接向用户说明，没有继续申请运行。
失败后的处理由父模型和宿主共同决定。本例没有自动重试子任务，也没有断点恢复；重新委派会新建历史。真实模型可能重复请求 task，所以父请求预算还不够覆盖全部成本：还需根据项目限制每轮工具数量、委派次数、总耗时与 token。子任务禁用 task 则先消除了递归扩张这条路径。
公开测试用真实临时文件和程序标记验证拒绝不会执行，同时检查父子历史分开、子工具表复制、越权和递归请求失败、模型错误、超长结论与原ID。固定 Anthropic SDK 1.13.0 的模拟 HTTP 验证允许/拒绝各7次请求，分别检查父子 system、工具定义与父结果。没有调用在线模型；显式 --live 并设置环境变量才会接入真实 API。
这一版保留了一个容易检验的边界：父流程能看到子任务的结论，却看不到它的完整探索历史；需要追责时由宿主查子记录，需要继续运行时仍走原权限链。要让多个子任务同时修改文件，必须另外解决副作用隔离、取消和冲突处理，不能只把这一层同步调用改成并发就结束设计。
资料核对于2026-10-19：课程 Subagent与官方子代理文档。上述函数是本册独立 Python 实现，未据此声称复现 Claude Code 的产品运行时。