网硕互联帮助中心周五下午，技术部会议室。
"这批新夹具从设计到首件合格，走了 23 天，"项目经理老刘把甘特图投影在墙上，"设计 5 天，工艺 4 天，加工 8 天，质检 6 天——看起来各干各的，串行排队。可实际上设计改了第三版的时候工艺就已经在审第二版了，加工那边提前看了三维模型在准备毛坯，质检也提前介入了公差评审。问题是这些'并行'全靠微信群里吼，没人说得清到底谁和谁在协同、协同强度多大、哪条链路是瓶颈。"
我翻了翻邮件和微信群记录。
"这有什么问题？"老刘问。
"问题是协同是隐性的，"我说，"设计改了尺寸，工艺要跟着改夹具方案，加工要重新选刀路，质检要重新定检具。这条'变更传播链'你看得见吗？哪个工程师跟哪个工程师交互最多？哪次变更触发了最多的下游返工？"
"我就想干一件事，"老刘说，"把设计、工艺、加工、质检四个角色的协同关系画出来——谁跟谁沟通最多、每次沟通涉及哪些变更、变更传播了几跳才稳定。我想知道并行工程里到底哪条协同链路最活跃、哪个人是枢纽节点。"
"比如设计改了定位销孔径，工艺要改定位方案，加工要换钻头，质检要换塞规，"我接话，"这就是一条传播链：设计→工艺→加工→质检，四跳。如果设计改了三次才定稿，每次都全链路传播，那就是 12 次交互。把这些交互频次建成网络图，谁的度数最高谁就是枢纽。"
"对，"老刘点头，"还有，我想看'变更传播深度'——有的变更只影响工艺，有的变更一路传到质检。传播越深的变更，说明前期设计评审没做到位。最好还能聚类，看哪些工程师是一个'协同社区'。"
"用 networkx 建四角色协同网络，节点是工程师，边是协同交互频次，边权=交互次数，节点大小=度数中心性，"我开工程，"pandas 管协同日志，numpy 算中心性指标，scipy 做社区发现后的模块度检验，scikit-learn 对工程师做聚类分群，matplotlib 画网络图+传播深度直方图。数据自包含，合成一批并行工程协同记录，下载就能跑。"
敲了行原型：
G.add_edge("设计-张三", "工艺-李四", weight=12, changes=["孔径变更","公差调整"])
centrality = nx.degree_centrality(G)
communities = nx.community.greedy_modularity_communities(G)
"完整版用 OOP 封好，"我说，"一个类管协同日志加载，一个类建网络，一个类算中心性，一个类做社区发现，一个类分析变更传播，一个类出图。输出协同网络图、枢纽节点排名、社区划分、变更传播深度分布，存 results/。"
老刘凑近看："那以后看报告：工艺-李四是枢纽（度数 8，介数中心性最高），设计-张三和传播深度最深的变更链（平均 3.2 跳），四个工程师形成两个社区（设计和工艺一组、加工和质检一组），模块度 0.42。说明设计和工艺黏性够，但加工和质检之间协同偏弱，变更到加工就断了，质检只能被动等结果。"
"对，"我接话，"并行工程不是'同时开工'就完了，是信息流要通畅。网络图就是信息流的 X 光片。数字孪生里协同设计平台要做流程优化，这张网络就是现状基线。"
一、实际应用场景（真实痛点）
场景设定：多品种工装夹具开发项目，采用并行工程模式，设计/工艺/加工/质检四角色同步推进。协同依赖微信群、邮件、口头沟通，无结构化记录。项目经理无法量化协同强度、识别信息孤岛、定位变更传播瓶颈。
现场原话（叙事化）：
"不是我们不并行，"老刘说，"是并行了但看不见。设计改了个尺寸，工艺那边不知道，照样按旧图做方案，等发现了已经白干了两天。后来我们规定'设计变更必须通知工艺和加工'，可通知了没跟踪，质检那边还是最后才知道。我想知道到底谁跟谁协同最密、哪次变更传播最远、哪个人是信息黑洞。"
"还有社区问题，"老刘补充，"设计三个人、工艺两个、加工三个、质检两个，一共十个人。我感觉设计和工艺是一伙的，加工自己玩，质检最后才进来。可这是感觉，我要数据证明，才能调整人员搭配。"
核心矛盾："隐式协同交互流水" 与 "协同网络拓扑 + 中心性分析 + 社区发现 + 变更传播深度 + 信息孤岛定位" 之间的断层。
二、痛点分析（映射到滨州职业学院《先进制造技术》课程模型）
《先进制造技术》模块 本篇痛点对应
先进制造新模式：并行工程(CE)、协同设计、DFX 多角色协同网络建模与优化
先进制造技术基础：系统工程、信息流分析 协同交互的结构化量化
CAD/CAM技术：设计-工艺-制造信息集成 变更传播链分析
智能制造与数字孪生：协同平台信息流仿真 网络拓扑作为流程优化基线
FMS与先进生产管理：项目管理、并行任务调度 枢纽节点识别与资源调配
一句话总结：我们需要一个"并行工程多角色协同网络分析程序"，用
"networkx" 建图+中心性+社区发现，
"pandas" 管协同日志，
"numpy" 算传播深度，
"scipy" 做模块度评估，
"scikit-learn" 聚类工程师分群，实现从"微信群吼"到"协同网络拓扑 + 枢纽定位 + 信息孤岛识别"。
三、核心逻辑讲解（大白话）
3.1 问题本质：把协同想成"朋友圈"
把并行工程团队想成一个微信群：
* 节点 = 群成员（设计张三、工艺李四、加工王五、质检赵六…）
* 边 = 聊天频次（张三@李四 12 次 = 边权 12）
* 度数 = 谁最活跃（被 @ 最多的人 = 枢纽）
* 介数中心性 = 谁是不可替代的中间人（绕过他就断联）
* 社区 = 谁跟谁是一伙的（设计+工艺黏一起，加工+质检各玩各）
* 变更传播 = 一条消息从设计发出，经过几跳才到质检（跳数=传播深度）
* 信息孤岛 = 度数低+介数低+不在主社区的节点（被遗忘的人）
3.2 业务逻辑 → 代码映射
导入协同交互日志
│
▼ CollabLoader (pandas)
读取 CSV：
source, target, role_source, role_target,
interaction_count, change_type, change_depth
校验角色合法(设计/工艺/加工/质检)
│
▼ CollabNetworkBuilder (networkx)
建图：
节点=工程师，边=交互频次
节点属性=角色，边属性=变更类型+传播深度
│
▼ CentralityAnalyzer (networkx/numpy)
中心性计算：
度数中心性(谁最活跃)
介数中心性(谁是不可替代的桥)
接近中心性(谁离所有人最近)
特征向量中心性(谁跟重要的人连接)
│
▼ CommunityDetector (networkx)
社区发现：
greedy_modularity_communities
模块度评估(scipy)
│
▼ PropagationAnalyzer
变更传播分析：
按变更类型统计传播深度分布
找最深传播链
│
▼ CollabVisualizer (matplotlib)
可视化：
1. 协同网络拓扑图(节点大小=度数, 颜色=角色, 边宽=频次)
2. 中心性雷达图(枢纽节点对比)
3. 社区分布条形图
4. 变更传播深度直方图
5. 角色间交互热力图
│
▼ SyntheticCollabGenerator (numpy)
合成数据：
10工程师×4角色，交互频次幂律分布
含枢纽节点+信息孤岛
变更传播深度 1~4 跳
3.3 为什么需要网络分析
方法 问题
数邮件/微信 只能看自己参与的，看不到全局
画组织架构图 是汇报关系，不是协同关系
网络拓扑 真实信息流，谁跟谁真在协作
3.4 分析前后对比
维度 现状 本程序
协同强度 感觉 边权定量
枢纽人物 领导指定 中心性排名
信息孤岛 没人提 低度数+低介数自动识别
社区划分 凭经验 模块度优化自动聚类
变更传播 无记录 传播深度分布+最长链
四、OOP 代码实现
4.1 项目结构
parallel_collab_network/
├── parallel_collab_network/
│ ├── __init__.py
│ ├── collab_loader.py # 协同日志加载
│ ├── network_builder.py # 网络构建
│ ├── centrality_analyzer.py # 中心性分析
│ ├── community_detector.py # 社区发现
│ ├── propagation_analyzer.py # 变更传播分析
│ ├── visualizer.py # 可视化
│ └── synthetic_data.py # 合成数据
├── tests/
│ ├── __init__.py
│ └── test_collab_network.py
├── results/
│ ├── collab_network.png
│ ├── centrality_radar.png
│ ├── community_bar.png
│ ├── propagation_hist.png
│ ├── role_heatmap.png
│ ├── centrality_ranking.csv
│ ├── communities.csv
│ ├── propagation_depth.csv
│ └── collab_report.txt
└── run_collab_network.py
4.2 核心源码
<details>
<summary></summary>
"""并行工程协同日志加载器"""
import pandas as pd
from pathlib import Path
from typing import Optional
VALID_ROLES = {"设计", "工艺", "加工", "质检"}
class CollabLoader:
"""加载协同交互日志"""
def __init__(self, filepath: str = "collab_log.csv",
encoding: str = "utf-8"):
self.filepath = Path(filepath)
self.encoding = encoding
self._raw: Optional[pd.DataFrame] = None
def load(self) -> pd.DataFrame:
if not self.filepath.exists():
raise FileNotFoundError(f"文件不存在: {self.filepath}")
self._raw = pd.read_csv(self.filepath, encoding=self.encoding)
rename = {}
for tgt, al in {
"source": ["source", "发起方", "from"],
"target": ["target", "接收方", "to"],
"role_source": ["role_source", "发起角色", "role_from"],
"role_target": ["role_target", "接收角色", "role_to"],
"interaction_count": ["interaction_count", "交互次数", "count"],
"change_type": ["change_type", "变更类型", "change"],
"change_depth": ["change_depth", "传播深度", "depth"],
}.items():
if tgt not in self._raw.columns:
for a in al:
if a in self._raw.columns:
rename[a] = tgt
break
self._raw = self._raw.rename(columns=rename)
req = ["source", "target", "role_source", "role_target",
"interaction_count"]
miss = [c for c in req if c not in self._raw.columns]
if miss:
raise ValueError(f"缺少必要列: {miss}")
self._raw["interaction_count"] = pd.to_numeric(
self._raw["interaction_count"], errors="coerce").fillna(1).astype(int)
self._raw["change_depth"] = pd.to_numeric(
self._raw.get("change_depth"), errors="coerce").fillna(0).astype(int)
self._raw["change_type"] = self._raw.get("change_type", "无变更").fillna("无变更")
# 角色校验
for col in ["role_source", "role_target"]:
if col not in self._raw.columns:
self._raw[col] = "未知"
self._raw[col] = self._raw[col].str.strip()
return self._raw.reset_index(drop=True)
</details>
<details>
<summary></summary>
"""协同网络构建 (networkx)"""
import networkx as nx
import pandas as pd
from typing import Optional
class CollabNetworkBuilder:
"""从协同日志构建有向/无向图"""
def __init__(self, directed: bool = False):
self.directed = directed
self.G = nx.DiGraph() if directed else nx.Graph()
def build(self, df: pd.DataFrame) -> nx.Graph:
self.G.clear()
for _, r in df.iterrows():
src = str(r["source"])
tgt = str(r["target"])
# 节点属性
self.G.add_node(src, role=str(r.get("role_source", "未知")))
self.G.add_node(tgt, role=str(r.get("role_target", "未知")))
# 边属性
w = int(r["interaction_count"])
if self.G.has_edge(src, tgt):
self.G[src][tgt]["weight"] += w
self.G[src][tgt]["changes"].append(str(r.get("change_type", "")))
else:
self.G.add_edge(src, tgt, weight=w,
changes=[str(r.get("change_type", ""))],
depth=int(r.get("change_depth", 0)))
return self.G
def basic_stats(self) -> dict:
return {
"nodes": self.G.number_of_nodes(),
"edges": self.G.number_of_edges(),
"density": round(nx.density(self.G), 4),
"avg_degree": round(
sum(dict(self.G.degree()).values()) / max(1, self.G.number_of_nodes()), 2
),
}
</details>
<details>
<summary></summary>
"""中心性分析 (networkx/numpy)"""
import numpy as np
import pandas as pd
import networkx as nx
from typing import Optional, Dict, List
class CentralityAnalyzer:
"""计算多种中心性指标"""
def __init__(self):
pass
def analyze(self, G: nx.Graph) -> pd.DataFrame:
deg = nx.degree_centrality(G)
betw = nx.betweenness_centrality(G, weight="weight", normalized=True)
close = nx.closeness_centrality(G)
try:
eig = nx.eigenvector_centrality_numpy(G, weight="weight")
except Exception:
eig = {n: 0.0 for n in G.nodes()}
rows = []
for n in G.nodes():
rows.append({
"node": n,
"role": G.nodes[n].get("role", "未知"),
"degree_centrality": round(deg.get(n, 0), 4),
"betweenness_centrality": round(betw.get(n, 0), 4),
"closeness_centrality": round(close.get(n, 0), 4),
"eigenvector_centrality": round(eig.get(n, 0), 4),
"degree": G.degree(n),
})
return pd.DataFrame(rows).sort_values("degree_centrality",
ascending=False).reset_index(drop=True)
def hub_nodes(self, df: pd.DataFrame, top_n: int = 3) -> pd.DataFrame:
"""枢纽节点 = 介数中心性最高"""
return df.nlargest(top_n, "betweenness_centrality").reset_index(drop=True)
def info_islands(self, df: pd.DataFrame,
deg_thresh: float = 0.15,
betw_thresh: float = 0.1) -> pd.DataFrame:
"""信息孤岛 = 度数和介数都低的节点"""
return df[
(df["degree_centrality"] < deg_thresh) &
(df["betweenness_centrality"] < betw_thresh)
].reset_index(drop=True)
</details>
<details>
<summary></summary>
"""社区发现 (networkx)"""
import networkx as nx
import pandas as pd
from typing import Optional, List, Dict
class CommunityDetector:
"""基于模块度优化的社区发现"""
def __init__(self):
pass
def detect(self, G: nx.Graph) -> List[set]:
"""贪婪模块度社区发现"""
if G.number_of_edges() == 0:
return [set(G.nodes())]
communities = nx.community.greedy_modularity_communities(G)
return [set(c) for c in communities]
def modularity(self, G: nx.Graph, communities: List[set]) -> float:
"""计算模块度"""
if len(communities) <= 1:
return 0.0
return round(nx.community.modularity(G, communities), 4)
def community_summary(self, G: nx.Graph,
communities: List[set]) -> pd.DataFrame:
"""社区汇总"""
rows = []
for i, comm in enumerate(communities):
roles = {}
for n in comm:
r = G.nodes[n].get("role", "未知")
roles[r] = roles.get(r, 0) + 1
rows.append({
"community_id": i + 1,
"size": len(comm),
"members": ", ".join(sorted(comm)),
"role_distribution": ", ".join(f"{k}:{v}" for k, v in roles.items()),
"internal_edges": sum(1 for u in comm
for v in comm if u != v and G.has_edge(u, v)) // 2,
})
return pd.DataFrame(rows)
</details>
<details>
<summary></summary>
"""变更传播分析"""
import numpy as np
import pandas as pd
from collections import defaultdict
from typing import Optional
class PropagationAnalyzer:
"""分析变更传播深度和路径"""
def __init__(self):
pass
def depth_distribution(self, df: pd.DataFrame) -> pd.DataFrame:
"""传播深度分布"""
depths = df[df["change_depth"] > 0]["change_depth"]
if depths.empty:
return pd.DataFrame()
dist = depths.value_counts().sort_index().reset_index()
dist.columns = ["depth", "count"]
dist["ratio"] = (dist["count"] / dist["count"].sum()).round(4)
return dist
def deepest_chains(self, df: pd.DataFrame,
top_n: int = 5) -> pd.DataFrame:
"""传播深度最大的交互"""
sub = df[df["change_depth"] > 0].copy()
if sub.empty:
return pd.DataFrame()
sub = sub.sort_values("change_depth", ascending=False).head(top_n)
return sub[["source", "target", "change_type",
"change_depth", "interaction_count"]].reset_index(drop=True)
def change_type_stats(self, df: pd.DataFrame) -> pd.DataFrame:
"""按变更类型统计"""
sub = df[df["change_type"] != "无变更"].copy()
if sub.empty:
return pd.DataFrame()
rows = []
for ct, g in sub.groupby("change_type"):
rows.append({
"change_type": ct,
"count": len(g),
"avg_depth": round(g["change_depth"].mean(), 2),
"max_depth": int(g["change_depth"].max()),
"total_interactions": int(g["interaction_count"].sum()),
})
return pd.DataFrame(rows).sort_values("avg_depth",
ascending=False).reset_index(drop=True)
</details>
<details>
<summary></summary>
"""可视化 (matplotlib)"""
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from pathlib import Path
plt.rcParams["font.sans-serif"] = ["SimHei", "DejaVu Sans"]
plt.rcParams["axes.unicode_minus"] = False
class CollabVisualizer:
def __init__(self, results_dir: str = "results"):
self.results_dir = Path(results_dir)
self.results_dir.mkdir(exist_ok=True)
self.role_colors = {
"设计": "#E74C3C", "工艺": "#3498DB",
"加工": "#27AE60", "质检": "#F39C12", "未知": "#95A5A6"
}
def network_plot(self, G, centrality_df):
fig, ax = plt.subplots(figsize=(16, 12))
pos = nx.spring_layout(G, seed=42, k=0.8)
# 节点颜色=角色
nc = [self.role_colors.get(G.nodes[n].get("role", "未知"), "#999")
for n in G.nodes()]
# 节点大小=度数中心性
deg_c = dict(zip(centrality_df["node"], centrality_df["degree_centrality"]))
ns = [max(300, deg_c.get(n, 0) * 5000) for n in G.nodes()]
# 边宽=权重
ew = [max(0.5, G[u][v].get("weight", 1) / 3) for u, v in G.edges()]
nx.draw_networkx_nodes(G, pos, node_color=nc, node_size=ns,
ax=ax, alpha=0.9, edgecolors="black", linewidths=1.0)
nx.draw_networkx_edges(G, pos, width=ew, alpha=0.35, ax=ax,
edge_color="#7F8C8D")
nx.draw_networkx_labels(G, pos, font_size=8, ax=ax,
font_weight="bold")
# 图例
for role, color in self.role_colors.items():
ax.plot([], [], "o", color=color, label=role, markersize=10)
ax.legend(loc="upper left", framealpha=0.9)
ax.set_title("并行工程协同网络\\n(颜色=角色 大小=度数中心性 线宽=交互频次)",
fontsize=14, fontweight="bold")
ax.axis("off")
plt.tight_layout()
plt.savefig(self.results_dir / "collab_network.png",
dpi=150, bbox_inches="tight")
plt.close()
def centrality_radar(self, centrality_df, top_n=5):
sub = centrality_df.head(top_n)
metrics = ["degree_centrality", "betweenness_centrality",
"closeness_centrality", "eigenvector_centrality"]
labels = ["度数", "介数", "接近", "特征向量"]
angles = np.linspace(0, 2 * np.pi, len(metrics), endpoint=False).tolist()
angles += angles[:1]
fig, ax = plt.subplots(figsize=(10, 10), subplot_kw=dict(polar=True))
for _, r in sub.iterrows():
vals = [r[m] for m in metrics]
max_val = centrality_df[m].max() if m in centrality_df.columns else 1
vals = [v / max(centrality_df[m].max(), 1e-6) for m in metrics]
vals += vals[:1]
ax.plot(angles, vals, "o-", label=r["node"], linewidth=2)
ax.fill(angles, vals, alpha=0.1)
ax.set_xticks(angles[:-1])
ax.set_xticklabels(labels, fontsize=12)
ax.set_title(f"Top{top_n} 枢纽节点中心性雷达图", fontsize=14, fontweight="bold")
ax.legend(loc="upper right", bbox_to_anchor=(1.3, 1.1))
plt.tight_layout()
plt.savefig(self.results_dir / "centrality_radar.png",
dpi=150, bbox_inches="tight")
plt.close()
def community_bar(self, comm_df):
if comm_df.empty:
return
fig, ax = plt.subplots(figsize=(10, 6))
x = range(len(comm_df))
ax.bar(x, comm_df["size"], color="#3498DB", alpha=0.8)
ax.set_xticks(list(x))
ax.set_xticklabels([f"C{i+1}" for i in range(len(comm_df))], fontsize=11)
ax.set_ylabel("社区人数")
ax.set_title("社区规模分布", fontsize=14, fontweight="bold")
for i, r in comm_df.iterrows():
ax.text(i, r["size"] + 0.1, r["role_distribution"],
ha="center", fontsize=8)
ax.grid(axis="y", alpha=0.3)
plt.tight_layout()
plt.savefig(self.results_dir / "community_bar.png",
dpi=150, bbox_inches="tight")
plt.close()
def propagation_hist(self, depth_df):
if depth_df.empty:
return
fig, ax = plt.subplots(figsize=(9, 6))
ax.bar(depth_df["depth"].astype(str), depth_df["count"],
color="#E67E22", alpha=0.8)
ax.set_xlabel("传播深度(跳数)")
ax.set_ylabel("变更次数")
ax.set_title("变更传播深度分布", fontsize=14, fontweight="bold")
ax.grid(axis="y", alpha=0.3)
total = depth_df["count"].sum()
for _, r in depth_df.iterrows():
pct = r["count"] / total * 100
ax.text(str(r["depth"]), r["count"] + 0.2, f"{pct:.1f}%",
ha="center", fontsize=10, fontweight="bold")
plt.tight_layout()
plt.savefig(self.results_dir / "propagation_hist.png",
dpi=150, bbox_inches="tight")
plt.close()
def role_heatmap(self, df):
"""角色间交互热力图"""
df2 = df.copy()
crosstab = pd.crosstab(
df2["role_source"], df2["role_target"],
values=df2["interaction_count"], aggfunc="sum"
).fillna(0)
fig, ax = plt.subplots(figsize=(8, 6))
im = ax.imshow(crosstab.values, cmap="YlOrRd")
ax.set_xticks(range(len(crosstab.columns)))
ax.set_xticklabels(crosstab.columns, fontsize=11)
ax.set_yticks(range(len(crosstab.index)))
ax.set_yticklabels(crosstab.index, fontsize=11)
ax.set_title("角色间交互热力图(频次)", fontsize=14, fontweight="bold")
for i in range(len(crosstab.index)):
for j in range(len(crosstab.columns)):
ax.text(j, i, int(crosstab.values[i, j]),
ha="center", va="center", fontsize=10,
color="black" if crosstab.values[i, j] < crosstab.values.max()/2 else "white")
plt.colorbar(im, ax=ax)
plt.tight_layout()
plt.savefig(self.results_dir / "role_heatmap.png",
dpi=150, bbox_inches="tight")
plt.close()
</details>
<details>
<summary></summary>
"""合成并行工程协同数据生成器"""
import numpy as np
import pandas as pd
from pathlib import Path
from typing import Optional
class SyntheticCollabGenerator:
"""
生成 10 工程师×4角色 协同日志
角色: 设计(3人) 工艺(2人) 加工(3人) 质检(2人)
交互频次: 幂律分布(枢纽多, 边缘少)
变更传播: 深度1~4跳
"""
def __init__(self, rng: Optional[np.random.RandomState] = None):
self.rng = rng or np.random.RandomState(42)
def generate(self, output_path: str = "collab_log.csv",
n_engineers: int = 10) -> pd.DataFrame:
roles = {
"设计": ["张工", "王工", "李工"],
"工艺": ["赵工", "孙工"],
"加工": ["周工", "吴工", "郑工"],
"质检": ["钱工", "冯工"],
}
engineers = []
for role, names in roles.items():
for name in names:
engineers.append((f"{role}-{name}", role))
change_types = ["尺寸变更", "公差调整", "材料变更",
"表面处理变更", "装配顺序变更", "无变更"]
records = []
# 生成交互(幂律: 部分人交互多)
for i, (s, sr) in enumerate(engineers):
f
利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！
评论前必须登录！
注册