genosis 1.0.1

Official Python SDK for the Genosis LLM cost optimization platform

genosis

Genosis reduces LLM inference costs by up to 75% through server-optimized prompt caching. The SDK wraps your existing API calls with one method — g.call() — and applies optimization transparently.

import anthropic
from genosis import Genosis

client = anthropic.Anthropic()
g = Genosis(api_key="gns_live_...")

result = g.call(
    {
        "model": "claude-sonnet-4-6",
        "system": [
            {"type": "text", "text": system_context},
            {"type": "text", "text": product_catalog},
        ],
        "messages": [{"role": "user", "content": question}],
        "max_tokens": 1024,
    },
    lambda params: client.messages.create(**params),
)

print(result.response)   # the Anthropic response object, unmodified
print(result.memoized)   # True if served from local cache

No schema changes. No new concepts. Your existing LLM code stays intact.

Installation

The package will be published to PyPI. Until then, install directly from GitHub:

pip install git+https://github.com/Genosis-Limited/genosis-python.git

Requires Python 3.10+. The only runtime dependency is httpx.

Once published to PyPI:

pip install genosis

Provider Examples

Anthropic

import anthropic
from genosis import Genosis

client = anthropic.Anthropic()
g = Genosis(api_key="gns_live_...")

result = g.call(
    {
        "model": "claude-sonnet-4-6",
        "system": [
            {"type": "text", "text": system_context},
            {"type": "text", "text": product_catalog},
        ],
        "messages": [{"role": "user", "content": question}],
        "max_tokens": 512,
    },
    lambda params: client.messages.create(**params),
)

Genosis adds cache_control breakpoints to your system blocks automatically. You do not need to add them yourself — any existing breakpoints you placed are replaced with the server-optimized set.

OpenAI

import openai
from genosis import Genosis

client = openai.OpenAI()
g = Genosis(api_key="gns_live_...")

result = g.call(
    {
        "model": "gpt-4o",
        "messages": [
            {
                "role": "system",
                "content": [
                    {"type": "text", "text": system_context},
                    {"type": "text", "text": product_catalog},
                ],
            },
            {"role": "user", "content": question},
        ],
        "max_tokens": 512,
    },
    lambda params: client.chat.completions.create(**params),
)

For OpenAI, Genosis reorders system content blocks to maximize prefix cache hits. No cache_control markers — OpenAI's prompt caching is automatic.

AWS Bedrock

import boto3
import json
from genosis import Genosis

bedrock = boto3.client("bedrock-runtime", region_name="us-east-1")
g = Genosis(api_key="gns_live_...")

# Bedrock ARNs are normalized automatically — the manifest lookup uses
# the canonical model name (e.g., claude-sonnet-4-6-20250514)
result = g.call(
    {
        "model": "anthropic.claude-sonnet-4-6-20250514-v1:0",
        "system": system_prompt,
        "messages": [{"role": "user", "content": question}],
        "max_tokens": 512,
        "anthropic_version": "bedrock-2023-05-31",
    },
    lambda params: json.loads(
        bedrock.invoke_model(
            modelId=params["model"],
            body=json.dumps(params),
        )["body"].read()
    ),
)

Cross-region inference ARNs (us.anthropic.claude-*) are also handled.

Azure OpenAI

import openai
from genosis import Genosis

azure = openai.AzureOpenAI(
    azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
    api_key=os.environ["AZURE_OPENAI_API_KEY"],
    api_version="2024-02-01",
)
g = Genosis(api_key="gns_live_...")

result = g.call(
    {
        "model": "gpt-4o",
        "messages": [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": question},
        ],
        "max_tokens": 512,
    },
    lambda params: azure.chat.completions.create(**params),
)

Supported Providers and Models

Provider	Models	Also works via
Anthropic	claude-opus-4, claude-sonnet-4-6, claude-haiku-4-5	AWS Bedrock
OpenAI	gpt-4.1, gpt-4.1-mini, gpt-4o, gpt-4o-mini, o1, o3, o4-mini	Azure OpenAI
Google (coming soon)	gemini-2.5-pro, gemini-2.5-flash, gemini-2.5-flash-lite	Vertex AI

Provider detection is automatic from the model name. Bedrock ARNs (anthropic.claude-*, us.anthropic.claude-*) are recognized and normalized to canonical model IDs for manifest lookup.

How It Works

On each g.call():

1. The SDK detects your provider from the model field.
2. It checks whether a server-optimized manifest exists for that provider/model. On the first call, the manifest is fetched in the background — your call goes through normally.
3. When a manifest is available, the SDK applies it: for Anthropic, it inserts cache_control breakpoints on high-value system blocks; for OpenAI, it reorders system content blocks to maximize prefix cache hits.
4. Your function is called with the (possibly modified) params.
5. Usage data is hashed and queued for telemetry. The background worker flushes it to api.usegenosis.ai — no synchronous network call on the hot path.
6. Manifests refresh every 5 minutes in the background. A stale manifest is better than no optimization.

If anything in the Genosis layer throws, your function is called with the original unmodified params and the error is reported silently. g.call() cannot break your LLM calls.

Configuration

from genosis import Genosis

g = Genosis(
    # Required
    api_key="gns_live_...",            # or "gns_test_..." for test keys

    # Optional — shown with defaults
    base_url="https://api.usegenosis.ai",
    max_retries=2,                     # retries on 429/5xx (exponential backoff)
    timeout=60.0,                      # per-request timeout in seconds
    manifest_refresh_interval=300,     # seconds between manifest refreshes; 0 = disabled
    memoization_enabled=True,          # see Memoization below
    memoization_max_entries=1000,      # max entries in the in-process LRU cache
    memo_storage=None,                 # plug in Redis, etc. (see Memoization below)
    buffer_path=None,                  # SQLite buffer path; default: ~/.genosis/buffer_<prefix>.db
    buffer_max_size=10_000,            # max buffered events before oldest are dropped
)

Memoization

Memoization serves identical requests from a local cache without calling the LLM at all. The server identifies which request patterns are worth memoizing based on your telemetry — the SDK just applies the decision.

When a memoized response is served, result.memoized is True and no LLM call is made.

The default storage is an in-process LRU map. For multi-process deployments (multiple workers, serverless), plug in a shared store:

from genosis import Genosis, MemoStorage
from typing import Any, Optional

class RedisMemoStorage(MemoStorage):
    def __init__(self, redis_client):
        self._redis = redis_client

    def get(self, fingerprint: str) -> Optional[Any]:
        import json
        val = self._redis.get(f"genosis:memo:{fingerprint}")
        return json.loads(val) if val else None

    def set(self, fingerprint: str, response: Any, ttl_seconds: int) -> None:
        import json
        self._redis.setex(
            f"genosis:memo:{fingerprint}",
            ttl_seconds,
            json.dumps(response),
        )

g = Genosis(api_key="gns_live_...", memo_storage=RedisMemoStorage(redis_client))

MemoStorage is a two-method abstract base class — any implementation works.

To disable memoization entirely:

g = Genosis(api_key="gns_live_...", memoization_enabled=False)

Serverless and Batch Jobs

The background worker flushes telemetry continuously in long-running processes. In serverless functions or batch jobs that exit after each invocation, call flush() before the process ends:

# At the end of your handler / job
remaining = g.flush(timeout=30.0)  # wait up to 30s for buffer to drain

flush() returns the number of events still in the buffer when the timeout is reached. A return value of 0 means the buffer was fully drained.

Example AWS Lambda handler:

import atexit
from genosis import Genosis

g = Genosis(api_key="gns_live_...")

def handler(event, context):
    result = g.call(params, lambda p: client.messages.create(**p))
    # ... process result ...
    g.flush(timeout=10.0)
    return response

Background Worker

The worker starts automatically when you construct Genosis. It handles:

• Telemetry batching and upload
• Manifest acknowledgement
• Error reporting

Telemetry is written to a local SQLite file first (~/.genosis/buffer_<keyprefix>.db). If the network is unavailable, events are held in the buffer and retried on the next worker cycle. Nothing is lost on transient failures.

Each API key prefix gets its own buffer file, so multiple apps on the same machine do not share state.

The worker thread is a daemon — it does not prevent the Python process from exiting. On clean process exit, atexit and SIGTERM handlers flush remaining events automatically. For hard kills or serverless functions, call g.flush() explicitly.

Content-Blind Security Model

Genosis never sees your prompts, responses, user data, or API keys.

What leaves the SDK:

• SHA-256 hashes of content blocks (one-way, irreversible)
• Token counts
• Usage numbers from the LLM response (input_tokens, output_tokens, cache_read_input_tokens, etc.)
• Provider and model name

What stays local:

• All prompt text
• All LLM responses
• The memoization cache

The hashing is done in the SDK before any network call. You can verify this in genosis/client.py — search for sha256. Error messages are also sanitized before logging: API keys and long base64 strings are redacted automatically.

Error Handling

Errors from the management API (g.account, g.manifest, etc.) raise typed exceptions:

from genosis import (
    GenosisError,
    AuthenticationError,
    RateLimitError,
    NotFoundError,
    ConnectionError,
    TimeoutError,
)

try:
    g.optimization.trigger("anthropic", "claude-sonnet-4-6")
except AuthenticationError:
    pass  # Invalid or revoked API key (HTTP 401)
except RateLimitError:
    pass  # Too many requests (HTTP 429) — back off
except ConnectionError:
    pass  # Network failure — no HTTP response received
except TimeoutError:
    pass  # Request exceeded the configured timeout
except GenosisError as e:
    print(e.status, e.code, str(e))

All typed errors extend GenosisError and expose status (HTTP status code) and code (machine-readable string).

g.call() does not raise Genosis errors. If the optimization layer fails for any reason, fn is called with the original unmodified params. LLM errors (rate limits, network failures, etc.) propagate normally — Genosis does not swallow them.

Error classes

Class	Status	Default code
BadRequestError	400	BAD_REQUEST
AuthenticationError	401	UNAUTHORIZED
PermissionDeniedError	403	FORBIDDEN
NotFoundError	404	NOT_FOUND
ConflictError	409	CONFLICT
UnprocessableEntityError	422	UNPROCESSABLE_ENTITY
RateLimitError	429	RATE_LIMITED
InternalServerError	500+	INTERNAL_SERVER_ERROR
ConnectionError	—	CONNECTION_ERROR
TimeoutError	—	TIMEOUT

Management API

Use these for dashboards, scripts, and setup tooling — not in the hot path.

# Account
account = g.account.get()
usage = g.account.get_usage()
keys = g.account.list_api_keys()
new_key = g.account.create_api_key("worker-prod", ["ingest", "manifest:read"])
g.account.revoke_api_key(key_id)

# Manifests
manifest_resp = g.manifest.get("anthropic", "claude-sonnet-4-6")  # manifest_resp["data"]
all_manifests = g.manifest.list_all()
history = g.manifest.get_history("anthropic", "claude-sonnet-4-6")

# Optimization (runs server-side)
run = g.optimization.trigger("anthropic", "claude-sonnet-4-6")
status = g.optimization.get_status("anthropic", "claude-sonnet-4-6")
results = g.optimization.get_results("anthropic", "claude-sonnet-4-6")

# Telemetry
summary = g.telemetry.get_summary(days=7)
costs = g.telemetry.get_cost_breakdown(days=30, provider="anthropic")
blocks = g.telemetry.get_block_frequencies(days=7, provider="anthropic", model="claude-sonnet-4-6")

Types

All public types are TypedDict subclasses from genosis.types. Import them for type checking:

from genosis.types import (
    CallResult,
    Account,
    AccountUsage,
    ApiKey,
    CreatedApiKey,
    CacheManifest,
    CacheTrainEntry,
    ManifestResponse,
    ManifestHistoryEntry,
    ManifestsListEntry,
    OptimizationResult,
    OptimizationStatusResponse,
    OptimizationResultsResponse,
    TelemetrySummary,
    CostBreakdownEntry,
    BlockFrequency,
    MemoizationCandidate,
)

Key types:

CallResult — returned by g.call()

class CallResult(Generic[R]):
    response: R      # the LLM response object, unmodified
    memoized: bool   # True if served from local memo cache

CacheManifest — the server-generated optimization manifest

class CacheManifest(TypedDict, total=False):
    manifest_version: str
    provider: str
    model: str
    cache_train: list[CacheTrainEntry]   # blocks to cache, by hash
    memoization: MemoizationSection
    provider_hints: ProviderHints

CacheTrainEntry — one block in the cache plan

class CacheTrainEntry(TypedDict):
    hash: str        # SHA-256 of the content block
    tokens: int
    priority: float
    position: int

MemoizationCandidate — a request pattern flagged for memoization

class MemoizationCandidate(TypedDict):
    fingerprint: str
    ttl_seconds: int
    block_hashes: list[str]
    estimated_savings_per_hit: float

MemoStorage — abstract base class for custom memo backends

class MemoStorage(ABC):
    def get(self, fingerprint: str) -> Optional[Any]: ...
    def set(self, fingerprint: str, response: Any, ttl_seconds: int) -> None: ...

Logging

The SDK logs to the genosis logger at DEBUG level. Errors in the optimization layer are also logged there. To see SDK diagnostics:

import logging
logging.getLogger("genosis").setLevel(logging.DEBUG)

In production, the default level is WARNING, so no output is produced unless something goes wrong.

License

Apache 2.0 — see LICENSE and NOTICE.

Patent pending. All patent inquiries: [e-mail]