hiphopsllm.pipeline

The whole analysis behind one object.

AgenticReliabilityStudy runs the five steps this package exists to connect, in order, and keeps every intermediate result addressable:

  1. Extract the architecture from a LangGraph application.

  2. Synthesise static fault trees for each system-level hazard, unrolling feedback loops so the trees are genuinely acyclic.

  3. Measure each component’s failure probability from observed outcomes under an explicit operational profile, using HIP-LLM’s hierarchical imprecise posterior — which returns an interval, because a few hundred items do not identify a point.

  4. Calibrate the trees’ basic events with those intervals, replacing engineering-judgement placeholders and recording what could not be matched.

  5. Convert a tree to conditional probability tables and a Bayesian network, for an exact top-event probability, a posterior over causes, and a picture.

End to end:

from hiphopsllm import AgenticReliabilityStudy, load_example, load_outcomes

study = AgenticReliabilityStudy(load_example("parallel_aggregator"))
study.observe(load_outcomes(), stratum_column="stratum",
              profile={"short": 0.3, "medium": 0.5, "long": 0.2})
study.run()
print(study.summary())
study.bayesnet("H2").show()

Each step can be run on its own, and nothing is done implicitly: a study that has not been given outcomes reports placeholder probabilities and says so, rather than quietly presenting judgement as measurement.

class hiphopsllm.pipeline.AgenticReliabilityStudy[source]

Bases: object

A reliability study of one agentic workflow, from graph to Bayesian network.

Parameters:

  • graph (Any) – A compiled LangGraph, the drawable from graph.get_graph(), mermaid text, a graph specification dict (see load_example()), or a SystemModel.

  • name (str) – What the report calls this system.

  • profile (Union[OperationalProfile, Mapping[str, float], None]) – The operational profile. It can also be supplied later, to observe().

  • globals_ns (Optional[Dict[str, Any]]) – Pass globals() in a notebook so the node functions and the real model objects can be found; that is what makes shared-snapshot detection — and therefore common-cause analysis — reliable.

  • node_functions (Optional[Dict[str, Callable[..., Any]]]) – Explicit {node_id: function} mapping when the functions are not in globals_ns. Use "<node>::router" for a conditional-edge function.

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

  • hazards (Optional[Sequence[Hazard]]) – System-level hazards to synthesise trees for. The default set is derived from the architecture.

  • settings (Any) – HIP-LLM inference settings; defaults to the interactive-speed ones.

  • exact_inference (bool) – Run HIP-LLM’s full hierarchical inference (default). False uses a Jeffreys approximation — much faster, and labelled as an approximation in every evidence string it writes.

graph: Any
name: str = 'agentic workflow'
profile: Union[OperationalProfile, Mapping[str, float], None] = None
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
hazards: Optional[Sequence[Hazard]] = None
settings: Any = None
exact_inference: bool = True
credible_level: float = 0.95
bound: str = 'credible'
report: Optional[SafetyReport] = None
evidence: Dict[str, ComponentEvidence]
calibration: Optional[CalibrationReport] = None
operational_failure: Any = None

the system-level HIP-LLM result, when whole-system outcomes were given

analyse(**kwargs)[source]

Extract the architecture and synthesise the fault trees.

Called automatically by run() and by anything that needs a tree, so it rarely has to be called by hand. It is worth calling on its own when you want the structural result — cut sets, single points of failure, the FMEA — before any measurement exists.

Return type:

SafetyReport

Parameters:

kwargs (Any)

observe(outcomes=None, strata=None, *, profile=None, component=None, stratum_column='stratum', component_columns=None, split_column='split', calibration_split='calibration', component_map=None)[source]

Record measured outcomes. Accepts three shapes.

A table — one row per item, a stratum column, and one 1/0 correctness column per component:

study.observe(load_outcomes(), profile={"short": .3, "medium": .5, "long": .2})

One component — outcomes and strata as sequences:

study.observe([1, 1, 0, 1], ["short"] * 4, component="react_agent")

The whole system — outcomes and strata with no component named. This runs HIP-LLM over the system’s end-to-end correctness and stores it as operational_failure, the direct analogue of calling OperationalFailureProb yourself:

study.observe(outcomes, strata, profile={"short": .3, "long": .7})

1 means the item was answered correctly, matching HIP-LLM.

When a split column is present, only the calibration rows are used; fitting basic-event probabilities on the evaluation set would make every downstream number optimistic and untestable.

Return type:

AgenticReliabilityStudy

Parameters:
run_and_observe(inputs, success, *, stratum=None, profile=None, invoke=None, calibration_fraction=0.75, on_error='skip', progress=True, calibrate=True)[source]

Run the graph over inputs, score every node, and calibrate.

This is the one-cell entry point for a notebook that already builds and runs a LangGraph application: append this and the whole reliability analysis follows from the runs it performs.

study = AgenticReliabilityStudy(graph, globals_ns=globals())
study.run_and_observe(
    inputs=[{"smiles": s} for s in SMILES],
    stratum=lambda item: "large" if len(item["smiles"]) > 40 else "small",
    success={
        "research_agent": lambda s: s.get("browser_status") == "ready",
        "pixelrag_agent": lambda s: s.get("capture_status") == "captured",
        "safety_agent":   lambda s: bool(s.get("workflow_succeeded")),
    },
    profile={"small": 0.6, "large": 0.4},
)
Parameters:

  • inputs (Sequence[Any]) – One graph input per benchmark item. Each is passed to graph.invoke (or to invoke).

  • success (Mapping[str, Callable[[Any], Any]]) –

    {node_id: predicate}. Each predicate receives the final state and returns True (that node did its job), False (it did not), or None.

    None means not exercised — a router sent the run to END before this node ran — and is recorded as a missing observation rather than a failure. That distinction matters: scoring an unreached node as failed would blame it for an upstream fault.

  • stratum (Union[Callable[[Any], str], Sequence[str], None]) – A callable mapping an input to its stratum label, or a ready-made sequence of labels, one per input. Defaults to a single stratum, which is honest but gives up the whole point of an operational profile — pass one if the workload is not uniform.

  • profile (Union[OperationalProfile, Mapping[str, float], None]) – The operational profile. Defaults to the observed frequencies of the stratum labels, recorded as such.

  • invoke (Optional[Callable[[Any], Any]]) – Custom runner, for a graph that needs streaming or a config. Receives one input and must return the final state.

  • calibration_fraction (float) – Fraction of items used to fit basic-event probabilities; the rest are marked test and left out, so a held-out set survives.

  • on_error (str) –

    What to do when a run raises. "skip" (the default) drops the item and reports how many were dropped; "record" keeps the row with every node unscored and a run_error message.

    Neither blames a node. A crashed run leaves no state to score against, so marking its nodes as failures would penalise components that had already succeeded before the exception — the same mistake as scoring an unreached node. What is lost either way is visible: the count is printed, and "record" keeps the message.

  • calibrate (bool) – Run calibrate() when the runs finish. False stops after recording the outcomes.

  • progress (bool)

Return type:

Any

Returns:

pandas.DataFrame – One row per item: item_id, stratum, one column per scored node, and split. Keep it — it is the measurement, and re-running the graph is the expensive part.

calibrate(**kwargs)[source]

Fit each observed component and write the intervals into the trees.

Re-synthesises the trees afterwards, so cut-set quantification and the Bayesian network both see the calibrated numbers.

Return type:

CalibrationReport

Parameters:

kwargs (Any)

run()[source]

Analyse, and calibrate if outcomes were given. Returns self.

This is the one call that makes the whole pipeline a chain:

study = AgenticReliabilityStudy(graph).observe(table).run()
Return type:

AgenticReliabilityStudy

bayesnet(hazard='H2', **kwargs)[source]

The Bayesian network for one hazard, built from its fault tree.

Return type:

BayesianNetwork

Parameters:
imprecise_bayesnet(hazard='H2', **kwargs)[source]

The lower/upper network pair, when basic events carry intervals.

Only meaningful after calibrate(): without measurement the intervals are degenerate and both networks are the same.

Return type:

ImpreciseBayesianNetwork

Parameters:
cpts(hazard='H2', **kwargs)[source]

The conditional probability tables for one hazard’s fault tree.

Parameters:
hazard_probability(hazard='H2')[source]

P(hazard) as an interval, exactly, by inference over the network.

Return type:

Envelope

Parameters:

hazard (str)

property system: SystemModel
property failure_model
hazards_found()[source]
Return type:

List[str]

cut_sets(hazard='H2')[source]
Return type:

List[List[str]]

Parameters:

hazard (str)

single_points()[source]
Return type:

List[Dict[str, str]]

fmea()[source]
plot(hazard='H2')[source]

Draw one fault tree with matplotlib.

Parameters:

hazard (str)

plot_architecture()[source]
plot_importance(hazard='H2', top_n=12)[source]

Fussell-Vesely contribution per basic event, ranked — what to fix first.

Parameters:
plot_cutset_orders()[source]

Cut sets per order, per hazard — how much defence in depth exists.

A tall order-1 bar is the finding: those are single points of failure.

show(hazard='H2', **kwargs)[source]

Draw the Bayesian network for a hazard.

Parameters:
operational_reliability(n_tasks=1)[source]

The measured claim, stated the way HIP-LLM defines it.

This is the measurement: for each component, the probability that it fails on one task drawn from the stated operational profile, and the probability of failure-free operation over n_tasks.

It is deliberately separate from the fault tree. The tree decomposes each of these numbers over a component’s internal failure modes so that it can be propagated through the architecture; that decomposition is a modelling step, not something that was observed. What was observed is here: a task either succeeded or it did not, whatever the reason.

Return type:

str

Parameters:

n_tasks (int)

summary()[source]

Everything the study currently knows, in one printable block.

Return type:

str

save(directory, prefix=None)[source]

Write the report, every tree export, the cut sets and the FMEA.

Return type:

List[str]

Parameters:

  • directory (str)

  • prefix (str | None)

__init__(graph, name='agentic workflow', profile=None, globals_ns=None, node_functions=None, role_overrides=<factory>, resource_overrides=<factory>, unroll=1, hazards=None, settings=None, exact_inference=True, credible_level=0.95, bound='credible')
Parameters:
Return type:

None

exception hiphopsllm.pipeline.StudyNotReady[source]

Bases: RuntimeError

A step was asked for before the step it depends on had been run.