◆ 博主名称： QuZhengRong
AI俘虏，样式苦手
⭐️
Agent专栏：
Agent
⭐️
LuckReport专栏：
LuckReport
⭐️
SpringBoot专栏：
SpringBoot
目录
- 一、拆分时机
- 二、拆分
- 三、调用
- 四、结果回收
- 五、共享数据
- 六、示例
- 七、LuckReport 项目推荐
-
- 1、项目简介
- 2、在线体验
Subgent 是多智能体协作里很常用的思路：主智能体把大任务拆分成多个小任务，派发给子智能体，各个子智能体跑完任务之后，再把结果汇总回来。很适合需要分开职责、子任务可以并行跑，又或者不同任务不能混用工具权限的场景。这篇文章就来讲一下如何设计 Subagent。
一、拆分时机
大多数场景下，单智能体配上一组工具就够完成任务了，引入多子智能体反而会添加额外的成本：调度、上下文复制、结果汇总逻辑都会变得更复杂。
只有出现下面的情况时，才考虑拆分：
二、拆分
常见三种切法如下：
1、按角色。Planner 只出计划，Executor 调工具执行，Reviewer 按验收标准检查。适合流程固定、角色边界清楚的任务。
2、按子任务。把总目标拆成互不依赖的几块，每块一个子智能体。例如门店经营排查：客流与竞对一块，成本结构一块，两块查完再汇总。
3、按权限面。Explorer 只有只读工具，Actor 才能写文件、改配置、发请求。探索阶段的误操作不会落到生产侧。
同一任务支持多种拆分方式叠加。比如先按子任务并行采集、查询数据，再交由 Reviewer 统一校验，就是「按子任务 + 按角色」的组合用法。
拆分后必须明确三个问题：该子智能体的成功判定标准、可用工具集合、任务结束后需要返回的输出物。如果无法清晰回答这三点，说明拆分还不到位。
三、调用
主智能体负责任务分发。下发子任务时，应当单独构造一份精简的任务规约，而不是直接转发完整对话上下文。
任务规约通常包含四项内容：
子智能体调用主要分为两种模式：
传参仅携带任务规约与必要的只读资料引用。子智能体内部的对话记录、中间推理思考过程，默认不回传给主智能体，也不在子智能体之间互相拷贝。
伪代码形态如下：
# handle 只表示一次派活；具体框架里可能是线程、进程或远程会话
handle = spawn(
agent="cost_analyst",
task={
"goal": "核算南山店本月成本结构",
"constraints": ["只读查询", "不改业务数据"],
"output": {"summary": "≤200字", "artifacts": ["shared/cost.json"]},
"tools": ["get_cost_detail", "get_month_revenue"],
},
)
result = await handle.wait() # 同步等结果；并行时用 gather 收多个 handle
四、结果回收
子智能体仅反馈 “任务完成” 是不够的，必须返回主智能体可结构化解析的结果。 可约定固定返回字段如下：
主智能体默认只读取 summary 和 artifacts；如需深入细节，再通过路径从共享区读取文件，禁止将子智能体完整推理轨迹并入主上下文。
根据 status 做分支处置：可恢复的错误自动重试；权限不足、数据源缺失这类问题升级人工处理；状态为 partial（部分成功）时，可先汇总已有成果，对未完成的子块单独补发任务。
# 主智能体侧的最小消费逻辑
def consume(result: dict) –> str:
if result["status"] == "failed":
return f"子任务失败：{result.get('error')}"
parts = [result["summary"]]
for path in result.get("artifacts", []):
parts.append(f"产物：{path}") # 需要细节时再 open(path)
return "\\n".join(parts)
五、共享数据
每个子智能体拥有独立会话与独立的工具白名单。数据流转依靠显式交接，而不是共用一份全局上下文。
交互关系示意：
主智能体
├─ 下发 task（目标 + 约束 + 输出契约）
├─ 共享区（本地目录 / 对象存储 / 挂载工作区）
│ └─ 子智能体产出的 json / md / 表格等文件
└─ 回收 result（status + summary + artifacts）
└─ 子智能体（独立上下文 + 独立工具白名单）
三层数据边界规则：
共享区需要配套权限管控：探索类子智能体仅允许写入自身对应的产物目录；具备业务变更能力的 Actor 单独分配写入授权，避免只读类任务意外获得变更权限。
六、示例
场景：勇哥说餐饮要排查「深圳南山店这个月为什么亏了」。主智能体拆成两个子任务并行：客流与竞对、成本结构。各自写结果到 shared/，主智能体回收后汇总。
# Subagent 编排示例：并行两个子任务，写共享区，主智能体按契约回收
import json
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
SHARED = Path("shared")
SHARED.mkdir(exist_ok=True)
def tool_traffic(store: str) –> dict:
"""模拟客流与周边查询。"""
return {
"store": store,
"traffic_wow": –0.25,
"nearby_new_stores": 2,
"note": "300 米内新开 2 家连锁快餐，其中一家在做开业大促",
}
def tool_cost(store: str) –> dict:
"""模拟成本与营业额查询。"""
return {
"store": store,
"revenue_wow": –0.18,
"food_cost_ratio": 0.32,
"food_cost_redline": 0.30,
"items": {"rent": 4.5, "labor": 5.2, "food": 6.8, "misc": 0.8},
}
def run_subagent(name: str, task: dict) –> dict:
"""子智能体：独立执行，只把契约结果交回；细节落盘到 artifacts。"""
store = task["goal_store"]
try:
if name == "traffic":
raw = tool_traffic(store)
path = SHARED / "traffic.json"
path.write_text(json.dumps(raw, ensure_ascii=False, indent=2), encoding="utf-8")
summary = (
f"{store} 客流环比 {raw['traffic_wow']:.0%}，"
f"周边新开 {raw['nearby_new_stores']} 家。"
)
return {
"status": "ok",
"summary": summary,
"artifacts": [str(path)],
"evidence": [raw["note"]],
}
if name == "cost":
raw = tool_cost(store)
path = SHARED / "cost.json"
path.write_text(json.dumps(raw, ensure_ascii=False, indent=2), encoding="utf-8")
over = raw["food_cost_ratio"] > raw["food_cost_redline"]
summary = (
f"{store} 营业额环比 {raw['revenue_wow']:.0%}，"
f"食材成本占比 {raw['food_cost_ratio']:.0%}"
+ ("，超过红线。" if over else "。")
)
return {
"status": "ok",
"summary": summary,
"artifacts": [str(path)],
"evidence": [f"食材红线 {raw['food_cost_redline']:.0%}"],
}
return {"status": "failed", "summary": "", "artifacts": [], "error": f"未知子智能体 {name}"}
except Exception as e:
return {"status": "failed", "summary": "", "artifacts": [], "error": str(e)}
def main():
# 1、拆分：两个互不依赖的子任务
tasks = {
"traffic": {
"goal_store": "深圳南山店",
"constraints": ["只读"],
"output": {"artifacts": ["shared/traffic.json"]},
},
"cost": {
"goal_store": "深圳南山店",
"constraints": ["只读"],
"output": {"artifacts": ["shared/cost.json"]},
},
}
# 2、并行调用
results = {}
with ThreadPoolExecutor(max_workers=2) as pool:
futures = {pool.submit(run_subagent, name, task): name for name, task in tasks.items()}
for fut in as_completed(futures):
name = futures[fut]
results[name] = fut.result()
# 3、按契约回收，只吃 summary + artifacts
lines = ["南山店亏损排查汇总："]
for name, result in results.items():
if result["status"] != "ok":
lines.append(f"- {name} 失败：{result.get('error')}")
continue
lines.append(f"- [{name}] {result['summary']}")
for p in result["artifacts"]:
lines.append(f" 产物：{p}")
print("\\n".join(lines))
if __name__ == "__main__":
main()
跑完后 shared/ 下会有 traffic.json 和 cost.json。主智能体汇总时读的是摘要；需要核对数字再打开产物文件。
接真实模型时，把 run_subagent 里的模拟查询换成带独立 system prompt 和工具白名单的 Agent 会话即可。派活字段、返回契约、共享目录这三处约定可以保持不变。
七、LuckReport 项目推荐
导航：LuckReport专栏
1、项目简介
Luck-Report 是一款面向中国式复杂报表的开源报表引擎。基于单元格迭代，可在浏览器中完成交叉表、分组表、主从表等复杂版式的设计、预览与导出。
Luck-Report 内置 AI 智能助手，可结合大模型配置与知识库，用自然语言完成制表、改表与答疑。后端基于 Spring Boot，支持独立部署。项目由开源 UReport2 重构而来，在保留中国式报表能力的基础上，补齐现代化架构与 AI 辅助能力。
本仓库为报表引擎后端源码，基于 Apache-2.0 开源，可免费商用。
2、在线体验
- 体验地址：https://www.tinyluck.cn:8060/login
- 源码地址：https://gitee.com/LuckyPools/luck-report-server
- 文档地址：https://www.tinyluck.cn:8099/se/docs
