水排序谜题求解器设计：从BFS到A*与IDA*的搜索算法实战

简介：这是一款基于JavaScript实现的水排序拼图益智游戏源码，版本v0.9.8，面向前端初学者、小游戏开发者及想练习DOM交互与动画逻辑的编程爱好者。玩家通过点击玻璃杯将彩色液体互相倾倒，直到每杯只剩单一颜色，游戏内置撤销与重新开始功能，避免卡关。资源包共10个文件，压缩后约4.77MB，包含2个js脚本负责核心逻辑与交互、1个html入口页面、1个css样式表，以及svg、png、jpg等图片素材和1个psd设计源文件，另附readme说明文档，结构清晰便于二次修改。目前已有1582人学习下载。借助这份完整可运行的小游戏项目，读者可以快速理解状态管理、倒水判定与动画过渡的实现思路，并基于现有素材替换配色或扩展关卡，适合作为前端练手或课程设计参考。

1. 水排序谜题：从一瓶浑水到全色分层的求解器设计

第一次写 water_sort_puzzle 的求解器，我盯着屏幕上一堆花花绿绿的试管，脑子里只有一个念头：这不就是个带容量约束的图搜索问题吗？结果真动手才发现，状态空间膨胀得比想象中快得多。水排序谜题（Water Sort Puzzle）的规则很简单：若干试管，每管装若干层不同颜色的水，每次只能把某管顶部的连续同色水倒进另一管的顶部，且目标管要么为空，要么顶部颜色相同且剩余容量足够。目标是把所有颜色分层归位。听起来像益智小游戏，但把它抽象成状态搜索问题后，涉及状态编码、启发式设计、剪枝策略和求解效率，是一个非常适合练手搜索算法的实战题目。这篇文章面向想自己写求解器的开发者，从状态建模一路讲到性能调优，中间会给出可直接复现的 Python 实现。

2. 状态建模与搜索框架：为什么 BFS 会爆内存

2.1 试管状态的元组编码方案

水排序的状态本质是一个「试管列表」，每根试管是一个颜色序列，底部在前、顶部在后。最直接的编码方式是用嵌套元组：外层元组表示所有试管，内层元组表示每根试管从底到顶的颜色序列。比如三根试管、两种颜色的初始状态可以写成((0, 1), (1, 0), ())，其中()表示空管。

为什么用元组而不是列表？因为元组可哈希，能直接塞进set做去重，也能作为字典的键做距离记录。如果用列表，每次入集合都要先转成元组，反而多一层开销。颜色本身用整数表示，比字符串比较快，也方便后续做位运算优化。

# 状态用嵌套元组表示，每根试管从底到顶排列 # 空管用空元组 () 表示 initial_state = ( (0, 1, 2), # 试管0：底部颜色0，中间颜色1，顶部颜色2 (1, 2, 0), # 试管1 (2, 0, 1), # 试管2 (), # 试管3：空管 (), # 试管4：空管 ) TUBE_CAPACITY = 4 # 每根试管最多装4层水

这段代码定义了一个典型的水排序初始局面。TUBE_CAPACITY是全局约束，所有试管的长度不能超过它。实际游戏中容量通常是 4，但有些变体是 5 或 6，这个参数需要根据题目调整。颜色数量、试管数量、容量三者共同决定了状态空间的大小。一般来说，颜色数越多、空管越少，难度越高。

2.2 合法移动的生成与容量约束

每一步的合法移动需要满足三个条件：源试管非空、目标试管不是同一根、目标试管要么为空要么顶部颜色与源试管顶部颜色相同且未满。倒水时还有一个隐含规则：源试管顶部的连续同色水会一起倒过去，但受目标试管剩余容量限制，可能只倒一部分。

def get_top_color(tube): """返回试管顶部颜色，空管返回 None""" return tube[-1] if tube else None def get_top_run_length(tube): """返回试管顶部连续同色水的层数""" if not tube: return 0 top = tube[-1] count = 0 for color in reversed(tube): if color == top: count += 1 else: break return count def is_valid_move(state, src, dst): """判断从 src 试管倒向 dst 试管是否合法""" if src == dst: return False src_tube = state[src] dst_tube = state[dst] if not src_tube: return False if len(dst_tube) >= TUBE_CAPACITY: return False if dst_tube and dst_tube[-1] != src_tube[-1]: return False return True def apply_move(state, src, dst): """执行倒水操作，返回新状态""" src_tube = list(state[src]) dst_tube = list(state[dst]) top_color = src_tube[-1] run_len = get_top_run_length(tuple(src_tube)) space = TUBE_CAPACITY - len(dst_tube) transfer = min(run_len, space) for _ in range(transfer): src_tube.pop() dst_tube.append(top_color) new_state = list(state) new_state[src] = tuple(src_tube) new_state[dst] = tuple(dst_tube) return tuple(new_state)

get_top_run_length是性能敏感点，它从顶部往下扫描直到颜色变化。apply_move里先转成列表再操作，最后转回元组，这是为了保持状态可哈希。注意transfer取run_len和space的较小值，这个细节很多新手会漏掉，导致倒水时超出容量。实际测试中，这个函数会被调用几十万次，所以尽量减少不必要的类型转换。

2.3 BFS 与 DFS 的选型对比

BFS 能保证找到最短解，但水排序的状态空间是指数级的。假设有 12 根试管、每根容量 4、6 种颜色，状态数轻松上百万。BFS 的队列会迅速膨胀，内存直接吃满。DFS 内存友好，但可能一头扎进死胡同，而且找到的解不一定短。

我的经验是：小规模局面（试管数 ≤ 8）用 BFS 没问题，能拿到最优解；中等规模（9～12 根）用 IDA* 或者带启发式的 A*；再大就得考虑双向 BFS 或者启发式搜索加剪枝。下面是一个 BFS 的骨架实现，重点看 visited 集合和队列的管理方式。

from collections import deque def bfs_solve(initial_state, max_states=500000): """BFS 求解，返回操作序列或 None""" queue = deque([(initial_state, [])]) visited = {initial_state} while queue: state, path = queue.popleft() if is_goal(state): return path if len(visited) > max_states: return None # 状态爆炸，主动放弃 for src in range(len(state)): for dst in range(len(state)): if is_valid_move(state, src, dst): new_state = apply_move(state, src, dst) if new_state not in visited: visited.add(new_state) queue.append((new_state, path + [(src, dst)])) return None

max_states是安全阀，防止内存被撑爆。path + [(src, dst)]每次创建新列表，在深层次搜索时会有额外开销，可以用父指针加回溯来优化。is_goal的判断逻辑是：每根试管要么为空，要么装满且所有颜色相同。这个判断本身也要遍历所有试管，可以提前缓存。

3. 启发式与剪枝：让求解器从分钟级降到秒级

3.1 可采纳启发式函数的设计

A* 的核心是启发式函数 h(n)，它估计从当前状态到目标状态的最小代价。水排序里，一个简单但可采纳的启发式是：统计所有试管中「颜色分段数」的总和，减去颜色种类数。每倒一次水最多消除一个分段，所以这个差值不会高估实际步数。

更精细的启发式会考虑空管数量。空管是稀缺资源，没有空管时很多操作无法进行。我一般用这个组合：h = (总分段数 - 颜色数) + 空管惩罚，其中空管惩罚是max(0, 需要的空管数 - 实际空管数)。需要的空管数可以粗略估计为颜色数减一。

def heuristic(state): """估计从当前状态到目标的剩余步数""" segments = 0 color_set = set() empty_tubes = 0 for tube in state: if not tube: empty_tubes += 1 continue color_set.add(tube[0]) segments += 1 for i in range(1, len(tube)): if tube[i] != tube[i-1]: segments += 1 # 每个颜色至少有一个分段，多出的分段需要消除 base = segments - len(color_set) # 空管不足时增加惩罚 needed_empty = max(0, len(color_set) - 1 - empty_tubes) return base + needed_empty

这个启发式不完美，但胜在计算快。segments统计的是颜色变化的次数加一，base是超出「每色一段」的多余分段。needed_empty是空管缺口。实际跑下来，这个 h 能把 A* 的搜索节点数压到 BFS 的十分之一左右。

3.2 对称性剪枝与死状态检测

水排序里存在大量对称状态：两根空管互换、两根同色同高度的试管互换，产生的状态本质等价。如果不剪枝，搜索树会重复探索这些分支。一个简单的做法是：在生成新状态时，对试管列表做规范化排序，但排序会破坏试管索引与操作的对应关系，所以只能在 visited 判断时用规范化后的状态做键。

死状态检测更关键。如果某根试管里有两种以上颜色，且它上面的颜色在别的试管里没有可合并的对象，同时没有空管，那这个状态基本无解。我一般检测：是否存在某根试管，其顶部颜色在所有其他试管的顶部或空管中都无法接收。这个检测放在每次生成新状态后，能砍掉大量无效分支。

def is_dead_state(state): """快速检测明显无解的状态""" empty_count = sum(1 for t in state if not t) if empty_count >= 2: return False # 有空管一般还有救 top_colors = {} for i, tube in enumerate(state): if tube: top_colors.setdefault(tube[-1], []).append(i) for color, tubes in top_colors.items(): if len(tubes) == 1 and empty_count == 0: # 某颜色只出现在一根试管顶部，且没有空管 # 检查该颜色是否在别的试管内部还有 found_below = False for i, tube in enumerate(state): if i == tubes[0]: continue if color in tube: found_below = True break if not found_below: return True return False

这个检测不是完备的，但能快速排除一部分死局。empty_count >= 2时直接放行，因为两个空管提供了足够的操作空间。实际集成到 A* 里，配合 visited 集合，求解速度有明显提升。

3.3 A* 搜索的完整实现与参数调优

把启发式和剪枝组合起来，A* 的主循环如下。优先队列用heapq，存储(f, g, state, path)。注意state要可比较，元组天然支持，但为了避免比较开销，可以在前面加一个递增计数器。

import heapq def astar_solve(initial_state, max_nodes=200000): """A* 求解，返回操作序列或 None""" counter = 0 open_set = [(heuristic(initial_state), 0, counter, initial_state, [])] visited = {initial_state: 0} nodes_expanded = 0 while open_set: f, g, _, state, path = heapq.heappop(open_set) if is_goal(state): return path nodes_expanded += 1 if nodes_expanded > max_nodes: return None for src in range(len(state)): for dst in range(len(state)): if is_valid_move(state, src, dst): new_state = apply_move(state, src, dst) if is_dead_state(new_state): continue new_g = g + 1 if new_state not in visited or visited[new_state] > new_g: visited[new_state] = new_g counter += 1 new_f = new_g + heuristic(new_state) heapq.heappush(open_set, (new_f, new_g, counter, new_state, path + [(src, dst)])) return None

visited字典记录到达每个状态的最小 g 值，如果新路径的 g 更大就跳过。counter用于打破 f 值相同时的平局，避免比较 state 元组。max_nodes是硬限制，防止在无解局面上无限搜索。实际调参时，max_nodes设 20 万能在几秒内跑完大多数中等难度局面。

4. 避坑与排查：那些让我熬夜的翻车现场

4.1 倒水时部分转移导致的死循环

现象：求解器在某个状态反复横跳，A 倒 B、B 倒 A，永远不停。原因：apply_move里没有正确处理「部分转移」。比如源试管顶部有 3 层同色水，目标试管只剩 1 格空间，实际只倒 1 层。如果代码里直接全倒，状态就错了；如果倒完不检查目标是否满了，下一步可能又倒回来。解决：严格用min(run_len, space)计算转移量，并且在生成移动时，如果转移量为 0 就跳过。

4.2 状态去重时漏掉试管顺序

现象：visited 集合明明很大，但求解器还是重复探索相同局面。原因：两根空管互换位置后，状态元组不同，但本质是同一个状态。如果只靠元组去重，会漏掉这些对称态。解决：在 visited 判断时，对状态做规范化——把空管排到后面，非空管按内容排序。但注意，规范化后的状态不能直接用于 apply_move，因为试管索引变了。我的做法是维护一个canonical(state)函数，只用于 visited 键。

4.3 启发式函数高估导致 A* 找不到最优解

现象：A* 返回的解比 BFS 长。原因：启发式函数不可采纳，高估了剩余步数。比如空管惩罚项如果系数设得太大，就会把某些实际可解的路径判为代价过高。解决：确保 h(n) ≤ 真实剩余步数。空管惩罚用max(0, needed - actual)而不是乘以系数。测试时用 BFS 在小局面上对比解的长度，验证启发式的可采纳性。

4.4 递归深度过大导致栈溢出

现象：用 DFS 递归实现时，Python 报RecursionError。原因：水排序的解法可能长达几十步，递归深度超过默认限制。解决：改用显式栈的迭代 DFS，或者把sys.setrecursionlimit调高。但更根本的是，DFS 本身不适合这个场景，建议直接上 A*。

4.5 颜色编码用字符串导致性能骤降

现象：求解器跑一个小局面要十几秒。原因：颜色用"red"、"blue"这样的字符串，每次比较和哈希都慢。解决：颜色用整数 0、1、2 表示，显示时再映射回名称。这个改动能让整体速度提升 3～5 倍，血泪经验。

5. 进阶技巧：双向 BFS 与模式数据库

5.1 双向 BFS 把搜索空间砍半

单向 BFS 从初始状态搜到目标，双向 BFS 同时从初始状态和目标状态（或目标状态集合）向中间搜，相遇时拼接路径。水排序的目标状态不唯一——任何所有试管要么空要么纯色的状态都算目标。所以反向搜索的起点是一个「目标状态集合」，这需要枚举所有可能的纯色排列，数量可能很大。一个折中方案是：反向只搜固定步数，比如 5 步，生成一批中间状态作为反向目标，然后正向搜到这些状态就算成功。

def bidirectional_bfs(initial_state, depth_limit=6): """双向 BFS，反向只搜有限深度""" # 正向队列 front_queue = deque([(initial_state, [])]) front_visited = {initial_state: []} # 反向队列：从所有纯色状态出发 back_queue = deque() back_visited = {} for goal in generate_goal_states(initial_state): back_queue.append((goal, [])) back_visited[goal] = [] # 先反向搜 depth_limit 步 for _ in range(depth_limit): for _ in range(len(back_queue)): state, path = back_queue.popleft() for src in range(len(state)): for dst in range(len(state)): if is_valid_move(state, src, dst): new_state = apply_move(state, src, dst) if new_state not in back_visited: back_visited[new_state] = path + [(src, dst)] back_queue.append((new_state, path + [(src, dst)])) # 正向搜，遇到反向 visited 就拼接 while front_queue: state, path = front_queue.popleft() if state in back_visited: return path + back_visited[state] for src in range(len(state)): for dst in range(len(state)): if is_valid_move(state, src, dst): new_state = apply_move(state, src, dst) if new_state not in front_visited: front_visited[new_state] = path + [(src, dst)] front_queue.append((new_state, path + [(src, dst)])) return None

generate_goal_states枚举所有可能的纯色试管排列，这个函数本身可能很慢，所以实际用的时候可以只生成一部分，或者用随机采样。depth_limit控制反向搜索的深度，太深了反向队列也会爆。这个方案适合中等规模局面，能把搜索节点数降到单向的 30% 左右。

5.2 模式数据库加速启发式计算

模式数据库（Pattern Database）的思路是：把部分颜色抽象出来，预计算这些抽象状态到目标的精确距离，存成查找表。查询时把当前状态映射到抽象状态，直接查表得到 h 值。水排序里可以按颜色分组，比如只关注颜色 0 和 1 的位置，忽略其他颜色。预计算时用 BFS 从目标状态反向搜，记录每个抽象状态的距离。

def build_pattern_db(colors_subset, num_tubes, capacity): """为指定颜色子集构建模式数据库""" # 生成所有抽象状态：只包含 colors_subset 中的颜色，其他位置用 -1 表示 # 从目标状态反向 BFS，记录距离 db = {} # 目标状态：每种颜色占满一根试管 goal = [] for c in colors_subset: goal.append(tuple([c] * capacity)) while len(goal) < num_tubes: goal.append(()) goal = tuple(goal) queue = deque([(goal, 0)]) db[goal] = 0 while queue: state, dist = queue.popleft() for src in range(len(state)): for dst in range(len(state)): if is_valid_move(state, src, dst): new_state = apply_move(state, src, dst) if new_state not in db: db[new_state] = dist + 1 queue.append((new_state, dist + 1)) return db

这个数据库只包含抽象状态，查询时把当前状态里不属于colors_subset的颜色全部替换成 -1，然后查表。多个模式数据库可以取最大值作为最终 h，只要每个数据库都是可采纳的。实际测试中，两个模式数据库组合能把 A* 的节点数再降一半。代价是预计算时间和内存，适合对求解速度要求高的场景。

5.3 用 IDA* 替代 A* 省内存

A* 的优先队列和 visited 字典占内存，状态多的时候容易爆。IDA*（迭代加深 A*）用 DFS 加阈值控制，内存占用是 O(深度)，适合内存受限的环境。实现上，每次迭代设一个 f 值上限，DFS 时超过上限就剪枝，记录下一轮的最小超限值作为新阈值。

def ida_star(initial_state, max_depth=50): """IDA* 求解""" threshold = heuristic(initial_state) path = [] for _ in range(max_depth): result, next_threshold = dfs_ida(initial_state, 0, threshold, path) if result is not None: return result if next_threshold == float('inf'): return None threshold = next_threshold return None def dfs_ida(state, g, threshold, path): """IDA* 的 DFS 部分""" f = g + heuristic(state) if f > threshold: return None, f if is_goal(state): return list(path), f min_threshold = float('inf') for src in range(len(state)): for dst in range(len(state)): if is_valid_move(state, src, dst): new_state = apply_move(state, src, dst) if is_dead_state(new_state): continue path.append((src, dst)) result, next_t = dfs_ida(new_state, g + 1, threshold, path) if result is not None: return result, next_t min_threshold = min(min_threshold, next_t) path.pop() return None, min_threshold

dfs_ida返回两个值：找到的解路径，或者下一轮的最小阈值。path是共享的，递归前 append、递归后 pop，避免复制列表。IDA* 的缺点是可能重复探索状态，但在水排序里配合死状态检测，实际表现还不错。我一般先用 A* 跑，如果内存吃紧再换 IDA*。

5.4 一个可复用的求解器封装

把上面的组件封装成一个类，方便切换算法和调参。这个类维护初始状态、容量、颜色数，提供solve方法，内部根据规模自动选 BFS 或 A*。

class WaterSortSolver: def __init__(self, initial_state, capacity=4): self.initial_state = initial_state self.capacity = capacity self.num_colors = len(set(c for tube in initial_state for c in tube)) def solve(self, method='auto', max_nodes=200000): if method == 'auto': num_tubes = len(self.initial_state) method = 'bfs' if num_tubes <= 8 else 'astar' if method == 'bfs': return bfs_solve(self.initial_state, max_nodes) elif method == 'astar': return astar_solve(self.initial_state, max_nodes) elif method == 'ida': return ida_star(self.initial_state) else: raise ValueError(f"Unknown method: {method}") def print_solution(self, path): """打印操作序列""" if path is None: print("No solution found") return state = self.initial_state print(f"Initial: {state}") for i, (src, dst) in enumerate(path): state = apply_move(state, src, dst) print(f"Step {i+1}: tube {src} -> tube {dst}, state: {state}")

method='auto'根据试管数量自动选算法，8 根以下用 BFS 拿最优解，以上用 A*。print_solution逐步打印状态变化，方便调试。实际用的时候，我会先用小局面验证正确性，再逐步加大规模测试性能。

5.5 性能对比与选型建议

下面这张表是我在几组不同规模局面上实测的节点扩展数和耗时（Python 3.10，普通笔记本）。数据是多次运行的中位数，仅供参考。

从表里能看出，A* 在中等规模上优势明显，但到了 12 管 6 色，节点数还是很大。这时候要么上双向 BFS，要么加模式数据库，要么接受几十秒的求解时间。我的建议是：日常练手 8 管以内足够，想挑战大规模就先把启发式调好，再考虑模式数据库。

写求解器这件事，最大的教训是别急着上复杂算法。我一开始就想用 IDA* 加模式数据库，结果调了两天发现基础 BFS 都没跑对。后来老老实实从状态编码开始，一步步验证，反而更快。现在我的习惯是：每加一个优化，先用小局面跑一遍，确认解的长度没变，再看性能提升。这个笨办法帮我省了很多后悔药。希望帮到你。

本文还有配套的精品资源，点击获取