ontocast.cli.plot_graph

Attributes

DEFAULT_FORMATS = ('svg', 'png') module-attribute

FORMATS = ('svg', 'png', 'pdf') module-attribute

NodeShape = Literal['process', 'decision', 'terminal'] module-attribute

frontmatter_config = {'config': {'theme': 'base', 'look': 'handDrawn', 'themeVariables': {'primaryColor': '#FFF3E0', 'primaryBorderColor': '#143642', 'primaryTextColor': '#372237', 'lineColor': '#FFAB91', 'fontFamily': "'Architects Daughter', cursive", 'fontSize': '20px'}, 'flowchart': {'curve': 'basis', 'htmlLabels': True, 'useMaxWidth': True}}} module-attribute

logger = logging.getLogger(__name__) module-attribute

Classes

FlowEdge dataclass

Source code in ontocast/cli/plot_graph.py

@dataclass(frozen=True)
class FlowEdge:
    start: str
    end: str
    label: str = ""
    conditional: bool = False

Attributes

conditional = False class-attribute instance-attribute

end instance-attribute

label = '' class-attribute instance-attribute

start instance-attribute

Methods:

__init__(start, end, label='', conditional=False)

FlowGraph dataclass

Source code in ontocast/cli/plot_graph.py

@dataclass(frozen=True)
class FlowGraph:
    nodes: tuple[FlowNode, ...]
    edges: tuple[FlowEdge, ...]
    start_node: str
    end_node: str

Attributes

edges instance-attribute

end_node instance-attribute

nodes instance-attribute

start_node instance-attribute

Methods:

__init__(nodes, edges, start_node, end_node)

FlowNode dataclass

Source code in ontocast/cli/plot_graph.py

@dataclass(frozen=True)
class FlowNode:
    node_id: str
    label: str
    shape: NodeShape = "process"

Attributes

label instance-attribute

node_id instance-attribute

shape = 'process' class-attribute instance-attribute

Methods:

__init__(node_id, label, shape='process')

Functions:

draw_flow_graphviz(pgv_module, flow, fname, extensions, rankdir='TB')

Source code in ontocast/cli/plot_graph.py

def draw_flow_graphviz(
    pgv_module: Any,
    flow: FlowGraph,
    fname: str,
    extensions: tuple[str, ...],
    rankdir: str = "TB",
) -> None:
    is_lr = rankdir == "LR"

    viz: Any = pgv_module.AGraph(directed=True, strict=False)
    viz.graph_attr.update(
        rankdir=rankdir,
        bgcolor=_BG_COLOR,
        pad="0.3" if is_lr else "0.6",
        nodesep="0.35" if is_lr else "0.7",
        ranksep="0.5" if is_lr else "0.9",
        fontname=_FONTNAME,
        splines="spline",
    )
    viz.node_attr.update(
        shape="box",
        style="rounded,filled",
        fillcolor=_NODE_FILL,
        color=_NODE_BORDER,
        fontcolor=_NODE_FONT,
        fontsize="11" if is_lr else "13",
        fontname=_FONTNAME,
        margin="0.15,0.08" if is_lr else "0.25,0.12",
        penwidth="1.5" if is_lr else "1.8",
    )
    viz.edge_attr.update(
        fontname=_FONTNAME,
        fontsize="9" if is_lr else "10",
        fontcolor=_EDGE_COLOR,
        color=_EDGE_COLOR,
        penwidth="1.2" if is_lr else "1.5",
        arrowsize="0.7" if is_lr else "0.9",
    )

    for node in flow.nodes:
        label = _flow_label_for_graphviz(node.label, is_lr)
        if node.shape == "decision":
            viz.add_node(
                node.node_id,
                label=label,
                shape="diamond",
                fontsize="10" if is_lr else "11",
                margin="0.12,0.06" if is_lr else "0.18,0.10",
            )
        else:
            viz.add_node(node.node_id, label=label)

    for edge in flow.edges:
        if edge.conditional:
            viz.add_edge(
                edge.start,
                edge.end,
                label=edge.label,
                style="dashed",
                color=_COND_EDGE_COLOR,
                fontcolor=_COND_EDGE_COLOR,
            )
        else:
            viz.add_edge(edge.start, edge.end, label=edge.label)

    accent_attrs = {
        "fillcolor": _ACCENT_FILL,
        "color": _ACCENT_BORDER,
        "fontcolor": _ACCENT_BORDER,
        "penwidth": "2.5",
    }
    viz.get_node(flow.start_node).attr.update(**accent_attrs)
    viz.get_node(flow.end_node).attr.update(**accent_attrs)

    _write(viz, fname, extensions, rankdir)

draw_graphviz(pgv_module, graph, fname, extensions, rankdir='TB')

Source code in ontocast/cli/plot_graph.py

def draw_graphviz(
    pgv_module: Any,
    graph: "Graph",
    fname: str,
    extensions: tuple[str, ...],
    rankdir: str = "TB",
) -> None:
    is_lr = rankdir == "LR"
    splines = "spline"

    viz: Any = pgv_module.AGraph(directed=True, strict=False)
    viz.graph_attr.update(
        rankdir=rankdir,
        bgcolor=_BG_COLOR,
        pad="0.3" if is_lr else "0.6",
        nodesep="0.35" if is_lr else "0.7",
        ranksep="0.28" if is_lr else "0.9",
        fontname=_FONTNAME,
        splines=splines,
    )
    viz.node_attr.update(
        shape="box",
        style="rounded,filled",
        fillcolor=_NODE_FILL,
        color=_NODE_BORDER,
        fontcolor=_NODE_FONT,
        fontsize="14" if is_lr else "13",
        fontname=_FONTNAME,
        margin="0.1,0.06" if is_lr else "0.25,0.12",
        penwidth="1.5" if is_lr else "1.8",
    )
    viz.edge_attr.update(
        fontname=_FONTNAME,
        fontsize="9" if is_lr else "10",
        fontcolor=_EDGE_COLOR,
        color=_EDGE_COLOR,
        penwidth="1.2" if is_lr else "1.5",
        arrowsize="0.7" if is_lr else "0.9",
    )

    hidden_nodes = {"__start__", "__end__"} if is_lr else set()

    for node in graph.nodes:
        if node in hidden_nodes:
            continue
        label = _NODE_LABELS.get(node, node)
        if is_lr:
            label = _wrap_label(label)
        viz.add_node(node, label=label)

    for start, end, data, conditional in graph.edges:
        if start in hidden_nodes or end in hidden_nodes:
            continue
        raw_label = str(data) if data is not None else ""
        label = _NODE_LABELS.get(raw_label, raw_label)
        if conditional:
            viz.add_edge(
                start,
                end,
                label=label,
                style="dashed",
                color=_COND_EDGE_COLOR,
                fontcolor=_COND_EDGE_COLOR,
            )
        else:
            viz.add_edge(start, end, label=label)

    accent_attrs = {
        "fillcolor": _ACCENT_FILL,
        "color": _ACCENT_BORDER,
        "fontcolor": _ACCENT_BORDER,
        "penwidth": "2.5",
    }
    if first := graph.first_node():
        if first.id not in hidden_nodes:
            viz.get_node(first.id).attr.update(**accent_attrs)
    if last := graph.last_node():
        if last.id not in hidden_nodes:
            viz.get_node(last.id).attr.update(**accent_attrs)

    _write(viz, fname, extensions, rankdir)

facts_loop_flow(*, include_evidence=False)

Source code in ontocast/cli/plot_graph.py

def facts_loop_flow(*, include_evidence: bool = False) -> FlowGraph:
    return unit_loop_flow(
        "facts",
        include_evidence=include_evidence,
        passes_var="FACTS_CRITIC_PASSES",
    )

flow_graph_to_mermaid(flow)

Source code in ontocast/cli/plot_graph.py

def flow_graph_to_mermaid(flow: FlowGraph) -> str:
    lines = [_mermaid_frontmatter(), "flowchart TD;"]
    for node in flow.nodes:
        lines.append(_mermaid_node(node))
    for edge in flow.edges:
        arrow = "-.->" if edge.conditional else "-->"
        suffix = f"|{edge.label}|" if edge.label else ""
        lines.append(f"\t{edge.start} {arrow}{suffix} {edge.end};")
    return "\n".join(lines) + "\n"

main(output_dir, formats)

Render the workflow graph and per-unit loop diagrams.

Rendering needs only the compiled graph topology: the LLM client is never set up and the stores are in memory, so no provider credentials or services are required.

Source code in ontocast/cli/plot_graph.py

@click.command()
@click.option(
    "--output-dir",
    type=click.Path(file_okay=False, path_type=Path),
    default=Path("docs/assets"),
    show_default=True,
    help="Directory to write diagrams into. Created if absent.",
)
@click.option(
    "--format",
    "formats",
    default=",".join(DEFAULT_FORMATS),
    show_default=True,
    help=f"Comma-separated output formats, any of: {', '.join(FORMATS)}.",
)
def main(output_dir: Path, formats: str) -> None:
    """Render the workflow graph and per-unit loop diagrams.

    Rendering needs only the compiled graph topology: the LLM client is never
    set up and the stores are in memory, so no provider credentials or
    services are required.
    """
    extensions = tuple(f.strip().lower() for f in formats.split(",") if f.strip())
    unknown = sorted(set(extensions) - set(FORMATS))
    if unknown:
        raise click.BadParameter(
            f"unknown format(s): {', '.join(unknown)}", param_hint="--format"
        )
    output_dir = Path(output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)

    config = Config.in_memory()
    config.tool_config.path_config.ontology_directory = None
    # Not set up, so no provider client is built; the graph never calls it.
    llm = LLMTool(config=config.tool_config.llm_config)
    toolbox = ToolBox(config, llm=llm)

    app = create_agent_graph(toolbox)
    graph = app.get_graph()
    mmd_data = graph.draw_mermaid(frontmatter_config=frontmatter_config)

    (output_dir / "graph.mmd").write_text(mmd_data)

    graph_stem = str(output_dir / "graph")
    try:
        pgv_module = importlib.import_module("pygraphviz")

        draw_graphviz(pgv_module, graph, graph_stem, extensions, rankdir="TB")
        draw_graphviz(pgv_module, graph, graph_stem, extensions, rankdir="LR")
        write_atomic_loop_diagrams(pgv_module, output_dir, extensions)
    except ImportError as e:
        logger.info(f"pygraphviz not available, skipping graphviz output: {e}")

ontology_loop_flow(*, include_evidence=False)

Source code in ontocast/cli/plot_graph.py

def ontology_loop_flow(*, include_evidence: bool = False) -> FlowGraph:
    return unit_loop_flow(
        "ontology",
        include_evidence=include_evidence,
        passes_var="ONTOLOGY_CRITIC_PASSES",
    )

unit_loop_flow(phase, *, include_evidence=False, passes_var)

The per-unit loop: one core shared by both phases, plus facts-only exits.

The facts phase adds three things the ontology phase does not have. A skip gate spends no critic call on a citation-metadata unit or an empty render. A critic that returns no critique leaves the loop unpatched and unreviewed. And an optional insert-only completion stage runs after the critic loop, however that loop ended -- except when every render failed.

Parameters:

Name	Type	Description	Default
phase	str	"facts" or "ontology".	required
include_evidence	bool	Draw the optional web-evidence branches.	False
passes_var	str	The setting that bounds the critic passes, named on the diagram so the picture and the configuration agree.	required

Source code in ontocast/cli/plot_graph.py

def unit_loop_flow(
    phase: str, *, include_evidence: bool = False, passes_var: str
) -> FlowGraph:
    """The per-unit loop: one core shared by both phases, plus facts-only exits.

    The facts phase adds three things the ontology phase does not have. A skip
    gate spends no critic call on a citation-metadata unit or an empty render.
    A critic that returns no critique leaves the loop unpatched and unreviewed.
    And an optional insert-only completion stage runs after the critic loop,
    however that loop ended -- except when every render failed.

    Args:
        phase: ``"facts"`` or ``"ontology"``.
        include_evidence: Draw the optional web-evidence branches.
        passes_var: The setting that bounds the critic passes, named on the
            diagram so the picture and the configuration agree.
    """
    render_node = f"render_{phase}"
    critic_node = f"criticise_{phase}"
    noun = "facts" if phase == "facts" else "ontology"
    is_facts = phase == "facts"
    review_entry = "skip_critic" if is_facts else critic_node
    # Where the critic loop hands over once it stops.
    loop_exit = "completion" if is_facts else "done"
    common = (
        FlowNode("start", "Unit start", "terminal"),
        FlowNode("ctx", "Resolve / apply<br/>ontology context"),
        FlowNode("render_loop", "failed render<br/>1 … MAX_VISITS", "decision"),
        FlowNode(render_node, f"Render {noun}"),
        FlowNode("pass_loop", f"critic pass<br/>1 … {passes_var}", "decision"),
        FlowNode("findings", "Deterministic checks<br/>(no LLM call)"),
        FlowNode(critic_node, f"Criticise {noun}<br/>(cites statement ids)"),
        FlowNode("patch", "Compile, screen, apply<br/>patch (no LLM call)"),
        FlowNode(
            "converged",
            "no fix kept, and a rollback<br/>or no mandatory findings left?",
            "decision",
        ),
        FlowNode("done", "Return unit state", "terminal"),
        FlowNode("exhausted", "Return (retries exhausted)", "terminal"),
    )
    facts_nodes = (
        FlowNode(
            "skip_critic",
            "citation metadata, or fewer than<br/>FACTS_CRITIC_MIN_TRIPLES triples?",
            "decision",
        ),
        FlowNode(
            "completion",
            "Completion passes<br/>0 … FACTS_COMPLETION_PASSES<br/>(insert-only)",
        ),
    )
    facts_edges = (
        FlowEdge("skip_critic", critic_node, "no", conditional=True),
        FlowEdge("skip_critic", "completion", "yes (no critic call)", conditional=True),
        FlowEdge(critic_node, "completion", "unavailable (no patch)", conditional=True),
        FlowEdge("completion", "done"),
    )
    if include_evidence:
        nodes = (
            *common,
            FlowNode("render_fail_search", "initiate_search?", "decision"),
            FlowNode("evid_r", "Plan + fetch<br/>web evidence"),
            FlowNode("rerender", f"Re-render {noun}"),
            FlowNode("critic_fail_search", "initiate_search?", "decision"),
            FlowNode("evid_c", "Plan + fetch<br/>web evidence"),
            FlowNode("recritic", f"Re-criticise {noun}"),
        )
        loop_edges = _unit_loop_evidence_edges(
            render_node=render_node,
            critic_node=critic_node,
            review_entry=review_entry,
        )
        if is_facts:
            facts_edges = (
                *facts_edges,
                FlowEdge(
                    "recritic",
                    "completion",
                    "unavailable (no patch)",
                    conditional=True,
                ),
            )
    else:
        nodes = common
        loop_edges = _unit_loop_core_edges(
            render_node=render_node,
            critic_node=critic_node,
            review_entry=review_entry,
        )
    if is_facts:
        nodes = (*nodes, *facts_nodes)
    edges = (
        FlowEdge("start", "ctx"),
        FlowEdge("ctx", "render_loop"),
        *loop_edges,
        FlowEdge("patch", "converged"),
        FlowEdge("converged", loop_exit, "yes", conditional=True),
        FlowEdge("converged", "pass_loop", "no", conditional=True),
        FlowEdge("pass_loop", loop_exit, "budget spent", conditional=True),
        *(facts_edges if is_facts else ()),
    )
    return FlowGraph(
        nodes=nodes,
        edges=edges,
        start_node="start",
        end_node="done",
    )

write_atomic_loop_diagrams(pgv_module, output_dir, extensions=DEFAULT_FORMATS)

Source code in ontocast/cli/plot_graph.py

def write_atomic_loop_diagrams(
    pgv_module: Any,
    output_dir: Path,
    extensions: tuple[str, ...] = DEFAULT_FORMATS,
) -> None:
    assets = Path(output_dir)
    assets.mkdir(parents=True, exist_ok=True)
    for name, builder in (
        ("facts_loop", lambda: facts_loop_flow(include_evidence=False)),
        ("facts_loop_evidence", lambda: facts_loop_flow(include_evidence=True)),
        ("ontology_loop", lambda: ontology_loop_flow(include_evidence=False)),
        ("ontology_loop_evidence", lambda: ontology_loop_flow(include_evidence=True)),
    ):
        flow = builder()
        mmd_path = assets / f"{name}.mmd"
        mmd_path.write_text(flow_graph_to_mermaid(flow))
        print(f"Wrote {mmd_path}")
        base = assets / name
        draw_flow_graphviz(pgv_module, flow, str(base), extensions, rankdir="TB")
        draw_flow_graphviz(pgv_module, flow, str(base), extensions, rankdir="LR")