# Project export: LLM Gateway This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small. ## Project metadata - Hackathon: OpenAI Build Week - Tagline: A smart AI control plane that routes, caches, and protects every LLM request. - Devpost: https://devpost.com/software/llm-gateway-fw29p1 - GitHub: https://github.com/Tvaibhav06/AI-Middleware - Video: https://www.youtube.com/embed/w7JfRfs0gYY?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent - Team: 1 GitHub contributor(s) — Tvaibhav06 (6 commits) ## Devpost submission (written by the team) ### Challenges we ran into # About the Project Inspiration Modern AI applications rely on multiple LLM providers, but integrating them directly into every application creates unnecessary complexity. Every team ends up solving the same problems such as authentication, rate limiting, provider routing, budget tracking, caching, resilience, and observability. The inspiration behind LLM Gateway was to build a unified control plane that sits between applications and AI providers. Instead of every application handling infrastructure concerns independently, the gateway centralizes them into a single platform. This makes AI applications easier to build, cheaper to operate, and more reliable. What it does LLM Gateway is an AI middleware platform that acts as a single entry point for LLM requests. It provides: Provider-agnostic request routing Semantic caching to reduce latency and cost Team-based authentication and access control Rate limiting and token budget enforcement Automatic retries, fallback routing, and circuit breaker protection Real-time metrics and telemetry Runtime policy updates through admin APIs OpenAI-compatible Chat Completions API for easy integration Applications only communicate with the gateway, while the gateway intelligently decides how every request should be processed. How we built it We built the gateway using FastAPI with a modular middleware architecture where each responsibility is isolated into its own component. The platform includes: Semantic caching powered by vector similarity search Redis-compatible storage for rate limiting and budgets Embedded Qdrant for local development OpenTelemetry-compatible metrics with Prometheus support Configurable policy management through YAML and admin APIs Dependency injection to switch seamlessly between development and production environments For development, we created an isolated runtime using injected provider doubles, embedded infrastructure, and mock transports. This allowed us to verify the complete middleware pipeline without requiring paid APIs or cloud infrastructure while keeping the production deployment path unchanged. Challenges we ran into The biggest challenge was designing a gateway that could remain provider agnostic while keeping the request flow consistent across different AI providers. Another challenge was balancing a production-ready architecture with an efficient local development experience. We wanted developers to run the complete platform without requiring Docker, paid API keys, or external monitoring systems, while ensuring the production deployment remained completely untouched. We also spent significant effort validating reliability features such as semantic caching, rate limiting, budget enforcement, retries, circuit breakers, and fallback routing under simulated load before moving toward production deployment. ### Accomplishments we're proud of Built a unified AI middleware platform instead of a single AI application. Designed a modular architecture that cleanly separates routing, caching, resilience, observability, and policy management. Created a development runtime that mirrors production behavior without modifying production code. Successfully validated thousands of simulated gateway requests using local infrastructure and injected provider doubles. Exposed an OpenAI-compatible API, making it easy for existing AI applications to integrate with the gateway. ### What we learned This project taught us that building reliable AI infrastructure involves much more than calling an LLM API. We gained hands-on experience with distributed systems concepts such as caching, resilience patterns, observability, dependency injection, and policy-driven infrastructure. We also learned how important it is to separate development and production environments so that new features can be tested safely without affecting deployment. Most importantly, we learned that developer experience is just as important as system architecture. A platform is far more useful when it is easy to understand, run, and extend. ### What's next Our next goal is to complete the production deployment by integrating Docker, Redis, Prometheus, Grafana, and production-grade Qdrant deployments. Beyond deployment, we plan to expand the gateway with intelligent provider selection based on latency and cost, multi-region support, advanced analytics dashboards, stronger security and audit capabilities, and support for additional LLM providers. We also want to make the gateway deployable with a single command so that teams can adopt it with minimal setup. Accomplishments that we're proud of What we learned What's next for LLM Gateway ## README (from the GitHub repository) # FlowGate -- Unified LLM Gateway & Control Plane A shared control layer for applications that call large language models (LLMs). Instead of every application separately handling provider access, caching, limits, budgets, and outages, they can call one OpenAI-compatible gateway. This README is written from the project state, verification reports, and source files. It distinguishes recorded results from items that are still open. ## Project Overview The FLowGate AI Middleware Platform is a Python 3.11+ and FastAPI service that sits between client applications and configured LLM providers. Clients send a Chat Completions request to `POST /v1/chat/completions`; the gateway applies the platform controls before it calls a provider. The platform includes: - Team authentication with `Authorization: Bearer `. - A semantic cache backed by Qdrant Server in production. - Redis-backed rate limiting and budget enforcement. - Provider routing, retries, fallback, and a circuit breaker. - Incremental response streaming using Server-Sent Events (SSE). - Telemetry, metrics, dashboards, alerts, and admin APIs. It is not a new LLM or a replacement for a provider. It is a control plane around existing provider calls. ## Why This Exists The source plan and PRD/TRD describe a common problem: teams can make repeated LLM calls, exceed a shared budget, and have little visibility when a provider slows down or fails. Rebuilding those controls separately in every application adds cost and operational risk. This project centralizes those controls. A cacheable repeat can return without another provider call, a request can be stopped before forwarding when a rate limit or budget is exceeded, and telemetry records what happened. ## How It Works ### Request Workflow The runtime order is fixed and verified as follows: ```mermaid flowchart TD C[Client application] --> A[Authentication and team lookup] A --> SC[Semantic cache] SC -- Cache hit --> CH[Return cached response] CH --> T[Telemetry] SC -- Cache miss --> RL[Redis rate limiting] RL -- Blocked --> R429[429 response] RL -- Allowed --> B[Redis budget enforcement] B -- Blocked --> BE[Budget error response] B -- Allowed --> PR[Provider routing] PR --> RT[Retries] RT --> CB[Circuit breaker] CB --> FB[Fallback] FB --> P[Provider call] P --> S[Response streaming] S --> CS[Store complete successful response in semantic cache] CS --> T ``` In plain language: 1. The gateway checks the incoming team API key and loads that team's allowed models and providers. 2. It looks for a usable answer in the semantic cache. 3. On a cache hit, it returns the cached response. This bypasses rate limiting and budget enforcement, as required by the project specification. 4. On a cache miss, Redis checks whether the team is within its request/token limit and budget before the request reaches a provider. 5. The provider router selects a permitted provider for the requested model. 6. The resilience stages handle retryable failures, circuit-breaker state, and configured fallback paths. 7. The provider response is returned normally or streamed in chunks. A complete successful response can be saved for later cache hits. 8. Telemetry records the request path, cache outcome, timing, provider outcome, and related metrics. ### Major Components | Component | What it does | | --- | --- | | FastAPI gateway | Exposes the OpenAI-compatible Chat Completions endpoint and health/metrics routes. | | Authentication | Looks up a team from its inbound Bearer API key. | | Semantic cache | Uses prompt similarity and cache policy to return a stored response when appropriate. Qdrant Server is the production cache store. | | Redis rate limiting | Uses token buckets to limit requests and tokens for each team. A priority queue component is also verified separately. | | Budget enforcement | Reserves the maximum configured cost before a provider call, then settles against provider usage or releases the reservation according to the recorded streaming rules. | | Provider routing | Applies the YAML policy that maps a requested model to a provider, while team policy controls which models/providers a team may request. | | Retries, fallback, and circuit breaker | Retries eligible failures, prevents repeated calls to an unhealthy route, and selects a configured fallback when available. | | Streaming | Forwards OpenAI-compatible SSE chunks incrementally. Complete successful streams may be cached after they finish. | | Telemetry | Produces request, cache, spend, rate-limit, fallback, and circuit-breaker metrics plus tracing spans. | | Admin APIs | Provide rate-limit status, limits/budgets updates, spend, alert-threshold updates, cache invalidation, and cache threshold testing. | | Prometheus | Scrapes the gateway `/metrics` endpoint and stores metrics for querying. | | Grafana | Displays the provisioned Operations, Business, Performance, and Cache dashboards. | | OpenTelemetry Collector | Receives deployment tracing over OTLP/HTTP. | | Qdrant Server | Stores semantic-cache vectors and metadata in the production deployment. | ## Architecture ```mermaid flowchart LR Client[Client applications] --> Gateway[Production gateway\nFastAPI + Gunicorn] Gateway --> Redis[Redis\nrate limiting and budget enforcement] Gateway --> Qdrant[Qdrant Server\nsemantic cache] Gateway --> Providers[Configured provider adapters] Gateway --> Collector[OpenTelemetry Collector] Prometheus[Prometheus] -->|scrapes /metrics| Gateway Grafana[Grafana] -->|queries| Prometheus Admin[Admin APIs] --> Gateway Demo[Demo gateway\ndemo profile] --> Mock[Mock provider\ndemo profile] Demo --> Redis Demo --> Qdrant Demo --> Collector ``` Redis and Qdrant Server have deliberately separate jobs: Redis is for rate limits and budgets; Qdrant Server is for the semantic cache. The platform does not merge those responsibilities. ## Development Runtime, Demo Gateway, and Production Gateway These are separate ways to run the project. They are not interchangeable. | Runtime | Purpose | Providers and secrets | Local services | Default address | | --- | --- | --- | --- | --- | | Development runtime | Local development and component exploration. | Injected provider doubles, mock Slack transport, and safe placeholders in `.env.local`. No real provider call. | Memurai at `127.0.0.1:6379` and Qdrant Embedded Mode under `.dev/qdrant`. | `http://127.0.0.1:8000` | | Demo gateway | Safe Compose demonstration of the assembled production factory. | Demo-only mock provider and demo values from `deploy/env/demo.env`. | Compose Redis, Qdrant Server, Prometheus, Grafana, and OpenTelemetry Collector. | `http://127.0.0.1:8001` | | Production gateway | Deployment path for real configuration. | Deployment-supplied provider credentials, Slack configuration, and team keys. Placeholders in `deploy/env/production.env.example` are not real production values. | Compose Redis, Qdrant Server, Prometheus, Grafana, and OpenTelemetry Collector. | `http://127.0.0.1:8000` | The development runtime is separate from the production image and production factory. The demo gateway uses the production image and factory but only the explicit `demo` Compose profile with a mock provider. The mock provider is not included in the production profile. ## Running Locally ### Option A: Development runtime Use this option to explore the gateway without real provider credentials. It requires a Redis-compatible server such as the documented local Memurai instance listening on `127.0.0.1:6379`. Qdrant runs in Embedded Mode for this runtime, so no Qdrant container is needed. 1. Create and activate a virtual environment in PowerShell: ```powershell python -m venv .venv .\.venv\Scripts\Activate.ps1 python -m pip install -r requirements-dev.txt ``` 2. Create or keep an ignored `.env.local` in the repository root. Th [README truncated for size] ## Detected evidence (automated analysis) Indexed codebase: 153 recognized source files, 525 KB. - FastAPI (technology) — detected in the code - Python (language) — detected in the code - Docker (technology) — claimed on Devpost, not found in the code - Redis (technology) — claimed on Devpost, not found in the code ## Codebase structure (from repository index) ### Files (120 of 184) ``` .dockerignore .gitignore ai-middleware-platform-plan.md ai-middleware-platform-prd-trd.md compose.yaml deploy/.gitkeep deploy/config/policy.demo.yaml deploy/config/policy.production.example.yaml deploy/demo/Dockerfile deploy/demo/mock_provider.py deploy/entrypoint.sh deploy/env/demo.env deploy/env/production.env.example deploy/grafana/dashboards/business.json deploy/grafana/dashboards/cache.json deploy/grafana/dashboards/operations.json deploy/grafana/dashboards/performance.json deploy/grafana/provisioning/dashboards/dashboards.yaml deploy/grafana/provisioning/datasources/prometheus.yaml deploy/otel/collector.yaml deploy/otel/lifespan.py deploy/prometheus/prometheus.yml deploy/tiktoken/preload.py dev_runtime/app.py dev_runtime/README.md dev_runtime/run.py Dockerfile docs/adr/ADR-001-fastapi-selection.md docs/adr/ADR-002-qdrant-semantic-cache.md docs/adr/ADR-003-redis-responsibility-boundary.md docs/adr/ADR-004-streaming-in-phase-1.md docs/adr/ADR-005-middleware-performance-metric.md docs/adr/ADR-006-frozen-package-boundaries.md docs/adr/ADR-007-request-enrichment-scope-resolution.md docs/adr/ADR-008-external-standard-reference-authorization-openai-chat-completions-contract.md docs/adr/ADR-009-team-api-key-authorization-header.md docs/adr/ADR-010-anthropic-messages-wire-reference-authorization.md docs/adr/ADR-011-redis-atomic-rate-limit-and-priority-queues.md docs/adr/ADR-012-memurai-development-test-redis-substitute.md docs/adr/ADR-013-budget-reservation-requires-max-tokens.md docs/adr/ADR-014-qdrant-embedded-mode-development.md docs/adr/ADR-015-openai-embeddings-contract-and-shared-cache.md docs/adr/ADR-016-ttl-tier-classification.md docs/adr/ADR-017-provisional-adaptive-cache-thresholds.md docs/adr/ADR-018-resilience-policy-injection.md docs/adr/ADR-019-standalone-observability-contract.md docs/adr/ADR-020-alert-evaluation-before-delivery.md docs/adr/ADR-021-runtime-cache-bypass-and-stream-settlement.md docs/adr/ADR-022-phase-11-yaml-policy-parser-and-hot-reload.md docs/adr/ADR-023-admin-audit-and-slack-delivery-boundary.md docs/adr/ADR-024-redis-persistent-connection-for-load-validation.md docs/adr/ADR-025-phase-12-asgi-gateway-overhead-measurement.md docs/adr/ADR-026-phase-12-grafana-validation-deferral.md docs/adr/ADR-027-production-asgi-server.md docs/adr/ADR-028-production-outbound-http-transport.md docs/adr/ADR-029-production-team-registry-configuration.md docs/adr/ADR-030-yaml-provider-routing-and-model-tiers.md docs/adr/ADR-031-streaming-usage-and-budget-settlement.md docs/adr/ADR-032-yaml-pricing-and-team-budget-policy.md docs/adr/ADR-033-provisional-resilience-policy-defaults.md docs/adr/ADR-034-deployment-otel-collector-topology.md docs/adr/ADR-035-slack-incoming-webhook-message-format.md docs/adr/ADR-036-provider-agnostic-input-token-estimation.md docs/adr/ADR-037-phase-13-container-image-pins.md docs/AI_MIDDLEWARE_PLATFORM_HANDOFF.md docs/architecture/admin-config-boundary.md docs/architecture/authentication-boundary.md docs/architecture/budget-enforcement-boundary.md docs/architecture/chat-completions-compatibility.md docs/architecture/observability-boundary.md docs/architecture/provider-routing-boundary.md docs/architecture/rate-limiting-boundary.md docs/architecture/semantic-cache-completion-boundary.md docs/architecture/semantic-cache-foundation-boundary.md docs/architecture/system-architecture.md docs/definition-of-done.md docs/DEMO_GUIDE.md docs/phase-13-implementation-plan.md docs/PORTFOLIO_NARRATIVE.md docs/repository-structure.md docs/verification/phase-0-verification-report.md docs/verification/phase-1-verification-report.md docs/verification/phase-10-verification-report.md docs/verification/phase-11-verification-report.md docs/verification/phase-12-load-results.json docs/verification/phase-12-verification-report.md docs/verification/phase-13-verification-report.md docs/verification/phase-14-verification-report.md docs/verification/phase-2-verification-report.md docs/verification/phase-3-verification-report.md docs/verification/phase-4-verification-report.md docs/verification/phase-5-verification-report.md docs/verification/phase-6-verification-report.md docs/verification/phase-7-verification-report.md docs/verification/phase-8-verification-report.md docs/verification/phase-9-verification-report.md PROJECT_PHASE_MAPPING.md PROJECT_STATE.md pyproject.toml qdrant_data/.lock qdrant_data/meta.json README.md requirements-dev.txt requirements-prod.txt src/ai_middleware/__init__.py src/ai_middleware/api/.gitkeep src/ai_middleware/api/admin.py src/ai_middleware/api/app.py src/ai_middleware/api/streaming.py src/ai_middleware/auth/.gitkeep src/ai_middleware/auth/production_registry.py src/ai_middleware/auth/team_authentication.py src/ai_middleware/budget/__init__.py src/ai_middleware/budget/redis_budget.py src/ai_middleware/budget/token_estimation.py src/ai_middleware/budgets/.gitkeep src/ai_middleware/cache/.gitkeep src/ai_middleware/config/__init__.py src/ai_middleware/config/.gitkeep src/ai_middleware/config/policy_store.py [64 more files omitted for size] ``` ### Dependencies - pyproject.toml: fastapi, opentelemetry-api, prometheus-client, PyYAML, qdrant-client ### Recent commits (newest first) - Delete Screenshot 2 - Delete Screenshotpng - Name update - Merge pull request #1 from Tvaibhav06/dev-runtime - Remove unwanted items from README - Phase 13 deployment and testing - phase 13 - phase 13 - Merge branch 'master' of https://github.com/Tvaibhav06/AI-Middleware - phase 12 - ReadMe Added - phase 0-11 completed ## Key source files (fetched from GitHub, selected and truncated for size) ### PROJECT_PHASE_MAPPING.md ```markdown # Project Phase Mapping This mapping links the user-approved Phase 0–14 execution order to the original implementation plan. The PRD/TRD remains the acceptance specification; this file prevents an unsupported claim that a phase merely "matches the plan." | Project phase | Scope | Original plan / PRD-TRD reference | Mapping note | |---|---|---|---| | 0 | Scope freeze, architecture setup, project mapping | No equivalent delivery phase | Governance baseline required by the approved build order before implementation. | | 1 | FastAPI gateway skeleton, OpenAI-compatible endpoint, streaming | Plan Phase 1.2; Plan Phase 2.4; PRD/TRD §10, §9.8; ADR-008 | **Intentional Architectural Deviation:** streaming is advanced from original Plan Phase 2 to Project Phase 1. The official OpenAI Chat Completions wire reference is used only within the ADR-008 authorization boundary. | | 2 | Authentication and team lookup | Plan Phase 1.3; PRD/TRD §9.1 | Implement independently; it is not assembled into the full runtime pipeline yet. | | 3 | Provider abstraction and router | Plan Phase 1.1 and Phase 2.3; PRD/TRD §5 Provider Routing, §9.6 | Normalized provider interface and model-based routing. | | 4 | Redis rate limiting and priority queues | Plan Phase 3.1; Plan Phase 3.3; PRD/TRD §5 Rate Limiting & Budget, §9.4 | Implement and independently verify atomic per-team token buckets (requests/min and tokens/min) plus tiered priority queues that serve high-priority real-time requests over low-priority batch requests near capacity. Redis remains limited to rate limits and budgets. | | 5 | Budget enforcement | Plan Phase 3.2; PRD/TRD §5 Rate Limiting & Budget, §9.5 | Redis-backed budget enforcement, independently verified. | | 6 | Qdrant semantic-cache foundation | Plan Phase 1.4–1.5; PRD/TRD §5 Semantic Cache, §11 | Cache data model, cache-key rules, and Qdrant-only cache boundary; no future cache-completion behavior. | | 7 | Cache completion | Plan Phase 2.1–2.2; Plan Phase 3.5–3.8; Plan Phase 5.4; PRD/TRD §5 Semantic Cache, §9.3, §9.9 | Similarity lookup, storage, TTL, invalidation, near-miss tracking, persistence, adaptive task-type thresholds, and internal historical replay/tuning. Admin HTTP exposure remains Phase 11. | | 8 | Resilience | Plan Phase 4.1–4.4; PRD/TRD §5 Fallback & Resilience, §9.7 | Health checks, retries, circuit breaker, and tier-based fallback. | | 9 | Observability | Plan Phase 5.1–5.3 and 5.5; PRD/TRD §5 Observability, §9.10 | OpenTelemetry, Prometheus, Grafana metrics, and alerts. | | 10 | Runtime pipeline integration | PRD/TRD §8–§9 | Integrates existing components only, in the required runtime order. | | 11 | Admin APIs and hot reload | Plan Phase 3.4; Plan Phase 3.6; PRD/TRD §5 Admin & Config, §10 | Includes the specified admin operations and YAML policy hot reload. | | 12 | Integration and load tests | Plan Phase 6.1–6.2; PRD/TRD §14 | Includes 5,000+ concurrent and 2,000+ cache-mix validations. | | 13 | Docker Compose deployment | Plan Phase 6 [truncated — 1167 more characters] ``` ### PROJECT_STATE.md ```markdown # Project State ## Current Phase Phase 14 - Documentation, README, Demo, and Portfolio Polish. PARTIAL. The written documentation deliverables are complete and regression-verified; the source plan's recorded under-four-minute demo artifact is not present. ## Completed Work - Phase 0 through Phase 11 are complete, including the FastAPI gateway, SSE serializer, authentication, provider routing, Redis rate/budget controls, Qdrant semantic cache, resilience, telemetry, runtime orchestration, YAML hot reload, and admin APIs. - Phase 12 added concurrent runtime budget-cap verification, simulated fallback/circuit-breaker integration coverage, and multi-chunk Unicode streaming integrity coverage. - Earlier Phase 13 full regression passed: 3 Phase 0 tests, 73 unit tests, and 31 integration tests (one environment-gated live-Qdrant test is skipped in the default suite and was separately verified as passing). The current regression result is recorded below. - Required load validation passed: 5,000 concurrent gateway tasks with five team keys/two models and exact 2,500 rate-limit blocks; 5,000 Redis priority items with real-time precedence/FIFO; 2,000 cache-mix requests with 990 hits and 9,900 configured microdollars avoided. - Measured full FastAPI/ASGI gateway-path middleware processing overhead was 0.6180 ms mean, 0.8419 ms P95, and 1.3167 ms P99, separately from Redis, Qdrant, cache, provider-double, and end-to-end latency. - Corrected the verified Redis direct-RESP socket-exhaustion defect with one locked reusable connection per client instance (ADR-024). - Phase 13 has recorded the production ASGI server pattern: Gunicorn with the pinned `uvicorn_worker.UvicornWorker` integration, using deployment-tunable worker count and Qdrant Server for multi-worker support (ADR-027). - Phase 13 has added the pinned asynchronous `httpx.AsyncClient` transport foundation, explicit timeout/connection-pool defaults, native provider SSE decoding, non-blocking Slack transport, and a Qdrant Server cache foundation adapter (ADR-028). - Phase 13 has added startup-validated `TEAM_REGISTRY_JSON` identity loading and hot-reload YAML provider routing, model tiers, and fallback-chain policy (ADR-029, ADR-030). - Phase 13 has added provider stream-usage capture and the successful-stream ceiling-settlement path: OpenAI requests usage chunks, Anthropic normalizes terminal usage, and a successful stream without usage retains its full reservation (ADR-031). - Phase 13 has extended hot-reload YAML with strict model-pricing and per-team budget-period policy, validating exact model/team coverage against routing and `TEAM_REGISTRY_JSON` (ADR-032). - The live gateway now recognizes an assembled async stream, returns the existing OpenAI-compatible SSE framing, and finalizes its budget/telemetry after client chunk delivery. - Phase 13 has added provisional, hot-reloadable resilience defaults: 20%/50% health thresholds over five minutes, 1000ms P99, retry from 250ms with full jitter, and a [truncated — 9631 more characters] ``` ### pyproject.toml ``` [project] name = "ai-middleware-platform" version = "0.1.0" description = "AI Middleware Platform architecture baseline" requires-python = ">=3.11" dependencies = [ "fastapi", "PyYAML", "qdrant-client", "opentelemetry-api", "prometheus-client", ] [tool.unittest] start-directory = "tests" ``` ### Dockerfile ``` FROM python:3.11-slim WORKDIR /app ENV PYTHONDONTWRITEBYTECODE=1 \ PYTHONUNBUFFERED=1 \ PYTHONPATH=/app/src \ TIKTOKEN_CACHE_DIR=/opt/tiktoken-cache COPY requirements-prod.txt ./ RUN pip install --no-cache-dir -r requirements-prod.txt COPY src ./src COPY deploy/entrypoint.sh ./deploy/entrypoint.sh COPY deploy/otel/lifespan.py ./deploy/otel/lifespan.py COPY deploy/tiktoken/preload.py ./deploy/tiktoken/preload.py COPY deploy/config/policy.production.example.yaml ./deploy/config/policy.production.example.yaml # ADR-036: fail image build if the fixed encoding cannot be present locally. RUN python deploy/tiktoken/preload.py RUN chmod +x /app/deploy/entrypoint.sh \ && useradd --create-home --uid 10001 gateway \ && chown -R gateway:gateway /app /opt/tiktoken-cache USER gateway EXPOSE 8000 ENTRYPOINT ["/app/deploy/entrypoint.sh"] ``` ### deploy/demo/Dockerfile ``` FROM python:3.11-slim WORKDIR /app RUN pip install --no-cache-dir fastapi==0.139.2 uvicorn==0.51.0 COPY deploy/demo/mock_provider.py ./mock_provider.py EXPOSE 8081 CMD ["uvicorn", "mock_provider:app", "--host", "0.0.0.0", "--port", "8081"] ``` ### dev_runtime/app.py ```python """Development-only composition of the existing AI Middleware Platform components.""" from __future__ import annotations import asyncio from collections.abc import AsyncIterator, Mapping from contextvars import ContextVar from dataclasses import dataclass from datetime import datetime, timedelta, timezone from hashlib import sha256 from pathlib import Path import sys from time import perf_counter, time from typing import Any from uuid import uuid4 from fastapi import APIRouter, Response from prometheus_client import CONTENT_TYPE_LATEST, generate_latest ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "src")) from ai_middleware.api.admin import AdminService from ai_middleware.api.app import create_app from ai_middleware.auth.team_authentication import TeamCredential, TeamProfile, TeamRegistry, authenticate from ai_middleware.budget.redis_budget import BudgetPeriod, BudgetPolicy, BudgetPricing, RedisBudgetEnforcer from ai_middleware.config.policy_store import YamlPolicyStore from ai_middleware.contracts.chat_completions import ChatCompletionRequest, parse_chat_completion_request from ai_middleware.observability import AlertSnapshot, MiddlewareTelemetry, SlackAlertDispatcher from ai_middleware.observability.alert_delivery import SlackAlertTransport from ai_middleware.providers.provider_routing import ProviderResponse, ProviderRouter from ai_middleware.rate_limiting.redis_rate_limiting import RateLimitPolicy, RedisPriorityQueue, RedisRespClient, RedisTokenBucketRateLimiter from ai_middleware.resilience.core import BreakerState, CircuitBreaker, CircuitBreakerPolicy, FallbackRouter, HealthStatus, ProviderFailure, RetryExecutor, RetryPolicy from ai_middleware.runtime.pipeline import RuntimePipeline from ai_middleware.semantic_cache.completion import CachePolicy, QdrantSemanticCache, ThresholdReplaySample from ai_middleware.semantic_cache.foundation import CacheEntry, CacheResponse, EmbeddedQdrantSemanticCache, TokenCounts, build_cache_key @dataclass class _RequestState: team: TeamProfile | None = None provider: str = "unknown" _request_state: ContextVar[_RequestState | None] = ContextVar("dev_request_state", default=None) class DevProvider: """Development double that satisfies the existing provider-router interface.""" def __init__(self, name: str, supported_models: frozenset[str]) -> None: self.name = name self.supported_models = supported_models async def complete(self, request: ChatCompletionRequest) -> ProviderResponse: return ProviderResponse(self.name, request.model, self.payload(request)) async def _stream(self) -> AsyncIterator[Mapping[str, Any]]: yield {"object": "chat.completion.chunk"} def stream(self, request: ChatCompletionRequest) -> AsyncIterator[Mapping[str, Any]]: return self._stream() def payload(self, request: ChatCompletionRequest) -> dict[str, Any]: return { "id": f"dev-{uuid4().hex}", "object": "chat.completion", "model": request.model, "choices": [{"index": 0, "message": {"role": "assistant", "content": f"development response via {self.name}"}, "finish_reason": "stop"}], "usage": {"prompt_tokens": 1, "completion_tokens": 8, "total_tokens": 9}, } class DevSlackTransport(SlackAlertTransport): """Records alert payloads locally; it never makes a network request.""" def __init__(self) -> None: self.deliveries: list[dict[str, Any]] = [] def post_alert(self, *, webhook_url: str, payload: Mapping[str, Any]) -> None: self.deliveries.append({"webhook_url": webhook_url, "payload": dict(payload)}) class DevPipeline: """A development wrapper around the unmodified RuntimePipeline.""" def __init__(self, pipeline: RuntimePipeline, telemetry: MiddlewareTelemetry) -> None: self._pipeline = pipeline self._telemetry = telemetry def execute(self, request: ChatCompletionRequest, authorization: str | None) -> Any: state = _RequestState() token = _request_state.set(state) started = perf_counter() try: with self._telemetry.span("development_gateway_request", model=request.model): result = self._pipeline.execute(request, authorization) team_id = state.team.team_id if state.team else "unknown" self._telemetry.record_request(team_id, request.model, state.provider, "success", perf_counter() - started) return result except Exception: team_id = state.team.team_id if state.team else "unknown" self._telemetry.record_request(team_id, request.model, state.provider, "error", perf_counter() - started) raise finally: _request_state.reset(token) def _load_env(path: Path) -> dict[str, str]: values: dict[str, str] = {} if not path.exists(): return values for line in path.read_text(encoding="utf-8").splitlines(): if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) values[key.strip()] = value.strip() return values def _policy_file(path: Path) -> None: if path.exists(): return path.parent.mkdir(parents=True, exist_ok=True) path.write_text( "teams:\n" " team-alpha:\n" " rate_limits:\n" " requests_per_minute: 2\n" " request_bucket_capacity: 2\n" " tokens_per_minute: 64\n" " token_bucket_capacity: 64\n" " budget:\n" " limit_microdollars: 1000\n" " team-beta:\n" " rate_limits:\n" " requests_per_minute: 4\n" " request_bucket_capacity: 4\n" " tokens_per_minute: 64\n" " token_bucket_capacity: 64\n" " budget:\n" " limit_microdollars: 1000\n" "alerts:\n" " provider_error_rate: 0. [truncated — 12452 more characters] ``` ### src/ai_middleware/api/app.py ```python """Phase 1 FastAPI gateway boundary.""" from __future__ import annotations from inspect import isawaitable from typing import Any from fastapi import FastAPI, Request from fastapi.openapi.utils import get_openapi from fastapi.responses import JSONResponse, StreamingResponse from ai_middleware.api.admin import AdminService, create_admin_router from ai_middleware.api.streaming import encode_chat_completion_stream from ai_middleware.contracts.chat_completions import ( CHAT_COMPLETION_REQUEST_OPENAPI_SCHEMA, ChatCompletionRequestError, openai_error, parse_chat_completion_request, ) def _error_response( *, status_code: int, message: str, error_type: str, parameter: str | None = None, code: str | None = None, ) -> JSONResponse: return JSONResponse( status_code=status_code, content=openai_error( message=message, error_type=error_type, parameter=parameter, code=code, ), ) def _add_bearer_authentication_documentation(app: FastAPI) -> None: """Document the existing inbound team Bearer header without enforcing it twice. Runtime authentication remains in the assembled pipeline so its existing OpenAI-compatible error behavior is unchanged. Adding a FastAPI security dependency here would alter that behavior before the route handler runs. """ def openapi() -> dict[str, Any]: if app.openapi_schema is not None: return app.openapi_schema schema = get_openapi( title=app.title, version=app.version, openapi_version=app.openapi_version, description=app.description, routes=app.routes, ) components = schema.setdefault("components", {}) security_schemes = components.setdefault("securitySchemes", {}) security_schemes["TeamBearerAuth"] = { "type": "http", "scheme": "bearer", "bearerFormat": "API key", "description": "Team API key sent as Authorization: Bearer .", } app.openapi_schema = schema return schema app.openapi = openapi # type: ignore[method-assign] def create_app( runtime_pipeline: Any | None = None, admin_service: AdminService | None = None, lifespan: Any | None = None, ) -> FastAPI: """Create the Phase 1 gateway without composing future runtime stages.""" app = FastAPI(title="AI Middleware Platform Gateway", lifespan=lifespan) _add_bearer_authentication_documentation(app) if admin_service is not None: app.include_router(create_admin_router(admin_service)) @app.post( "/v1/chat/completions", openapi_extra={ "security": [{"TeamBearerAuth": []}], "requestBody": { "required": True, "content": { "application/json": { "schema": CHAT_COMPLETION_REQUEST_OPENAPI_SCHEMA, } }, } }, ) async def create_chat_completion(request: Request) -> Any: try: payload: Any = await request.json() except ValueError: return _error_response( status_code=400, message="Invalid JSON in request body.", error_type="invalid_request_error", ) try: parsed = parse_chat_completion_request(payload) except ChatCompletionRequestError as error: return _error_response( status_code=400, message=str(error), error_type="invalid_request_error", parameter=error.parameter, ) if runtime_pipeline is not None: try: result = runtime_pipeline.execute(parsed, request.headers.get("authorization")) if isawaitable(result): result = await result if result.streaming: return StreamingResponse( encode_chat_completion_stream(result.payload), media_type="text/event-stream", ) return JSONResponse(content=result.payload) except PermissionError as error: return _error_response(status_code=429, message=str(error), error_type="rate_limit_error") except ValueError as error: return _error_response(status_code=400, message=str(error), error_type="invalid_request_error") return _error_response( status_code=503, message="No provider route is configured for this gateway.", error_type="server_error", ) return app app = create_app() ``` ### compose.yaml ```yaml name: ai-middleware services: gateway: build: context: . dockerfile: Dockerfile image: ai-middleware-gateway:local profiles: [production] env_file: - ${PRODUCTION_ENV_FILE:-./deploy/env/production.env.example} volumes: - ${PRODUCTION_POLICY_FILE:-./deploy/config/policy.production.example.yaml}:/app/deploy/config/policy.production.example.yaml:ro depends_on: redis: condition: service_healthy qdrant: condition: service_healthy otel-collector: condition: service_healthy ports: - "8000:8000" restart: unless-stopped healthcheck: test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8000/health', timeout=2)"] interval: 10s timeout: 5s retries: 12 start_period: 20s gateway-demo: build: context: . dockerfile: Dockerfile image: ai-middleware-gateway:local profiles: [demo] env_file: - ./deploy/env/demo.env volumes: - ./deploy/config/policy.demo.yaml:/app/deploy/config/policy.demo.yaml:ro depends_on: redis: condition: service_healthy qdrant: condition: service_healthy otel-collector: condition: service_healthy mock-provider: condition: service_healthy ports: - "8001:8000" restart: "no" healthcheck: test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8000/health', timeout=2)"] interval: 10s timeout: 5s retries: 12 start_period: 20s redis: image: redis:7.4.2-alpine command: ["redis-server", "--appendonly", "yes"] volumes: - redis-data:/data restart: unless-stopped healthcheck: test: ["CMD", "redis-cli", "ping"] interval: 5s timeout: 3s retries: 12 qdrant: image: qdrant/qdrant:v1.18.0 volumes: - qdrant-data:/qdrant/storage restart: unless-stopped healthcheck: test: ["CMD", "/qdrant/qdrant", "--version"] interval: 5s timeout: 3s retries: 12 prometheus: image: prom/prometheus:v3.1.0 command: ["--config.file=/etc/prometheus/prometheus.yml", "--storage.tsdb.path=/prometheus"] volumes: - ./deploy/prometheus/prometheus.yml:/etc/prometheus/prometheus.yml:ro - prometheus-data:/prometheus ports: - "9090:9090" restart: unless-stopped healthcheck: test: ["CMD-SHELL", "wget -qO- http://127.0.0.1:9090/-/ready >/dev/null"] interval: 10s timeout: 5s retries: 12 grafana: image: grafana/grafana:11.4.0 environment: GF_SECURITY_ADMIN_PASSWORD: ${GRAFANA_ADMIN_PASSWORD:-change-me-before-production} volumes: - grafana-data:/var/lib/grafana - ./deploy/grafana/provisioning:/etc/grafana/provisioning:ro - ./deploy/grafana/dashboards:/var/lib/grafana/dashboards:ro ports: - "3000:3000" depends_on: prometheus: condition: service_healthy restart: unless-stopped healthcheck: test: ["CMD-SHELL", "wget -qO- http://127.0.0.1:3000/api/health >/dev/null"] interval: 10s timeout: 5s retries: 12 otel-collector: image: otel/opentelemetry-collector:0.116.1 command: ["--config=/etc/otelcol/config.yaml"] volumes: - ./deploy/otel/collector.yaml:/etc/otelcol/config.yaml:ro ports: - "4318:4318" restart: unless-stopped healthcheck: test: ["CMD", "/otelcol", "--version"] interval: 10s timeout: 5s retries: 12 mock-provider: build: context: . dockerfile: deploy/demo/Dockerfile profiles: [demo] restart: "no" healthcheck: test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8081/health', timeout=2)"] interval: 5s timeout: 3s retries: 12 volumes: redis-data: qdrant-data: prometheus-data: grafana-data: ``` ### deploy/entrypoint.sh ```shell #!/bin/sh set -eu if [ -n "${GUNICORN_WORKERS:-}" ]; then workers="$GUNICORN_WORKERS" else workers="$(( $(getconf _NPROCESSORS_ONLN) * 2 + 1 ))" fi exec gunicorn ai_middleware.deployment.asgi:app \ --bind "${GUNICORN_BIND:-0.0.0.0:8000}" \ --worker-class uvicorn_worker.UvicornWorker \ --workers "$workers" \ --access-logfile - \ --error-logfile - ``` ### dev_runtime/run.py ```python """Local launcher for the development-only runtime.""" from __future__ import annotations from pathlib import Path import sys ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(Path(__file__).resolve().parent)) from app import _load_env, build_dev_app def main() -> None: env = _load_env(ROOT / ".env.local") import uvicorn uvicorn.run( build_dev_app(ROOT / ".env.local"), host=env.get("DEV_HOST", "127.0.0.1"), port=int(env.get("DEV_PORT", "8000")), ) if __name__ == "__main__": main() ``` [144 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]