智能制造产线协作：设备依赖编排 + 角色分工 + 异构通信安全
当一条产线上同时存在 OPC-UA 的机械臂、MTConnect 的 CNC、MQTT 的 AGV、gRPC 的视觉相机，甚至还要和“供应商的 Agent”打交道时，谁来把这些异构设备“编排”成一支能自己跑的团队？本文以 JiuwenSwarm 最新蜂群（Swarm）能力为底座，拆解三个关键能力在制造场景里的落地：设备依赖编排（Swarmflow）、角色分工（Agent Team）、异构通信安全（A2A + 分层权限策略）。
目录
- 产线协作的三道老难题
- JiuwenSwarm 凭什么能“编排产线”
- 产线协作全景
- 让设备依赖自己跑起来
- Agent Team 怎么分工
- 几十个技能怎么选、怎么串
- 异构设备怎么安全地说话
- 一张大屏看全链路
- 跨机器部署产线 Agent
- 落地建议与总结
- 相关引用
产线协作的三道老难题
把一条真实产线“数字化协作”起来，传统方案几乎都会撞上三堵墙：
| 难题 | 典型表现 | 传统解法的痛点 |
|---|---|---|
| ① 设备依赖编排 | “取料没完成，CNC 不能开工；质检没过，不能入库”——任务之间有严格的先后 / 并行 / 条件依赖 | 写死在 MES 的状态机里，需求一变就要改代码、停线升级 |
| ② 角色分工 | 机械臂、CNC、AGV、相机各有专长，一个“大而全”的程序既难维护又容易出错 | 单体脚本，谁出了问题难定位，扩展新设备要重写 |
| ③ 异构通信安全 | 设备协议五花八门（OPC-UA / MTConnect / MQTT / gRPC / EtherCAT），还要连外部供应商系统 | 协议适配堆叠如山；更要命的是“谁能让机械臂动起来”缺乏统一管控 |
JiuwenSwarm 给出的答案是：把这些设备变成一支“Agent 团队”，用自然语言下达产线目标，由 Leader 自动拆解、分配、推进，全程在统一的安全策略下执行。
JiuwenSwarm 凭什么能“编排产线”
JiuwenSwarm 是一款让多智能体真正协作起来的 Agent 系统，面向需要自动化处理复杂任务的开发者和团队，帮助用户通过自然语言驱动多 Agent 协作、Skill 自演进和工具调用，实现从意图到结果的端到端交付。
它和“再写一个 RPA / 再写一个调度脚本”的根本区别在于下面这几项最新特性，本文会逐一把它们映射到制造场景：
| 特性 | 一句话 | 本文用在哪 |
|---|---|---|
| Swarmflow（蜂群流） | 把产线目标自动拆成可执行的 DAG 工作流，支持并行 / 串行 / 条件分支 / 人工算子 | 设备依赖编排（取料→加工→质检→入库） |
| Agent Team | Leader 统筹 + Teammate 专业执行，分级自主协同；持久团队自动沉淀双层团队记忆（个人记忆 + TEAM_MEMORY.md） | 机械臂 / CNC / AGV / 视觉质检的角色分工 |
| Symphony 双核 | 检索用树、编排用图——面对几十上百个 Skill，自动选出并串成可执行的技能链 | 设备 Skill 太多时怎么选、怎么串 |
| A2A 协议 | Agent-to-Agent 通信标准，让本厂 Agent 与外部供应商 Agent 异构互通 | 跨企业物料协同、外部系统接入 |
| 工具权限三级防护 | allow / ask / deny 三级动作 + 分层策略，高危动作必须人工审批 | 机械臂取放 / 急停等动作的安全管控 |
| 分布式 Agent Swarm | Leader / Teammate 可跨进程、跨机器部署，突破单机算力瓶颈 | 跨厂区把设备 Agent 下沉到各工控机 |
一句话理解：JiuwenSwarm 不是“再写一个调度引擎”，而是把产线协作从 “写死的代码” 变成 “可对话、可演进、可审计的 Agent 团队”。
切到集群模式后，运行界面长这样——左侧是 Leader 对用户产线目标的拆解（Swarmflow DAG 可视化），下方展开的「团队成员」面板里能看到每个 Teammate 的认领、工具调用、权限审批、汇报全过程：
产线协作全景
我们先看全貌。下图把一条“电机壳体产线”映射成了三层架构：
- 顶层 · 蜂群协作层（Agent Team）：一个 Leader（产线调度官）+ 多个专业 Teammate（机械臂 / CNC / AGV / 视觉质检 / 供应商协同 Agent），由 Swarmflow 编排，沉淀团队记忆。
- 中层 · 异构通信适配层（Gateway A2A Channel）：屏蔽设备协议差异，统一收发消息（E2A 内层协议），并在出口处做权限 / 审批拦截。
- 底层 · 物理产线层：各种异构设备 + 外部供应商 Agent（通过 A2A 接入）。
注意图里两条线的语义不同：
- Leader → Teammate 的实线：是任务下发、结果汇报的控制流（事件驱动，自动推进）。
- Teammate → A2A 适配层 的虚线：是每个 Agent 通过自己的 Skill 去操作真实设备的数据流（双向协议适配）。
让设备依赖自己跑起来
产线上最常见的诉求其实是一句话：“把这批电机壳体加工完并质检入库”。但这句话背后藏着一个 DAG（有向无环图）：
JiuwenSwarm 的 Swarmflow 做的事，就是把这句自然语言自动拆成上面这个 DAG，并让各阶段 Agent 自动衔接。下图标出了一个真实批次的依赖图：
几个关键点值得在制造场景里特别强调：
- 并行 + 串行混合：T1（物料）、T2（夹具）、T3（取料）可以并行启动；T4（CNC 加工）必须等 T2、T3 都完成。Swarmflow 会自动判断“前置任务是否就绪”，而不是写死时序。
- 条件分支 + 人工算子（Roadmap 中的有状态算子）：T5（视觉质检）判定“疑似不良”时，不会盲目继续，而是进入 H1 人工复核节点——人作为“有状态算子”参与流程，根据中间结果审批、修正或接管后续步骤。复核通过则继续 T6 入库，复核拒绝则 Leader 重派 T4。
- 事件驱动持续推进：不是“分完工就结束”。T4 完成 → 自动触发 T5；T5 完成 → 自动判断走哪条分支。整个过程用户可观察、可介入。
Swarmflow 是一段可执行的 Python
很多人以为 Swarmflow 只是“Leader 在脑子里规划”，其实不是——Swarmflow 提供了一组可执行的工作流原语，产线 DAG 是一段真实的 Python 代码。下面是仓库里 SwarmSkill 工作流模板 workflow.py.template 的真实结构，改成“电机壳体装配”产线后：
from swarmflow import agent, compact, log, map_parallel, parallel, phase
META = {
"name": "motor-housing-line",
"description": "电机壳体产线：取料→加工→质检→入库（含人工复核分支）",
"whenToUse": "当用户要求生产/加工电机壳体批次时复用此工作流",
"phases": [
{"title": "备料", "detail": "物料到货校验 + 夹具就位 + 机械臂取料（并行）"},
{"title": "加工", "detail": "CNC 加工，依赖夹具与取料完成"},
{"title": "质检", "detail": "视觉质检，疑似不良触发人工复核"},
{"title": "入库", "detail": "复核通过后 AGV 入库"},
],
}
ROLE_RESULT_SCHEMA = { # 每个 agent() 的结构化输出
"type": "object",
"properties": {"result": {"type": "string"}, "verdict": {"type": "string"}},
}
async def run(args):
# —— 阶段1 备料：三件事并行启动 ——
phase("备料")
log("并行启动：物料校验 / 夹具就位 / 机械臂取料")
prep = await parallel([
lambda: agent(build_prompt("物料校验", args), label="物料", phase="备料", schema=ROLE_RESULT_SCHEMA),
lambda: agent(build_prompt("夹具就位", args), label="夹具", phase="备料", schema=ROLE_RESULT_SCHEMA),
lambda: agent(build_prompt("机械臂取料", args), label="取料", phase="备料", schema=ROLE_RESULT_SCHEMA),
])
prep = compact(prep) # 容错：某分支失败时过滤掉 None
# —— 阶段2 加工：必须等夹具+取料都完成 ——
phase("加工")
machined = await agent(
build_prompt("CNC加工", {**args, "prep": prep}),
label="加工", phase="加工", schema=ROLE_RESULT_SCHEMA,
)
# —— 阶段3 质检：条件分支 ——
phase("质检")
qc = await agent(build_prompt("视觉质检", machined), label="质检", phase="质检", schema=ROLE_RESULT_SCHEMA)
if extract_json(qc).get("verdict") == "疑似不良":
log("质检判定疑似不良 → 进入人工复核算子")
review = await agent(build_prompt("人工复核", qc), label="复核", phase="质检", schema=ROLE_RESULT_SCHEMA)
if extract_json(review).get("verdict") != "通过":
return {"status": "degraded", "reason": "复核未通过，需重派加工"}
# —— 阶段4 入库 ——
phase("入库")
await agent(build_prompt("AGV入库", args), label="入库", phase="入库", schema=ROLE_RESULT_SCHEMA)
return {"status": "complete", "batch": args.get("batch_id")}
这段代码里用到的原语，全部来自 swarmflow 模块（仓库模板的约束清单里明确列出了支持集合）：
| 原语 | 语义 | 产线对应 |
|---|---|---|
phase("...") | 稳定的阶段标记，必须与 META["phases"] 里的 title 一一对应 | 备料 / 加工 / 质检 / 入库 |
parallel([...]) | 屏障式并行：等所有分支都回来再继续（适合“都需要才能下一步”） | 备料阶段三件事并行 |
map_parallel(items, fn) | 同构扇出：对一批同质任务并行 | 一批壳体同时取料 |
pipeline(items, s1, s2) | 每件物品有序多阶段 | 逐件：取料→加工→质检 |
agent(prompt, label, phase, schema) | 派一个子 Agent 执行，结构化输出 | 每个设备动作 |
compact([...]) / extract_json(...) | 容错：过滤失败的分支、解析 LLM 输出 | 某台设备超时不阻断整线 |
工作流跑起来后，会持续吐出进度事件。仓库里 workflow_state.py 的 WorkflowRunState 就是消费这些事件的状态机，它的派发表就是真实的事件流：
_KIND_HANDLERS = {
"workflow_started": "_on_workflow_started",
"phase": "_on_phase", # 进入新阶段（备料→加工）
"agent_started": "_on_agent_started", # 某设备 Agent 开始
"agent_completed": "_on_agent_completed",# 某设备 Agent 完成
"agent_failed": "_on_agent_failed", # 某设备 Agent 失败
"workflow_completed":"_on_workflow_completed",
"workflow_failed": "_on_workflow_failed",
"log": "_on_log",
}完整的生命周期是：workflow_started → phase → agent_started → agent_completed/failed → … → workflow_completed/failed。前端大屏上看到的“T4 加工 68%”进度条，就是这条事件流实时驱动的。
这意味着：过去要改 MES 状态机、停线升级才能调整的工艺流程，现在可以通过一段 Python 工作流（或一句自然语言让 Leader 生成它）重新编排，产线的“柔性”第一次真正落到可执行的代码层。
Agent Team 怎么分工
有了 DAG，还需要有人去“干每个节点”。Agent Team 的思路是 “让多个专业化的 Agent 组成团队，每个 Agent 负责自己擅长的部分”，而不是用一个“大而全”的程序包打天下。
Leader（产线调度官）的职责：目标理解、团队组建、DAG 编排、关键决策（如审批高危动作）、整体推进、在每个 round 结束后自动提取团队记忆。
Teammate（专业执行者）的职责：认领任务 → 独立执行 → 困难时求助 Leader → 汇报结果 → 提交中间产物。
产线经验自动沉淀
制造场景有一个刚需：经验必须能沉淀（哪台 CNC 在什么转速下振动大、AGV 几点该返航充电）。Agent Team 提供了双层记忆：
| 层级 | 访问权限 | 写入方 | 产线场景含义 |
|---|---|---|---|
| 个人记忆 | 该成员独占 | 成员自身 | 机械臂 Agent 记住自己各关节的零位偏移 |
团队记忆 TEAM_MEMORY.md | 所有成员只读 | Leader 在 round 结束后由提取 Agent 自动写入 | “CNC-02 主轴 >11500rpm 振动偏大，建议降速”这类跨设备经验 |
团队记忆按 [decision] / [lesson] / [member] / [context] 分类，跨 round 累积——这意味着产线跑得越久，团队越“懂”这条线。
真实的团队配置
“组建团队”在 JiuwenSwarm 里不是写代码 new 一堆对象，而是声明式配置。下面是从仓库 config.team.distributed.leader.yaml 里摘出来的真实结构（省略了无关字段），改成了产线角色：
modes:
team:
jiuwen_team: # 团队名
team_name: jiuwen_team
lifecycle: persistent # 持久团队：记忆跨 round 累积（临时团队则用 temporary）
teammate_mode: build_mode
spawn_mode: inprocess
enable_swarmflow: true # 开启 Swarmflow 编排（默认 true）
leader:
member_name: team_leader
display_name: 产线调度官
persona: "资深生产调度专家，擅长产线任务分解、设备依赖编排与异常协调"
agents: # 每个 teammate 的运行参数
leader: { max_iterations: 200, completion_timeout: 600.0 }
teammate: { max_iterations: 200, completion_timeout: 600.0 }
memory:
enabled: true
scenario: "general" # 产线属 general 场景，用个人长期记忆
auto_extract: true # round 结束自动提取团队记忆
shared_memory: true # 写入 TEAM_MEMORY.md
member_memory_prompt_mode: "proactive"
timezone_offset_hours: 8.0关键字段对应前面的概念：
lifecycle: persistent→ 触发双层记忆（个人记忆读写 + 团队记忆自动提取累积）；若写temporary则只读父 workspace、不留痕。enable_swarmflow: true→ 让这个团队跑前面那套可执行工作流；关掉就退回普通多 Agent 协作。memory.auto_extract: true+shared_memory: true→ 就是 Leader 在 round 结束自动写TEAM_MEMORY.md的开关。
而真正“让团队跑起来”的入口，是 SDK 的流式执行器（仓库 team_helpers.py）：
async for chunk in Runner.run_agent_team_streaming(
agent_team=team_spec, # 上面 YAML 经 load_team_spec_dict() 解析出的 TeamAgentSpec
inputs={"query": initial_query}, # 用户的自然语言产线目标
session=session_id,
envs=envs,
stream_logger=lg,
):
... # 每个 chunk 是一条进度/消息事件，推给前端大屏Leader 收到 initial_query 后，会用 SDK 提供的团队工具去干活——仓库 code_rails.py 里给 Leader 注入的系统提示就点名了这些工具：
“Start the approved team workflow with team tools such as build_team, create_task, spawn_teammate, and send_message。”
也就是说，Leader 通过 build_team 组队、create_task 建任务并指定 assignee、spawn_teammate 拉起成员、send_message 协调，这套调用链是系统提示约束的真实工具，不是泛泛而谈。
关键认知：Agent Team 的核心不是“更强的 Agent”，而是“更好的协作”。重点在于如何组织团队、分配任务、推进流程。
几十个技能怎么选、怎么串
每个 Teammate 要操作真实设备，靠的是 Skill（技能）：机械臂 Skill 封装 OPC-UA 读写、CNC Skill 封装 MTConnect、AGV Skill 封装 MQTT……当 Skill 数量一多（几十上百个），就会撞上两个问题：上下文被塞爆、模型注意力分散选错技能。
最新的 Symphony（技能交响乐） 用“双核心架构”解决：
- ① 技能检索（Skill Tree，怎么选）：把平铺的技能列表组织成可逐层浏览的技能树，Agent 像查目录一样
skill_branch_explore/skill_branch_peek，只展开相关分支，避免把所有SKILL.md一次性塞进上下文。 - ② 技能编排（Skill Score，怎么用）：基于候选技能、输入输出结构和技能总谱（一张描述
can_feed可衔接关系的关系图），生成一条可确认、可执行的技能链。
可以把 Symphony 理解为“双核心架构”：检索用树，编排用图。它判断的不只是“技能是否相关”，更是“上游输出能否真正接到下游输入”。
在产线上的典型链路：取料 Skill → 加工 Skill → 测量 Skill → 质检 Skill → 入库 Skill。图里每条边的数字是置信度，低于 min_edge_confidence（默认 0.3）的边不会被优先用于编排。这样生成出来的不是“看起来相关”的技能组合，而是能解释依赖、可确认、可执行的技能链。
一个机械臂控制技能长什么样
很多人觉得 Symphony 很玄，那是因为没见过 Skill 的真身。一个 Skill 就是一个文件夹 + 一个 SKILL.md，元数据用 YAML frontmatter 声明。仓库内置技能 delayed-restart-app（重启服务）是“shell+脚本”类技能的最小范例，照它的结构写一个产线的“机械臂取放”Skill：
opcua-arm-pickplace/
├── SKILL.md # 技能定义（必需）
├── scripts/
│ └── arm_pickplace.py # 封装 OPC-UA 读写，被 mcp_exec_command 调起
└── references/
└── opcua-points.md # 机械臂各关节的节点地址表SKILL.md（frontmatter 字段是机器强校验的，name 必须匹配 ^[a-z0-9-]+$、description 禁止 </> 字符）：
---
name: opcua-arm-pickplace
description: 控制产线机械臂完成取放动作。读取关节坐标、执行取料/放料、必要时急停。通过 mcp_exec_command 执行 scripts/arm_pickplace.py，经 OPC-UA 与 ABB 机械臂通信。
allowed_tools: [mcp_exec_command]
---
# 机械臂取放控制
## 执行方式
必须使用 `mcp_exec_command` 执行脚本，不要只口头回答：
python ~/.jiuwenswarm/agent/skills/opcua-arm-pickplace/scripts/arm_pickplace.py \
--host 192.168.1.11 --action pick --from A1 --to B2
## 参数
- `--action`：pick（取料）/ place（放料）/ home（归位）/ estop（急停）
- `--from` / `--to`：料位编号
## 边界
- 急停（estop）属于高危动作，会触发权限引擎 ask，需人工确认
- 关节坐标读取是只读，配置为 allow，无需确认scripts/arm_pickplace.py（封装 OPC-UA 的部分，用 asyncua 库）：
import argparse, asyncio
from asyncua import Client
async def run(host, action, src=None, dst=None):
async with Client(f"opc.tcp://{host}:4840") as client:
ns = await client.get_namespace_index("urn:ABB")
joint = await client.nodes.objects.get_child([f"{ns}:Robot", f"{ns}:JointPos"])
if action == "pick":
await move(client, ns, src) # 移到取料位
await gripper(client, ns, True) # 闭合夹爪
await move(client, ns, dst)
elif action == "estop":
await client.nodes.objects.get_child([f"{ns}:EStop"]).call_method()
print(f"{{\"verdict\":\"ok\",\"action\":\"{action}\"}}") # 结构化输出喂给工作流
if __name__ == "__main__":
p = argparse.ArgumentParser()
p.add_argument("--host", required=True); p.add_argument("--action", required=True)
p.add_argument("--from", dest="src"); p.add_argument("--to", dest="dst")
a = p.parse_args()
asyncio.run(run(a.host, a.action, a.src, a.dst))这就是“技能怎么被选、怎么被串”的全部落点：当已安装技能很少时，模型直接看列表选；当技能多了（产线扩到几十个），开启 Symphony 后：
- 技能检索把上面这种 Skill 组织进技能树（
设备控制 → 机械臂 → opcua-arm-pickplace），Leader 用skill_branch_explore逐层找到它； - 技能编排读技能总谱，看到
opcua-arm-pickplace的输出（JSON{verdict, action}）能can_feed给下游的mtconnect-cnc（加工 Skill），就自动生成取料→加工→质检这条链。
仓库 frontmatter 校验器（validator.py）里允许的字段就这些，超出会被硬拒：
ALLOWED_FRONTMATTER_KEYS = {
"name", "description", "license",
"allowed-tools", "allowed_tools",
"metadata", "compatibility",
}所以“Skill”不是一个营销词，它有确定的文件结构、强校验的元数据、明确的执行入口（bash/mcp_exec_command 调脚本），Symphony 的树和图都是基于这些真实 Skill 构建出来的。
异构设备怎么安全地说话
让 Agent 能“指挥”机械臂动起来，是件既强大又危险的事。JiuwenSwarm 在通信层和权限层做了两道闸。
让本厂和供应商 Agent 说同一种话
产线不仅要“对内协作”，还要“对外协同”：物料什么时候到、供应商系统能否提前预警。JiuwenSwarm 的 A2A（Agent-to-Agent）协议 提供了标准化的对外通道：对外暴露标准 JSON-RPC 入口与 Agent Card（/.well-known/agent-card.json），外部 A2A 客户端可统一发现并调用。
这不是文档上的承诺，而是 a2a_connect.py 里的真实代码。启动时构造的 AgentCard 长这样：
agent_card = AgentCard(
name=self.config.app_name,
description=self.config.app_description,
version=self.config.app_version,
supported_interfaces=[
AgentInterface(
url=f"http://{self.config.host}:{self.config.port}{self.config.rpc_path}",
protocol_binding="jsonrpc",
protocol_version=self.config.protocol_version, # 默认 1.0.0
)
],
capabilities=AgentCapabilities(streaming=True, push_notifications=False),
default_input_modes=["text/plain"],
default_output_modes=["text/plain"],
skills=[
AgentSkill(
id="chat", name="chat",
description="Send user prompt to JiuwenSwarm via Gateway",
tags=["chat", "gateway", "jiuwenswarm"],
examples=["Hello", "Summarize this"],
input_modes=["text/plain"], output_modes=["text/plain"],
)
],
)启用它只需在 ~/.jiuwenswarm/config/.env 里加几个环境变量（app_gateway.py 启动时读取），然后装可选依赖：
pip install "jiuwenswarm[a2a]" # 装的是 a2a-sdk[http-server]==1.0.0# ~/.jiuwenswarm/config/.env
A2A_SERVER_ENABLED=true # 不设默认关闭；1/true/yes/on 都算开
A2A_SERVER_HOST=0.0.0.0 # 对供应商暴露用 0.0.0.0，本机调试用 127.0.0.1
A2A_SERVER_PORT=19100 # 别和 Web/ACP 端口冲突
A2A_SERVER_PATH=/a2a # JSON-RPC 入口路径
A2A_SERVER_CARD_PATH=/.well-known/agent-card.json
A2A_SERVER_APP_NAME=电机壳体产线Agent供应商那边的 Agent 怎么调用你的产线？一个标准的 A2A JSON-RPC 调用（仓库 《A2A 接入说明》 的真实示例）：
# 非流式：问“下一批物料几点到”
curl -sS -X POST "http://产线IP:19100/a2a" -H 'Content-Type: application/json' \
-d '{"jsonrpc":"2.0","id":"m1","method":"SendMessage","params":{"message":{
"messageId":"req-1","contextId":"ctx-1","role":"ROLE_USER",
"parts":[{"text":"下一批 H7 电机壳体 ETA？"}]}}}'
# 流式：边产出边收（capabilities.streaming=true 才支持）
curl -sS -N -X POST "http://产线IP:19100/a2a" -H 'Content-Type: application/json' \
-d '{"jsonrpc":"2.0","id":"m2","method":"SendStreamingMessage","params":{
"message":{"messageId":"req-2","contextId":"ctx-1","role":"ROLE_USER",
"parts":[{"text":"查询今日各工位良品率"}]}}}'入站时 _A2AAgentExecutor 把 A2A 的 message.parts 映射成内部 Message.params（文本合并为 query、非文本落 files[]）；回包时把内部 payload 反向映射成 A2A Part（支持多模态 url/data/raw）。终结态映射：CHAT_ERROR→FAILED、CHAT_INTERRUPT_RESULT→CANCELED、其余→COMPLETED。
换句话说，供应商的 Agent 不需要懂你的 OPC-UA / MTConnect，只要会说 A2A，就能接入你的产线协作——这正是“异构通信”在跨企业边界上的延伸。（注：当前仓库实现的是入站 A2A Server；出站去调外部 A2A Agent 的能力在本仓库尚未包含，跨企业双向协同时由对方作 Server。）
每一步都在掌控中
更关键的是“谁能做什么”。JiuwenSwarm 的工具权限采用 allow / ask / deny 三级动作 + 分层策略（tiered_policy）：
上半部分是 severity × 模式 的动作映射矩阵，并配了产线场景举例：
| severity | normal 模式 | strict 模式 | 产线举例 |
|---|---|---|---|
| LOW（读状态） | allow | allow | 读取机械臂当前坐标 |
| MEDIUM（参数下发） | allow | ask | 下发主轴转速 12000rpm |
| HIGH（动作类） | ask | ask | 机械臂取放 / AGV 移动 |
| CRITICAL（高危） | ask | deny | 整线急停 / 修改安全围栏 |
下半部分是 一次工具调用如何定级的六步流程（evaluate_tiered_policy 的真实分支顺序）：
- 整工具基线：若某工具被设为
deny，立即返回拒绝，不再看任何参数规则。 - 内置安全规则（
builtin_rules.yaml）：内置deny优先于同层其它级别——例如bash(re:.*rm -rf.*)会被直接拦下。 - 用户参数规则：用户
deny在approval_overrides之前生效，可以拦住本应命中的“总是允许”。 - approval_overrides：“总是允许”命中即
allow——把高频低危操作（如每次都要点确认的“读关节坐标”）持久化为 allow，省掉重复审批；但无法绕过内置 deny。 - external_directory：MES 数据库、产线配方目录属于 workspace 外路径，单独走路径校验，与当前级别做 strictest 合并。
- shell_operators 升阶：含管道 / 提权 / 远程下载的命令，即使配置了 allow 也会被升为 ask。
真实的权限配置与内置规则
上面这些不是理论，是 config.yaml 里 permissions: 段的真实字段（产线版精简）：
permissions:
enabled: true # 产线务必开启；关闭后所有工具返回 allow
schema: tiered_policy # 启用分层策略
permission_mode: strict # 产线建议 strict：MEDIUM→ask, CRITICAL→deny
defaults:
"*": "allow" # 未命中的兜底（strict 下仍会被 severity 覆盖）
tools: # 整工具基线
bash: ask # 任何 shell 默认要确认
mcp_exec_command: ask # 调脚本（含我们的 OPC-UA 脚本）默认确认
write_file: ask
read_memory: allow # 读记忆/待办免确认，减少打断
todo_create: allow
rules: # 参数级规则（更细粒度）
- id: shell_allow_dir
tools: [bash, mcp_exec_command, create_terminal]
pattern: "dir *"
severity: LOW # 只读目录 → allow
- id: shell_ask_rm
tools: [bash, mcp_exec_command, create_terminal]
pattern: "rm *"
severity: HIGH # 删除 → ask
- id: path_ask_env
tools: [read_file, write_file, grep, list_dir]
pattern: "**/.env*"
severity: HIGH # 读 .env → ask
external_directory:
"*": "ask" # 访问产线配方/MES 等 workspace 外路径要确认
"D:/产线配方/": allow # 可对特定可信目录放行更狠的是内置安全规则（builtin_rules.yaml，即使用户不配也生效，且用户无法用 override 绕过）。仓库里真实存在的高危规则（节选）：
- id: shell_fs_recursive_or_forced_delete
description: "递归、强制或批量删除关键路径，可能造成不可恢复的数据破坏"
tools: [bash, mcp_exec_command, create_terminal]
match_type: command
pattern: 're:(?i)(rm\s+[^;&|]*(-[A-Za-z]*[rR][A-Za-z]*[fF]|--recursive|--force)[^;&|]*(/|\*|~))'
severity: CRITICAL # strict 模式下 → deny
- id: shell_privilege_escalation
description: "通过 sudo/su/doas/pkexec/runas 提升权限"
pattern: 're:(?i)(sudo|doas|pkexec)\s+|su\s+(-|root|\w)|runas\b'
severity: CRITICAL
- id: shell_system_shutdown_or_reboot
description: "关机、重启或切换运行级别"
pattern: 're:(?i)(shutdown|reboot|halt|poweroff)\b|(init|telinit)\s+(0|6)\b'
action: deny # 这条直接硬拒，不分模式在产线上的实际效果：当机械臂 Agent 想执行急停、AGV Agent 的脚本里出现 sudo、或者有人想让 Agent 跑 rm -rf 清理产线日志，这些规则会在用户配置之上先拦下——CRITICAL 在 strict 模式变 deny，带 action: deny 的规则任何模式都拒。
在 TUI 里还可以用 /permissions 命令实时管理（写入的就是上面 permissions.tools / permissions.rules）：
# 允许读取机械臂状态（持久化到 approval_overrides）
/permissions allow read_state
# 任何含 rm -rf 的命令一律拒绝（内置 deny 之上再加双保险）
/permissions deny bash(re:.*rm -rf.*)
# 写产线配方前必须确认
/permissions ask write_file(re:.*recipe.*\.yaml$)一张大屏看全链路
把上面四个能力拼到一起，运行起来是什么样？下面这张是“电机壳体产线监控大屏”：
大屏分三栏，恰好对应本文的三大主题：
- 左栏 · Agent Team 实时状态：可以看到 Leader 正在“编排 T4→T5 依赖”，各 Teammate 的认领 / 执行 / 等待 / 条件分支状态一目了然；底部是当轮自动沉淀的
TEAM_MEMORY.md（[lesson]CNC 降速建议、[decision]质检走 H1 复核、[member]AGV 返航规则）。 - 中栏 · 产线吞吐与节拍：KPI（良品 / 待复核 / 不良率 / 节拍）+ 近 60 分钟产出曲线（实际 vs 计划）+ 异常事件标记；下方是当前 Swarmflow DAG 的实时进度条
T4加工(68%) → T5质检→H1复核 → T6入库。 - 右栏 · 异构设备状态 + 权限审批日志：每台设备的协议（OPC-UA / MTConnect / MQTT / gRPC / EtherCAT）、温度 / 电量 / 负载；以及按
allow/ask/deny染色的权限日志——set_spindle(12000rpm)被 strict 模式升为 ask、rm -rf被内置规则 deny、外部 MES 目录访问被 ask。
这张屏背后没有任何“写死的调度脚本”——所有任务流转、记忆沉淀、权限判定，都由 JiuwenSwarm 的 Agent Team + Swarmflow + Symphony + 分层策略自动完成。
跨机器部署产线 Agent
产线规模一大，单机肯定扛不住。JiuwenSwarm 的分布式 Agent Swarm 让 Leader / Teammate 可以跨进程、跨机器部署：
- 控制面：Teammate 启动后向 A2X 注册中心注册为“空闲节点”，Leader 组队时通过
reserve_blank_agents预约，再用 direct ZMQ 发送 bootstrap 接管——Leader 不需要预先知道 Teammate 的地址。 - 数据面：任务、成员状态、消息走共享存储（推荐 PostgreSQL）；team-workspace 通过 NFS 等共享目录让成员产出的文件彼此可见。
- 会话语义：分布式模式保持单活 session，确保远程成员的 bootstrap、传输连接和运行时资源不会跨会话复用。
Leader 与 Teammate 的关键差异，就体现在 team.runtime 和 team.transport 两段配置上。下面是仓库 config.team.distributed.leader.yaml / config.team.distributed.teammate.yaml 里的真实字段对比（省略相同部分）：
# —— Leader（中央机房）——
team:
runtime:
mode: distributed
role: leader
transport:
type: pyzmq
params:
direct_addr: tcp://0.0.0.0:28555 # Leader 绑定，等 Teammate 来连
pubsub_publish_addr: tcp://127.0.0.1:28556
pubsub_subscribe_addr: tcp://127.0.0.1:28557
metadata:
pubsub_bind: true # Leader 绑定 pubsub
storage:
type: postgresql
params:
connection_string: postgresql+asyncpg://postgres:postgres@127.0.0.1:5432/jiuwen_team
# —— Teammate（某台产线工控机）——
team:
runtime:
mode: distributed
role: teammate
member_name: teammate_1 # 本进程默认身份
transport:
type: pyzmq
params:
direct_addr: tcp://0.0.0.0:28611 # 自己的通信地址
bootstrap_direct_addr: tcp://127.0.0.1:28610 # 向注册中心发布的可连接地址
pubsub_publish_addr: tcp://127.0.0.1:28556 # 连 Leader 的 pub
pubsub_subscribe_addr: tcp://127.0.0.1:28557
# 注意 teammate 不写 pubsub_bind: true注意 Leader/Teammate 双方都要指向同一个 react.a2x_registry（A2X 注册中心）和同一个 team.storage（PostgreSQL），这是分布式协作的前提：
react:
a2x_registry:
base_url: "http://127.0.0.1:8000" # Leader 和 Teammate 必须指向同一注册中心
dataset: "team_pool"
# Leader 端：role: teamleader
# Teammate 端：role: teammate + endpoint: "tcp://127.0.0.1:28610"（自己发布的地址）Teammate 启动时由
AsyncA2XRegistryClient.register_blank_agent(...)把自己注册成空闲节点；Leader 组队时调reserve_blank_agents预约、拿到 endpoint 后用 direct ZMQ 发 bootstrap 接管。dataset和endpoint两者必须同时配置，否则自动注册会被跳过。
最小启动形态（四个进程）：
# 1) A2X 注册中心（独立部署，从 agent-protocol 仓装）
a2x-registry # 默认 127.0.0.1:8000
# 2) Teammate（某台产线工控机）
JIUWENSWARM_DATA_DIR="<TEAMMATE_DATA_DIR>" \
AGENT_SERVER_PORT=28193 \
python -m jiuwenswarm.server.app_agentserver
# 3) Leader（中央机房）
JIUWENSWARM_DATA_DIR="<LEADER_DATA_DIR>" \
AGENT_SERVER_PORT=28192 GATEWAY_PORT=29101 WEB_PORT=29100 \
python -m jiuwenswarm.app
# 4) Web 前端（可选）
cd jiuwenswarm/channels/web/frontend && npm run dev -- --host 0.0.0.0 --port 5173在制造场景里，这意味着：每台关键设备的工控机都可以跑一个专属 Teammate Agent，中央机房跑 Leader 统一编排，跨厂区的供应商 Agent 通过 A2A 接入——物理拓扑和协作拓扑终于可以分离了。
落地建议与总结
如果你也想在自己的产线上试一把，建议按这个节奏来——四个阶段层层递进，后一阶段继承前一阶段的配置与记忆，不必推翻重来：
- 先单机跑通一个“最小产线”：1 个 Leader + 2~3 个 Teammate（比如机械臂 + CNC + 视觉），用本地模式验证 Swarmflow DAG 和团队记忆。
- 用 Symphony 管理技能：当设备 Skill 超过 20 个，开启技能检索 + 技能总谱，让编排自动选链，而不是硬编码调用顺序。
- 上线前配好安全策略：把
permission_mode切到strict，高危动作（动作类 / 急停 / 改安全参数）显式deny或ask；用approval_overrides把高频低危操作持久化为 allow。 - 需要跨厂区协同时再上分布式：部署 A2X 注册中心 + 共享 PostgreSQL + NFS 工作区，把设备 Agent 分布到各工控机。
三句话总结
- 设备依赖编排：Swarmflow 把自然语言目标自动拆成 DAG，并行 / 串行 / 条件分支 / 人工算子全覆盖，产线流程不再写死在 MES 里。
- 角色分工：Agent Team 的 Leader/Teammate 分级自主协同 + 双层团队记忆，让产线经验越跑越沉淀。
- 异构通信安全：A2A 让本厂与供应商 Agent 异构互通；
allow/ask/deny三级动作 + 分层策略，让“自主跑”和“关键动作人批”同时成立。
未来，随着 AgentSwarms 在更多制造场景中的深入应用，人机协作的产线模式将成为提升制造效率与柔性的关键杠杆——而 JiuwenSwarm 正在把这条路铺平。
相关引用
- JiuwenSwarm 官网：https://openjiuwen.com
- JiuwenSwarm 仓库（GitCode）：https://gitcode.com/openJiuwen/jiuwenswarm
- 官方文档：
- Swarm Skills Hub：https://swarmskills.openjiuwen.com/