Metadata-Version: 2.4
Name: chain-probe
Version: 0.1.0
Summary: Step-level semantic fault isolation for multi-step LLM pipelines. CASCADE analysis. Framework-agnostic.
Project-URL: Homepage, https://github.com/Rowusuduah/chain-probe
Project-URL: Repository, https://github.com/Rowusuduah/chain-probe
Project-URL: Issues, https://github.com/Rowusuduah/chain-probe/issues
Author-email: Richmond Owusu Duah <Rowusuduah@users.noreply.github.com>
License: MIT
License-File: LICENSE
Keywords: agent,ai,cascade,ci-cd,debugging,fault-isolation,llm,observability,pipeline,testing
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Classifier: Topic :: Software Development :: Debuggers
Classifier: Topic :: Software Development :: Testing
Requires-Python: >=3.10
Provides-Extra: dev
Requires-Dist: pytest-cov; extra == 'dev'
Requires-Dist: pytest>=7.0; extra == 'dev'
Requires-Dist: ruff; extra == 'dev'
Description-Content-Type: text/markdown

# 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.

```bash
pip install chain-probe
```

## Quick Start

```python
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

```python
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

```python
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

```python
from chain_probe import ProbeStore

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

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

## License

MIT
