PromptGuard 🛡️

The Framework for Reliable LLM Orchestration

PromptGuard is a Python library that brings production-grade reliability, type safety, and observability to LLM applications. Think of it as "Pydantic meets Circuit Breaker for AI" - reducing boilerplate by 80% while making your AI apps bulletproof.

---

🎯 The Problem We Solve

Every AI engineer writes the same boilerplate:

• Manual retry logic with exponential backoff
• Model fallback chains when primary fails
• Response parsing with regex/string manipulation
• Token counting and cost tracking
• Response validation and error handling

PromptGuard eliminates all of this.

---

✨ Core Features

1. Smart Execution with Auto-Retry & Fallbacks

from promptguard_pro import PromptChain

chain = PromptChain(
    models=["anthropic/claude-3-5-sonnet", "openai/gpt-4o", "groq/llama-70b"],
    strategy="cascade",
    max_retries=3,
    retry_delay="exponential"
)

result = await chain.execute("Analyze this document...")
print(result.response)  # Guaranteed to succeed or raise clear error

2. Type-Safe Response Validation

from pydantic import BaseModel, Field
from promptguard_pro import PromptChain

class EvaluationResponse(BaseModel):
    evaluation: str = Field(description="Overall evaluation")
    score: int = Field(ge=0, le=100)
    reason: str

chain = PromptChain(
    models=["anthropic/claude-3-5-sonnet"],
    response_schema=EvaluationResponse,
    validation_mode="strict"
)

result = await chain.execute(prompt)
print(result.response.score)  # Type-safe!

3. Multi-Provider Support

chain = PromptChain(
    models=[
        "anthropic/claude-3-5-sonnet",
        "openai/gpt-4o",
        "groq/llama-3-70b",
        "cohere/command-r-plus",
        "google/gemini-1.5-pro"
    ]
)

4. Automatic Token Tracking & Cost Estimation

result = await chain.execute(prompt)

print(result.metadata.tokens_used)
print(result.metadata.estimated_cost)
print(result.metadata.model_used)
print(result.metadata.execution_time_ms)

5. Response Caching

from promptguard_pro import CacheBackend

chain = PromptChain(
    models=["anthropic/claude-3-5-sonnet"],
    cache=CacheBackend.memory(),  # or .redis() or .disk()
    cache_ttl=3600
)

result1 = await chain.execute("What is AI?", cache_key="ai_def_v1")
result2 = await chain.execute("What is AI?", cache_key="ai_def_v1")
assert result2.metadata.cached == True

6. Semantic Response Validation

from promptguard_pro import validators

chain = PromptChain(
    models=["anthropic/claude-3-5-sonnet"],
    validators=[
        validators.length_range(min_chars=100, max_chars=5000),
        validators.contains_keywords(["risk", "evaluation"]),
        validators.has_citations(required=True),
        validators.sentiment_check(allowed=["neutral", "positive"])
    ]
)

7. Streaming Support

chain = PromptChain(models=["anthropic/claude-3-5-sonnet"])

async for chunk in chain.stream("Write a long essay..."):
    print(chunk.delta, end="", flush=True)

8. Batch Processing

prompts = ["Evaluate doc 1...", "Evaluate doc 2...", ...]

results = await chain.batch_execute(
    prompts,
    max_concurrent=5,
    show_progress=True
)

---

📦 Installation

# Basic installation
pip install promptguard-pro

# With all features
pip install promptguard-pro[all]

# With specific features
pip install promptguard-pro[cache]      # Redis caching
pip install promptguard-pro[validation]  # Semantic validators
pip install promptguard-pro[metrics]     # Prometheus metrics

---

🚀 Quick Start

import asyncio
from promptguard_pro import PromptChain, validators, CacheBackend
from pydantic import BaseModel

class Analysis(BaseModel):
    summary: str
    score: int
    recommendations: list[str]

async def main():
    chain = PromptChain(
        models=[
            "anthropic/claude-3-5-sonnet",
            "openai/gpt-4o",
            "groq/llama-70b"
        ],
        strategy="cascade",
        max_retries=3,
        response_schema=Analysis,
        validators=[
            validators.length_range(min_chars=100),
            validators.has_citations()
        ],
        cache=CacheBackend.memory(),
        cache_ttl=3600
    )
    
    result = await chain.execute(
        prompt="Analyze this document: ...",
        cache_key="analysis_v1"
    )
    
    print(f"Score: {result.response.score}")
    print(f"Cost: ${result.metadata.estimated_cost:.4f}")
    print(f"Time: {result.metadata.execution_time_ms:.0f}ms")

asyncio.run(main())

---

📚 Documentation

• Getting Started
• API Reference
• Examples
• Advanced Usage
• Contributing

---

🧪 Testing

# Run all tests
pytest tests/ -v

# Run with coverage
pytest tests/ --cov=promptguard --cov-report=html

# Run specific test file
pytest tests/unit/test_core.py -v

---

🔧 Configuration

Environment Variables

# API Keys
export ANTHROPIC_API_KEY=sk-ant-...
export OPENAI_API_KEY=sk-...
export GROQ_API_KEY=gsk-...
export COHERE_API_KEY=...
export GOOGLE_API_KEY=...

# Redis
export REDIS_URL=redis://localhost:6379

---

🤝 Contributing

Contributions are welcome! Please see CONTRIBUTING.md for guidelines.

---

📄 License

MIT License - See LICENSE file for details

---