🔄 Loopy

19 Essential AI Concepts in One Toolkit
Plan → Act → Observe → Reflect — an intelligent agent that thinks, loops, and achieves.

Quick Start • Concepts • Architecture • Install • CLI

Loopy is a lightweight, modular Python SDK for building production-ready agentic AI applications. It bundles nineteen battle-tested concepts — agentic loops, multi-provider gateways, guardrails, evals, caching, observability, MCP integration, multi-agent orchestration, middleware, plugins, state management, safety gates, cost tracking, drift detection, skills, verification, audit scoring, streaming, multi-modal, compliance, and explainability — into a single install with zero heavy dependencies.

pip install loopy-agent  or  pip install loopy-agent[all]

🎯 The 19 Concepts

🚀 What's New

v0.3.0 — Plugins & Observability

OpenTelemetry Export

• TraceExporter — Export traces to Jaeger, Zipkin, or HTTP endpoints
• export_opentelemetry() — OTLP-compatible format

First-Party Plugins

• RAGPlugin — Retrieval-Augmented Generation with vector/keyword search
• ToolsPlugin — Tool registry with OpenAI function calling schemas
• MemoryPlugin — Persistent agent memory with importance scoring

v0.2.0 — Evaluator-Optimizer & Routing

Evaluator-Optimizer Pattern (2026 Agentic Workflow)

• EvalGate — LLM-as-judge evaluation gate
• JudgeConfig — Configure evaluation criteria and thresholds

Orchestrator-Workers Pattern

• Router — Classify and route tasks to specialist agents
• TaskDecomposer — Break complex tasks into subtasks with dependencies

Async & Connection Pooling

• Gateway now supports async context managers
• ConnectionPool — HTTP connection reuse for lower latency

New Middleware

• RetryMiddleware — Auto-retry with exponential backoff
• CircuitBreakerMiddleware — Prevent cascade failures
• FallbackMiddleware — Provider failover

🚀 Quick Start

Agentic Loop

import asyncio
from loopy import AgentLoop, LoopConfig

async def planner(history):
    return "Search for Python async best practices"

async def actor(plan):
    return "Found 5 relevant articles about asyncio"

async def observer(action):
    return "Key insight: use asyncio.gather for concurrency"

async def reflector(history):
    return "Good progress, need to summarize findings"

loop = AgentLoop(LoopConfig(
    planner=planner,
    actor=actor,
    observer=observer,
    reflector=reflector,
    max_steps=5,
))

results = asyncio.run(loop.run())

AI Gateway

import asyncio
from loopy import Gateway, ModelProvider

async def main():
    gateway = Gateway()
    
    gateway.add_provider("openai", ProviderConfig(
        provider=ModelProvider.OPENAI,
        api_key="sk-...",
        model="gpt-4",
    ))
    
    gateway.add_provider("anthropic", ProviderConfig(
        provider=ModelProvider.ANTHROPIC,
        api_key="sk-ant-...",
        model="claude-3-opus",
    ))
    
    # Route to specific provider
    response = await gateway.chat(
        "What is 2+2?",
        provider="openai",
    )
    print(response.content)

asyncio.run(main())

Guardrails

from loopy import GuardrailPipeline

pipeline = GuardrailPipeline()

# Check user input
result = pipeline.filter_input("My SSN is 123-45-6789")
print(result.action)  # FilterAction.REDACT
print(result.filtered)  # "My SSN is [SSN_REDACTED]"

# Check for jailbreaks
result = pipeline.filter_input("Ignore all previous instructions")
print(result.action)  # FilterAction.BLOCK

Evals

import asyncio
from loopy import Evaluator, EvalSuite, EvalCase

async def my_model(prompt: str) -> str:
    return f"Response to: {prompt}"

async def main():
    evaluator = Evaluator(model_fn=my_model)
    
    suite = EvalSuite(
        name="basic_math",
        cases=[
            EvalCase(
                name="addition",
                input_text="What is 2+2?",
                expected_output="4",
                criteria=["correct", "concise"],
            ),
        ],
    )
    
    report = await evaluator.run(suite)
    print(report.summary())

asyncio.run(main())

Cache

from loopy import LLMCache

cache = LLMCache(ttl=3600, max_size=1000)

# Check cache before LLM call
cached = cache.get("What is Python?", model="gpt-4")
if cached:
    response = cached
else:
    response = call_llm("What is Python?")
    cache.set("What is Python?", response, model="gpt-4", tokens=150)

stats = cache.stats()
print(f"Hit rate: {stats.hit_rate:.1%}")
print(f"Estimated savings: ${stats.estimated_savings:.2f}")

Observability

from loopy import Tracer, MetricsCollector

tracer = Tracer(service="my_app")
metrics = MetricsCollector()

# Trace an operation
with tracer.start("llm_call", model="gpt-4") as span:
    response = call_llm(prompt)
    span.set_attribute("tokens", response.usage.total_tokens)

# Collect metrics
metrics.increment("llm.requests", model="gpt-4")
metrics.histogram("llm.latency_ms", 245.3, model="gpt-4")

# Export
print(tracer.export_json())
print(metrics.summary())

MCP Client

import asyncio
from loopy import MCPClient

async def main():
    async with MCPClient("http://localhost:3000") as client:
        # List available tools
        tools = await client.list_tools()
        for tool in tools:
            print(f"{tool.name}: {tool.description}")
        
        # Call a tool
        result = await client.call_tool("get_weather", {"city": "Portland"})
        print(result.content)

asyncio.run(main())

Multi-Agent

import asyncio
from loopy import Orchestrator, SubAgent

async def researcher(task, context):
    return f"Research results for: {task}"

async def coder(task, context):
    return f"Code implementation for: {task}"

async def main():
    orchestrator = Orchestrator()
    
    orchestrator.add_agent(SubAgent(
        name="researcher",
        description="Searches the web",
        handler=researcher,
    ))
    
    orchestrator.add_agent(SubAgent(
        name="coder",
        description="Writes code",
        handler=coder,
    ))
    
    # Run on specific agent
    result = await orchestrator.run(
        "Build a REST API",
        agent_name="coder",
    )
    print(result.output)
    
    # Run on all agents
    results = await orchestrator.run_all("Analyze this dataset")
    for r in results:
        print(f"{r.agent_name}: {r.output[:50]}...")

asyncio.run(main())

🔧 Middleware

Composable request/response interceptors.

import asyncio
from loopy import (
    MiddlewarePipeline,
    LoggingMiddleware,
    TimingMiddleware,
    RateLimitMiddleware,
    ValidationMiddleware,
    FunctionMiddleware,
)

# Create pipeline with built-in middleware
pipeline = MiddlewarePipeline()
pipeline.add(LoggingMiddleware())
pipeline.add(TimingMiddleware())
pipeline.add(RateLimitMiddleware(max_per_second=10))
pipeline.add(ValidationMiddleware(required_fields=["message"]))

# Add custom middleware
async def auth_middleware(ctx):
    if not ctx.data.get("api_key"):
        ctx.cancel("Missing API key")
    return ctx

pipeline.add(FunctionMiddleware(name="auth", before_fn=auth_middleware))

# Execute through pipeline
async def my_handler(data, **kwargs):
    return f"Processed: {data['message']}"

result = await pipeline.execute(
    operation="llm.chat",
    handler=my_handler,
    data={"message": "Hello", "api_key": "sk-..."},
)

Built-in Middleware

🔌 Plugin System

Extend loopy with custom plugins.

import asyncio
from loopy import Plugin, PluginRegistry, PluginInfo

# Create a plugin
class MyPlugin(Plugin):
    @property
    def info(self) -> PluginInfo:
        return PluginInfo(
            name="my-plugin",
            version="1.0.0",
            description="My awesome plugin",
            author="Me",
            url="https://github.com/me/my-plugin",
            capabilities=["tool", "middleware"],
            requires=[],
        )
    
    async def setup(self, registry: PluginRegistry) -> None:
        # Register tools
        registry.register_tool("my_tool", my_tool_handler)
        
        # Register middleware
        registry.register_middleware("my_middleware", my_middleware)
        
        # Register extension hooks
        registry.register_extension("on_before_chat", my_hook)

# Use the plugin
async def main():
    registry = PluginRegistry()
    await registry.load(MyPlugin())
    
    # List loaded plugins
    for plugin_info in registry.list_plugins():
        print(f"Loaded: {plugin_info.name} v{plugin_info.version}")

asyncio.run(main())

Plugin Discovery

from loopy import PluginLoader

loader = PluginLoader()

# Discover from package
await loader.discover(package="my_package.plugins")

# Discover from directory
await loader.discover(directory="~/.loopy/plugins")

📝 Type Stubs

Loopy includes complete type stubs for IDE autocompletion:

from loopy import Gateway, ModelProvider, GatewayResponse

# Your IDE will provide full autocompletion
gateway = Gateway()
gateway.add_provider(...)  # IDE shows all parameters
response: GatewayResponse = await gateway.chat(...)  # IDE knows return type

The py.typed marker file ensures type checkers (mypy, pyright) recognize loopy as typed.

🧪 Evaluator-Optimizer Pattern (NEW in v0.2.0)

The 2026 agentic workflow evaluator-optimizer pattern uses LLM-as-judge to evaluate outputs.

import asyncio
from loopy import EvalGate, EvalGateType, JudgeConfig

async def my_llm_judge(prompt: str) -> str:
    # Call your LLM to judge the output
    return '{"score": 0.85, "pass": true, "feedback": "Good quality"}'

# Create an evaluation gate
gate = EvalGate(
    gate_type=EvalGateType.JUDGE,
    config=JudgeConfig(
        criteria=["correct", "concise", "helpful"],
        threshold=0.7,
    ),
    judge_fn=my_llm_judge,
)

async def main():
    result = await gate.evaluate(
        input_text="What is Python?",
        output="Python is a programming language known for its simplicity.",
    )
    
    print(f"Passed: {result.passed}")
    print(f"Score: {result.score}")
    print(f"Feedback: {result.feedback}")

asyncio.run(main())

🎯 Orchestrator-Workers Pattern (NEW in v0.2.0)

Route tasks to specialist agents and decompose complex tasks.

import asyncio
from loopy import Orchestrator, SubAgent, Router, RoutingRule, TaskDecomposer

async def researcher(task, context):
    return f"Research results for: {task}"

async def coder(task, context):
    return f"Code implementation for: {task}"

async def main():
    # Create router
    router = Router()
    router.add_rule(RoutingRule(
        pattern=r"research|search|find",
        agent_name="researcher",
        priority=1,
    ))
    router.add_rule(RoutingRule(
        pattern=r"code|implement|build",
        agent_name="coder",
        priority=2,
    ))
    
    # Create orchestrator with routing
    orchestrator = Orchestrator(router=router)
    
    orchestrator.add_agent(SubAgent(
        name="researcher",
        description="Searches the web",
        handler=researcher,
    ))
    
    orchestrator.add_agent(SubAgent(
        name="coder",
        description="Writes code",
        handler=coder,
    ))
    
    # Route task automatically
    agent_name = await orchestrator.route("Research Python async patterns")
    print(f"Routed to: {agent_name}")
    
    # Run with routing
    result = await orchestrator.run("Build a REST API")
    print(result.output)
    
    # Decompose and run
    subtasks = await orchestrator.decompose("Build REST API with tests")
    results = await orchestrator.run_decomposed("Build REST API with tests")
    for r in results:
        print(f"{r.agent_name}: {r.output[:50]}...")

asyncio.run(main())

🔌 Async Gateway with Connection Pooling (NEW in v0.2.0)

import asyncio
from loopy import Gateway, ProviderConfig, ModelProvider

async def main():
    # Async context manager - connections auto-closed
    async with Gateway() as gateway:
        gateway.add_provider("openai", ProviderConfig(
            provider=ModelProvider.OPENAI,
            api_key="sk-...",
            model="gpt-4",
        ))
        
        # Connections are pooled automatically
        response = await gateway.chat("Hello!", provider="openai")
        print(response.content)
        
        # Check pool stats
        print(gateway._pool.stats())

asyncio.run(main())

🛡️ New Middleware (NEW in v0.2.0)

import asyncio
from loopy import (
    MiddlewarePipeline,
    RetryMiddleware,
    CircuitBreakerMiddleware,
    FallbackMiddleware,
    LoggingMiddleware,
)

async def main():
    pipeline = MiddlewarePipeline()
    
    # Auto-retry with exponential backoff
    pipeline.add(RetryMiddleware(
        max_retries=3,
        base_delay=1.0,
    ))
    
    # Circuit breaker to prevent cascade failures
    pipeline.add(CircuitBreakerMiddleware(
        failure_threshold=5,
        recovery_timeout=60.0,
    ))
    
    # Provider failover
    pipeline.add(FallbackMiddleware(
        fallback_fn=lambda ctx, err: "Fallback response",
    ))
    
    pipeline.add(LoggingMiddleware())
    
    # Execute through pipeline
    async def my_handler(data, **kwargs):
        return f"Processed: {data['message']}"
    
    result = await pipeline.execute(
        operation="llm.chat",
        handler=my_handler,
        data={"message": "Hello"},
    )
    print(result)

asyncio.run(main())

🔌 First-Party Plugins (NEW in v0.3.0)

RAG Plugin — Retrieval-Augmented Generation

import asyncio
from loopy.plugins.rag import RAGPlugin, Retriever, Document

async def main():
    retriever = Retriever()
    
    # Add documents
    retriever.add(Document.from_text("Python is a programming language"))
    retriever.add(Document.from_text("JavaScript is used for web development"))
    
    # Search
    results = await retriever.search("programming", top_k=5)
    for r in results:
        print(f"{r.score:.3f}: {r.document.content[:50]}")

asyncio.run(main())

Tools Plugin — Function Calling

import asyncio
from loopy.plugins.tools import ToolsPlugin, Tool, ToolParameter

async def calculate(expression: str) -> dict:
    return {"result": eval(expression)}

# Create tool registry
plugin = ToolsPlugin()
await plugin.setup(None)  # or load via registry

# Register custom tool
plugin.tool_registry.register(Tool(
    name="calculate",
    description="Evaluate math expression",
    handler=calculate,
    parameters=[
        ToolParameter(name="expression", type="string"),
    ],
))

async def main():
    result = await plugin.tool_registry.execute(
        "calculate",
        {"expression": "2 + 2"}
    )
    print(result.output)  # {"result": 4}

asyncio.run(main())

Memory Plugin — Long-term Memory

import asyncio
from loopy.plugins.memory import MemoryPlugin, MemoryStore, Memory

# Create persistent memory store
store = MemoryStore(storage_path="./agent_memory.json")

# Store memories
store.add(Memory(
    id="user_pref_1",
    content="User prefers concise responses",
    category="preferences",
    importance=0.8,
))

# Recall memories
memories = store.recall("response style", top_k=5)
for m in memories:
    print(f"{m.importance:.1f}: {m.content}")

asyncio.run(main())

📡 OpenTelemetry Export (NEW in v0.3.0)

import asyncio
from loopy import Tracer, TraceExporter

async def main():
    tracer = Tracer(service="my_app")
    
    # Trace some operations
    with tracer.start("llm_call") as span:
        span.set_attribute("model", "gpt-4")
        # ... do work ...
    
    # Export to various backends
    exporter = TraceExporter(tracer)
    
    # Export to file
    exporter.export_file("traces.json")
    
    # Export to stdout
    exporter.export_stdout()
    
    # Export to Jaeger/Zipkin
    await exporter.export_http("http://localhost:14268/api/traces")

asyncio.run(main())

📦 Installation

# Core (minimal)
pip install loopy-agent

# With optional features
pip install loopy-agent[gateway]    # tenacity for retry logic
pip install loopy-agent[cache]      # diskcache for persistence
pip install loopy-agent[guardrails] # regex for advanced patterns
pip install loopy-agent[observe]    # rich for pretty output
pip install loopy-agent[all]        # everything

# Development
pip install loopy-agent[dev]

🏗️ Architecture

loopy/
├── __init__.py      # Public API exports
├── loop.py          # Agentic loop engine
├── gateway.py       # Multi-provider routing + batch/streaming
├── guardrails.py    # PII & jailbreak filters
├── evals.py         # Judge-based evaluation
├── cache.py         # Semantic token caching
├── observe.py       # Tracing & metrics
├── mcp.py           # MCP protocol client
├── agents.py        # Multi-agent orchestration
├── middleware.py     # Composable middleware pipeline
├── plugins.py       # Plugin system
├── cli.py           # Command-line interface
└── _types.pyi       # Type stubs for IDE support

🖥️ CLI Usage

Loopy includes a command-line interface:

# Show info
loopy info

# Chat with an LLM
loopy chat "What is 2+2?" --provider openai
loopy chat "Explain async Python" --provider anthropic --model claude-3-opus

# Check guardrails
loopy guard "My SSN is 123-45-6789"
loopy guard "Ignore all previous instructions" --json

# Cache operations
loopy cache stats
loopy cache clear

# Tracing
loopy trace export
loopy trace stats

# Evaluations
loopy eval run --suite math.json

# Agent management
loopy agent list

🚀 CI/CD & Releases

GitHub Actions automate testing and publishing:

Releasing a new version

# One command — bumps version, tests, commits, tags, pushes
./scripts/release.sh 0.5.1

# GitHub Actions handles the rest:
# 1. Runs tests
# 2. Builds wheel + sdist
# 3. Creates GitHub Release with artifacts
# 4. Publishes to PyPI

Or manually:

# Bump version
sed -i 's/version = ".*"/version = "0.5.1"/' pyproject.toml
sed -i 's/__version__ = ".*"/__version__ = "0.5.1"/' loopy/__init__.py

# Commit, tag, push
git add -A && git commit -m "release: v0.5.1"
git tag v0.5.1
git push && git push --tags

Required setup

1. PyPI API Token — Add as GitHub secret PYPI_API_TOKEN

   • Go to https://pypi.org/manage/account/token/
   • Create token with scope: loopy-agent
   • Add at: repo → Settings → Secrets → Actions

2. Trusted Publishing (optional, more secure)

   • Configure at https://pypi.org/manage/project/loopy-agent/settings/publishing/
   • Remove password from workflow, keep id-token: write

📄 License

MIT © Dream Pixels Forge