Track and reduce the cost of every AI call

Installation

CogsLayer has no required dependencies. It uses Python's standard library for HTTP transport, so it won't interfere with your existing dependency tree.

pip install cogslayer

Requires Python 3.10 or later. Works on Linux, macOS, and Windows.

Quick Start

Two changes to your existing code: swap the provider import, and decorate functions you want to track. CogsLayer handles the rest. Every call inside a tracked function is captured with full token counts, cost, and latency.

import cogslayer
from cogslayer.openai import OpenAI          # 1. Replace the import

cogslayer.init(api_key="cl_live_xxx",        # 2. Init once at startup
             service="my-api")
client = OpenAI()

@cogslayer.track(feature="chat", team="growth")
def ask(prompt: str) -> str:
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
    )
    return resp.choices[0].message.content

answer = ask("Explain quantum computing")
# -> Event captured: model=gpt-4o, cost=$0.0032, feature=chat, team=growth

answer = ask("Explain quantum computing")
# -> Deduped! Returned cached response, saved $0.0032

Provider Wrappers

CogsLayer provides drop-in replacements for every major LLM provider. The wrapped clients are API-identical to the originals. You change the import line and nothing else breaks. Async clients are included.

from cogslayer.openai import OpenAI, AsyncOpenAI

# Standard OpenAI
client = OpenAI()

# Groq, Together, Fireworks, xAI, Ollama: anything OpenAI-compatible
groq = OpenAI(base_url="https://api.groq.com/openai/v1")
together = OpenAI(base_url="https://api.together.xyz/v1")

from cogslayer.anthropic import Anthropic, AsyncAnthropic

client = Anthropic()

from cogslayer.gemini import Client

client = Client(api_key="...")

All wrappers support both sync and async usage. The async variants (AsyncOpenAI, AsyncAnthropic) work seamlessly with asyncio and the decorator is async-safe with no special configuration needed.

import cogslayer
from openai import OpenAI  # Keep your original import

cogslayer.init(api_key="cl_live_xxx")
client = cogslayer.wrap(OpenAI())  # Same client, now tracked + optimized

cogslayer.wrap() patches the underlying provider SDK and returns the same client instance. Use it when you prefer not to change your import lines.

Automatic Optimization: Guaranteed Savings

CogsLayer automatically applies four zero-risk optimizations to every LLM call. These never change your model or affect output quality. They only prevent wasted spend. Enabled by default when you call cogslayer.init().

Response Dedup

When your application sends the same prompt twice, CogsLayer returns the cached response instead of calling the provider. The cache is keyed on the SHA-256 fingerprint of the prompt content with a configurable TTL (default: 5 minutes).

import cogslayer
from cogslayer.openai import OpenAI

cogslayer.init(api_key="cl_live_xxx")
client = OpenAI()

@cogslayer.track(feature="faq")
def answer(question: str) -> str:
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": question}],
    )
    return resp.choices[0].message.content

# First call: hits OpenAI ($0.003)
answer("What is your return policy?")

# Second call: returns cached response ($0.00)
answer("What is your return policy?")
# -> CogsLayer logs: saved $0.003 via response_dedup

Prompt Caching

Providers offer discounted pricing for cached input tokens. Anthropic gives a 90% discount, OpenAI 50%, and Gemini 75%. CogsLayer automatically injects the required cache control headers so your repeated system prompts and prefixes get the discount without any code changes.

Retry Elimination

When the same user triggers the same feature with the same prompt within a short window (default: 5 seconds), CogsLayer recognizes it as a retry and returns the original response. This catches application-level retries, double-clicks, and race conditions.

# User clicks "Generate" twice rapidly
# First click: calls GPT-4o ($0.004)
# Second click: CogsLayer returns cached response ($0.00)
# -> Saved $0.004 via retry_dedup

Output Capping

Most applications don't set max_tokens, so models generate until they're done, often producing 2 to 3x more output than needed. CogsLayer learns your feature's output patterns and automatically sets a max_tokens ceiling at 120% of the historical p95 output length, after observing at least 10 calls.

from cogslayer._optimizer import configure

# Set explicit ceilings per feature
configure(max_tokens_profiles={
    "summarization": 300,
    "classification": 50,
    "chat": 1000,
})

Optimizer Configuration

All optimizations are enabled by default. You can disable the entire optimizer or configure individual techniques.

cogslayer.init(api_key="cl_live_xxx", optimize=False)

from cogslayer._optimizer import configure

configure(
    dedup_ttl=600.0,           # Cache responses for 10 minutes
    dedup_max_size=2000,       # LRU cache holds 2000 entries
    retry_dedup_window=10.0,   # Detect retries within 10 seconds
)

# Disable specific techniques
configure(
    dedup_enabled=False,       # Disable response dedup
    cache_headers_enabled=False,  # Disable cache header injection
)

Wrap your agent run with cogslayer.session()

Group every LLM call in one user or agent flow into a single run in the dashboard. CogsLayer does not store prompt text, only fingerprints and technical metadata.

import cogslayer
from cogslayer.openai import OpenAI

cogslayer.init(api_key="cl_live_xxx")
client = OpenAI()

with cogslayer.session():
    draft = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": user_input}],
    )
    refined = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": draft.choices[0].message.content}],
    )
# Both calls appear under one run: total cost, timeline, and call breakdown.

View run costs in the dashboard

Open Runs to see each flow’s total cost, call count, tokens, and duration. Expand a run for a run summary, timeline of calls (duration and cost), and a nested call breakdown when your SDK emits span metadata. Older data without spans still shows accurate costs; structure appears for newer SDK versions.

Estimate savings with model swaps and cache assumptions

On a run, use Estimate savings to map models to cheaper alternatives and optionally assume a cache hit rate. Results are hypothetical using published list prices; your actual invoice may differ. The Savings page ranks org-wide opportunities so you know what to validate first.

API: POST /v1/sessions/:session_id/replay with model_overrides and optional cache_rate (0 to 1).

Attribution with @track()

Attribution is what separates CogsLayer from billing scrapers. Instead of seeing “you spent $400 on GPT-4o this month”, you see “the summarization feature in the growth team spent $180 on GPT-4o for user X.” Every LLM call inside a tracked function inherits the attribution context.

@cogslayer.track(
    feature="summarization",   # Which product feature
    team="growth",             # Which team owns it
    user_id="usr_123",         # Which end-user triggered it
    tenant="acme-corp",        # Multi-tenant: which customer
)
def summarize(text: str) -> str:
    # Every LLM call in here is tagged with all four labels
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": text}],
    )
    return resp.choices[0].message.content

You can pass any additional keyword arguments. They're stored as custom metadata on the event and available in exports and the dashboard.

Streaming Support

Streaming responses work out of the box. CogsLayer captures usage from the final chunk, which contains the complete token counts. Both sync generators and async generators are supported with no additional configuration.

@cogslayer.track(feature="chat")
def stream_response(prompt: str):
    stream = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        stream=True,
    )
    for chunk in stream:
        yield chunk.choices[0].delta.content or ""

# Async streaming works the same way
@cogslayer.track(feature="chat")
async def stream_async(prompt: str):
    stream = await async_client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        stream=True,
    )
    async for chunk in stream:
        yield chunk.choices[0].delta.content or ""

Custom Model Pricing

CogsLayer ships with a built-in pricing registry for all major models. For fine-tuned models, self-hosted endpoints, or new models not yet in the registry, register custom pricing so cost estimates stay accurate.

cogslayer.register_model(
    "ft:gpt-4o:my-org:custom-model:abc123",
    prompt_price_per_1k=0.005,     # $ per 1K input tokens
    completion_price_per_1k=0.015, # $ per 1K output tokens
)

# Now calls to this model are priced correctly
@cogslayer.track(feature="extraction")
def extract(text: str):
    return client.chat.completions.create(
        model="ft:gpt-4o:my-org:custom-model:abc123",
        messages=[{"role": "user", "content": text}],
    )

Configuration

All configuration is passed to cogslayer.init(). There are no config files or environment variable conventions. Everything is explicit.

cogslayer.init(
    api_key="cl_live_abc123def456",
    service="recommendation-api",
    environment="production",
    base_url="https://cogslayer.example.com",  # self-hosted
)

Budget Alerts

Set spend thresholds per feature, team, model, or total spend. Alerts are evaluated on-demand. Call check_alerts() from the SDK, hit the API endpoint, or use the “Run Check” button in the dashboard. When spend exceeds a threshold, the alert triggers and logs to the alert history.

import cogslayer

cogslayer.init(api_key="cl_live_xxx")

# Check which alerts are currently triggered
result = cogslayer.check_alerts()
for alert in result["triggered"]:
    name = alert["alert_name"]
    spend = alert["spend_usd"]
    limit = alert["threshold_usd"]
    print(f"Warning: {name}: spent {spend:.2f} / {limit:.2f} USD")

POST /v1/alerts
Content-Type: application/json
Authorization: Bearer cl_live_xxx

{
  "name": "Daily GPT-4o budget",
  "threshold_usd": 50.00,
  "period": "daily",
  "scope": "model",
  "scope_value": "gpt-4o"
}

Cost Insights

CogsLayer analyzes your usage patterns and generates actionable recommendations to reduce spend. Because it tracks code-level attribution, recommendations are scoped to specific features, not generic “use a cheaper model” advice.

insights = cogslayer.get_insights(days=30)

total = insights["total_estimated_savings_usd"]
print(f"Total estimated savings: {total:.2f} USD")

for rec in insights["insights"]:
    print(f"[{rec['severity']}] {rec['message']}")
    print(f"  -> Save ~{rec['estimated_savings_usd']:.2f} USD")
    print()

CSV Export

Export raw events as CSV for offline analysis, compliance, or piping into your own BI tools. Every column, including all attribution fields, is included. Available from the SDK, the API, or the dashboard's Events page.

from datetime import datetime

csv_data = cogslayer.export_csv(
    start=datetime(2025, 1, 1),
    end=datetime(2025, 1, 31),
    feature="chat",
)

with open("january_chat_events.csv", "w") as f:
    f.write(csv_data)

GET /v1/events/export?start=2025-01-01&end=2025-01-31&feature=chat
Authorization: Bearer cl_live_xxx

# Response: text/csv with all event columns

Time-series Data

Query bucketed cost data over time for building charts, tracking trends, or feeding into monitoring pipelines. Supports hourly, daily, and weekly intervals with optional grouping.

from datetime import datetime

data = cogslayer.get_timeseries(
    interval="day",
    group_by="feature",
    start=datetime(2025, 1, 1),
    end=datetime(2025, 1, 31),
)

for bucket in data["data"]:
    cost = bucket["cost_usd"]
    count = bucket["event_count"]
    print(f"{bucket['bucket']}: {cost:.4f} USD ({count} events)")

SDK Methods

All public methods available after import cogslayer. Platform query methods require a valid API key set via init().

API Endpoints

The CogsLayer platform exposes a REST API. All endpoints require authentication via the Authorization header: either an API key (cl_live_...) or a JWT from the dashboard. All request and response bodies are JSON unless noted otherwise.

Tracked Metrics

Every LLM call captured by CogsLayer records these fields. Usage metrics are extracted automatically from provider responses. Attribution fields are set through @cogslayer.track() and cogslayer.init().

Error Handling

CogsLayer is designed to never break your application. Tracking failures are silently swallowed. If the platform is unreachable or returns an error, your LLM calls still execute normally. Failed events are logged at debug level and dropped to avoid unbounded memory growth.

Platform query methods (get_insights, check_alerts, etc.) do raise exceptions on failure, since they're explicit data fetches. Wrap them in try/except if your application needs to handle platform downtime gracefully.