Skip to content

the-shadow-0/Pactum

Folders and files

NameName
Last commit message
Last commit date

Latest commit

Β 

History

97 Commits
Β 
Β 
Β 
Β 
Β 
Β 
Β 
Β 
Β 
Β 
Β 
Β 
Β 
Β 
Β 
Β 
Β 
Β 
Β 
Β 

Repository files navigation

πŸ” Pactum

The AI Contract Runtime

First-class, versioned, testable interfaces and deterministic runtimes for AI components.

Quickstart β€’ Features β€’ Examples β€’ Documentation

Version Python License Status


πŸ€” The Problem

AI bugs are non-reproducible. LLM-powered features are a nightmare to test, debug, and audit:

  • πŸ› Non-reproducible bugs β€” randomness, model updates, temperature all conspire against you
  • πŸ§ͺ Can't unit-test AI logic β€” no clear input/output boundaries, no determinism
  • πŸ’₯ Silent failures β€” prompt template or memory schema changes break things without warning
  • πŸ“‹ No compliance trail β€” who called what LLM, with what data, when?
  • 🐌 Slow feedback loops β€” no way to quickly iterate on AI behavior changes

πŸ’‘ The Solution

Pactum introduces AI Contracts β€” declarative, versioned, enforceable interfaces for AI components:

from pactum import contract, PactRuntime, MemorySchema

@contract(
    name="customer_support_reply:v1",
    inputs={"query": str, "customer_id": str},
    outputs={"reply": str, "intent": str},
    memory=MemorySchema(keys={"customer_profile": {"type": "json", "version": 1}}),
    allowed_tools=["kb_retriever", "crm_get"],
    nondet_budget={"tokens": 8}
)
def support_reply(ctx, inputs):
    snippets = ctx.tools.kb_retriever(inputs["query"], top_k=3)
    prompt = f"Query: {inputs['query']}\nContext: {snippets}\nAnswer concisely."
    result = ctx.llm.complete(prompt, temperature=0.7, max_tokens=256)
    return {"reply": result.text, "intent": result.classification}

# Run with full tracing and snapshotting
runtime = PactRuntime(seed=42)
result = runtime.run(support_reply, {"query": "Where's my order?", "customer_id": "C-12345"})
print(f"Snapshot: {result.snapshot_id}")  # Replay this anytime!

✨ Features

Feature Description
πŸ” AI Contracts Declarative input/output schemas, memory schemas, tool access controls
🎯 Deterministic Runtime Seeded PRNG, token-level tracing, reproducible execution
πŸ“Έ Snapshot Store Content-addressed (SHA-256), Git-friendly execution snapshots
πŸ”„ Replay Engine Deterministic replay from any snapshot β€” reproduce any bug
πŸ§ͺ Test Harness Mock generation, regression testing from snapshots
πŸŒ€ Fuzzing Auto-generate random inputs to find contract violations
πŸ”Œ Plugin System Hooks for LLMs, tools, memory, validators (before_run, after_run, etc.)
πŸ–₯️ CLI init, run, replay, test, mock, fuzz β€” everything from the terminal

πŸš€ Quickstart

Installation

pip install -e ".[dev]"

Initialize a Project

pactum init --name my-ai-project

This creates:

  • pactum.yaml β€” configuration
  • .pactum/snapshots/ β€” snapshot storage
  • contracts/example.py β€” starter contract
  • tests/test_example.py β€” starter test

Run a Contract

pactum run contracts/example.py:hello_world --input-file input.json --seed 42

Replay from Snapshot

pactum replay --snapshot 9f2c3a

Run Tests

pactum test --ci

Fuzz Testing

pactum fuzz contracts/example.py:hello_world --iterations 1000 --seed 42

πŸ—οΈ Architecture

β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”
β”‚   AI Component   β”‚
β”‚  (@contract)     β”‚
β””β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜
         β”‚
β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β–Όβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”
β”‚  Pactum Runtime   β”‚ ← Seeded PRNG, Tool Access, Tracer
β”‚                   β”‚
β”‚  β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”  β”‚
β”‚  β”‚  Validator   β”‚  β”‚ ← Input/Output, Memory, Tools, Budget
β”‚  β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜  β”‚
β”‚  β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”  β”‚
β”‚  β”‚   Tracer     β”‚  β”‚ ← Token-level event recording
β”‚  β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜  β”‚
β””β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜
         β”‚
β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β–Όβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”
β”‚  Snapshot Store   β”‚ ← Content-addressed (SHA-256)
β”‚  (.pactum/)       β”‚
β””β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜
         β”‚
β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β–Όβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”
β”‚  Replay Engine / Test Harness        β”‚
β”‚  Mocks Β· Fuzzing Β· CI Integration    β”‚
β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜
         β”‚
β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β–Όβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”
β”‚  Plugins                              β”‚
β”‚  LLM Β· Memory Β· Tool Β· Validator     β”‚
β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜

πŸ“‚ Project Structure

pactum/
β”œβ”€β”€ core/           # Runtime, tracer, contract DSL, validator
β”œβ”€β”€ cli/            # Command-line interface
β”œβ”€β”€ plugins/        # LLM adapters, tools, memory, plugin base
β”œβ”€β”€ snapshot/       # Content-addressed snapshot store
β”œβ”€β”€ testing/        # Test harness, mock generator, fuzzer
examples/
β”œβ”€β”€ support-bot/    # Customer support contract example
β”œβ”€β”€ rag-app/        # RAG retrieval example
docs/
β”œβ”€β”€ concepts/       # What are contracts, runtime, snapshots
β”œβ”€β”€ cli/            # CLI reference
β”œβ”€β”€ api/            # Python SDK reference
tests/              # Full test suite

πŸ“ Examples

Support Bot

from pactum import contract, PactRuntime, MemorySchema
from pactum.plugins.llm_adapter import StubAdapter
from pactum.plugins.tool_adapter import ToolRegistry

@contract(
    name="support:v1",
    inputs={"query": str, "customer_id": str},
    outputs={"reply": str, "intent": str},
    allowed_tools=["kb_retriever"],
    nondet_budget={"tokens": 256},
)
def support_reply(ctx, inputs):
    docs = ctx.tools.kb_retriever(inputs["query"])
    result = ctx.llm.complete(f"Answer: {inputs['query']}\nContext: {docs}")
    return {"reply": result.text, "intent": "general"}

# Test with stub adapter β€” no API key needed!
tools = ToolRegistry()
tools.register("kb_retriever", lambda q, **kw: ["FAQ: Check tracking page."])

runtime = PactRuntime(
    llm_adapter=StubAdapter(default_response="Your order is on its way!"),
    tool_registry=tools,
    seed=42,
)
result = runtime.run(support_reply, {"query": "Where's my order?", "customer_id": "C-123"})
print(result.outputs)   # {"reply": "Your order is on its way!", "intent": "general"}
print(result.snapshot_id)  # Replay this anytime!

πŸ”Œ Plugin System

from pactum.plugins.base import PactumPlugin

class MetricsPlugin(PactumPlugin):
    @property
    def name(self):
        return "metrics"

    def after_run(self, ctx, inputs, outputs, trace):
        tokens = sum(e.get("token_count", 0) or 0 for e in trace)
        print(f"Total tokens used: {tokens}")

Plugin Hooks:

  • before_run(context, inputs) β€” Before contract execution
  • after_run(context, inputs, outputs, trace) β€” After execution
  • on_trace(event) β€” On each trace event
  • on_snapshot_commit(snapshot_id, data) β€” When snapshot is stored

🀝 Contributing

Contributions are welcome! See our docs for architecture details.

# Install in dev mode
pip install -e ".[dev]"

# Run tests
pytest tests/ -v

# Run with coverage
pytest tests/ -v --cov=pactum --cov-report=term-missing

πŸ“„ License

MIT License β€” see LICENSE for details.


Pactum β€” Build AI components that are observable, reproducible, testable, and safe.

Releases

No releases published

Sponsor this project

 

Packages

 
 
 

Contributors

Languages