Attractor — Knowledge Graph - Source

Projects / wikigen-code-excerpts-readme

Attractor — Knowledge Graph - Source

wikigencontext-and-memory

A look at concept extraction, consolidation, and layout — the steps that turn conversation files into a graph.

README · ARCHITECTURE · TECHNICAL · source

Find an excerpt

1. Extraction

One call per conversation. Three of the comments in this function are bugs that took real time to find.

def extract_concepts(client, messages, conversation_name, model=DEFAULT_MODEL) -> list[str]:
    """Ask Claude for 3-5 key concepts from a conversation. Returns concept names."""
    if not messages:
        return []

    truncated = [
        {"role": m["role"], "content": m["content"][:800] + ("..." if len(m["content"]) > 800 else "")}
        for m in messages[:40]
    ]

    system = """You are a knowledge curator. Identify the 3-5 most distinct, memorable concepts
from this conversation — things like ideas, techniques, frameworks, questions, or insights
that would be worth their own wiki article.

Return ONLY a JSON array of short concept names (2-5 words each). No explanation, no markdown.
Example: ["context window optimization", "residual stream", "attention heads"]

If the conversation has no notable concepts, return []."""

    try:
        response = client.messages.create(
            model=model,
            # Generous for a 3-5 item list, because reasoning models spend
            # this budget on a thinking block before writing any text — 300 was
            # enough for Haiku and truncated Opus mid-JSON.
            max_tokens=2000,
            # Only where the model accepts it. SDK 1.x removed temperature from
            # the messages.create() signature, so it goes via extra_body — but
            # newer models reject the parameter outright ("`temperature` is
            # deprecated for this model"), and extra_body passes it straight
            # through to that rejection. The workaround for the SDK change is
            # exactly what breaks the request on Opus 5.
            **({"extra_body": {"temperature": 0.2}} if _accepts_temperature(model) else {}),
            system=system,
            messages=truncated + [{"role": "user", "content": "List the key concepts from this conversation."}],
        )
        # Not content[0]: reasoning models put a ThinkingBlock first, which has
        # no .text. Take the first actual text block.
        text = next((b.text for b in response.content if getattr(b, "type", "") == "text"), "").strip()
        text = text.replace("```json", "").replace("```", "").strip()
        concepts = json.loads(text)
        if isinstance(concepts, list):
            return [str(c).strip() for c in concepts if c][:5]
    except Exception as e:
        print(f"  [warn] concept extraction failed for {conversation_name}: {e}")

    return []

All three comments describe the same class of problem: code that works on one model and silently fails on another. A max_tokens that is generous for Haiku truncates Opus mid-JSON. A content[0] that is correct for a non-reasoning model returns a thinking block that has no .text. A workaround for an SDK change is itself what the newer model rejects.

Every one of those produced an empty concept list, not an exception — a conversation that quietly contributed nothing to the graph. Extraction failures degrade to a smaller graph rather than a crash, which is the right behavior and also the reason they are hard to notice.

2. Consolidation

The second kind of model call: one pass over every concept name in the corpus.

def consolidate_concepts(client, concepts, model):
    """
    Merge near-synonymous concept names into canonical ones.

    Each conversation is analysed in isolation, so the same idea comes back
    phrased differently every time — "local JSON state", "...persistence",
    "...storage" and "...management" are four nodes for one idea. Because they
    share neighbours the layout correctly stacks them, and the graph becomes a
    pile of overlapping labels. No amount of layout tuning fixes that: the
    duplicates have to be merged before the graph is built.

    Returns {original: canonical}, containing only the names being changed.
    """
    if len(concepts) < 2:
        return {}

    system = """You are consolidating concept names extracted from separate conversations.
The same idea is often phrased differently across them.

Group names that mean the same thing and pick the clearest, shortest name for each group.
Merge only genuine synonyms — "local JSON state" and "local JSON state persistence" are the
same idea; "model routing" and "model consistency" are not. When in doubt, leave a name alone.

Return ONLY a JSON object mapping each original name to its canonical name. Include only the
names you are merging; omit anything that stays as-is. No explanation, no markdown."""

    try:
        response = client.messages.create(
            model=model, max_tokens=8000, system=system,
            messages=[{"role": "user", "content": json.dumps(sorted(concepts))}],
        )
        text = next((b.text for b in response.content if getattr(b, "type", "") == "text"), "")
        clean = text.replace("```json", "").replace("```", "").strip()
        mapping = json.loads(clean)
        return {k: v for k, v in mapping.items() if isinstance(v, str) and k != v}
    except Exception as e:
        print(f"  consolidation failed ({e}) — continuing with unmerged concepts")
        return {}

Two design choices worth naming. The prompt asks for only the names being changed, so the return value is a diff rather than a full mapping — cheaper, and it makes an over-eager merge visible by reading the response. And the failure path continues with unmerged concepts rather than aborting: a cluttered graph beats no graph.

when in doubt, leave a name alone is doing more work than its length suggests. A missed merge shows up as visible duplicate labels. A wrong merge silently destroys the distinction the graph exists to show.

3. Layout packing

A long comment that earns its length.

def compute_layout(G: nx.Graph, seed: int = 42) -> dict:
    if len(G.nodes) == 0:
        return {}
    if len(G.nodes) == 1:
        return {list(G.nodes)[0]: (0.5, 0.5)}

    components = sorted(nx.connected_components(G), key=len, reverse=True)

    def _spring(g, spread=1.5):
        k = spread / (len(g.nodes) ** 0.5)
        return nx.spring_layout(g, k=k, iterations=80, seed=seed)

    if len(components) == 1:
        return _spring(G)

    # One conversation per cluster means the graph is usually disconnected.
    # spring_layout flings whole components apart with nothing between them, so
    # each cluster collapses into an unreadable knot in a mostly-empty frame.
    #
    # Lay each component out on its own, then pack the boxes. Box size scales
    # with sqrt(node count): a uniform grid gives a forty-node cluster the same
    # canvas as a three-node one, which compresses it far harder and produces a
    # pile of overlapping labels — the small clusters look fine and the one
    # that matters most is illegible.
    #
    # Packing is a simple shelf algorithm: place boxes left to right, wrap to a
    # new row when the row is full, row height set by its tallest box.

seed=42 makes a repeated layout reproducible for the same graph. In the Haiku-versus-Opus comparison in the README, the extracted graph structure changes too, so positions can still differ. The fixed seed removes one source of randomness; it does not make node position an independent model measure.

The sqrt(node count) box sizing is the specific insight. A uniform grid is the obvious implementation and it fails in the worst possible way — the clusters that matter most are the biggest, so uniform boxes compress exactly the part of the picture you most wanted to read, while the trivial three-node clusters look fine.

4. The noise filter

def is_noisy(messages: list[dict], threshold: float = 0.55) -> bool:
    """Return True if the conversation is mostly terminal output / error logs."""
    if not messages:
        return True
    all_text = "\n".join(m["content"] for m in messages)
    noise_lines = len(_NOISE_RE.findall(all_text))
    total_lines = max(all_text.count("\n"), 1)
    total_chars = len(all_text)
    # Also skip very short conversations (< 500 chars of real content)
    if total_chars < 500:
        return True
    return (noise_lines / total_lines) > threshold

0.55 and 500 are tuned against my own corpus and are the first two numbers to change on someone else's. The filter runs before the API call, not after, because not paying for concept extraction on a stack trace is the entire purpose.

5. The chat client that made the corpus

Fifty-two lines. It exists because retyping a script to send one more prompt was intolerable, and the timestamped transcripts it writes are the only reason there is anything to graph.

# src/chat.py — writes chat_history/<timestamp>.txt, every session, unasked.

Worth including because it is the actual origin of the project: the tool was built to escape an annoyance, and the by-product turned out to be the dataset.