Skip to main content

Production-ready framework for reliable LLM orchestration

Project description

PromptGuard ๐Ÿ›ก๏ธ

The Production-Ready 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.

License: MIT Python 3.9+


๐ŸŽฏ 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 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 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 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 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

# With all features
pip install promptguard[all]

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

๐Ÿš€ Quick Start

import asyncio
from promptguard 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())

๐Ÿ—๏ธ Architecture

promptguard/
โ”œโ”€โ”€ core/              # Main orchestration
โ”‚   โ”œโ”€โ”€ chain.py       # PromptChain orchestrator
โ”‚   โ”œโ”€โ”€ executor.py    # Execution engine
โ”‚   โ”œโ”€โ”€ response.py    # Response models
โ”‚   โ””โ”€โ”€ models.py      # Model registry
โ”œโ”€โ”€ providers/         # LLM provider integrations
โ”‚   โ”œโ”€โ”€ base.py
โ”‚   โ”œโ”€โ”€ anthropic_provider.py
โ”‚   โ”œโ”€โ”€ openai_provider.py
โ”‚   โ”œโ”€โ”€ groq_provider.py
โ”‚   โ”œโ”€โ”€ cohere_provider.py
โ”‚   โ””โ”€โ”€ google_provider.py
โ”œโ”€โ”€ validation/        # Response validation
โ”‚   โ”œโ”€โ”€ semantic.py    # Semantic validators
โ”‚   โ””โ”€โ”€ schema.py      # Pydantic schema validation
โ”œโ”€โ”€ caching/           # Response caching
โ”‚   โ”œโ”€โ”€ base.py
โ”‚   โ”œโ”€โ”€ memory.py
โ”‚   โ”œโ”€โ”€ redis.py
โ”‚   โ””โ”€โ”€ disk.py
โ”œโ”€โ”€ retry/             # Retry strategies
โ”‚   โ””โ”€โ”€ strategies.py
โ”œโ”€โ”€ observability/     # Logging and metrics
โ”‚   โ”œโ”€โ”€ logger.py
โ”‚   โ”œโ”€โ”€ metrics.py
โ”‚   โ””โ”€โ”€ tracing.py
โ””โ”€โ”€ utils/             # Utilities

๐Ÿ“š Documentation


๐Ÿงช 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


๐Ÿ™ Acknowledgments

Built with โค๏ธ for the AI engineering community.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

promptguard_pro-0.1.0.tar.gz (23.4 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

promptguard_pro-0.1.0-py3-none-any.whl (34.2 kB view details)

Uploaded Python 3

File details

Details for the file promptguard_pro-0.1.0.tar.gz.

File metadata

  • Download URL: promptguard_pro-0.1.0.tar.gz
  • Upload date:
  • Size: 23.4 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.12.0

File hashes

Hashes for promptguard_pro-0.1.0.tar.gz
Algorithm Hash digest
SHA256 826155df9514e4d1fc3c1dc15f8e2f8f4c8f0b7e68f8e3b90943b6264a3e7bc2
MD5 30cf17d8722322dba8f25e89c851ff89
BLAKE2b-256 e6baf20d77ff5087475dc1ba3cc9a16734e6bef5a7a709906a6d55da1c495d85

See more details on using hashes here.

File details

Details for the file promptguard_pro-0.1.0-py3-none-any.whl.

File metadata

File hashes

Hashes for promptguard_pro-0.1.0-py3-none-any.whl
Algorithm Hash digest
SHA256 40e8503b176d0f8e8ec46d3197d7dd864f08a88f26f9682bcbed8c98cb686dad
MD5 5373f99822f2c3aae6ccdb6fd7562a3a
BLAKE2b-256 0c8e8fbccdeb239f756a671b42c9c6e81ddbd5cc3f6fb841c05d67b6410f3edb

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page