1. 从实习需求说起合并关联集合到底难在哪先明确一下我们要解决的问题是什么。假设你手里有一批集合集合之间可能存在交集只要两个集合有公共元素它们就应该被合并成一个更大的集合。这个需求在数据清洗、用户画像合并、图连通性分析里非常常见。拿一个具体例子来说初始列表长这样[A, B, C, D] [E, F, G] [H, I] [A, Q, E]前三行彼此独立但第四行同时包含 A 和 E于是第一行、第二行、第四行全部连通最终应该合并为[A, B, C, D, E, F, G, Q] [H, I]最直觉的做法是两两比较拿每个集合去和已有集合比对有交集就合并合并完再从头扫一遍直到某一轮没有任何合并发生。这个暴力解法的时间复杂度接近 O(n²)而且合并过程中还要反复重建列表数据量一上来就会明显卡顿。我在实际项目里踩过的坑是一开始用 Python 的 list 做嵌套遍历几千个集合跑起来还行到了几十万级别直接超时。后来才意识到这类合并关联集合的问题本质上就是**并查集Union-Find和连通分量Connected Components**这两个经典算法要解决的事情。这篇文章会给出两条可落地的路线一条是并查集适合在线增量合并、动态加边另一条是连通分量适合离线批处理、一次性把整个图跑完。两条路线我都会给出完整可复制的代码并且用 TaoToken 的统一 Key 通道调用模型来生成测试用例验证合并结果是否正确。如果你正在学算法、准备面试或者工程里真的遇到了集合合并的性能瓶颈这篇内容可以直接跟着敲一遍。2. TaoToken 统一 Key 前置准备一次配置调用多个模型在写算法之前先把调用通道准备好。我选择用 TaoToken 的原因是它提供统一的 API Key 和 Base URL不用为每个模型单独维护一套鉴权和地址切换模型只改一个 Model ID 就行。对于我们要做的生成测试用例 校验合并结果这种辅助任务来说非常省事。你需要准备三样东西Base URLhttps://taotoken.net/apiAPI Key在控制台创建地址是https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentapi-keysModel ID按你需要的模型填比如对话类、代码类都可以如果你用的是 Claude Code 这类编码工具接入方式是把 Base URL 和 Key 写进它的配置里Model ID 选一个擅长代码的即可。想先体验模型对话效果可以直接打开https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmodel-chat试一句。对于长期要跑编码和 Agent 任务的场景可以考虑 Coding Plan入口在https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentcoding-plan。接入文档在https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentdoc里面有各语言的调用示例。这里要强调一点TaoToken 只是模型调用的统一通道它不替代你的编辑器也不替代你的算法实现。算法代码还是老老实实写在本地TaoToken 负责的是帮你生成测试数据、解释报错、校验结果。配置的时候记住三件套缺一不可Base URL API Key Model ID。少任何一个都会在请求时报鉴权或路由错误。下面这段是通用的调用配置Python 环境可以直接用import os from openai import OpenAI client OpenAI( base_urlhttps://taotoken.net/api, api_keyos.environ.get(TAOTOKEN_API_KEY), ) resp client.chat.completions.create( modelyour-model-id, messages[ {role: user, content: 生成 5 组用于测试并查集合并的集合数据} ], ) print(resp.choices[0].message.content)把TAOTOKEN_API_KEY设成环境变量不要硬编码在代码里。your-model-id换成你实际选用的模型标识。跑通这一步后面生成测试用例就顺了。3. 可复制配置并查集路径压缩 按秩合并完整实现并查集的核心思想是每个元素先各自成一个集合用一棵树来表示一个集合树根就是集合的代表元。合并两个集合就是把一棵树的根挂到另一棵树的根下面。查找某个元素属于哪个集合就顺着父指针一路找到根。朴素并查集在极端情况下会退化成一条链查找变成 O(n)。所以工程实现里必须加两个优化路径压缩查找时把沿途节点的父指针直接指向根下次查找就是 O(1)。按秩合并合并时把矮树挂到高树下面避免树高增长过快。两个优化一起用单次操作的均摊复杂度接近 O(α(n))α 是反阿克曼函数实际场景里几乎等于常数。下面是完整可复制的 Python 实现class UnionFind: def __init__(self, n): self.parent list(range(n)) self.rank [0] * n self.count n # 当前连通分量个数 def find(self, x): # 路径压缩迭代写法避免递归爆栈 root x while self.parent[root] ! root: root self.parent[root] while self.parent[x] ! root: self.parent[x], x root, self.parent[x] return root def union(self, x, y): rx, ry self.find(x), self.find(y) if rx ry: return False # 按秩合并矮树挂到高树下 if self.rank[rx] self.rank[ry]: rx, ry ry, rx self.parent[ry] rx if self.rank[rx] self.rank[ry]: self.rank[rx] 1 self.count - 1 return True def connected(self, x, y): return self.find(x) self.find(y)把元素映射成 0 到 n-1 的整数下标就能直接用这个结构。对于前面那个字母集合的例子先收集所有出现过的字母建立映射然后对每个集合内部的所有元素两两 union或者更省事把集合内第一个元素和其余每个元素 union最后按根分组即可。def merge_with_union_find(groups): # 收集所有元素 all_items [] seen set() for g in groups: for item in g: if item not in seen: seen.add(item) all_items.append(item) idx {item: i for i, item in enumerate(all_items)} uf UnionFind(len(all_items)) for g in groups: g list(g) if not g: continue first idx[g[0]] for item in g[1:]: uf.union(first, idx[item]) # 按根分组 from collections import defaultdict buckets defaultdict(list) for item in all_items: buckets[uf.find(idx[item])].append(item) return list(buckets.values())跑一下groups [ [A, B, C, D], [E, F, G], [H, I], [A, Q, E], ] print(merge_with_union_find(groups)) # [[A, B, C, D, Q, E, F, G], [H, I]]结果正确。并查集的优势在于如果数据是流式到来的每来一个新集合你只需要对集合内元素做几次 union不需要重新扫描全部历史数据。这就是它适合在线增量场景的原因。4. 连通分量 BFS/DFS 模板与结果验证请求如果数据是一次性给全的或者你更习惯图论视角那么连通分量是另一条清晰路线。把每个元素看成图的一个节点同一个集合内的元素两两连边实际实现里连成一条链就够不必全连接然后对整个图做一次 BFS 或 DFS每个连通块就是一个合并后的集合。先建图再遍历。下面是 BFS 版本from collections import defaultdict, deque def merge_with_connected_components(groups): graph defaultdict(set) for g in groups: g list(g) for i in range(len(g) - 1): graph[g[i]].add(g[i 1]) graph[g[i 1]].add(g[i]) visited set() result [] for node in graph: if node in visited: continue comp [] queue deque([node]) visited.add(node) while queue: cur queue.popleft() comp.append(cur) for nxt in graph[cur]: if nxt not in visited: visited.add(nxt) queue.append(nxt) result.append(comp) return resultDFS 版本把队列换成栈即可逻辑一致。两者复杂度都是 O(V E)V 是元素数E 是边数。现在用 TaoToken 生成一批测试用例验证两种算法结果一致。请求可以这样写prompt 请生成 5 组测试数据每组是一个二维列表 内层列表是若干字符串集合集合之间可能存在交集。 要求覆盖完全独立、两两相交、链式相交、单个大集合、空集合。 只输出 JSON不要解释。 resp client.chat.completions.create( modelyour-model-id, messages[{role: user, content: prompt}], ) print(resp.choices[0].message.content)拿到 JSON 后把每组数据分别喂给merge_with_union_find和merge_with_connected_components比较两者输出的集合划分是否等价把每个结果转成 frozenset 的集合再比对def normalize(result): return {frozenset(g) for g in result} for case in test_cases: a normalize(merge_with_union_find(case)) b normalize(merge_with_connected_components(case)) assert a b, f不一致: {case} print(全部用例通过)如果两种算法在随机数据上都能对齐说明实现没有漏边或漏合并。这一步用模型批量造数据比手写几个用例覆盖度高得多。5. 本篇常见报错排查401、local proxy failed 与 reading choices调通过程中容易撞到几类错误这里逐个对照。401 Unauthorized最常见的原因是 API Key 没设对或者环境变量名写错。检查TAOTOKEN_API_KEY是否真的注入到了当前进程可以用print(os.environ.get(TAOTOKEN_API_KEY)[:8])打印前几位确认。另外注意 Base URL 结尾不要多加/v1之类的路径按文档给的https://taotoken.net/api填。local proxy failed / connection error这类报错通常是本地网络配置或代理设置干扰了请求。先确认你的运行环境没有残留的代理环境变量比如HTTP_PROXY、HTTPS_PROXY有的话临时清掉再试。如果是在容器里跑检查容器网络是否正常。reading choices 相关报错一般是响应结构和你解析的字段不匹配。比如你按resp[choices]取值但实际返回对象需要用resp.choices[0].message.content。用 SDK 时优先用属性访问别自己拼字典。如果返回内容为空检查model字段是不是填了不存在的 Model ID。OAuth / 鉴权失败如果你用的是 Claude Code 或类似工具报 OAuth 相关错误多半是配置里 Base URL、Key、Model ID 三件套没对齐。重新核对一遍Base URL 用https://taotoken.net/apiKey 用控制台新建的Model ID 用工具支持的标识。三者任意一个错位都会导致鉴权链路断掉。算法侧的结果不一致如果两种算法输出对不上先检查建图时是不是漏了单元素集合。一个只含单个元素的集合在并查集里会自成一个分量但在建图时如果不给它加任何边它可能不会出现在graph里导致被漏掉。修法是在建图前把所有元素都初始化成节点哪怕没有边。排查顺序建议先确认请求能通打印一次简单响应再确认数据正确打印中间结果最后才怀疑算法逻辑。大部分算法不对其实是数据没喂对。6. 语义一致 CTA把统一 Key 用进你的算法工作流两种算法各有适用面。并查集适合增量、动态、流式的合并场景加一个集合就是几次 union代价极低连通分量适合离线批处理一次遍历出全部结果代码直观。实际工程里我经常两个都写用连通分量做基准用并查集做在线版本互相校验。把 TaoToken 接进这套工作流之后最省时间的环节是测试数据生成和结果解释。你可以让它批量造边界用例也可以把报错原文贴过去让它帮你定位。统一 Key 的好处是不用为每个模型单独配一套鉴权切换只改 Model ID。需要新建 Key 就去https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentapi-keys接入细节看https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentdoc。想先验证模型输出质量直接开https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmodel-chat对话即可。长期跑编码和 Agent 任务的话Coding Plan 在https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentcoding-plan。最后留一个实用技巧把并查集的find写成迭代版而不是递归版数据量大时能避免递归深度超限连通分量建图时用链式连边而不是全连接边数从 O(k²) 降到 O(k)k 是集合大小内存和遍历时间都会明显下降。这两点在你处理真实数据时比算法本身更容易决定成败。 
长河君 · 2026/10/7 7:11:18 · 阅读 1.2k 并查集与连通分量：合并关联集合的两种算法实战（TaoToken 统一 Key 调用示例）
1. 从实习需求说起合并关联集合到底难在哪先明确一下我们要解决的问题是什么。假设你手里有一批集合集合之间可能存在交集只要两个集合有公共元素它们就应该被合并成一个更大的集合。这个需求在数据清洗、用户画像合并、图连通性分析里非常常见。拿一个具体例子来说初始列表长这样[A, B, C, D] [E, F, G] [H, I] [A, Q, E]前三行彼此独立但第四行同时包含 A 和 E于是第一行、第二行、第四行全部连通最终应该合并为[A, B, C, D, E, F, G, Q] [H, I]最直觉的做法是两两比较拿每个集合去和已有集合比对有交集就合并合并完再从头扫一遍直到某一轮没有任何合并发生。这个暴力解法的时间复杂度接近 O(n²)而且合并过程中还要反复重建列表数据量一上来就会明显卡顿。我在实际项目里踩过的坑是一开始用 Python 的 list 做嵌套遍历几千个集合跑起来还行到了几十万级别直接超时。后来才意识到这类合并关联集合的问题本质上就是**并查集Union-Find和连通分量Connected Components**这两个经典算法要解决的事情。这篇文章会给出两条可落地的路线一条是并查集适合在线增量合并、动态加边另一条是连通分量适合离线批处理、一次性把整个图跑完。两条路线我都会给出完整可复制的代码并且用 TaoToken 的统一 Key 通道调用模型来生成测试用例验证合并结果是否正确。如果你正在学算法、准备面试或者工程里真的遇到了集合合并的性能瓶颈这篇内容可以直接跟着敲一遍。2. TaoToken 统一 Key 前置准备一次配置调用多个模型在写算法之前先把调用通道准备好。我选择用 TaoToken 的原因是它提供统一的 API Key 和 Base URL不用为每个模型单独维护一套鉴权和地址切换模型只改一个 Model ID 就行。对于我们要做的生成测试用例 校验合并结果这种辅助任务来说非常省事。你需要准备三样东西Base URLhttps://taotoken.net/apiAPI Key在控制台创建地址是https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentapi-keysModel ID按你需要的模型填比如对话类、代码类都可以如果你用的是 Claude Code 这类编码工具接入方式是把 Base URL 和 Key 写进它的配置里Model ID 选一个擅长代码的即可。想先体验模型对话效果可以直接打开https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmodel-chat试一句。对于长期要跑编码和 Agent 任务的场景可以考虑 Coding Plan入口在https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentcoding-plan。接入文档在https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentdoc里面有各语言的调用示例。这里要强调一点TaoToken 只是模型调用的统一通道它不替代你的编辑器也不替代你的算法实现。算法代码还是老老实实写在本地TaoToken 负责的是帮你生成测试数据、解释报错、校验结果。配置的时候记住三件套缺一不可Base URL API Key Model ID。少任何一个都会在请求时报鉴权或路由错误。下面这段是通用的调用配置Python 环境可以直接用import os from openai import OpenAI client OpenAI( base_urlhttps://taotoken.net/api, api_keyos.environ.get(TAOTOKEN_API_KEY), ) resp client.chat.completions.create( modelyour-model-id, messages[ {role: user, content: 生成 5 组用于测试并查集合并的集合数据} ], ) print(resp.choices[0].message.content)把TAOTOKEN_API_KEY设成环境变量不要硬编码在代码里。your-model-id换成你实际选用的模型标识。跑通这一步后面生成测试用例就顺了。3. 可复制配置并查集路径压缩 按秩合并完整实现并查集的核心思想是每个元素先各自成一个集合用一棵树来表示一个集合树根就是集合的代表元。合并两个集合就是把一棵树的根挂到另一棵树的根下面。查找某个元素属于哪个集合就顺着父指针一路找到根。朴素并查集在极端情况下会退化成一条链查找变成 O(n)。所以工程实现里必须加两个优化路径压缩查找时把沿途节点的父指针直接指向根下次查找就是 O(1)。按秩合并合并时把矮树挂到高树下面避免树高增长过快。两个优化一起用单次操作的均摊复杂度接近 O(α(n))α 是反阿克曼函数实际场景里几乎等于常数。下面是完整可复制的 Python 实现class UnionFind: def __init__(self, n): self.parent list(range(n)) self.rank [0] * n self.count n # 当前连通分量个数 def find(self, x): # 路径压缩迭代写法避免递归爆栈 root x while self.parent[root] ! root: root self.parent[root] while self.parent[x] ! root: self.parent[x], x root, self.parent[x] return root def union(self, x, y): rx, ry self.find(x), self.find(y) if rx ry: return False # 按秩合并矮树挂到高树下 if self.rank[rx] self.rank[ry]: rx, ry ry, rx self.parent[ry] rx if self.rank[rx] self.rank[ry]: self.rank[rx] 1 self.count - 1 return True def connected(self, x, y): return self.find(x) self.find(y)把元素映射成 0 到 n-1 的整数下标就能直接用这个结构。对于前面那个字母集合的例子先收集所有出现过的字母建立映射然后对每个集合内部的所有元素两两 union或者更省事把集合内第一个元素和其余每个元素 union最后按根分组即可。def merge_with_union_find(groups): # 收集所有元素 all_items [] seen set() for g in groups: for item in g: if item not in seen: seen.add(item) all_items.append(item) idx {item: i for i, item in enumerate(all_items)} uf UnionFind(len(all_items)) for g in groups: g list(g) if not g: continue first idx[g[0]] for item in g[1:]: uf.union(first, idx[item]) # 按根分组 from collections import defaultdict buckets defaultdict(list) for item in all_items: buckets[uf.find(idx[item])].append(item) return list(buckets.values())跑一下groups [ [A, B, C, D], [E, F, G], [H, I], [A, Q, E], ] print(merge_with_union_find(groups)) # [[A, B, C, D, Q, E, F, G], [H, I]]结果正确。并查集的优势在于如果数据是流式到来的每来一个新集合你只需要对集合内元素做几次 union不需要重新扫描全部历史数据。这就是它适合在线增量场景的原因。4. 连通分量 BFS/DFS 模板与结果验证请求如果数据是一次性给全的或者你更习惯图论视角那么连通分量是另一条清晰路线。把每个元素看成图的一个节点同一个集合内的元素两两连边实际实现里连成一条链就够不必全连接然后对整个图做一次 BFS 或 DFS每个连通块就是一个合并后的集合。先建图再遍历。下面是 BFS 版本from collections import defaultdict, deque def merge_with_connected_components(groups): graph defaultdict(set) for g in groups: g list(g) for i in range(len(g) - 1): graph[g[i]].add(g[i 1]) graph[g[i 1]].add(g[i]) visited set() result [] for node in graph: if node in visited: continue comp [] queue deque([node]) visited.add(node) while queue: cur queue.popleft() comp.append(cur) for nxt in graph[cur]: if nxt not in visited: visited.add(nxt) queue.append(nxt) result.append(comp) return resultDFS 版本把队列换成栈即可逻辑一致。两者复杂度都是 O(V E)V 是元素数E 是边数。现在用 TaoToken 生成一批测试用例验证两种算法结果一致。请求可以这样写prompt 请生成 5 组测试数据每组是一个二维列表 内层列表是若干字符串集合集合之间可能存在交集。 要求覆盖完全独立、两两相交、链式相交、单个大集合、空集合。 只输出 JSON不要解释。 resp client.chat.completions.create( modelyour-model-id, messages[{role: user, content: prompt}], ) print(resp.choices[0].message.content)拿到 JSON 后把每组数据分别喂给merge_with_union_find和merge_with_connected_components比较两者输出的集合划分是否等价把每个结果转成 frozenset 的集合再比对def normalize(result): return {frozenset(g) for g in result} for case in test_cases: a normalize(merge_with_union_find(case)) b normalize(merge_with_connected_components(case)) assert a b, f不一致: {case} print(全部用例通过)如果两种算法在随机数据上都能对齐说明实现没有漏边或漏合并。这一步用模型批量造数据比手写几个用例覆盖度高得多。5. 本篇常见报错排查401、local proxy failed 与 reading choices调通过程中容易撞到几类错误这里逐个对照。401 Unauthorized最常见的原因是 API Key 没设对或者环境变量名写错。检查TAOTOKEN_API_KEY是否真的注入到了当前进程可以用print(os.environ.get(TAOTOKEN_API_KEY)[:8])打印前几位确认。另外注意 Base URL 结尾不要多加/v1之类的路径按文档给的https://taotoken.net/api填。local proxy failed / connection error这类报错通常是本地网络配置或代理设置干扰了请求。先确认你的运行环境没有残留的代理环境变量比如HTTP_PROXY、HTTPS_PROXY有的话临时清掉再试。如果是在容器里跑检查容器网络是否正常。reading choices 相关报错一般是响应结构和你解析的字段不匹配。比如你按resp[choices]取值但实际返回对象需要用resp.choices[0].message.content。用 SDK 时优先用属性访问别自己拼字典。如果返回内容为空检查model字段是不是填了不存在的 Model ID。OAuth / 鉴权失败如果你用的是 Claude Code 或类似工具报 OAuth 相关错误多半是配置里 Base URL、Key、Model ID 三件套没对齐。重新核对一遍Base URL 用https://taotoken.net/apiKey 用控制台新建的Model ID 用工具支持的标识。三者任意一个错位都会导致鉴权链路断掉。算法侧的结果不一致如果两种算法输出对不上先检查建图时是不是漏了单元素集合。一个只含单个元素的集合在并查集里会自成一个分量但在建图时如果不给它加任何边它可能不会出现在graph里导致被漏掉。修法是在建图前把所有元素都初始化成节点哪怕没有边。排查顺序建议先确认请求能通打印一次简单响应再确认数据正确打印中间结果最后才怀疑算法逻辑。大部分算法不对其实是数据没喂对。6. 语义一致 CTA把统一 Key 用进你的算法工作流两种算法各有适用面。并查集适合增量、动态、流式的合并场景加一个集合就是几次 union代价极低连通分量适合离线批处理一次遍历出全部结果代码直观。实际工程里我经常两个都写用连通分量做基准用并查集做在线版本互相校验。把 TaoToken 接进这套工作流之后最省时间的环节是测试数据生成和结果解释。你可以让它批量造边界用例也可以把报错原文贴过去让它帮你定位。统一 Key 的好处是不用为每个模型单独配一套鉴权切换只改 Model ID。需要新建 Key 就去https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentapi-keys接入细节看https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentdoc。想先验证模型输出质量直接开https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmodel-chat对话即可。长期跑编码和 Agent 任务的话Coding Plan 在https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentcoding-plan。最后留一个实用技巧把并查集的find写成迭代版而不是递归版数据量大时能避免递归深度超限连通分量建图时用链式连边而不是全连接边数从 O(k²) 降到 O(k)k 是集合大小内存和遍历时间都会明显下降。这两点在你处理真实数据时比算法本身更容易决定成败。