别急着上多智能体:345 行 Sub-Agent 代码里的取舍
主 Agent 明明能自己读写文件,为什么还要拆一个子 Agent?拆了之后,父层看不到中间过程,出了问题怎么查?这份三百多行的实现给出的答案很具体:子 Agent 不共享对话历史,只把最终结论交回来。本文沿着源码拆开这个取舍,你会看到它省下了什么、又付出了什么,以及拿去生产前必须补的几处护栏。文中代码只做静态阅读,未运行验证。
一、先说结论:子 Agent 买的是上下文,不是智力
子 Agent 用的还是同一个模型、同一套文件工具,它并不会因为"是子 Agent"就变聪明。它真正提供的是一份全新的 messages:
- 子任务的试错、读文件、跑命令,全部留在子 Agent 内部;
- 父 Agent 只收到一段最终文本,上下文窗口不被中间过程占用;
- 子任务失败时,异常和噪声被限制在子循环里。
代价同样明确:父层失去过程可见性,调用链变长,预算与超时要在两层分别计算。
二、先看边界:路径沙箱挡不住什么
所有文件工具的第一步都一样:把路径规范化,再确认它没有跑出工作目录。
def safe_path(p: str) -> Path:
path = (WORKDIR / p).resolve()
if not path.is_relative_to(WORKDIR):
raise ValueError(f"Path escapes workspace: {p}")
return path
resolve() 负责把 ..、. 和符号链接展开;is_relative_to() 负责按路径段判断包含关系。少任何一个都不行:只做字符串前缀比较,/work-evil 会被当成 /work 的子目录;不做 resolve(),../../etc/passwd 能直接绕过。
但这层沙箱有两个缺口。第一,WORKDIR = Path.cwd() 没有 resolve(),启动目录本身是符号链接时,两边归一化形态可能不一致。第二,它只保护文件工具,run_bash 里的命令路径不经过它。静态推断,未运行验证。
三、工具层:三件套比提示词更重要
新增能力时,改提示词是最慢的方式;改工具表才是结构性的方式。这份实现把工具拆成三件套:JSON Schema 负责"告诉模型怎么调",handler 负责"真正执行",路由表负责"名字到函数"。
TOOL_HANDLERS = {
"bash": lambda **kw: run_bash(kw["command"]),
"read_file": lambda **kw: run_read(kw["path"], kw.get("limit")),
"write_file": lambda **kw: run_write(kw["path"], kw["content"]),
"edit_file": lambda **kw: run_edit(kw["path"], kw["old_text"], kw["new_text"]),
}
三个约定值得保留:
- 父子工具集用列表相加复用,子 Agent 只有四个文件工具,父 Agent 多一个 task,能力差异只体现在一张表上。
- lambda **kw 把 JSON 对象解包成关键字参数;可选参数用 kw.get("limit"),避免模型不传时 KeyError。
- 路由找不到时返回字符串,保证模型编造工具名时循环继续。
反过来,handler(**args) 本身没有 try。模型漏传必填字段时,TypeError 直接冒到循环外,进程结束。文件工具内部的异常捕获管不到这一步。
四、子 Agent:一份独立的 messages
子 Agent 的入口只有一个函数,逻辑也很短:新开一个列表,循环调用模型,内部消化工具结果。
def run_subagent(prompt: str) -> str:
sub_messages = [{"role": "user", "content": prompt}]
for _ in range(30):
response = client.chat.completions.create(
model=MODEL,
messages=[{"role": "system", "content": SUB_SYSTEM}] + sub_messages,
tools=CHILD_TOOLS,
max_tokens=8000,
)
msg = response.choices[0].message
sub_messages.append(msg.model_dump())
if response.choices[0].finish_reason != "tool_calls":
break
results = []
for tool_call in msg.tool_calls:
args = json.loads(tool_call.function.arguments)
handler = TOOL_HANDLERS.get(tool_call.function.name)
output = handler(**args) if handler else "Unknown tool"
results.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": str(output)[:50000],
})
sub_messages.extend(results)
return msg.content or "(no summary)"
这里有三件事常被误解:
- 隔离的是上下文,不是进程。源码注释说子 Agent 在子进程运行,实际没有创建子进程;文件系统、内存、异常都是共享的。
- 轮数上限不等于结论完整。跑满三十轮时,循环自然结束并返回最后一轮内容,模型可能还在"准备调工具"。
- 父层拿到的是结论,不是证据。这是优点也是缺点:省窗口,但丢审计。
五、主循环:为什么 while True 不够
父层循环比子层更短,也更危险。
def agent_loop(messages: list):
while True:
response = client.chat.completions.create(
model=MODEL,
messages=[{"role": "system", "content": SYSTEM}] + messages,
tools=PARENT_TOOLS,
max_tokens=8000,
)
msg = response.choices[0].message
messages.append(msg.model_dump())
if response.choices[0].finish_reason != "tool_calls":
return
results = []
for tool_call in msg.tool_calls:
args = json.loads(tool_call.function.arguments)
if tool_call.function.name == "task":
output = run_subagent(args.get("prompt", ""))
else:
handler = TOOL_HANDLERS.get(tool_call.function.name)
output = handler(**args) if handler else "Unknown tool"
results.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": str(output),
})
messages.extend(results)
它只有停止条件,没有停止上限。子层有轮数限制,父层没有;子层工具结果截断到固定长度,父层 str(output) 不截断。两个不对称叠在一起,长会话的成本会失控。
六、两处安全错觉与四个坑
错觉一:子 Agent 是进程隔离。 实际只是 messages 隔离。
错觉二:命令黑名单能挡住危险操作。 ["rm -rf /", "sudo", "shutdown", "reboot", "> /dev/"] 是子串匹配,等价写法、变量拼接都能绕过;换到 Windows,rm 根本不存在,黑名单几乎不覆盖真实危险命令。
七、拿去生产前,先补七件事
- 父循环加最大轮数与 token 预算
- 工具调用统一包 try,返回结构化结果
- 父层输出做与子层一致的截断
- WORKDIR 初始化为 Path.cwd().resolve()
- 命令执行改为白名单 + shlex 参数校验
- 用 logging 替换 print
- 实现最小上下文压缩
前两件决定了它能不能稳定跑完一个任务,后五件决定了它能不能被排查、被复现、被控制成本。
八、最后:怎么判断要不要拆子 Agent
一个简单的判断标准:如果父 Agent 需要看到过程,就不要拆;如果父 Agent 只需要结论,就拆。 拆分的收益来自上下文隔离,成本来自过程不可见。两者都要,就得靠日志与 Trace 补足可见性。
如果现在就要动手,先给父循环加一个最大轮数。这是改动最小、收益最直接的一处护栏。
本文由 AI图鉴自动抓取并翻译,版权归原作者所有;如需查看原文请点下方链接。