stream.analysis

stream.analysis

Project-scoped analysis over grounded notes.

Two capabilities, both reusing the zettelkasten matrix engine (tables.py) as the read-time projection of a project's notes:

• :func:build_matrix / :func:matrix_to_csv / :func:export_csv -- assemble a comparison grid for a (project, schema) and emit it as CSV. Deterministic by default (no LLM): cells fill from the dimension-tagged notes the extraction pipeline wrote.
• :func:synthesize_narrative -- hand the grounded matrix (values + verbatim evidence quotes) to a :class:~stream.protocols.Driver and ask for a narrative synthesis, optionally written back as an agent-authored _cross synthesis note (tagged origin:agent for governance).

The matrix is the same one the dashboard renders, so CSV/synthesis stay consistent with the UI. tables.py and the store are imported lazily.

build_matrix

build_matrix(project: str, *, schema: str = '', columns: list[dict[str, Any]] | None = None, row_axis: dict[str, Any] | None = None, ai: bool = False, driver: Driver | None = None, model: str = '', table_id: str = 'stream-matrix', title: str = '') -> dict[str, Any]

Build the synthesis matrix for a project.

By default (ai=False) this is a cost-free deterministic projection: dimension-tagged notes route into schema_tag columns, one row per source. Pass ai=True with a driver to also fill free-text prompt columns and per-cell summaries via the agent (this persists the grid).

Source code in stream/analysis.py

def build_matrix(
    project: str,
    *,
    schema: str = "",
    columns: list[dict[str, Any]] | None = None,
    row_axis: dict[str, Any] | None = None,
    ai: bool = False,
    driver: Driver | None = None,
    model: str = "",
    table_id: str = "stream-matrix",
    title: str = "",
) -> dict[str, Any]:
    """Build the synthesis matrix for a project.

    By default (``ai=False``) this is a cost-free deterministic projection:
    dimension-tagged notes route into ``schema_tag`` columns, one row per source.
    Pass ``ai=True`` with a ``driver`` to also fill free-text ``prompt`` columns
    and per-cell summaries via the agent (this persists the grid).
    """
    from zettelkasten import server as zk
    # ``zettelkasten.tables.build_matrix`` (formerly ``build_table``) is aliased to
    # ``_build_matrix`` on purpose: it collides by name with this module's own
    # ``build_matrix`` above. The alias is what keeps the call below delegating to
    # the tables engine rather than recursing into this function.
    from zettelkasten.tables import GRAPHS_DIR, build_matrix as _build_matrix

    if columns is None:
        if not schema:
            raise ValueError("build_matrix needs either explicit columns or a schema.")
        columns = _schema_columns(schema)

    extract_fn = summarize_fn = None
    if ai:
        if driver is None:
            raise ValueError("ai=True requires a driver.")
        extract_fn = _driver_text_fn(driver, model)
        summarize_fn = _driver_text_fn(driver, model)

    return _build_matrix(
        zk._get_graph,
        project=project,
        name=project,
        table_id=table_id,
        title=title or f"{project} ({schema})" if schema else project,
        columns=columns,
        row_axis=row_axis,
        preview=not ai,  # deterministic, no persist, unless AI fills are requested
        force=ai,
        extract_fn=extract_fn,
        summarize_fn=summarize_fn,
        graphs_dir=GRAPHS_DIR,
    )

matrix_to_csv

matrix_to_csv(table: dict[str, Any]) -> str

Render a built matrix as CSV text (one row per source, columns as headers).

Source code in stream/analysis.py

def matrix_to_csv(table: dict[str, Any]) -> str:
    """Render a built matrix as CSV text (one row per source, columns as headers)."""
    columns = table.get("columns", [])
    rows = table.get("rows", [])
    buf = io.StringIO()
    writer = csv.writer(buf)
    writer.writerow(["source", *[c.get("label", c.get("key", "")) for c in columns]])
    for row in rows:
        cells = row.get("cells", {})
        line = [row.get("label", row.get("id", ""))]
        for col in columns:
            cell = cells.get(col.get("key"), {})
            value = cell.get("value") or cell.get("summary") or ""
            if cell.get("gap"):
                value = ""
            line.append(value)
        writer.writerow(line)
    return buf.getvalue()

export_csv

export_csv(project: str, path: str, *, schema: str = '', columns: list[dict[str, Any]] | None = None, **kwargs: Any) -> str

Build the matrix and write it to path as CSV. Returns the path.

Source code in stream/analysis.py

def export_csv(
    project: str,
    path: str,
    *,
    schema: str = "",
    columns: list[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> str:
    """Build the matrix and write it to ``path`` as CSV. Returns the path."""
    table = build_matrix(project, schema=schema, columns=columns, **kwargs)
    with open(path, "w", encoding="utf-8", newline="") as fh:
        fh.write(matrix_to_csv(table))
    return path

render_grounded_matrix

render_grounded_matrix(table: dict[str, Any], *, max_quote: int = 240) -> str

Render the matrix as grounded text: values plus verbatim evidence quotes.

Source code in stream/analysis.py

def render_grounded_matrix(table: dict[str, Any], *, max_quote: int = 240) -> str:
    """Render the matrix as grounded text: values plus verbatim evidence quotes."""
    columns = table.get("columns", [])
    rows = table.get("rows", [])
    lines: list[str] = []
    header = " | ".join(["source", *[c.get("label", "") for c in columns]])
    lines.append(header)
    lines.append("-" * len(header))
    evidence: list[str] = []
    for row in rows:
        cells = row.get("cells", {})
        label = row.get("label", row.get("id", ""))
        values = []
        for col in columns:
            cell = cells.get(col.get("key"), {})
            if cell.get("gap"):
                values.append("")
            else:
                val = cell.get("value") or ""
                # Surface advisory low-confidence so the synthesizing agent can
                # hedge on weakly-extracted cells. Confidence is 0..1; flag only.
                if cell.get("low_confidence") and val:
                    conf = cell.get("confidence")
                    pct = f"{round(float(conf) * 100)}%" if isinstance(conf, (int, float)) else "low"
                    val = f"{val} [⚠ low confidence: {pct}]"
                values.append(val)
            ev = cell.get("evidence") or {}
            quote = (ev.get("quote") or "").strip() if isinstance(ev, dict) else ""
            if quote:
                evidence.append(
                    f"[{label} / {col.get('label','')}] \"{quote[:max_quote]}\""
                )
        lines.append(" | ".join([label, *values]))
    if evidence:
        lines.append("\nEVIDENCE (verbatim):")
        lines.extend(evidence)
    return "\n".join(lines)

synthesize_narrative

synthesize_narrative(project: str, *, driver: Driver, schema: str = '', columns: list[dict[str, Any]] | None = None, question: str = '', model: str = '', write_back: bool = False, cross_graph: str = _CROSS_GRAPH, title: str = '') -> dict[str, Any]

Build the grounded matrix and ask a Driver to synthesize a narrative.

Returns {narrative, matrix, note_id}. When write_back is set the narrative is committed as a synthesis note in cross_graph, claimed by project and tagged origin:agent so agent-authored writeback is distinguishable from human notes (governance).

Source code in stream/analysis.py

def synthesize_narrative(
    project: str,
    *,
    driver: Driver,
    schema: str = "",
    columns: list[dict[str, Any]] | None = None,
    question: str = "",
    model: str = "",
    write_back: bool = False,
    cross_graph: str = _CROSS_GRAPH,
    title: str = "",
) -> dict[str, Any]:
    """Build the grounded matrix and ask a Driver to synthesize a narrative.

    Returns ``{narrative, matrix, note_id}``. When ``write_back`` is set the
    narrative is committed as a ``synthesis`` note in ``cross_graph``, claimed by
    ``project`` and tagged ``origin:agent`` so agent-authored writeback is
    distinguishable from human notes (governance).
    """
    table = build_matrix(project, schema=schema, columns=columns)
    grounded = render_grounded_matrix(table)

    ask = question or (
        f"Synthesize the key cross-source findings for project '{project}'"
        + (f" under the '{schema}' lens" if schema else "")
        + ". Compare and contrast the sources, call out agreements, divergences, "
        "and notable gaps. Ground every claim in the evidence provided."
    )
    persona = (
        "You are a research analyst. Write a tight, well-structured narrative "
        "synthesis grounded ONLY in the matrix and verbatim evidence given. Do "
        "not invent facts. Note where evidence is missing rather than guessing."
    )
    prompt = f"{ask}\n\nGROUNDED MATRIX:\n{grounded}"
    narrative = driver.run_agent(
        role="checker", persona=persona, prompt=prompt, tools=[], model=model
    ).text

    note_id = ""
    if write_back:
        from zettelkasten import server as zk

        res = zk.add_note(
            graph=cross_graph,
            title=title or f"Synthesis: {project}{(' / ' + schema) if schema else ''}",
            type="synthesis",
            body=narrative,
            tags=["stream-synthesis", "origin:agent"],
            project=project,
        )
        if isinstance(res, dict):
            note_id = res.get("id") or res.get("note_id") or ""

    return {"narrative": narrative, "matrix": table, "note_id": note_id}