周五下午技术部会议室。这批新夹具从设计到首件合格走了 23 天项目经理老刘把甘特图投影在墙上设计 5 天工艺 4 天加工 8 天质检 6 天——看起来各干各的串行排队。可实际上设计改了第三版的时候工艺就已经在审第二版了加工那边提前看了三维模型在准备毛坯质检也提前介入了公差评审。问题是这些并行全靠微信群里吼没人说得清到底谁和谁在协同、协同强度多大、哪条链路是瓶颈。我翻了翻邮件和微信群记录。这有什么问题老刘问。问题是协同是隐性的我说设计改了尺寸工艺要跟着改夹具方案加工要重新选刀路质检要重新定检具。这条变更传播链你看得见吗哪个工程师跟哪个工程师交互最多哪次变更触发了最多的下游返工我就想干一件事老刘说把设计、工艺、加工、质检四个角色的协同关系画出来——谁跟谁沟通最多、每次沟通涉及哪些变更、变更传播了几跳才稳定。我想知道并行工程里到底哪条协同链路最活跃、哪个人是枢纽节点。比如设计改了定位销孔径工艺要改定位方案加工要换钻头质检要换塞规我接话这就是一条传播链设计→工艺→加工→质检四跳。如果设计改了三次才定稿每次都全链路传播那就是 12 次交互。把这些交互频次建成网络图谁的度数最高谁就是枢纽。对老刘点头还有我想看变更传播深度——有的变更只影响工艺有的变更一路传到质检。传播越深的变更说明前期设计评审没做到位。最好还能聚类看哪些工程师是一个协同社区。用 networkx 建四角色协同网络节点是工程师边是协同交互频次边权交互次数节点大小度数中心性我开工程pandas 管协同日志numpy 算中心性指标scipy 做社区发现后的模块度检验scikit-learn 对工程师做聚类分群matplotlib 画网络图传播深度直方图。数据自包含合成一批并行工程协同记录下载就能跑。敲了行原型G.add_edge(设计-张三, 工艺-李四, weight12, 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)读取 CSVsource, 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.py4.2 核心源码detailssummary/summary并行工程协同日志加载器import pandas as pdfrom pathlib import Pathfrom typing import OptionalVALID_ROLES {设计, 工艺, 加工, 质检}class CollabLoader:加载协同交互日志def __init__(self, filepath: str collab_log.csv,encoding: str utf-8):self.filepath Path(filepath)self.encoding encodingself._raw: Optional[pd.DataFrame] Nonedef load(self) - pd.DataFrame:if not self.filepath.exists():raise FileNotFoundError(f文件不存在: {self.filepath})self._raw pd.read_csv(self.filepath, encodingself.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] tgtbreakself._raw self._raw.rename(columnsrename)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], errorscoerce).fillna(1).astype(int)self._raw[change_depth] pd.to_numeric(self._raw.get(change_depth), errorscoerce).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(dropTrue)/detailsdetailssummary/summary协同网络构建 (networkx)import networkx as nximport pandas as pdfrom typing import Optionalclass CollabNetworkBuilder:从协同日志构建有向/无向图def __init__(self, directed: bool False):self.directed directedself.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, rolestr(r.get(role_source, 未知)))self.G.add_node(tgt, rolestr(r.get(role_target, 未知)))# 边属性w int(r[interaction_count])if self.G.has_edge(src, tgt):self.G[src][tgt][weight] wself.G[src][tgt][changes].append(str(r.get(change_type, )))else:self.G.add_edge(src, tgt, weightw,changes[str(r.get(change_type, ))],depthint(r.get(change_depth, 0)))return self.Gdef 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),}/detailsdetailssummary/summary中心性分析 (networkx/numpy)import numpy as npimport pandas as pdimport networkx as nxfrom typing import Optional, Dict, Listclass CentralityAnalyzer:计算多种中心性指标def __init__(self):passdef analyze(self, G: nx.Graph) - pd.DataFrame:deg nx.degree_centrality(G)betw nx.betweenness_centrality(G, weightweight, normalizedTrue)close nx.closeness_centrality(G)try:eig nx.eigenvector_centrality_numpy(G, weightweight)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,ascendingFalse).reset_index(dropTrue)def hub_nodes(self, df: pd.DataFrame, top_n: int 3) - pd.DataFrame:枢纽节点 介数中心性最高return df.nlargest(top_n, betweenness_centrality).reset_index(dropTrue)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(dropTrue)/detailsdetailssummary/summary社区发现 (networkx)import networkx as nximport pandas as pdfrom typing import Optional, List, Dictclass CommunityDetector:基于模块度优化的社区发现def __init__(self):passdef 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.0return 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) 1rows.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 commfor v in comm if u ! v and G.has_edge(u, v)) // 2,})return pd.DataFrame(rows)/detailsdetailssummary/summary变更传播分析import numpy as npimport pandas as pdfrom collections import defaultdictfrom typing import Optionalclass PropagationAnalyzer:分析变更传播深度和路径def __init__(self):passdef 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 distdef 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, ascendingFalse).head(top_n)return sub[[source, target, change_type,change_depth, interaction_count]].reset_index(dropTrue)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,ascendingFalse).reset_index(dropTrue)/detailsdetailssummary/summary可视化 (matplotlib)import numpy as npimport pandas as pdimport matplotlib.pyplot as pltfrom pathlib import Pathplt.rcParams[font.sans-serif] [SimHei, DejaVu Sans]plt.rcParams[axes.unicode_minus] Falseclass CollabVisualizer:def __init__(self, results_dir: str results):self.results_dir Path(results_dir)self.results_dir.mkdir(exist_okTrue)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, seed42, k0.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_colornc, node_sizens,axax, alpha0.9, edgecolorsblack, linewidths1.0)nx.draw_networkx_edges(G, pos, widthew, alpha0.35, axax,edge_color#7F8C8D)nx.draw_networkx_labels(G, pos, font_size8, axax,font_weightbold)# 图例for role, color in self.role_colors.items():ax.plot([], [], o, colorcolor, labelrole, markersize10)ax.legend(locupper left, framealpha0.9)ax.set_title(并行工程协同网络\n(颜色角色 大小度数中心性 线宽交互频次),fontsize14, fontweightbold)ax.axis(off)plt.tight_layout()plt.savefig(self.results_dir / collab_network.png,dpi150, bbox_inchestight)plt.close()def centrality_radar(self, centrality_df, top_n5):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), endpointFalse).tolist()angles angles[:1]fig, ax plt.subplots(figsize(10, 10), subplot_kwdict(polarTrue))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 1vals [v / max(centrality_df[m].max(), 1e-6) for m in metrics]vals vals[:1]ax.plot(angles, vals, o-, labelr[node], linewidth2)ax.fill(angles, vals, alpha0.1)ax.set_xticks(angles[:-1])ax.set_xticklabels(labels, fontsize12)ax.set_title(fTop{top_n} 枢纽节点中心性雷达图, fontsize14, fontweightbold)ax.legend(locupper right, bbox_to_anchor(1.3, 1.1))plt.tight_layout()plt.savefig(self.results_dir / centrality_radar.png,dpi150, bbox_inchestight)plt.close()def community_bar(self, comm_df):if comm_df.empty:returnfig, ax plt.subplots(figsize(10, 6))x range(len(comm_df))ax.bar(x, comm_df[size], color#3498DB, alpha0.8)ax.set_xticks(list(x))ax.set_xticklabels([fC{i1} for i in range(len(comm_df))], fontsize11)ax.set_ylabel(社区人数)ax.set_title(社区规模分布, fontsize14, fontweightbold)for i, r in comm_df.iterrows():ax.text(i, r[size] 0.1, r[role_distribution],hacenter, fontsize8)ax.grid(axisy, alpha0.3)plt.tight_layout()plt.savefig(self.results_dir / community_bar.png,dpi150, bbox_inchestight)plt.close()def propagation_hist(self, depth_df):if depth_df.empty:returnfig, ax plt.subplots(figsize(9, 6))ax.bar(depth_df[depth].astype(str), depth_df[count],color#E67E22, alpha0.8)ax.set_xlabel(传播深度(跳数))ax.set_ylabel(变更次数)ax.set_title(变更传播深度分布, fontsize14, fontweightbold)ax.grid(axisy, alpha0.3)total depth_df[count].sum()for _, r in depth_df.iterrows():pct r[count] / total * 100ax.text(str(r[depth]), r[count] 0.2, f{pct:.1f}%,hacenter, fontsize10, fontweightbold)plt.tight_layout()plt.savefig(self.results_dir / propagation_hist.png,dpi150, bbox_inchestight)plt.close()def role_heatmap(self, df):角色间交互热力图df2 df.copy()crosstab pd.crosstab(df2[role_source], df2[role_target],valuesdf2[interaction_count], aggfuncsum).fillna(0)fig, ax plt.subplots(figsize(8, 6))im ax.imshow(crosstab.values, cmapYlOrRd)ax.set_xticks(range(len(crosstab.columns)))ax.set_xticklabels(crosstab.columns, fontsize11)ax.set_yticks(range(len(crosstab.index)))ax.set_yticklabels(crosstab.index, fontsize11)ax.set_title(角色间交互热力图(频次), fontsize14, fontweightbold)for i in range(len(crosstab.index)):for j in range(len(crosstab.columns)):ax.text(j, i, int(crosstab.values[i, j]),hacenter, vacenter, fontsize10,colorblack if crosstab.values[i, j] crosstab.values.max()/2 else white)plt.colorbar(im, axax)plt.tight_layout()plt.savefig(self.results_dir / role_heatmap.png,dpi150, bbox_inchestight)plt.close()/detailsdetailssummary/summary合成并行工程协同数据生成器import numpy as npimport pandas as pdfrom pathlib import Pathfrom typing import Optionalclass 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解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛