hiphopsllm.architecture

Architecture extraction: a LangGraph application as an analysable system model.

The first HiP-HOPS phase is model the system: components with ports, and the connections between them. Here that model is read out of the object a LangGraph notebook already renders — graph.get_graph() — rather than being drawn by hand, so the analysis cannot drift away from the code it describes.

Agent graphs contain feedback (ReAct loops); fault trees cannot. acyclic unrolls each loop to a stated depth and closes it with a feedback-cut component, which keeps the loop’s contribution in the tree instead of silently deleting it.

The model

hiphopsllm.architecture.model — architecture meta-model and extraction from a LangGraph graph.

This is Phase 0 of the HiP-HOPS process (“model the architecture”). HiP-HOPS analyses a hierarchical model of components connected through ports; before any failure logic can be attached we need that model. In a LangGraph application the architecture is already there — it is exactly the object the notebook renders with:

display(Image(graph.get_graph().draw_mermaid_png()))

graph.get_graph() returns a drawable graph carrying .nodes and .edges. This module turns it into a SystemModel: typed components, explicit input/output ports, and directed connections between ports. Three input routes are supported so the analysis works inside the notebook and offline:

1. a compiled LangGraph / any object exposing get_graph();

2. the mermaid source string (graph.get_graph().draw_mermaid());

3. a plain dict specification (SystemModel.from_spec()).

Two modelling decisions are worth stating explicitly because they shape every fault tree produced downstream:

• Conditional edges become a component. In LangGraph a router function is not a node — it is a callable attached to add_conditional_edges. It is nevertheless a real piece of software with its own failure modes (a regular expression that matches the wrong branch, or no branch at all). We therefore materialise it as a distinct ROUTER component sitting between the deciding node and its successors.

• Fan-in is a shared-state hazard. Where two nodes write the same LangGraph state channel in one super-step, a non-reducer channel raises InvalidUpdateError and the run dies. Fan-in connections are flagged here (Connection.fan_in) so the failure library can attach channel-contention events to them.

class hiphopsllm.architecture.model.Role

Bases: str, Enum

Archetype of a component, which selects its default failure logic.

SOURCE = 'source'

SINK = 'sink'

LLM_AGENT = 'llm_agent'

TOOL = 'tool'

ROUTER = 'router'

AGGREGATOR = 'aggregator'

TRANSFORM = 'transform'

FEEDBACK = 'feedback_cut'

__new__(value)

class hiphopsllm.architecture.model.Component

Bases: object

One architectural block with ports, akin to a HiP-HOPS component.

id: str

label: str

role: Role

ports_in: List[str]

ports_out: List[str]

resources: Dict[str, str]

shared physical/logical resources — the basis for common-cause grouping e.g. {"llm": "Qwen/Qwen2.5-Math-1.5B-Instruct", "runtime": "gpu:0"}

source_code: str = ''

branches: List[str]

notes: List[str]

metadata: Dict[str, Any]

property is_boundary: bool

port_in(index=0)

Return type:

str

Parameters:

index (int)

port_out(index=0)

Return type:

str

Parameters:

index (int)

__init__(id, label, role, ports_in=<factory>, ports_out=<factory>, resources=<factory>, source_code='', branches=<factory>, notes=<factory>, metadata=<factory>)

Parameters:

• id (str)

• label (str)

• role (Role)

• ports_in (List[str])

• ports_out (List[str])

• resources (Dict[str, str])

• source_code (str)

• branches (List[str])

• notes (List[str])

• metadata (Dict[str, Any])

Return type:

None

class hiphopsllm.architecture.model.Connection

Bases: object

A directed port-to-port link (a HiP-HOPS ‘connection’/channel).

src: str

src_port: str

dst: str

dst_port: str

label: str = ''

conditional: bool = False

fan_in: bool = False

parallel: bool = False

property id: str

__init__(src, src_port, dst, dst_port, label='', conditional=False, fan_in=False, parallel=False)

Parameters:

• src (str)

• src_port (str)

• dst (str)

• dst_port (str)

• label (str)

• conditional (bool)

• fan_in (bool)

• parallel (bool)

Return type:

None

class hiphopsllm.architecture.model.SystemModel

Bases: object

The full architecture: components + connections, plus lookup helpers.

name: str

components: Dict[str, Component]

connections: List[Connection]

metadata: Dict[str, Any]

component(cid)

Return type:

Component

Parameters:

cid (str)

incoming(cid, port=None)

Return type:

List[Connection]

Parameters:

• cid (str)

• port (str | None)

outgoing(cid, port=None)

Return type:

List[Connection]

Parameters:

• cid (str)

• port (str | None)

predecessors(cid)

Return type:

List[str]

Parameters:

cid (str)

successors(cid)

Return type:

List[str]

Parameters:

cid (str)

sinks()

Return type:

List[str]

sources()

Return type:

List[str]

by_role(role)

Return type:

List[Component]

Parameters:

role (Role)

common_cause_groups()

Group components that share a resource.

A shared resource (the same model snapshot, the same GPU, the same tokenizer, the same prompt template) is a common-cause failure (CCF) candidate: redundancy built from components in one group does not deliver the failure independence the architecture appears to promise.

Return type:

Dict[Tuple[str, str], List[str]]

classmethod from_spec(spec, **kwargs)

Build a model from a plain dict (offline / unit-test route).

spec schema:

{"name": str,
 "nodes": {node_id: {"role": str|Role, "resources": {...},
                     "source_code": str, "label": str}},
 "edges": [(src, dst) | (src, dst, label, conditional)]}

kwargs are forwarded to build_system_model(), so role and resource overrides work on a specification exactly as they do on a live LangGraph.

Return type:

SystemModel

Parameters:

• spec (Dict[str, Any])

• kwargs (Any)

architecture_table()

Return type:

List[Dict[str, str]]

to_mermaid()

Return type:

str

__init__(name, components=<factory>, connections=<factory>, metadata=<factory>)

Parameters:

• name (str)

• components (Dict[str, Component])

• connections (List[Connection])

• metadata (Dict[str, Any])

Return type:

None

class hiphopsllm.architecture.model.RawGraph

Bases: object

Node ids/labels and raw edges, before ports and roles are assigned.

name: str = 'system'

nodes: Dict[str, str]

edges: List[Tuple[str, str, str, bool]]

node_meta: Dict[str, Dict[str, Any]]

__init__(name='system', nodes=<factory>, edges=<factory>, node_meta=<factory>)

Parameters:

• name (str)

• nodes (Dict[str, str])

• edges (List[Tuple[str, str, str, bool]])

• node_meta (Dict[str, Dict[str, Any]])

Return type:

None

hiphopsllm.architecture.model.parse_mermaid(mermaid, name='system')

Parse the mermaid source produced by draw_mermaid().

Used when the drawable graph object is unavailable (e.g. offline analysis of a notebook, or a mermaid diagram pasted from a report).

Return type:

RawGraph

Parameters:

• mermaid (str)

• name (str)

hiphopsllm.architecture.model.extract_architecture(graph_like, name='langgraph_system', role_overrides=None, resource_overrides=None, globals_ns=None, node_functions=None, materialise_routers=True)

Extract a SystemModel from a LangGraph object or mermaid text.

This is the entry point that replaces graph.get_graph().draw_mermaid_png() as the source of truth for the analysis:

model = extract_architecture(graph, globals_ns=globals())

Return type:

SystemModel

Parameters:

• graph_like (Any)

• name (str)

• role_overrides (Dict[str, Role | str] | None)

• resource_overrides (Dict[str, Dict[str, str]] | None)

• globals_ns (Dict[str, Any] | None)

• node_functions (Dict[str, Callable[[...], Any]] | None)

• materialise_routers (bool)

hiphopsllm.architecture.model.raw_from_spec(spec)

Read a plain dict specification into a RawGraph.

Kept separate from SystemModel.from_spec() so the specification route goes through exactly the same build_system_model() call as a live LangGraph, and therefore honours the same overrides.

Return type:

RawGraph

Parameters:

spec (Dict[str, Any])

hiphopsllm.architecture.model.source_of_function(fn)

inspect.getsource, but only when the text really is that function.

In a notebook whose cell has been re-executed, linecache can hand back stale or unrelated lines. Classifying a component from someone else’s source is worse than classifying it from none, so the result is discarded unless it contains the function’s own def.

Return type:

str

Parameters:

fn (Callable[[...], Any] | None)

Extraction

A configurable, reusable extractor for LangGraph architectures.

extract_architecture() is a function that takes ten keyword arguments. That is fine for a single call in a notebook, but awkward when the same conventions — the same role overrides, the same shared model snapshots, the same unroll depth — have to be applied to several graphs and compared. LangGraphExtractor holds those conventions as state and applies them to any number of graphs:

extractor = LangGraphExtractor(
    globals_ns=globals(),
    role_overrides={"coder": "tool"},
    unroll=2,
)
approach_1 = extractor.extract(graph_1, name="Approach 1")
approach_2 = extractor.extract(graph_2, name="Approach 2")

The extractor also carries the loop-elimination step, so extract_acyclic returns the model the fault tree synthesiser can actually consume.

class hiphopsllm.architecture.extract.LangGraphExtractor

Bases: object

Turn LangGraph objects into SystemModel objects, under fixed conventions.

Parameters:

• globals_ns (Optional[Dict[str, Any]]) – Pass globals() from a notebook. Node functions are then found by name and the actual model objects are interrogated, which is what makes shared-snapshot (common-cause) detection reliable. Without it, role classification falls back to node names and edge topology alone.

• node_functions (Optional[Dict[str, Callable[..., Any]]]) – Explicit {node_id: function} mapping, for when the functions are not in globals_ns (a script, a class, an imported module). Takes precedence over globals_ns.

• role_overrides (Dict[str, Role | str]) – Force a component’s archetype, e.g. {"verifier": Role.AGGREGATOR}. Use this when a node’s name and source do not reveal what it really is.

• resource_overrides (Dict[str, Dict[str, str]]) – Declare shared resources the source does not name, e.g. {"critic": {"llm": "gpt-4o-2024-11-20"}}. Components sharing a resource become a common-cause group, which is usually the difference between a redundant architecture and one that only looks redundant.

• unroll (int) – Iterations of each feedback loop represented explicitly (default 1).

• materialise_routers (bool) – Turn add_conditional_edges into an explicit router component with its own failure logic, rather than an anonymous branch (default True).

globals_ns: Optional[Dict[str, Any]] = None

node_functions: Optional[Dict[str, Callable[..., Any]]] = None

role_overrides: Dict[str, Role | str]

resource_overrides: Dict[str, Dict[str, str]]

unroll: int = 1

materialise_routers: bool = True

with_roles(**roles)

Return a copy with additional role overrides.

Return type:

LangGraphExtractor

Parameters:

roles (Role | str)

with_resources(**resources)

Return a copy with additional shared-resource declarations.

Return type:

LangGraphExtractor

Parameters:

resources (Dict[str, str])

extract(graph, name='langgraph_system')

Read the architecture as it is, feedback loops and all.

graph may be a compiled LangGraph, the drawable from graph.get_graph(), mermaid text, a dict specification, or an already-built SystemModel (returned unchanged).

Return type:

SystemModel

Parameters:

• graph (Any)

• name (str)

extract_acyclic(graph, name='langgraph_system')

Extract, then unroll feedback loops to unroll iterations.

Returns the analysable (acyclic) model and the CycleReport recording what was cut, so the loop handling is visible in the report rather than hidden in the tree.

Return type:

Tuple[SystemModel, CycleReport]

Parameters:

• graph (Any)

• name (str)

__init__(globals_ns=None, node_functions=None, role_overrides=<factory>, resource_overrides=<factory>, unroll=1, materialise_routers=True)

Parameters:

• globals_ns (Dict[str, Any] | None)

• node_functions (Dict[str, Callable[[...], Any]] | None)

• role_overrides (Dict[str, Role | str])

• resource_overrides (Dict[str, Dict[str, str]])

• unroll (int)

• materialise_routers (bool)

Return type:

None

Loop elimination

hiphopsllm.architecture.acyclic — turning a cyclic agent graph into a loop-free analysis model.

Fault trees are, by definition, acyclic: a top event is refined into causes, and no event may be its own cause. Agentic LangGraph applications are not acyclic — the ReAct pattern is a feedback loop (generator -> coder -> generator). Classical HiP-HOPS meets the same problem with control loops in engineered systems and resolves it by breaking the circular dependency explicitly rather than letting the synthesis algorithm recurse for ever.

This module implements that step and records it, so the resulting fault tree can be read as a static structure without hiding what was done to obtain it.

Two policies are offered, both producing a directed acyclic graph:

unroll=k (default k = 1)

Iterations 1..k of the loop body are represented explicitly as distinct component instances (generator#1, generator#2, …). This exposes iteration-dependent behaviour (e.g. prompt growth) at the cost of a larger tree.

Feedback cut with contribution preserved (always applied at depth k)

The last back edge is replaced by a FEEDBACK pseudo-component that consumes the deviations the loop would have carried and delivers them to the system boundary, together with a loop-exhaustion basic event. This is the conservative choice: simply deleting the back edge would silently remove the tool’s contribution from the tree and produce an optimistic — that is, unsafe — result.

The intent is that no analysis result depends on an arbitrary recursion cut-off: every loop is either unrolled a stated number of times or represented by a named basic event that appears in the cut sets.

class hiphopsllm.architecture.acyclic.CycleReport

Bases: object

What was found and what was done about it — carried into the report.

cycles: List[List[str]]

back_edges: List[Tuple[str, str]]

unroll: int = 1

feedback_components: List[str]

replicated: Dict[str, List[str]]

notes: List[str]

property had_cycles: bool

summary()

Return type:

str

__init__(cycles=<factory>, back_edges=<factory>, unroll=1, feedback_components=<factory>, replicated=<factory>, notes=<factory>)

Parameters:

• cycles (List[List[str]])

• back_edges (List[Tuple[str, str]])

• unroll (int)

• feedback_components (List[str])

• replicated (Dict[str, List[str]])

• notes (List[str])

Return type:

None

hiphopsllm.architecture.acyclic.find_cycles(model)

Components that lie on a cycle, grouped by strongly connected component.

Return type:

List[List[str]]

Parameters:

model (SystemModel)

hiphopsllm.architecture.acyclic.find_back_edges(nodes, edges)

Return the back edges of a depth-first forest (iterative DFS).

A back edge points to a node currently on the DFS stack; removing the set of back edges is sufficient to make the graph acyclic.

Return type:

List[Tuple[str, str, str, bool]]

Parameters:

• nodes (Sequence[str])

• edges (Sequence[Tuple[str, str, str, bool]])

hiphopsllm.architecture.acyclic.make_acyclic(model, unroll=1, boundary=None)

Return a loop-free copy of model plus a CycleReport.

Parameters:

• unroll (int) – Number of loop iterations represented explicitly (>= 1).

• boundary (Optional[str]) – Component that receives the feedback-cut output. Defaults to the graph’s sink (__end__), because an unresolved loop manifests at the system boundary as “no answer” or “answer too late”.

• model (SystemModel)

Return type:

Tuple[SystemModel, CycleReport]

hiphopsllm.architecture.acyclic.is_acyclic(model)

Return type:

bool

Parameters:

model (SystemModel)