chain-probe 0.1.0

Step-level semantic fault isolation for multi-step LLM pipelines. CASCADE analysis. Framework-agnostic.

chain-probe

Step-level semantic fault isolation for multi-step LLM pipelines.

Your 4-step LLM pipeline failed. Which step caused it? Existing tools (Langfuse, Arize, Promptfoo) trace metadata or evaluate final output — none do step-level semantic fault isolation. chain-probe does.

pip install chain-probe

Quick Start

from chain_probe import Pipeline

pipeline = Pipeline("my_rag_pipeline")

@pipeline.probe()
def retrieve(query):
    return search_documents(query)

@pipeline.probe(
    validator=lambda inp, out: (0.9, "ok") if not hallucinated(out) else (0.1, "hallucinated"),
    threshold=0.5,
)
def generate(docs):
    return call_llm(docs)

@pipeline.probe()
def format(answer):
    return format_response(answer)

result = pipeline.cascade(initial_input="What is our refund policy?")

print(result.summary())
# Pipeline: my_rag_pipeline
# Verdict: FAILED
# Steps: 3 total, 1 failed
# Root cause: step 1 (generate)
#
#   ✓ [0] retrieve: pass
#   ✗ [1] generate: fail [root_cause] score=0.10
#       reason: hallucinated
#   ✓ [2] format: pass

CASCADE Fault Analysis

The key innovation: when multiple steps fail, chain-probe distinguishes root cause from inherited failure.

• ROOT_CAUSE — This step introduced the failure
• INHERITED — This step failed because it received bad input from an upstream failure
• INDEPENDENT — This step failed for its own reasons, unrelated to upstream failures

result = pipeline.cascade(initial_input="test")

for step in result.steps:
    if step.verdict != "pass":
        print(f"Step {step.step_name}: {step.fault_type.value}")
        # "retrieve: root_cause"
        # "generate: inherited"
        # "format: inherited"

Features

• @probe decorator — Zero-config step registration
• CASCADE analysis — Automatic root cause vs inherited fault classification
• Validators — Per-step semantic validation with score + reason
• Framework-agnostic — Works with LangChain, LlamaIndex, raw API calls, anything
• Zero dependencies — Pure Python, nothing to install
• SQLite history — Track cascade results over time
• CI gate — Exit code 0/1 for pipeline health

Convenience API

from chain_probe import run_cascade

result = run_cascade(
    steps=[retrieve, generate, format, deliver],
    initial_input="user query",
    pipeline_name="rag_v2",
)

if not result.passed:
    print(f"Root cause: {result.root_cause.step_name}")

Store & History

from chain_probe import ProbeStore

store = ProbeStore("pipeline_runs.db")
store.save(result)

history = store.get_history("my_pipeline", limit=10)

License

MIT