python -m venv venv
source venv/bin/activate
pip install redis openai tiktoken numpy
export OPENAI_API_KEY="sk-your-key"
export REDIS_URL="redis://localhost:6379"

# caching/semantic_cache.py
import redis
import hashlib
import json
from typing import Optional
import numpy as np

class SemanticCache:
    def __init__(self, redis_url: str = "redis://localhost:6379", threshold: float = 0.92):
        self.client = redis.from_url(redis_url, decode_responses=True)
        self.threshold = threshold
        self.embedding_cache = {}

    def _hash_query(self, query: str) -> str:
        return hashlib.md5(query.encode()).hexdigest()

    def _get_embedding(self, text: str) -> list:
        import openai
        client = openai.OpenAI()
        response = client.embeddings.create(model="text-embedding-3-small", input=text)
        return response.data[0].embedding

    def _cosine_similarity(self, a: list, b: list) -> float:
        a, b = np.array(a), np.array(b)
        return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

    def get(self, query: str) -> Optional[dict]:
        exact_key = f"cache:exact:{self._hash_query(query)}"
        cached = self.client.get(exact_key)
        if cached:
            return json.loads(cached)
        query_embedding = self._get_embedding(query)
        keys = self.client.keys("cache:semantic:*")
        for key in keys:
            data = json.loads(self.client.get(key))
            similarity = self._cosine_similarity(query_embedding, data.get("embedding", []))
            if similarity >= self.threshold:
                return data.get("response")
        return None

    def set(self, query: str, response: dict, ttl: int = 3600) -> None:
        exact_key = f"cache:exact:{self._hash_query(query)}"
        self.client.setex(exact_key, ttl, json.dumps(response))
        semantic_key = f"cache:semantic:{self._hash_query(query)}"
        embedding = self._get_embedding(query)
        self.client.setex(semantic_key, ttl, json.dumps({"embedding": embedding, "response": response}))

    def stats(self) -> dict:
        keys = self.client.keys("cache:*")
        return {"total_entries": len(keys), "threshold": self.threshold}

# routing/model_router.py
from openai import OpenAI
from typing import Dict

class ModelRouter:
    MODEL_TIERS = {
        "simple": {"model": "gpt-3.5-turbo", "cost_per_1k": 0.0005},
        "moderate": {"model": "gpt-4-turbo-preview", "cost_per_1k": 0.01},
        "complex": {"model": "gpt-4-turbo-preview", "cost_per_1k": 0.01},
    }

    def __init__(self, model: str = "gpt-4-turbo-preview"):
        self.client = OpenAI()
        self.classifier_model = model

    def classify_complexity(self, query: str) -> str:
        response = self.client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[
                {"role": "system", "content": """Classify query complexity.
                Return ONLY one word: simple, moderate, or complex.
                Simple: factual questions, simple calculations, basic lookups
                Moderate: analysis, comparisons, explanations
                Complex: multi-step reasoning, creative tasks, code generation"""},
                {"role": "user", "content": query},
            ],
            temperature=0.0,
            max_tokens=10,
        )
        complexity = response.choices[0].message.content.strip().lower()
        return complexity if complexity in self.MODEL_TIERS else "moderate"

    def route(self, query: str) -> Dict:
        complexity = self.classify_complexity(query)
        tier = self.MODEL_TIERS[complexity]
        return {"model": tier["model"], "complexity": complexity, "cost_per_1k": tier["cost_per_1k"]}

# optimization/token_optimizer.py
import tiktoken
from typing import Dict

class TokenOptimizer:
    def __init__(self, encoding_name: str = "cl100k_base"):
        self.enc = tiktoken.get_encoding(encoding_name)

    def count_tokens(self, text: str) -> int:
        return len(self.enc.encode(text))

    def compress_prompt(self, system_prompt: str, user_input: str, max_tokens: int = 4000) -> Dict:
        system_tokens = self.count_tokens(system_prompt)
        user_tokens = self.count_tokens(user_input)
        total = system_tokens + user_tokens
        if total <= max_tokens:
            return {"system": system_prompt, "user": user_input, "tokens_saved": 0}
        ratio = max_tokens / total
        compressed_system = self._truncate_to_tokens(system_prompt, int(system_tokens * ratio))
        compressed_user = self._truncate_to_tokens(user_input, int(user_tokens * ratio))
        saved = total - (self.count_tokens(compressed_system) + self.count_tokens(compressed_user))
        return {"system": compressed_system, "user": compressed_user, "tokens_saved": saved}

    def _truncate_to_tokens(self, text: str, max_tokens: int) -> str:
        tokens = self.enc.encode(text)
        return self.enc.decode(tokens[:max_tokens])

    def optimize_output(self, text: str, max_length: int = 500) -> str:
        if self.count_tokens(text) <= max_length:
            return text
        tokens = self.enc.encode(text)
        return self.enc.decode(tokens[:max_length]) + "..."

# tracking/cost_tracker.py
from typing import Dict, List
from datetime import datetime, timedelta
import json

class CostTracker:
    PRICING = {
        "gpt-4-turbo-preview": {"input": 0.01, "output": 0.03},
        "gpt-3.5-turbo": {"input": 0.0005, "output": 0.0015},
        "text-embedding-3-small": {"input": 0.00002, "output": 0},
    }

    def __init__(self):
        self.records: List[Dict] = []

    def record(
        self, model: str, input_tokens: int, output_tokens: int, user_id: str = "system"
    ) -> float:
        pricing = self.PRICING.get(model, {"input": 0.01, "output": 0.03})
        cost = (input_tokens * pricing["input"] + output_tokens * pricing["output"]) / 1000
        record = {
            "timestamp": datetime.now().isoformat(),
            "model": model,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "cost": cost,
            "user_id": user_id,
        }
        self.records.append(record)
        return cost

    def get_total_cost(self, period: timedelta = timedelta(hours=24)) -> float:
        cutoff = datetime.now() - period
        return sum(
            r["cost"] for r in self.records
            if datetime.fromisoformat(r["timestamp"]) > cutoff
        )

    def get_cost_by_model(self) -> Dict[str, float]:
        costs = {}
        for r in self.records:
            model = r["model"]
            costs[model] = costs.get(model, 0) + r["cost"]
        return costs

    def get_cost_by_user(self) -> Dict[str, float]:
        costs = {}
        for r in self.records:
            user = r["user_id"]
            costs[user] = costs.get(user, 0) + r["cost"]
        return costs

    def check_budget(self, budget: float, period: timedelta = timedelta(hours=24)) -> Dict:
        current = self.get_total_cost(period)
        return {"budget": budget, "current": current, "remaining": budget - current, "alert": current > budget * 0.8}

# optimizer.py
from caching.semantic_cache import SemanticCache
from routing.model_router import ModelRouter
from optimization.token_optimizer import TokenOptimizer
from tracking.cost_tracker import CostTracker
from openai import OpenAI
from typing import Dict

class CostOptimizedAgent:
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.cache = SemanticCache(redis_url)
        self.router = ModelRouter()
        self.optimizer = TokenOptimizer()
        self.tracker = CostTracker()
        self.client = OpenAI()

    async def process(self, query: str, user_id: str = "system") -> Dict:
        cached = self.cache.get(query)
        if cached:
            return {**cached, "cache_hit": True}
        route = self.router.route(query)
        compressed = self.optimizer.compress_prompt("You are a helpful assistant.", query)
        response = await self.client.chat.completions.create(
            model=route["model"],
            messages=[
                {"role": "system", "content": compressed["system"]},
                {"role": "user", "content": compressed["user"]},
            ],
            temperature=0.7,
        )
        answer = response.choices[0].message.content
        cost = self.tracker.record(
            route["model"],
            response.usage.prompt_tokens,
            response.usage.completion_tokens,
            user_id,
        )
        result = {
            "answer": answer,
            "model": route["model"],
            "complexity": route["complexity"],
            "tokens_used": response.usage.total_tokens,
            "cost": cost,
            "cache_hit": False,
            "tokens_saved": compressed["tokens_saved"],
        }
        self.cache.set(query, result)
        return result

    def get_analytics(self) -> Dict:
        return {
            "total_cost_24h": self.tracker.get_total_cost(),
            "cost_by_model": self.tracker.get_cost_by_model(),
            "cost_by_user": self.tracker.get_cost_by_user(),
            "cache_stats": self.cache.stats(),
        }

import pytest
from optimization.token_optimizer import TokenOptimizer

def test_token_counting():
    optimizer = TokenOptimizer()
    count = optimizer.count_tokens("Hello world")
    assert count > 0

def test_prompt_compression():
    optimizer = TokenOptimizer()
    result = optimizer.compress_prompt("System prompt", "User input " * 1000, max_tokens=100)
    assert result["tokens_saved"] > 0

# main.py
from optimizer import CostOptimizedAgent
import asyncio

async def main():
    agent = CostOptimizedAgent()
    queries = [
        "What is the capital of France?",
        "Explain quantum computing",
        "Write a Python function",
    ]
    for q in queries:
        result = await agent.process(q)
        print(f"Query: {q[:50]}...")
        print(f"Model: {result['model']}, Cost: ${result['cost']:.6f}")
        print(f"Cache hit: {result['cache_hit']}\n")
    print("Analytics:", agent.get_analytics())

if __name__ == "__main__":
    asyncio.run(main())