← All Problems

35. Group Similar Transcripts by Edit Distance

General Pattern Hard Union-Find + DP — Edit-Distance Clustering
Grounding: General industry practice: deduplicating near-identical ASR output (a retried recognition pass, or two overlapping partial transcripts of the same utterance) via edit-distance clustering is standard in streaming voice pipelines. This is not a confirmed detail of Cartesia's own Ink-2 transcription pipeline — the research behind this problem set found published WER and latency numbers for Ink-2 (cartesia.ai/blog/ink-2) but no public description of how, or whether, Cartesia deduplicates transcripts internally.

Problem

A real-time voice pipeline can emit more than one transcript for the same underlying utterance: a retried ASR pass, a partial-then-corrected hypothesis, or two barge-in attempts of the same command. Before these get logged as distinct events, group the transcripts that are near-duplicates of each other.

Two transcripts belong to the same group if their Levenshtein edit distance is at most max_distance, and group membership is transitive: if A is close to B and B is close to C, all three land in one group even if A and C alone exceed the threshold.

Source: src/35_group_similar_transcripts.py

def group_similar_transcripts(transcripts: list[str], max_distance: int) -> list[list[str]]:
    ...

Examples:
>>> group_similar_transcripts(
...     ["turn on the lights", "turn on the light", "call mom", "call mum"], 1
... )
[['call mom', 'call mum'], ['turn on the light', 'turn on the lights']]

>>> group_similar_transcripts(["hi", "hi", "bye"], 0)
[['bye'], ['hi', 'hi']]

Step-by-Step Approach

  1. Recognize the two-layer structure: a pairwise similarity test (edit distance) feeding a connectivity problem (union-find), not a single algorithm on its own.
  2. Compute the Levenshtein edit distance between every pair of transcripts with the standard O(len(a) · len(b)) DP: build a table where each cell is the min edits to turn one prefix into the other, using insert/delete/substitute.
  3. Whenever a pair's distance is within max_distance, union their indices in a union-find (disjoint set) structure instead of grouping them directly.
  4. Use path-halving (or path compression) in find so union-find operations stay near O(α(n)) each, making the whole clustering step dominated by the O(n²) pairwise distance computation.
  5. After all pairs are checked, bucket every transcript by its union-find root — transcripts sharing a root are one connected component, i.e. one group.
  6. Sort each group's contents alphabetically, then sort the list of groups by each group's first (smallest) element, to get a deterministic output order.

The key insight is that similarity is not required to be pairwise-consistent at output time — grouping is defined by the transitive closure of the "close enough" relation, which is exactly what union-find gives you for free, without needing to define a single distance threshold between every pair inside a cluster.

Reference solution

def _edit_distance(a: str, b: str) -> int:
    # standard Levenshtein DP, O(len(a) * len(b)) time, O(len(b)) space (rolling rows)
    m, n = len(a), len(b)
    prev = list(range(n + 1))
    for i in range(1, m + 1):
        curr = [i] + [0] * n
        for j in range(1, n + 1):
            if a[i - 1] == b[j - 1]:
                curr[j] = prev[j - 1]
            else:
                curr[j] = 1 + min(prev[j], curr[j - 1], prev[j - 1])
        prev = curr
    return prev[n]


def group_similar_transcripts(transcripts: list[str], max_distance: int) -> list[list[str]]:
    n = len(transcripts)
    parent = list(range(n))

    def find(x: int) -> int:
        # path-halving find, keeps future lookups near O(1)
        while parent[x] != x:
            parent[x] = parent[parent[x]]
            x = parent[x]
        return x

    def union(x: int, y: int) -> None:
        rx, ry = find(x), find(y)
        if rx != ry:
            parent[ry] = rx

    # O(n^2) pairwise edit-distance check, then union any pair within threshold
    for i in range(n):
        for j in range(i + 1, n):
            if _edit_distance(transcripts[i], transcripts[j]) <= max_distance:
                union(i, j)

    groups: dict[int, list[str]] = {}
    for i in range(n):
        groups.setdefault(find(i), []).append(transcripts[i])

    result = [sorted(group) for group in groups.values()]
    result.sort(key=lambda group: group[0])
    return result


TEST_CASES = [
    {
        "input": {
            "transcripts": ["turn on the lights", "turn on the light", "call mom", "call mum"],
            "max_distance": 1,
        },
        "expected": [["call mom", "call mum"], ["turn on the light", "turn on the lights"]],
    },
    {
        "input": {"transcripts": ["hi", "hi", "bye"], "max_distance": 0},
        "expected": [["bye"], ["hi", "hi"]],
    },
    {
        "input": {"transcripts": ["only one"], "max_distance": 2},
        "expected": [["only one"]],
    },
    {
        "input": {"transcripts": [], "max_distance": 1},
        "expected": [],
    },
    {
        "input": {"transcripts": ["cat", "cot", "dot"], "max_distance": 1},
        "expected": [["cat", "cot", "dot"]],
    },
    {
        "input": {"transcripts": ["abc", "xyz"], "max_distance": 1},
        "expected": [["abc"], ["xyz"]],
    },
]


def main():
    for i, case in enumerate(TEST_CASES):
        transcripts = case["input"]["transcripts"]
        max_distance = case["input"]["max_distance"]
        expected = case["expected"]
        print(f"Test {i}: group_similar_transcripts(transcripts={transcripts}, max_distance={max_distance})")
        n = len(transcripts)
        for a in range(n):
            for b in range(a + 1, n):
                d = _edit_distance(transcripts[a], transcripts[b])
                verdict = "UNION" if d <= max_distance else "skip"
                print(f"  dist({transcripts[a]!r}, {transcripts[b]!r}) = {d} -> {verdict}")
        result = group_similar_transcripts(transcripts=transcripts, max_distance=max_distance)
        assert result == expected
        print(f"PASSED: {result}")
    print(f"All {len(TEST_CASES)} test cases passed.")


if __name__ == "__main__":
    main()

Key Functions & Tricks

  • _edit_distance(a, b) — rolling-row Levenshtein DP, O(len(a) · len(b)) time, O(len(b)) space
  • parent = list(range(n)) — union-find init: every index starts as its own root
  • find(x) — path-halving root lookup: parent[x] = parent[parent[x]] keeps future lookups fast
  • union(x, y) — merges two components by pointing one root at the other
  • groups.setdefault(find(i), []).append(...) — buckets each transcript by its final union-find root
  • result.sort(key=lambda group: group[0]) — deterministic output order after alphabetizing each group

How to Recognize This Pattern

The tell is language like “group items that are similar to each other” where similarity isn't transitive by definition but the grouping should behave as if it were — that's a signal to reach for union-find layered on top of whatever pairwise similarity check the domain calls for (edit distance here, but it could be embedding cosine similarity, shared keys, or anything else). A common variation is being handed the similar pairs directly instead of computing them, which drops the O(n²) distance pass entirely. A common pitfall is grouping greedily pair by pair without union-find: that produces inconsistent clusters as soon as a single similarity relation isn't already an equivalence relation, which real-world thresholded distances almost never are.