Multi-Model AI Cost Optimizer
Build a FastAPI FinOps engine that detects LLM cost waste across providers.
Introduction
30 Second Summary
Every time a company asks an AI model a question, it costs money. Some questions cost 25x more than they need to because they're routed to the most expensive model available, even when a cheaper one would give the same answer.
In this project, you will build a Python API that analyzes AI usage across multiple providers, detects cost waste, and generates optimization recommendations that translate directly to dollar savings.
What You'll Build
A FastAPI application that ingests multi-provider LLM usage logs, classifies prompt complexity, calculates real dollar spend using verified API pricing, and runs a "Dream Engine" analysis powered by Claude Haiku 4.5 to identify cost-saving model swaps.
By the end of this project, you'll have:
- A cost calculation engine that computes real dollar spend across Anthropic and OpenAI models using verified current pricing.
- A prompt complexity classifier that categorizes AI tasks into three tiers to detect expensive models handling simple work.
- A Dream Engine powered by Claude Haiku 4.5 that generates actionable FinOps recommendations with projected dollar savings, served through a production-ready REST API.
- Secret Mission: Add a Model Swap Simulator that calculates exact projected annual savings if all Dream Engine recommendations are implemented.
Are there any prerequisites?
You'll need Python 3.12 installed, VS Code as your editor, and API keys for both Anthropic and OpenAI. The project costs approximately $0.05 to $0.15 in Claude Haiku API calls for the analysis step.
Before We Start
Organizations spend thousands of dollars per month on AI model API calls. Without visibility into which models handle which tasks, teams default to the most expensive option for everything.
In this step, you will set up the project environment for your FinOps Dream Engine. This gets you ready to build a cost optimization system that detects waste across multi-model deployments.
In this step, get ready to:
- Create the project folder and Python virtual environment.
- Install all required dependencies from a pinned requirements file.
- Configure API keys and verify both SDKs load correctly.
Create the project directory and virtual environment
Every Python project needs its own isolated environment. A virtual environment keeps this project's packages separate from your system Python, preventing version conflicts.
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type Terminal and press Enter to open it.
- Move to your Desktop by running this command:
cd ~/Desktop- Create your project folder and move into it by running these commands:
mkdir dream-engine
cd dream-engine- Create a Python virtual environment by running:
python3 -m venv venvWhat does this command do?
python3 -m venv venv tells Python to create a new virtual environment in a folder called venv. This folder contains a dedicated Python interpreter and a clean package space just for this project.
- Activate the virtual environment by running:
macOS
source venv/bin/activateWindows
venv\Scripts\activateYou should see (venv) appear at the start of your terminal prompt. This confirms the virtual environment is active.
Don't see (venv) in your prompt?
On Windows, if you get a permissions error in PowerShell, run Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSigned first, then try the activate command again.
On macOS, make sure you typed source before the path. Without it, the shell won't execute the activation script.
Help me activate my Python virtual environment.
- Now open this folder in VS Code. Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type VS Code and press Enter to open it.
- In VS Code, click File in the top menu bar.
- Click Open Folder.
- Navigate to your Desktop, select the dream-engine folder, and click Open.
Create requirements.txt and install dependencies
Pinning dependency versions in a requirements.txt file ensures every environment uses the exact same package versions. This prevents subtle bugs from version drift.
- In VS Code, click the New File icon in the Explorer sidebar (or press Cmd+N on macOS / Ctrl+N on Windows).
- Name the file requirements.txt and paste the following contents:
anthropic==0.109.2
openai==2.43.0
fastapi[standard]==0.137.2
python-dotenv>=1.0.0What do these packages do?
- anthropic==0.109.2 is the official Anthropic SDK for calling Claude models.
- openai==2.43.0 is the official OpenAI SDK for tracking GPT model usage.
- fastapi[standard]==0.137.2 is the web framework for building your API. The [standard] extra bundles uvicorn (the server) automatically.
- python-dotenv>=1.0.0 loads environment variables from .env files so your API keys stay out of code.
- Save requirements.txt (Cmd+S on macOS / Ctrl+S on Windows).
- Install all the dependencies by running this command in your terminal:
pip install -r requirements.txtYou should see a stream of download and install messages. When it finishes, your terminal returns to the (venv) prompt with no errors.
- Verify that uvicorn (the server bundled with FastAPI) is available by running:
uvicorn --versionYou should see a version number printed. This confirms FastAPI's standard bundle installed correctly.
Seeing 'command not found' for pip or uvicorn?
Make sure your virtual environment is active. You should see (venv) in your terminal prompt. If not, re-run the activation command from the previous substep.
If pip is not found, try pip3 instead. Some systems require the explicit pip3 command.
Help me install packages in my Python virtual environment.
Create the .env file and verify connectivity
Your API keys need to live in a .env file that stays out of version control. The python-dotenv package reads this file at runtime so your code never hard-codes credentials.
- In VS Code, create a new file called .env in the dream-engine folder.
- Paste the following contents, replacing the placeholder values with your actual keys:
ANTHROPIC_API_KEY=your-api-key-here
OPENAI_API_KEY=your-api-key-hereWhere do I find my API keys?
Your Anthropic API key comes from console.anthropic.com under the API Keys section. Your OpenAI API key comes from platform.openai.com under API keys in your account settings.
Copy the full key value (starting with sk- for both providers) and paste it directly after the = sign with no spaces.
- Save .env (Cmd+S on macOS / Ctrl+S on Windows).
- Verify that both SDKs and all packages load correctly by running this command in your terminal:
python -c "import anthropic, openai, fastapi; print('All packages loaded')"You should see All packages loaded printed to your terminal with no errors.
Seeing a ModuleNotFoundError?
This means Python cannot find one of the packages. Double check that (venv) appears in your prompt. If you opened a new terminal window, you need to re-activate the virtual environment.
If you see an error for a specific package, run pip install -r requirements.txt again to retry the installation.
Help me fix a ModuleNotFoundError in my Python project.
✔️ Awesome, I've got everything!
Great. Double check you have saved both requirements.txt and .env in your dream-engine folder.
ⓧ I'd like to double check the full code
Here is what each file should contain at this point:
anthropic==0.109.2
openai==2.43.0
fastapi[standard]==0.137.2
python-dotenv>=1.0.0ANTHROPIC_API_KEY=your-api-key-here
OPENAI_API_KEY=your-api-key-hereYour environment is ready. Next up, you will build the data models and mock usage log generator that powers the entire cost analysis system.
Build Data Models and Mock Log Generator
Your environment is ready with all dependencies installed. Now you need the foundation every analytics system starts with: structured data models and realistic test data to validate your logic against.
In this step, you'll create three files that form the backbone of the Dream Engine. A configuration file holds verified API pricing for six models across two providers. A models file defines the Pydantic schemas every component will share. A mock data generator produces 150 realistic usage log entries spanning 7 days.
In this step, get ready to:
- Create a pricing configuration with verified per-model costs for Anthropic and OpenAI.
- Define Pydantic data models for usage logs, cost reports, and analysis outputs.
- Build a mock data generator that produces 7 days of realistic multi-provider usage logs.
Create the pricing configuration
Every cost calculation in the Dream Engine traces back to a single source of truth: the per-model pricing table. This file stores verified pricing from Anthropic and OpenAI's official documentation, maps model IDs to their providers, and defines keyword lists for classifying prompt complexity.
- Create a new file called config.py in your dream-engine folder.
- Add the following pricing configuration:
from dotenv import load_dotenv
load_dotenv()
# Verified pricing per million tokens (USD) as of June 2026
# Source: docs.anthropic.com/en/docs/about-claude/pricing
# Source: openai.com/api/pricing/
MODEL_PRICING = {
"claude-haiku-4-5": {"input": 1.00, "output": 5.00},
"claude-sonnet-4-6": {"input": 3.00, "output": 15.00},
"claude-opus-4-8": {"input": 5.00, "output": 25.00},
"gpt-4.1-mini": {"input": 0.40, "output": 1.60},
"gpt-5.4-mini": {"input": 0.75, "output": 4.50},
"gpt-5.4": {"input": 2.50, "output": 15.00},
}
PROVIDER_MAP = {
"claude-haiku-4-5": "anthropic",
"claude-sonnet-4-6": "anthropic",
"claude-opus-4-8": "anthropic",
"gpt-4.1-mini": "openai",
"gpt-5.4-mini": "openai",
"gpt-5.4": "openai",
}
COMPLEXITY_KEYWORDS = {
"simple": [
"summarize", "classify", "extract", "translate", "answer",
"list", "define", "explain briefly", "yes or no", "categorize",
],
"moderate": [
"compare", "analyze", "draft", "rewrite", "outline",
"suggest", "review", "evaluate", "improve", "edit",
],
"complex": [
"architect", "implement", "debug", "generate code", "design system",
"multi-step", "refactor", "optimize", "build", "create application",
],
}What does this code do?
MODEL_PRICING stores the cost per million tokens (input and output) for each model the system tracks. Claude Haiku 4.5 at $1/$5 per MTok is the cheapest Anthropic model, while Opus 4.8 at $5/$25 is the most expensive.
PROVIDER_MAP links each model ID to its vendor (Anthropic or OpenAI) so the system can group costs by provider. COMPLEXITY_KEYWORDS defines keyword lists used later to classify prompts into simple, moderate, or complex tiers.
- Save config.py.
Why pricing is stored as constants
Hardcoding verified pricing into a config file (rather than fetching it from an API) gives you a stable baseline for cost calculations. In production, you would periodically update these values when providers change their pricing.
Define the Pydantic data models
Data models act as contracts between every component in the system. When the mock data generator creates a log entry, the cost engine processes it, and the Dream Engine reports on it, they all share the same structured shapes defined here.
- Create a new file called models.py in your dream-engine folder.
- Add the following Pydantic model definitions:
from pydantic import BaseModel
from datetime import datetime
class UsageLog(BaseModel):
id: str
timestamp: datetime
provider: str
model: str
prompt_text: str
input_tokens: int
output_tokens: int
complexity: str = "unclassified"
class ModelCost(BaseModel):
model: str
provider: str
total_calls: int
total_input_tokens: int
total_output_tokens: int
input_cost_usd: float
output_cost_usd: float
total_cost_usd: float
percentage_of_spend: float
class CostReport(BaseModel):
total_spend_usd: float
period_days: int
model_breakdown: list[ModelCost]
cost_by_complexity: dict[str, dict[str, float]]
class Recommendation(BaseModel):
priority: int
category: str
current_model: str
recommended_model: str
affected_tasks: int
weekly_savings_usd: float
reasoning: str
class DreamReport(BaseModel):
analysis_timestamp: datetime
period_analyzed_days: int
total_current_spend_usd: float
total_projected_savings_usd: float
recommendations: list[Recommendation]
executive_summary: str
class SimulationReport(BaseModel):
current_weekly_spend_usd: float
optimized_weekly_spend_usd: float
weekly_savings_usd: float
monthly_savings_usd: float
annual_savings_usd: float
roi_percentage: float
swaps_applied: list[dict[str, str]]What does each model represent?
- UsageLog captures a single API call: which model was used, how many tokens were consumed, and the prompt text.
- ModelCost and CostReport aggregate spending across all logs into per-model breakdowns and complexity-based summaries.
- Recommendation and DreamReport structure the Dream Engine's AI-generated cost-saving advice.
- SimulationReport holds the results of a model swap simulation with projected annual savings.
- Save models.py.
Seeing a red underline on list[ModelCost]?
The list[ModelCost] syntax requires Python 3.9 or higher. If your editor shows an error, confirm you're using Python 3.12 by running python --version in your terminal.
help me fix a Python version mismatch in my dream-engine project
Build the mock data generator
A cost optimization engine is only as useful as the data it analyzes. The mock data generator simulates 7 days of real-world LLM usage across an organization where developers route tasks of varying complexity to different models.
Critically, the generator intentionally routes some simple tasks (like summarization) to expensive models (like Opus 4.8). This creates the realistic "waste" pattern that the Dream Engine will later detect and recommend fixing.
- Create a new file called mock_data.py in your dream-engine folder.
- Add the imports and sample prompt data by pasting this code:
import random
import uuid
from datetime import datetime, timedelta
from models import UsageLog
from config import PROVIDER_MAP
SAMPLE_PROMPTS = {
"simple": [
"Summarize this paragraph in one sentence.",
"Classify this email as spam or not spam.",
"Extract the person's name from this text.",
"Translate 'hello world' to French.",
"Answer yes or no: is Python a compiled language?",
"List the top 3 benefits of cloud computing.",
"Define the term 'microservice' in one sentence.",
"Categorize this support ticket as billing, technical, or general.",
"Explain briefly what an API is.",
"Extract the date from this invoice text.",
],
"moderate": [
"Compare REST and GraphQL for a mobile application backend.",
"Analyze this customer review for sentiment and key themes.",
"Draft a professional email responding to a client complaint.",
"Rewrite this paragraph to be more concise and formal.",
"Outline a content marketing strategy for a SaaS product.",
"Suggest improvements to this database schema.",
"Review this job description for inclusive language.",
"Evaluate the pros and cons of serverless vs containers.",
"Improve the error handling in this code snippet.",
"Edit this blog post for clarity and SEO optimization.",
],
"complex": [
"Architect a microservices system for a real-time trading platform.",
"Implement a rate limiter using the token bucket algorithm in Python.",
"Debug this async Python code that causes a race condition.",
"Generate code for a FastAPI endpoint with OAuth2 authentication.",
"Design system for handling 10M concurrent WebSocket connections.",
"Create a multi-step data pipeline with error recovery.",
"Refactor this monolithic Django app into domain-driven services.",
"Optimize this SQL query that takes 30 seconds on 1M rows.",
"Build a CI/CD pipeline config for a multi-service deployment.",
"Create application architecture for a real-time collaboration tool.",
],
}
MODELS_BY_COMPLEXITY_REALISTIC = {
"simple": ["claude-opus-4-8", "claude-haiku-4-5", "gpt-5.4", "gpt-4.1-mini", "claude-sonnet-4-6"],
"moderate": ["claude-sonnet-4-6", "claude-opus-4-8", "gpt-5.4", "gpt-5.4-mini"],
"complex": ["claude-opus-4-8", "gpt-5.4", "claude-sonnet-4-6"],
}
TOKEN_RANGES = {
"simple": {"input": (50, 200), "output": (20, 150)},
"moderate": {"input": (200, 800), "output": (150, 600)},
"complex": {"input": (500, 2000), "output": (400, 2000)},
}What does this data define?
SAMPLE_PROMPTS contains 30 realistic prompts across three complexity tiers. These represent the kind of tasks a real engineering team sends to LLMs daily.
MODELS_BY_COMPLEXITY_REALISTIC controls which models get assigned to each tier. Notice that simple tasks include claude-opus-4-8 and gpt-5.4 in their list. This is the waste pattern the Dream Engine will catch.
TOKEN_RANGES sets realistic input/output token counts per complexity tier. Simple tasks use fewer tokens, complex tasks use more.
- Now add the generator function below the data constants in the same file:
def generate_mock_logs(num_entries: int = 150, days: int = 7) -> list[UsageLog]:
logs = []
now = datetime.now()
start = now - timedelta(days=days)
complexity_weights = {"simple": 0.45, "moderate": 0.35, "complex": 0.20}
for _ in range(num_entries):
complexity = random.choices(
list(complexity_weights.keys()),
weights=list(complexity_weights.values()),
)[0]
model = random.choice(MODELS_BY_COMPLEXITY_REALISTIC[complexity])
prompt = random.choice(SAMPLE_PROMPTS[complexity])
input_range = TOKEN_RANGES[complexity]["input"]
output_range = TOKEN_RANGES[complexity]["output"]
timestamp = start + timedelta(
seconds=random.randint(0, days * 86400)
)
log = UsageLog(
id=str(uuid.uuid4()),
timestamp=timestamp,
provider=PROVIDER_MAP[model],
model=model,
prompt_text=prompt,
input_tokens=random.randint(*input_range),
output_tokens=random.randint(*output_range),
complexity=complexity,
)
logs.append(log)
return sorted(logs, key=lambda x: x.timestamp)
if __name__ == "__main__":
logs = generate_mock_logs()
print(f"Generated {len(logs)} usage logs over 7 days\n")
for log in logs[:5]:
print(f" [{log.complexity:8}] {log.model:20} | {log.prompt_text[:50]}...")
print(f" ... and {len(logs) - 5} more entries")What does generate_mock_logs() do?
The function creates 150 usage log entries distributed over 7 days. 45% of entries are simple tasks, 35% moderate, and 20% complex. This distribution mirrors real-world usage patterns where most API calls are routine.
The if __name__ == "__main__" block lets you run the file directly to preview generated data. It prints the first 5 logs so you can verify variety in models and prompt types.
- Save mock_data.py.
- Test the generator by running this command in your terminal:
python mock_data.pyYou should see output showing 150 generated logs with a sample of 5 entries displaying varied complexity levels, model names, and prompt text.
Seeing a ModuleNotFoundError?
If the error mentions models or config, make sure all three files (config.py, models.py, mock_data.py) are saved in the same dream-engine folder. Also confirm your terminal is in the dream-engine directory and the virtual environment is active.
help me debug a ModuleNotFoundError in my mock_data.py file
✔️ Awesome, I've got everything!
Great. Double check you've saved all three files before moving on.
ⓧ I'd like to double check the full code
Here is the complete code for each file. Compare your files against these:
from dotenv import load_dotenv
load_dotenv()
# Verified pricing per million tokens (USD) as of June 2026
# Source: docs.anthropic.com/en/docs/about-claude/pricing
# Source: openai.com/api/pricing/
MODEL_PRICING = {
"claude-haiku-4-5": {"input": 1.00, "output": 5.00},
"claude-sonnet-4-6": {"input": 3.00, "output": 15.00},
"claude-opus-4-8": {"input": 5.00, "output": 25.00},
"gpt-4.1-mini": {"input": 0.40, "output": 1.60},
"gpt-5.4-mini": {"input": 0.75, "output": 4.50},
"gpt-5.4": {"input": 2.50, "output": 15.00},
}
PROVIDER_MAP = {
"claude-haiku-4-5": "anthropic",
"claude-sonnet-4-6": "anthropic",
"claude-opus-4-8": "anthropic",
"gpt-4.1-mini": "openai",
"gpt-5.4-mini": "openai",
"gpt-5.4": "openai",
}
COMPLEXITY_KEYWORDS = {
"simple": [
"summarize", "classify", "extract", "translate", "answer",
"list", "define", "explain briefly", "yes or no", "categorize",
],
"moderate": [
"compare", "analyze", "draft", "rewrite", "outline",
"suggest", "review", "evaluate", "improve", "edit",
],
"complex": [
"architect", "implement", "debug", "generate code", "design system",
"multi-step", "refactor", "optimize", "build", "create application",
],
}from pydantic import BaseModel
from datetime import datetime
class UsageLog(BaseModel):
id: str
timestamp: datetime
provider: str
model: str
prompt_text: str
input_tokens: int
output_tokens: int
complexity: str = "unclassified"
class ModelCost(BaseModel):
model: str
provider: str
total_calls: int
total_input_tokens: int
total_output_tokens: int
input_cost_usd: float
output_cost_usd: float
total_cost_usd: float
percentage_of_spend: float
class CostReport(BaseModel):
total_spend_usd: float
period_days: int
model_breakdown: list[ModelCost]
cost_by_complexity: dict[str, dict[str, float]]
class Recommendation(BaseModel):
priority: int
category: str
current_model: str
recommended_model: str
affected_tasks: int
weekly_savings_usd: float
reasoning: str
class DreamReport(BaseModel):
analysis_timestamp: datetime
period_analyzed_days: int
total_current_spend_usd: float
total_projected_savings_usd: float
recommendations: list[Recommendation]
executive_summary: str
class SimulationReport(BaseModel):
current_weekly_spend_usd: float
optimized_weekly_spend_usd: float
weekly_savings_usd: float
monthly_savings_usd: float
annual_savings_usd: float
roi_percentage: float
swaps_applied: list[dict[str, str]]import random
import uuid
from datetime import datetime, timedelta
from models import UsageLog
from config import PROVIDER_MAP
SAMPLE_PROMPTS = {
"simple": [
"Summarize this paragraph in one sentence.",
"Classify this email as spam or not spam.",
"Extract the person's name from this text.",
"Translate 'hello world' to French.",
"Answer yes or no: is Python a compiled language?",
"List the top 3 benefits of cloud computing.",
"Define the term 'microservice' in one sentence.",
"Categorize this support ticket as billing, technical, or general.",
"Explain briefly what an API is.",
"Extract the date from this invoice text.",
],
"moderate": [
"Compare REST and GraphQL for a mobile application backend.",
"Analyze this customer review for sentiment and key themes.",
"Draft a professional email responding to a client complaint.",
"Rewrite this paragraph to be more concise and formal.",
"Outline a content marketing strategy for a SaaS product.",
"Suggest improvements to this database schema.",
"Review this job description for inclusive language.",
"Evaluate the pros and cons of serverless vs containers.",
"Improve the error handling in this code snippet.",
"Edit this blog post for clarity and SEO optimization.",
],
"complex": [
"Architect a microservices system for a real-time trading platform.",
"Implement a rate limiter using the token bucket algorithm in Python.",
"Debug this async Python code that causes a race condition.",
"Generate code for a FastAPI endpoint with OAuth2 authentication.",
"Design system for handling 10M concurrent WebSocket connections.",
"Create a multi-step data pipeline with error recovery.",
"Refactor this monolithic Django app into domain-driven services.",
"Optimize this SQL query that takes 30 seconds on 1M rows.",
"Build a CI/CD pipeline config for a multi-service deployment.",
"Create application architecture for a real-time collaboration tool.",
],
}
MODELS_BY_COMPLEXITY_REALISTIC = {
"simple": ["claude-opus-4-8", "claude-haiku-4-5", "gpt-5.4", "gpt-4.1-mini", "claude-sonnet-4-6"],
"moderate": ["claude-sonnet-4-6", "claude-opus-4-8", "gpt-5.4", "gpt-5.4-mini"],
"complex": ["claude-opus-4-8", "gpt-5.4", "claude-sonnet-4-6"],
}
TOKEN_RANGES = {
"simple": {"input": (50, 200), "output": (20, 150)},
"moderate": {"input": (200, 800), "output": (150, 600)},
"complex": {"input": (500, 2000), "output": (400, 2000)},
}
def generate_mock_logs(num_entries: int = 150, days: int = 7) -> list[UsageLog]:
logs = []
now = datetime.now()
start = now - timedelta(days=days)
complexity_weights = {"simple": 0.45, "moderate": 0.35, "complex": 0.20}
for _ in range(num_entries):
complexity = random.choices(
list(complexity_weights.keys()),
weights=list(complexity_weights.values()),
)[0]
model = random.choice(MODELS_BY_COMPLEXITY_REALISTIC[complexity])
prompt = random.choice(SAMPLE_PROMPTS[complexity])
input_range = TOKEN_RANGES[complexity]["input"]
output_range = TOKEN_RANGES[complexity]["output"]
timestamp = start + timedelta(
seconds=random.randint(0, days * 86400)
)
log = UsageLog(
id=str(uuid.uuid4()),
timestamp=timestamp,
provider=PROVIDER_MAP[model],
model=model,
prompt_text=prompt,
input_tokens=random.randint(*input_range),
output_tokens=random.randint(*output_range),
complexity=complexity,
)
logs.append(log)
return sorted(logs, key=lambda x: x.timestamp)
if __name__ == "__main__":
logs = generate_mock_logs()
print(f"Generated {len(logs)} usage logs over 7 days\n")
for log in logs[:5]:
print(f" [{log.complexity:8}] {log.model:20} | {log.prompt_text[:50]}...")
print(f" ... and {len(logs) - 5} more entries")Your data foundation is solid. You have pricing constants, structured models, and a generator producing realistic waste patterns. Next up, you'll build the cost calculation engine that turns these raw logs into dollar amounts and detects where expensive models are being used for simple tasks.
Build the Cost Calculation Engine
You have 150 usage logs spanning 7 days of multi-model AI usage. The data is there, but it means nothing without dollar signs attached.
In this step, you will build the cost engine that transforms raw token counts into real spend per model. More importantly, it will cross-reference spend against task complexity to reveal the core FinOps insight: expensive models being used for tasks that cheaper ones handle equally well.
In this step, get ready to:
- Build a prompt complexity classifier using keyword heuristics.
- Calculate real dollar spend per model using verified API pricing.
- Generate a cost-by-complexity breakdown that reveals waste.
Add the prompt complexity classifier
Before calculating costs, you need a way to classify each prompt by difficulty. A simple summarization task does not require the same model power as an architecture design task.
- Create a new file called cost_engine.py in your dream-engine folder.
- Add the imports and the complexity classifier function by pasting this code:
from collections import defaultdict
from models import UsageLog, ModelCost, CostReport
from config import MODEL_PRICING, PROVIDER_MAP, COMPLEXITY_KEYWORDS
def classify_prompt_complexity(prompt_text: str) -> str:
prompt_lower = prompt_text.lower()
for complexity, keywords in COMPLEXITY_KEYWORDS.items():
for keyword in keywords:
if keyword in prompt_lower:
return complexity
return "moderate"What does this code do?
The classify_prompt_complexity() function scans a prompt's text for keywords that signal difficulty. It checks against the COMPLEXITY_KEYWORDS dictionary you defined in config.py earlier.
If a prompt contains "summarize" or "classify", it is tagged as simple. If it contains "architect" or "generate code", it is tagged as complex. If no keywords match, it defaults to moderate.
- Save cost_engine.py.
Seeing an import error?
Make sure models.py and config.py are in the same folder as cost_engine.py. All three files must be at the top level of your dream-engine directory.
help me debug an import error in cost_engine.py
Calculate per-model costs
This function is the heart of the cost engine. It aggregates token counts per model, multiplies against the verified pricing table, and computes each model's share of total spend.
- In cost_engine.py, add the calculate_costs() function below your classifier by pasting this code:
def calculate_costs(logs: list[UsageLog]) -> CostReport:
model_stats = defaultdict(lambda: {
"total_calls": 0,
"total_input_tokens": 0,
"total_output_tokens": 0,
})
for log in logs:
stats = model_stats[log.model]
stats["total_calls"] += 1
stats["total_input_tokens"] += log.input_tokens
stats["total_output_tokens"] += log.output_tokens
model_costs = []
total_spend = 0.0
for model, stats in model_stats.items():
pricing = MODEL_PRICING.get(model)
if not pricing:
continue
input_cost = (stats["total_input_tokens"] / 1_000_000) * pricing["input"]
output_cost = (stats["total_output_tokens"] / 1_000_000) * pricing["output"]
total_cost = input_cost + output_cost
total_spend += total_cost
model_costs.append(ModelCost(
model=model,
provider=PROVIDER_MAP.get(model, "unknown"),
total_calls=stats["total_calls"],
total_input_tokens=stats["total_input_tokens"],
total_output_tokens=stats["total_output_tokens"],
input_cost_usd=round(input_cost, 6),
output_cost_usd=round(output_cost, 6),
total_cost_usd=round(total_cost, 6),
percentage_of_spend=0.0,
))
for mc in model_costs:
if total_spend > 0:
mc.percentage_of_spend = round((mc.total_cost_usd / total_spend) * 100, 1)
model_costs.sort(key=lambda x: x.total_cost_usd, reverse=True)
cost_by_complexity = generate_cost_by_complexity(logs)
days = 7
if logs:
time_span = (logs[-1].timestamp - logs[0].timestamp).days
if time_span > 0:
days = time_span
return CostReport(
total_spend_usd=round(total_spend, 6),
period_days=days,
model_breakdown=model_costs,
cost_by_complexity=cost_by_complexity,
)What does this code do?
- The first loop aggregates total calls and token counts per model using a defaultdict.
- The second loop multiplies token counts against MODEL_PRICING to get dollar amounts. The formula divides tokens by 1,000,000 because pricing is per million tokens.
- After totals are computed, a third pass calculates each model's percentage of total spend and sorts them highest-cost first.
- The function calls generate_cost_by_complexity() (which you will add next) and returns a complete CostReport.
- Save cost_engine.py.
Seeing a NameError for generate_cost_by_complexity?
That is expected. The calculate_costs() function calls generate_cost_by_complexity() which you have not written yet. You will add it in the next substep.
help me understand why I'm getting a NameError in cost_engine.py
Cross-reference costs with task complexity
This is where the key FinOps insight emerges. By mapping dollar spend against task complexity, you can see exactly how much money goes to expensive models for tasks that cheaper ones handle equally well.
- In cost_engine.py, add the generate_cost_by_complexity() function below calculate_costs() by pasting this code:
def generate_cost_by_complexity(logs: list[UsageLog]) -> dict[str, dict[str, float]]:
complexity_model_spend = defaultdict(lambda: defaultdict(float))
for log in logs:
pricing = MODEL_PRICING.get(log.model)
if not pricing:
continue
cost = (
(log.input_tokens / 1_000_000) * pricing["input"]
+ (log.output_tokens / 1_000_000) * pricing["output"]
)
complexity_model_spend[log.complexity][log.model] += cost
result = {}
for complexity, model_costs in complexity_model_spend.items():
result[complexity] = {
model: round(cost, 6)
for model, cost in sorted(model_costs.items(), key=lambda x: x[1], reverse=True)
}
return resultWhat does this code do?
This function groups every usage log by its complexity field (simple, moderate, complex) and then by model within each group. It calculates the dollar spend for each combination.
The output reveals waste patterns. If claude-opus-4-8 appears under "simple" with significant spend, that means expensive models are being used for tasks that claude-haiku-4-5 could handle at a fraction of the cost.
Now add the test runner that lets you execute the cost engine standalone.
- At the bottom of cost_engine.py, add this __main__ block:
if __name__ == "__main__":
from mock_data import generate_mock_logs
logs = generate_mock_logs()
report = calculate_costs(logs)
print(f"Total Spend: ${report.total_spend_usd:.4f} over {report.period_days} days\n")
print("Model Breakdown:")
for mc in report.model_breakdown:
print(f" {mc.model:20} | ${mc.total_cost_usd:.4f} ({mc.percentage_of_spend}%) | {mc.total_calls} calls")
print("\nCost by Complexity:")
for complexity, models in report.cost_by_complexity.items():
print(f" {complexity}:")
for model, cost in models.items():
print(f" {model:20} ${cost:.4f}")What does the main block do?
The __main__ block generates mock logs and runs the full cost pipeline. It prints a formatted report showing which models cost the most and how spend distributes across task complexity levels.
- Save cost_engine.py.
- Run the cost engine by executing this command in your terminal:
python cost_engine.py- You should see a report showing total spend, a model breakdown sorted by cost, and a cost-by-complexity section. Look for claude-opus-4-8 or gpt-5.4 appearing under "simple" tasks. That is the waste your Dream Engine will target.
Seeing a ModuleNotFoundError?
Make sure your virtual environment is activated. You should see (venv) or similar at the start of your terminal prompt. If not, activate it again before running the script.
Also confirm you are running the command from inside the dream-engine folder where all your Python files live.
help me fix a ModuleNotFoundError when running cost_engine.py
✔️ Awesome, I've got everything!
Great. Double check you have saved cost_engine.py and that your output shows both the model breakdown and the cost-by-complexity sections.
ⓧ I'd like to double check the full code
Here is the complete cost_engine.py file. Compare yours against this:
from collections import defaultdict
from models import UsageLog, ModelCost, CostReport
from config import MODEL_PRICING, PROVIDER_MAP, COMPLEXITY_KEYWORDS
def classify_prompt_complexity(prompt_text: str) -> str:
prompt_lower = prompt_text.lower()
for complexity, keywords in COMPLEXITY_KEYWORDS.items():
for keyword in keywords:
if keyword in prompt_lower:
return complexity
return "moderate"
def calculate_costs(logs: list[UsageLog]) -> CostReport:
model_stats = defaultdict(lambda: {
"total_calls": 0,
"total_input_tokens": 0,
"total_output_tokens": 0,
})
for log in logs:
stats = model_stats[log.model]
stats["total_calls"] += 1
stats["total_input_tokens"] += log.input_tokens
stats["total_output_tokens"] += log.output_tokens
model_costs = []
total_spend = 0.0
for model, stats in model_stats.items():
pricing = MODEL_PRICING.get(model)
if not pricing:
continue
input_cost = (stats["total_input_tokens"] / 1_000_000) * pricing["input"]
output_cost = (stats["total_output_tokens"] / 1_000_000) * pricing["output"]
total_cost = input_cost + output_cost
total_spend += total_cost
model_costs.append(ModelCost(
model=model,
provider=PROVIDER_MAP.get(model, "unknown"),
total_calls=stats["total_calls"],
total_input_tokens=stats["total_input_tokens"],
total_output_tokens=stats["total_output_tokens"],
input_cost_usd=round(input_cost, 6),
output_cost_usd=round(output_cost, 6),
total_cost_usd=round(total_cost, 6),
percentage_of_spend=0.0,
))
for mc in model_costs:
if total_spend > 0:
mc.percentage_of_spend = round((mc.total_cost_usd / total_spend) * 100, 1)
model_costs.sort(key=lambda x: x.total_cost_usd, reverse=True)
cost_by_complexity = generate_cost_by_complexity(logs)
days = 7
if logs:
time_span = (logs[-1].timestamp - logs[0].timestamp).days
if time_span > 0:
days = time_span
return CostReport(
total_spend_usd=round(total_spend, 6),
period_days=days,
model_breakdown=model_costs,
cost_by_complexity=cost_by_complexity,
)
def generate_cost_by_complexity(logs: list[UsageLog]) -> dict[str, dict[str, float]]:
complexity_model_spend = defaultdict(lambda: defaultdict(float))
for log in logs:
pricing = MODEL_PRICING.get(log.model)
if not pricing:
continue
cost = (
(log.input_tokens / 1_000_000) * pricing["input"]
+ (log.output_tokens / 1_000_000) * pricing["output"]
)
complexity_model_spend[log.complexity][log.model] += cost
result = {}
for complexity, model_costs in complexity_model_spend.items():
result[complexity] = {
model: round(cost, 6)
for model, cost in sorted(model_costs.items(), key=lambda x: x[1], reverse=True)
}
return result
if __name__ == "__main__":
from mock_data import generate_mock_logs
logs = generate_mock_logs()
report = calculate_costs(logs)
print(f"Total Spend: ${report.total_spend_usd:.4f} over {report.period_days} days\n")
print("Model Breakdown:")
for mc in report.model_breakdown:
print(f" {mc.model:20} | ${mc.total_cost_usd:.4f} ({mc.percentage_of_spend}%) | {mc.total_calls} calls")
print("\nCost by Complexity:")
for complexity, models in report.cost_by_complexity.items():
print(f" {complexity}:")
for model, cost in models.items():
print(f" {model:20} ${cost:.4f}")Your cost engine is live. You can now see exactly where money is being wasted on overpowered models. Next up, you will build the Dream Engine that uses Claude Haiku 4.5 to read this report and generate natural-language optimization recommendations.
Build the Dream Engine Analyzer
Your cost engine now knows exactly where every dollar goes across your multi-model deployment. The next question is: what should you do about it?
In this step, you will build the Dream Engine itself. This is a Claude Haiku 4.5-powered analyst that reads your cost-by-complexity report and generates actionable recommendations for model swaps with projected dollar savings.
In this step, get ready to:
- Define the FinOps analyst system prompt and build a structured analysis prompt formatter.
- Implement the Claude API call that sends usage data and parses structured JSON recommendations.
- Run the Dream Engine against your mock data and verify it produces actionable cost-saving recommendations.
Define the FinOps analyst persona and prompt builder
The Dream Engine works by giving Claude a persona (a senior FinOps analyst) and a structured data summary. The system prompt constrains Claude to respond only with valid JSON matching your Recommendation model.
- Create a new file called dream_engine.py in your dream-engine folder.
- Add the imports and FINOPS_SYSTEM_PROMPT constant by pasting this code:
import os
import json
from datetime import datetime
from anthropic import Anthropic
from models import UsageLog, CostReport, DreamReport, Recommendation
from config import MODEL_PRICING
FINOPS_SYSTEM_PROMPT = """You are a senior AI FinOps analyst specializing in LLM cost optimization for enterprise clients. Your role is to analyze multi-model usage patterns and identify specific, actionable cost-saving opportunities.
You always provide:
1. Specific model swap recommendations (from expensive to cheaper models for appropriate tasks)
2. Exact dollar savings projections based on the data provided
3. Priority rankings (1 = highest impact)
4. Clear reasoning for each recommendation
You respond ONLY with valid JSON matching this exact structure:
{
"recommendations": [
{
"priority": 1,
"category": "model_downgrade",
"current_model": "model-name",
"recommended_model": "cheaper-model-name",
"affected_tasks": 0,
"weekly_savings_usd": 0.0,
"reasoning": "explanation"
}
],
"executive_summary": "A 2-3 sentence summary for C-level stakeholders"
}"""What does this code do?
The system prompt establishes Claude's persona as a FinOps analyst and constrains its output to a strict JSON schema. This is critical for reliable parsing.
- The Anthropic import is the official SDK client you installed earlier.
- The JSON structure in the prompt maps directly to your Recommendation Pydantic model in models.py.
- By asking Claude to respond ONLY with JSON, you avoid having to strip prose from the response before parsing.
Next, the Dream Engine needs a function that formats your cost report and waste detection data into a structured prompt that Claude can analyze.
- In dream_engine.py, add the build_analysis_prompt() function below the system prompt:
def build_analysis_prompt(cost_report: CostReport, logs: list[UsageLog]) -> str:
simple_on_expensive = []
for log in logs:
if log.complexity == "simple" and log.model in ("claude-opus-4-8", "gpt-5.4"):
simple_on_expensive.append(log)
moderate_on_expensive = []
for log in logs:
if log.complexity == "moderate" and log.model in ("claude-opus-4-8", "gpt-5.4"):
moderate_on_expensive.append(log)
prompt = f"""Analyze this 7-day AI usage report and recommend cost optimizations.
## Current Spend Summary
- Total spend: ${cost_report.total_spend_usd:.4f} over {cost_report.period_days} days
- Projected monthly: ${(cost_report.total_spend_usd / cost_report.period_days * 30):.2f}
## Model Breakdown
"""
for mc in cost_report.model_breakdown:
prompt += f"- {mc.model}: ${mc.total_cost_usd:.4f} ({mc.percentage_of_spend}% of spend, {mc.total_calls} calls)\n"
prompt += f"""
## Cost by Task Complexity
"""
for complexity, models in cost_report.cost_by_complexity.items():
prompt += f"\n### {complexity.upper()} tasks:\n"
for model, cost in models.items():
prompt += f" - {model}: ${cost:.4f}\n"
prompt += f"""
## Waste Detection
- Simple tasks routed to expensive models: {len(simple_on_expensive)} calls
- Moderate tasks routed to expensive models: {len(moderate_on_expensive)} calls
## Available Cheaper Alternatives (verified pricing per MTok)
- claude-haiku-4-5: $1.00 input / $5.00 output (fastest, ideal for simple tasks)
- gpt-4.1-mini: $0.40 input / $1.60 output (cheapest, good for classification)
- gpt-5.4-mini: $0.75 input / $4.50 output (balanced speed/cost)
- claude-sonnet-4-6: $3.00 input / $15.00 output (good for moderate tasks)
## Your Task
Identify the top 3-5 cost-saving opportunities. For each, specify the exact model swap, number of affected tasks, and projected weekly savings in USD. Be specific and data-driven.
"""
return promptWhat does this code do?
This function builds a rich, structured prompt that gives Claude everything it needs to make data-driven recommendations.
- It counts how many simple and moderate tasks are being routed to expensive models (Opus 4.8 and GPT-5.4). This is the "waste detection" signal.
- It includes the full model breakdown with spend percentages so Claude can identify the highest-cost areas.
- It lists cheaper alternatives with their exact pricing so Claude can calculate precise savings projections.
- Save dream_engine.py.
Seeing a syntax error on the f-string?
The triple-quoted f-strings with dollar signs can be tricky. Make sure the opening and closing triple quotes are on their own lines.
Check that you have not accidentally added an extra curly brace. The ${cost_report.total_spend_usd:.4f} pattern uses a literal $ followed by an f-string expression in braces.
Help me debug a syntax error in my build_analysis_prompt function
Implement the Claude API call and response parser
Now you need the function that actually calls the Anthropic API and parses Claude's JSON response into your DreamReport model. The executive summary is generated by Claude as part of the same analysis call.
- In dream_engine.py, add the run_dream_analysis() function below build_analysis_prompt():
def run_dream_analysis(logs: list[UsageLog], cost_report: CostReport) -> DreamReport:
client = Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
analysis_prompt = build_analysis_prompt(cost_report, logs)
message = client.messages.create(
model="claude-haiku-4-5",
max_tokens=2048,
system=FINOPS_SYSTEM_PROMPT,
messages=[
{"role": "user", "content": analysis_prompt}
],
)
response_text = message.content[0].text
try:
parsed = json.loads(response_text)
except json.JSONDecodeError:
start = response_text.find("{")
end = response_text.rfind("}") + 1
if start != -1 and end > start:
parsed = json.loads(response_text[start:end])
else:
parsed = {
"recommendations": [],
"executive_summary": "Analysis could not be parsed. Raw response stored.",
}
recommendations = [
Recommendation(**rec) for rec in parsed.get("recommendations", [])
]
total_savings = sum(r.weekly_savings_usd for r in recommendations)
return DreamReport(
analysis_timestamp=datetime.now(),
period_analyzed_days=cost_report.period_days,
total_current_spend_usd=cost_report.total_spend_usd,
total_projected_savings_usd=round(total_savings, 4),
recommendations=recommendations,
executive_summary=parsed.get("executive_summary", ""),
)What does this code do?
- Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY")) creates the SDK client using the key from your .env file.
- client.messages.create() sends the analysis to claude-haiku-4-5. This is the cheapest Claude model at $1/$5 per MTok, which is intentional. You are using the cheapest model to find savings on expensive ones.
- The try/except block handles cases where Claude wraps its JSON in markdown or extra text. The fallback searches for the outermost curly braces.
- The executive summary is part of Claude's JSON response. By embedding it in the same call, you avoid a second API request.
Why Claude Haiku 4.5 for the analyzer?
Using the cheapest model to analyze cost optimization is intentional dogfooding. Haiku 4.5 at $1/$5 per MTok is fully capable of structured JSON output and analytical reasoning. Spending $25/MTok on Opus to tell you to stop spending $25/MTok on Opus would defeat the purpose.
- Save dream_engine.py.
Seeing a red underline on Anthropic?
Make sure your virtual environment is activated. The anthropic package must be installed in the active environment for VS Code to resolve the import.
Help me fix an import error with the Anthropic SDK in my dream_engine.py
Run the Dream Engine and verify recommendations
Time to wire everything together with a test runner that generates mock data, calculates costs, and feeds it all to the Dream Engine.
- At the bottom of dream_engine.py, add the test runner block:
if __name__ == "__main__":
from mock_data import generate_mock_logs
from cost_engine import calculate_costs
logs = generate_mock_logs()
cost_report = calculate_costs(logs)
dream_report = run_dream_analysis(logs, cost_report)
print(f"Dream Engine Analysis Complete")
print(f"{'=' * 60}")
print(f"Period: {dream_report.period_analyzed_days} days")
print(f"Current spend: ${dream_report.total_current_spend_usd:.4f}")
print(f"Projected weekly savings: ${dream_report.total_projected_savings_usd:.4f}")
print(f"\nRecommendations:")
for rec in dream_report.recommendations:
print(f" [{rec.priority}] {rec.current_model} -> {rec.recommended_model}")
print(f" {rec.affected_tasks} tasks | saves ${rec.weekly_savings_usd:.4f}/week")
print(f" {rec.reasoning}")
print(f"\nExecutive Summary:\n {dream_report.executive_summary}")What does this code do?
- It generates 150 mock usage logs, calculates costs with your cost engine, then passes everything to the Dream Engine for analysis.
- The output prints each recommendation showing the model swap, number of affected tasks, and projected weekly dollar savings.
- The executive summary at the end is Claude's C-level stakeholder briefing generated from the same API call.
- Save dream_engine.py.
- Run the Dream Engine by running this command:
python dream_engine.pyYou should see output showing at least 3 recommendations with specific model swaps and dollar savings. The recommendations will look something like [1] claude-opus-4-8 -> claude-haiku-4-5 with a task count and weekly savings figure.
Why is the output different each time?
Two sources of randomness are at play. Your mock data generator uses random sampling, so the usage logs differ each run. Claude also generates slightly different recommendations depending on the exact data it receives. The key insight stays consistent: expensive models are being used for simple tasks.
Getting an authentication error?
Make sure your .env file contains a valid ANTHROPIC_API_KEY value. The key should start with sk-ant-.
If you see a JSONDecodeError after the API call succeeds, the fallback parser should handle it. If it still fails, check that your system prompt matches exactly (Claude may add markdown formatting if the JSON-only instruction is missing).
Help me debug my Dream Engine API call or JSON parsing
This API call costs real money
Each run of the Dream Engine sends roughly 1,000-2,000 tokens to Claude Haiku 4.5 and receives about 500-1,000 tokens back. At $1/$5 per MTok, each run costs well under $0.01. Running it 10+ times during development will still cost less than a penny.
✔️ Awesome, I've got everything!
Great. Make sure you have saved your file before moving on.
ⓧ I'd like to double check the full code
Here is the complete dream_engine.py file for comparison:
import os
import json
from datetime import datetime
from anthropic import Anthropic
from models import UsageLog, CostReport, DreamReport, Recommendation
from config import MODEL_PRICING
FINOPS_SYSTEM_PROMPT = """You are a senior AI FinOps analyst specializing in LLM cost optimization for enterprise clients. Your role is to analyze multi-model usage patterns and identify specific, actionable cost-saving opportunities.
You always provide:
1. Specific model swap recommendations (from expensive to cheaper models for appropriate tasks)
2. Exact dollar savings projections based on the data provided
3. Priority rankings (1 = highest impact)
4. Clear reasoning for each recommendation
You respond ONLY with valid JSON matching this exact structure:
{
"recommendations": [
{
"priority": 1,
"category": "model_downgrade",
"current_model": "model-name",
"recommended_model": "cheaper-model-name",
"affected_tasks": 0,
"weekly_savings_usd": 0.0,
"reasoning": "explanation"
}
],
"executive_summary": "A 2-3 sentence summary for C-level stakeholders"
}"""
def build_analysis_prompt(cost_report: CostReport, logs: list[UsageLog]) -> str:
simple_on_expensive = []
for log in logs:
if log.complexity == "simple" and log.model in ("claude-opus-4-8", "gpt-5.4"):
simple_on_expensive.append(log)
moderate_on_expensive = []
for log in logs:
if log.complexity == "moderate" and log.model in ("claude-opus-4-8", "gpt-5.4"):
moderate_on_expensive.append(log)
prompt = f"""Analyze this 7-day AI usage report and recommend cost optimizations.
## Current Spend Summary
- Total spend: ${cost_report.total_spend_usd:.4f} over {cost_report.period_days} days
- Projected monthly: ${(cost_report.total_spend_usd / cost_report.period_days * 30):.2f}
## Model Breakdown
"""
for mc in cost_report.model_breakdown:
prompt += f"- {mc.model}: ${mc.total_cost_usd:.4f} ({mc.percentage_of_spend}% of spend, {mc.total_calls} calls)\n"
prompt += f"""
## Cost by Task Complexity
"""
for complexity, models in cost_report.cost_by_complexity.items():
prompt += f"\n### {complexity.upper()} tasks:\n"
for model, cost in models.items():
prompt += f" - {model}: ${cost:.4f}\n"
prompt += f"""
## Waste Detection
- Simple tasks routed to expensive models: {len(simple_on_expensive)} calls
- Moderate tasks routed to expensive models: {len(moderate_on_expensive)} calls
## Available Cheaper Alternatives (verified pricing per MTok)
- claude-haiku-4-5: $1.00 input / $5.00 output (fastest, ideal for simple tasks)
- gpt-4.1-mini: $0.40 input / $1.60 output (cheapest, good for classification)
- gpt-5.4-mini: $0.75 input / $4.50 output (balanced speed/cost)
- claude-sonnet-4-6: $3.00 input / $15.00 output (good for moderate tasks)
## Your Task
Identify the top 3-5 cost-saving opportunities. For each, specify the exact model swap, number of affected tasks, and projected weekly savings in USD. Be specific and data-driven.
"""
return prompt
def run_dream_analysis(logs: list[UsageLog], cost_report: CostReport) -> DreamReport:
client = Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
analysis_prompt = build_analysis_prompt(cost_report, logs)
message = client.messages.create(
model="claude-haiku-4-5",
max_tokens=2048,
system=FINOPS_SYSTEM_PROMPT,
messages=[
{"role": "user", "content": analysis_prompt}
],
)
response_text = message.content[0].text
try:
parsed = json.loads(response_text)
except json.JSONDecodeError:
start = response_text.find("{")
end = response_text.rfind("}") + 1
if start != -1 and end > start:
parsed = json.loads(response_text[start:end])
else:
parsed = {
"recommendations": [],
"executive_summary": "Analysis could not be parsed. Raw response stored.",
}
recommendations = [
Recommendation(**rec) for rec in parsed.get("recommendations", [])
]
total_savings = sum(r.weekly_savings_usd for r in recommendations)
return DreamReport(
analysis_timestamp=datetime.now(),
period_analyzed_days=cost_report.period_days,
total_current_spend_usd=cost_report.total_spend_usd,
total_projected_savings_usd=round(total_savings, 4),
recommendations=recommendations,
executive_summary=parsed.get("executive_summary", ""),
)
if __name__ == "__main__":
from mock_data import generate_mock_logs
from cost_engine import calculate_costs
logs = generate_mock_logs()
cost_report = calculate_costs(logs)
dream_report = run_dream_analysis(logs, cost_report)
print(f"Dream Engine Analysis Complete")
print(f"{'=' * 60}")
print(f"Period: {dream_report.period_analyzed_days} days")
print(f"Current spend: ${dream_report.total_current_spend_usd:.4f}")
print(f"Projected weekly savings: ${dream_report.total_projected_savings_usd:.4f}")
print(f"\nRecommendations:")
for rec in dream_report.recommendations:
print(f" [{rec.priority}] {rec.current_model} -> {rec.recommended_model}")
print(f" {rec.affected_tasks} tasks | saves ${rec.weekly_savings_usd:.4f}/week")
print(f" {rec.reasoning}")
print(f"\nExecutive Summary:\n {dream_report.executive_summary}")Your Dream Engine is alive. It reads cost data, identifies waste, and generates actionable recommendations with dollar amounts. Next up, you will wire everything together into a production-ready FastAPI application with documented endpoints.
Assemble the FastAPI Application
You have a complete data pipeline: mock logs, a cost engine, and a Dream Engine analyzer. Right now, each module runs as a standalone script.
To make this system useful to enterprise clients, you need to expose it as an API. In this step, you will wire all modules together into a FastAPI application with REST endpoints that any dashboard, scheduler, or frontend can call.
In this step, get ready to:
- Create the FastAPI app with a lifespan context manager and CORS middleware.
- Add REST endpoints for usage logs, cost reports, pattern analysis, and Dream Engine execution.
- Run and test the application using the interactive Swagger UI.
Create the FastAPI application and configure middleware
The app needs to load mock usage data before it starts handling requests. FastAPI's lifespan context manager lets you run setup code exactly once at startup.
- Create a new file called main.py in your dream-engine folder.
- Add the imports, global state, and lifespan context manager by pasting the following code:
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from dotenv import load_dotenv
from models import UsageLog, CostReport, DreamReport, SimulationReport
from mock_data import generate_mock_logs
from cost_engine import calculate_costs
from dream_engine import run_dream_analysis
from config import MODEL_PRICING
load_dotenv()
# Global list that holds the generated usage logs for all endpoints
usage_logs: list[UsageLog] = []
@asynccontextmanager
async def lifespan(app: FastAPI):
# Generate mock data once when the server starts
global usage_logs
usage_logs = generate_mock_logs(num_entries=150, days=7)
print(f"Loaded {len(usage_logs)} usage logs for analysis")
yieldWhat does this code do?
The lifespan function is an async context manager that runs before the app starts accepting requests. Code before yield executes on startup. Code after yield would run on shutdown.
By generating 150 mock logs at startup, every endpoint can reference the same dataset without regenerating it on each request.
- Directly below the lifespan function, add the FastAPI app instance with OpenAPI metadata and CORS middleware:
app = FastAPI(
title="AI FinOps Dream Engine",
description="Multi-model cost optimization and governance API. "
"Analyzes LLM usage patterns across providers and recommends "
"cost-saving model swaps with projected ROI.",
version="1.0.0",
lifespan=lifespan,
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)What does this code do?
The FastAPI() constructor takes metadata fields (title, description, version) that populate the auto-generated Swagger UI documentation page. The lifespan parameter connects your startup logic.
CORS middleware allows requests from any origin. In production you would restrict allow_origins to specific domains, but for local development the wildcard is fine.
- Save main.py.
You won't see a change yet. The next section adds the endpoints that bring the app to life.
Seeing an import error?
Make sure your virtual environment is activated. If you see ModuleNotFoundError, confirm you ran pip install -r requirements.txt in Step 1.
Help me debug this import error
Add the API endpoints
Each endpoint exposes one capability of your FinOps system as a REST route. The GET endpoints return data. The POST endpoints trigger analysis.
- Below the CORS middleware block in main.py, add the usage and cost report endpoints:
@app.get("/api/v1/usage", response_model=list[UsageLog])
def get_usage_logs():
"""Return all usage logs for the analysis period."""
return usage_logs
@app.get("/api/v1/costs", response_model=CostReport)
def get_cost_report():
"""Calculate and return the full cost breakdown report."""
if not usage_logs:
raise HTTPException(status_code=404, detail="No usage logs available")
return calculate_costs(usage_logs)What does this code do?
- GET /api/v1/usage returns the raw list of 150 usage logs as JSON. The response_model parameter tells FastAPI to validate and serialize the output using the Pydantic model.
- GET /api/v1/costs calls your calculate_costs() function from earlier and returns the full cost report. If no logs exist, it raises a 404 error.
- Add the patterns endpoint below the cost report endpoint:
@app.get("/api/v1/patterns")
def get_patterns():
"""Return usage patterns grouped by complexity and model."""
if not usage_logs:
raise HTTPException(status_code=404, detail="No usage logs available")
patterns = {"simple": {}, "moderate": {}, "complex": {}}
for log in usage_logs:
if log.complexity not in patterns:
patterns[log.complexity] = {}
if log.model not in patterns[log.complexity]:
patterns[log.complexity][log.model] = 0
patterns[log.complexity][log.model] += 1
return {
"total_logs": len(usage_logs),
"patterns": patterns,
"waste_indicators": {
"simple_tasks_on_opus": sum(
1 for log in usage_logs
if log.complexity == "simple" and log.model == "claude-opus-4-8"
),
"simple_tasks_on_gpt54": sum(
1 for log in usage_logs
if log.complexity == "simple" and log.model == "gpt-5.4"
),
},
}What does this code do?
This endpoint groups usage logs by complexity tier and model, then counts how many calls each model received per tier. The waste_indicators section surfaces the key FinOps insight: how many simple tasks are being routed to expensive models like Opus or GPT-5.4.
- Add the Dream Engine endpoint that triggers the Claude Haiku 4.5 analysis:
@app.post("/api/v1/dream", response_model=DreamReport)
def run_dream_engine():
"""Execute the Dream Engine analysis using Claude Haiku 4.5."""
if not usage_logs:
raise HTTPException(status_code=404, detail="No usage logs available")
if not os.environ.get("ANTHROPIC_API_KEY"):
raise HTTPException(
status_code=500,
detail="ANTHROPIC_API_KEY not configured",
)
cost_report = calculate_costs(usage_logs)
return run_dream_analysis(usage_logs, cost_report)What does this code do?
This POST endpoint is the core of the system. It first validates that the API key is configured, then calculates costs and passes everything to your run_dream_analysis() function.
The double validation (logs exist AND API key present) prevents confusing errors. Without the key check, you would get an opaque SDK error instead of a clear message.
- Add the model swap simulator endpoint below the dream endpoint:
@app.post("/api/v1/simulate", response_model=SimulationReport)
def simulate_swaps():
"""Simulate cost savings if all simple tasks are routed to cheaper models."""
if not usage_logs:
raise HTTPException(status_code=404, detail="No usage logs available")
# Define which models should be swapped for each complexity tier
swap_rules = {
"simple": {
"claude-opus-4-8": "claude-haiku-4-5",
"claude-sonnet-4-6": "claude-haiku-4-5",
"gpt-5.4": "gpt-4.1-mini",
"gpt-5.4-mini": "gpt-4.1-mini",
},
"moderate": {
"claude-opus-4-8": "claude-sonnet-4-6",
"gpt-5.4": "gpt-5.4-mini",
},
}
current_cost = 0.0
optimized_cost = 0.0
swaps_applied = []
for log in usage_logs:
pricing = MODEL_PRICING.get(log.model)
if not pricing:
continue
# Calculate original cost for this log entry
cost = (
(log.input_tokens / 1_000_000) * pricing["input"]
+ (log.output_tokens / 1_000_000) * pricing["output"]
)
current_cost += cost
# Check if this log qualifies for a model swap
swapped_model = swap_rules.get(log.complexity, {}).get(log.model)
if swapped_model:
new_pricing = MODEL_PRICING[swapped_model]
new_cost = (
(log.input_tokens / 1_000_000) * new_pricing["input"]
+ (log.output_tokens / 1_000_000) * new_pricing["output"]
)
optimized_cost += new_cost
# Track unique swaps applied
swap_key = f"{log.model}->{swapped_model}"
if swap_key not in [s.get("swap") for s in swaps_applied]:
swaps_applied.append({
"swap": swap_key,
"from_model": log.model,
"to_model": swapped_model,
"complexity": log.complexity,
})
else:
optimized_cost += cost
weekly_savings = current_cost - optimized_cost
return SimulationReport(
current_weekly_spend_usd=round(current_cost, 4),
optimized_weekly_spend_usd=round(optimized_cost, 4),
weekly_savings_usd=round(weekly_savings, 4),
monthly_savings_usd=round(weekly_savings * 4.33, 2),
annual_savings_usd=round(weekly_savings * 52, 2),
roi_percentage=round((weekly_savings / current_cost * 100) if current_cost > 0 else 0, 1),
swaps_applied=swaps_applied,
)What does this code do?
The simulate endpoint re-costs every usage log with cheaper model alternatives applied. For simple tasks, it swaps Opus to Haiku and GPT-5.4 to GPT-4.1-mini. For moderate tasks, it downgrades Opus to Sonnet and GPT-5.4 to GPT-5.4-mini.
The result shows before-and-after spend, weekly/monthly/annual savings projections, and the ROI percentage. This is the number enterprise clients care about most.
- Save main.py.
Seeing a red underline on SimulationReport?
Check that your import line at the top includes SimulationReport. The full import should read from models import UsageLog, CostReport, DreamReport, SimulationReport.
Help me fix this import issue
✔️ Awesome, I've got everything!
Great. Double check you have saved main.py before moving on.
ⓧ I'd like to double check the full code
Here is the complete main.py file for comparison:
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from dotenv import load_dotenv
from models import UsageLog, CostReport, DreamReport, SimulationReport
from mock_data import generate_mock_logs
from cost_engine import calculate_costs
from dream_engine import run_dream_analysis
from config import MODEL_PRICING
load_dotenv()
usage_logs: list[UsageLog] = []
@asynccontextmanager
async def lifespan(app: FastAPI):
global usage_logs
usage_logs = generate_mock_logs(num_entries=150, days=7)
print(f"Loaded {len(usage_logs)} usage logs for analysis")
yield
app = FastAPI(
title="AI FinOps Dream Engine",
description="Multi-model cost optimization and governance API. "
"Analyzes LLM usage patterns across providers and recommends "
"cost-saving model swaps with projected ROI.",
version="1.0.0",
lifespan=lifespan,
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/api/v1/usage", response_model=list[UsageLog])
def get_usage_logs():
"""Return all usage logs for the analysis period."""
return usage_logs
@app.get("/api/v1/costs", response_model=CostReport)
def get_cost_report():
"""Calculate and return the full cost breakdown report."""
if not usage_logs:
raise HTTPException(status_code=404, detail="No usage logs available")
return calculate_costs(usage_logs)
@app.get("/api/v1/patterns")
def get_patterns():
"""Return usage patterns grouped by complexity and model."""
if not usage_logs:
raise HTTPException(status_code=404, detail="No usage logs available")
patterns = {"simple": {}, "moderate": {}, "complex": {}}
for log in usage_logs:
if log.complexity not in patterns:
patterns[log.complexity] = {}
if log.model not in patterns[log.complexity]:
patterns[log.complexity][log.model] = 0
patterns[log.complexity][log.model] += 1
return {
"total_logs": len(usage_logs),
"patterns": patterns,
"waste_indicators": {
"simple_tasks_on_opus": sum(
1 for log in usage_logs
if log.complexity == "simple" and log.model == "claude-opus-4-8"
),
"simple_tasks_on_gpt54": sum(
1 for log in usage_logs
if log.complexity == "simple" and log.model == "gpt-5.4"
),
},
}
@app.post("/api/v1/dream", response_model=DreamReport)
def run_dream_engine():
"""Execute the Dream Engine analysis using Claude Haiku 4.5."""
if not usage_logs:
raise HTTPException(status_code=404, detail="No usage logs available")
if not os.environ.get("ANTHROPIC_API_KEY"):
raise HTTPException(
status_code=500,
detail="ANTHROPIC_API_KEY not configured",
)
cost_report = calculate_costs(usage_logs)
return run_dream_analysis(usage_logs, cost_report)
@app.post("/api/v1/simulate", response_model=SimulationReport)
def simulate_swaps():
"""Simulate cost savings if all simple tasks are routed to cheaper models."""
if not usage_logs:
raise HTTPException(status_code=404, detail="No usage logs available")
swap_rules = {
"simple": {
"claude-opus-4-8": "claude-haiku-4-5",
"claude-sonnet-4-6": "claude-haiku-4-5",
"gpt-5.4": "gpt-4.1-mini",
"gpt-5.4-mini": "gpt-4.1-mini",
},
"moderate": {
"claude-opus-4-8": "claude-sonnet-4-6",
"gpt-5.4": "gpt-5.4-mini",
},
}
current_cost = 0.0
optimized_cost = 0.0
swaps_applied = []
for log in usage_logs:
pricing = MODEL_PRICING.get(log.model)
if not pricing:
continue
cost = (
(log.input_tokens / 1_000_000) * pricing["input"]
+ (log.output_tokens / 1_000_000) * pricing["output"]
)
current_cost += cost
swapped_model = swap_rules.get(log.complexity, {}).get(log.model)
if swapped_model:
new_pricing = MODEL_PRICING[swapped_model]
new_cost = (
(log.input_tokens / 1_000_000) * new_pricing["input"]
+ (log.output_tokens / 1_000_000) * new_pricing["output"]
)
optimized_cost += new_cost
swap_key = f"{log.model}->{swapped_model}"
if swap_key not in [s.get("swap") for s in swaps_applied]:
swaps_applied.append({
"swap": swap_key,
"from_model": log.model,
"to_model": swapped_model,
"complexity": log.complexity,
})
else:
optimized_cost += cost
weekly_savings = current_cost - optimized_cost
return SimulationReport(
current_weekly_spend_usd=round(current_cost, 4),
optimized_weekly_spend_usd=round(optimized_cost, 4),
weekly_savings_usd=round(weekly_savings, 4),
monthly_savings_usd=round(weekly_savings * 4.33, 2),
annual_savings_usd=round(weekly_savings * 52, 2),
roi_percentage=round((weekly_savings / current_cost * 100) if current_cost > 0 else 0, 1),
swaps_applied=swaps_applied,
)Run and test the application
Time to see the entire system working as a live API. Uvicorn serves your FastAPI app and auto-reloads when you change code.
- Start the development server by running this command in your terminal:
uvicorn main:app --reload- You should see output that includes Loaded 150 usage logs for analysis followed by Uvicorn running on http://127.0.0.1:8000.
What does this command do?
- uvicorn is the ASGI server bundled with fastapi[standard].
- main:app tells Uvicorn to import the app object from main.py.
- --reload watches for file changes and restarts the server automatically. Useful for development only.
- Open your browser and navigate to http://127.0.0.1:8000/docs.
- You should see the Swagger UI with your API title "AI FinOps Dream Engine" and all five endpoints listed.
- Click GET /api/v1/costs to expand it.
- Click Try it out, then click Execute.
- You should see a JSON response with total_spend_usd, model_breakdown, and cost_by_complexity fields.
- Now test the Dream Engine. Click POST /api/v1/dream, then Try it out, then Execute.
- You should see a DreamReport response with recommendations, total_projected_savings_usd, and an executive_summary field.
The Dream endpoint takes a few seconds
The /api/v1/dream endpoint calls the Claude API in real time. Expect a 2-5 second wait while Claude Haiku 4.5 analyzes your usage patterns and generates recommendations.
Seeing a 500 error on /api/v1/dream?
Check that your .env file contains a valid ANTHROPIC_API_KEY. The error detail in the response body will tell you if the key is missing versus an authentication failure.
If the server started but the endpoint fails, try restarting Uvicorn with Ctrl+C then running the start command again so the .env file is re-read.
Help me debug this 500 error
Your FinOps Dream Engine is live. You have a production-ready API that ingests usage data, calculates costs, detects waste, and generates AI-powered optimization recommendations. Next up, the Secret Mission challenges you to explore the Model Swap Simulator in depth.
Secret mission
Model Swap Simulator
Your Dream Engine identifies waste, but can you quantify the exact annual savings if every recommendation is implemented? The Model Swap Simulator takes your optimization rules and projects real dollar ROI across an entire year.
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely locally on your machine. The only cost was the one-time Claude Haiku API call (roughly $0.05-0.15), which has already been charged. There are no ongoing costs to worry about.
Decide whether to keep your resources running, pause them to come back later, or delete them entirely.
Resources you used:
- The dream-engine project folder on your Desktop (contains all Python files and virtual environment).
- Python virtual environment inside the project folder.
- Anthropic API key (used for the Dream Engine analysis).
- OpenAI API key (referenced in .env but not called directly in this project).
Keep everything running
No action needed. Choose this if you plan to extend the Dream Engine with real API log ingestion or add more providers.
- Your dream-engine folder stays on your Desktop with all files intact.
- Your virtual environment remains ready to activate whenever you return.
- Your API keys remain active on both Anthropic and OpenAI.
Pause - I'll come back to this later
Stop the running server to free up your terminal. Everything else stays in place for when you return.
- If the FastAPI server is still running, press Ctrl+C in the terminal to stop it.
- Deactivate your virtual environment by running:
deactivate- When you return, reactivate the environment and start the server again:
cd ~/Desktop/dream-engine
source venv/bin/activate
uvicorn main:app --reloadDelete - I don't want to use this again
Remove all project files from your machine. This cannot be undone.
- If the FastAPI server is still running, press Ctrl+C in the terminal to stop it.
- Deactivate your virtual environment by running:
deactivate- Delete the entire project folder by running:
macOS
rm -rf ~/Desktop/dream-engineWindows
rmdir /s /q %USERPROFILE%\Desktop\dream-engine- If you created your Anthropic API key specifically for this project, you can revoke it from the API Keys page in your Anthropic console at console.anthropic.com.
- If you created your OpenAI API key specifically for this project, you can revoke it from the API Keys page in your OpenAI dashboard at platform.openai.com.
Nice Work!
Nice Work!
Nice work! You just built a production-grade AI FinOps governance engine that detects cost waste across multi-model LLM deployments and generates actionable optimization recommendations.
You've learned how to:
- Build a cost calculation engine using verified API pricing across Anthropic and OpenAI models, with a prompt complexity classifier that categorizes tasks by required capability level.
- Create a Dream Engine powered by Claude Haiku 4.5 that analyzes 7-day usage patterns and outputs specific model swap recommendations with projected dollar savings.
- Assemble a production-ready FastAPI application with self-documenting Swagger UI, serving four REST endpoints for usage ingestion, cost reporting, pattern analysis, and AI-powered optimization.
- Secret Mission: Added a Model Swap Simulator that calculates exact projected annual savings if all Dream Engine recommendations are implemented.
Ready to quiz yourself?