ignis-observability 0.2.0

Non-Blocking AI Observability Hub for multi-model LLM applications

Ignis Observability: AI Observability Hub

A non-blocking AI observability hub that provides comprehensive monitoring, tracing, and cost tracking for LLM applications. Built on a producer-consumer pattern, it ensures zero impact on your application's response time while capturing detailed metrics across multiple observability platforms.

Installation

pip install ignis_observability

With optional provider integrations:

# Anthropic Claude support
pip install ignis_observability[anthropic]

# Google Gemini support
pip install ignis_observability[gemini]

# LangWatch integration
pip install ignis_observability[langwatch]

# Arize Phoenix / OpenTelemetry integration
pip install ignis_observability[arize]

# All optional integrations
pip install ignis_observability[all]

Key Features

• Zero-Latency Observability: Fire-and-forget pattern ensures LLM responses reach users instantly
• Multi-Platform Tracing: Send traces to LangSmith, LangWatch, Arize Phoenix, LangFuse, Laminar, or OpenTelemetry collectors
• Multi-Model Support: OpenAI GPT, Anthropic Claude, and Google Gemini out of the box
• Local Persistence: SQLite database for offline trace storage and cost analysis
• Automatic Cost Tracking: Built-in pricing tables calculate costs per request
• Streaming Support: Full support for streaming responses with TTFT (Time-to-First-Token) metrics
• Universal Decorator: Single @observe decorator works with both streaming and non-streaming functions
• Modern Tech Stack: SQLAlchemy 2.0, asyncio, type-safe with Mapped columns

📋 Table of Contents

• Architecture
• Supported Platforms
• Quick Start
• Configuration
• Usage Examples
• Database Schema
• Cost Tracking
• API Reference
• Troubleshooting

🏛️ Architecture

Producer-Consumer Pattern

┌─────────────┐         ┌──────────────┐         ┌─────────────────┐
│             │         │              │         │                 │
│  Your App   │───────▶│ asyncio.Queue │───────▶│ Background      │
│  (Producer) │  fast!  │              │         │ Worker          │
│             │         │              │         │ (Consumer)      │
└─────────────┘         └──────────────┘         └────────┬────────┘
                                                          │
                              ┌───────────────────────────┴─────────┐
                              ▼                                     ▼
                    ┌──────────────────┐                 ┌──────────────┐
                    │   LangSmith      │                 │   SQLite     │
                    │   (Cloud)        │                 │   (Local)    │
                    └──────────────────┘                 └──────────────┘

Tech Stack

Layer	Technology	Purpose
Language	Python 3.10+	Modern async/await support
Concurrency	asyncio	Non-blocking I/O operations
Database	SQLite + aiosqlite	Lightweight local storage
ORM	SQLAlchemy 2.0	Type-safe database models
Tracing Platforms	LangSmith, LangWatch, Arize, OTel	Cloud observability targets
Tokenization	tiktoken	Accurate token counting
Configuration	pydantic-settings	Type-safe environment config

Supported Platforms

Platform	Install Extra	Notes
LangSmith	(core)	Built-in, no extra needed
LangWatch	[langwatch]	pip install ignis_observability[langwatch]
Arize Phoenix	[arize]	Includes OpenTelemetry
OpenTelemetry	[arize]	OTLP-compatible collectors
LangFuse	[all]	Included in full install
Laminar	[all]	Included in full install

Quick Start

1. Installation

pip install ignis_observability

# For development (includes testing tools)
pip install ignis_observability[dev]

2. Get Your API Keys

OpenAI API Key

1. Go to https://platform.openai.com/api-keys
2. Click "Create new secret key"
3. Copy the key (starts with sk-proj-...)

LangSmith API Key

1. Go to https://smith.langchain.com (sign up if needed)
2. Navigate to Settings → API Keys
3. Click "Create API Key"
4. Copy the key (starts with lsv2_pt_...)

Create a LangSmith Project

1. In LangSmith, go to Projects
2. Click "New Project"
3. Name it (e.g., my-llm-app or production)
4. Copy the project name

3. Configure Environment

Create a .env file in your project root:

touch .env   # Linux/Mac
New-Item .env # Windows PowerShell

Add the following values:

# Required: Your OpenAI API key
OPENAI_API_KEY=sk-proj-YOUR_ACTUAL_KEY_HERE

# Required: Your LangSmith API key
LANGCHAIN_API_KEY=lsv2_pt_YOUR_ACTUAL_KEY_HERE

# Required: Your LangSmith project name
LANGCHAIN_PROJECT=my-llm-app

# Optional: Other settings (defaults usually fine)
LANGCHAIN_ENDPOINT=https://api.smith.langchain.com
LANGCHAIN_TRACING_V2=true
DATABASE_URL=sqlite+aiosqlite:///./obs_hub.db
LOG_LEVEL=info

4. Run Your First Trace

Create a main.py with the quick-start code from the Usage Examples section below and run it:

python main.py

You should see:

Database ready
Worker running
Response: Why do programmers prefer dark mode? Because light attracts bugs!
Done! Check your LangSmith dashboard and local database.

5. View Your Data

LangSmith Dashboard

1. Go to https://smith.langchain.com
2. Click on your project (e.g., my-llm-app)
3. See your traced LLM calls with full details!

Local Database

# Query your local data
python

import asyncio
from sqlalchemy import select
from ignis_observability import DBManager, UnifiedTrace

async def view_traces():
    db = DBManager()
    await db.init_db()
    async with db.session_factory() as session:
        result = await session.execute(
            select(UnifiedTrace).order_by(UnifiedTrace.timestamp.desc()).limit(5)
        )
        for trace in result.scalars():
            print(f"{trace.name}: {trace.latency_ms:.2f}ms, {trace.prompt_tokens + trace.completion_tokens} tokens, ${trace.total_cost:.6f}")

asyncio.run(view_traces())

Configuration

All configuration is managed through environment variables (loaded from .env file).

Required Variables

Variable	Description	Example	Where to Get
OPENAI_API_KEY	OpenAI API key for LLM calls	sk-proj-abc123...	OpenAI Platform
LANGCHAIN_API_KEY	LangSmith API key for tracing	lsv2_pt_xyz789...	LangSmith Settings
LANGCHAIN_PROJECT	LangSmith project name	my-production-app	LangSmith Projects

Optional Variables

Variable	Default	Description
LANGCHAIN_ENDPOINT	https://api.smith.langchain.com	LangSmith API endpoint
LANGCHAIN_TRACING_V2	true	Enable/disable LangSmith tracing
DATABASE_URL	sqlite+aiosqlite:///./obs_hub.db	Local database connection string
LOG_LEVEL	info	Logging level (debug/info/warning/error)

Usage Examples

Basic Usage (OpenAI + LangSmith)

import asyncio
from dotenv import load_dotenv
from openai import AsyncOpenAI
from ignis_observability import ObsHub, DBManager, settings

load_dotenv()

async def main():
    # Initialize database
    db = DBManager(settings.DATABASE_URL)
    await db.init_db()
    await db.seed_prices()

    # Initialize hub
    hub = ObsHub(
        langsmith_api_key=settings.LANGCHAIN_API_KEY,
        db_manager=db,
        project_name=settings.LANGCHAIN_PROJECT,
    )

    worker_task = asyncio.create_task(hub.worker())
    client = AsyncOpenAI(api_key=settings.OPENAI_API_KEY)

    # Decorate your LLM function — observability happens automatically
    @hub.observe(name="my_llm_call")
    async def ask_llm(question: str):
        return await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": question}],
        )

    response = await ask_llm("What is observability?")
    print(response.choices[0].message.content)

    # Allow background worker to flush traces
    await asyncio.sleep(2)
    worker_task.cancel()

asyncio.run(main())

Streaming Example

@hub.observe(name="streaming_response")
async def stream_response():
    return await client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": "Write a poem"}],
        stream=True,
    )

stream = await stream_response()
async for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

Anthropic Claude Example

pip install ignis_observability[anthropic]

import asyncio
from ignis_observability import ObsHub, DBManager, settings
from ignis_observability.providers.anthropic import AnthropicProvider

# Initialize with Anthropic provider
hub = ObsHub(langsmith_api_key=settings.LANGCHAIN_API_KEY, db_manager=db)

@hub.observe(name="claude_call")
async def ask_claude(prompt: str):
    provider = AnthropicProvider(api_key=settings.ANTHROPIC_API_KEY)
    return await provider.complete(prompt)

Google Gemini Example

pip install ignis_observability[gemini]

from ignis_observability import ObsHub, DBManager, settings
from ignis_observability.providers.gemini import GeminiProvider

@hub.observe(name="gemini_call")
async def ask_gemini(prompt: str):
    provider = GeminiProvider(api_key=settings.GEMINI_API_KEY)
    return await provider.complete(prompt)

🗄️ Database Schema

llm_traces Table

Column	Type	Description
id	String (PK)	Unique trace identifier
timestamp	DateTime	When the trace was created
name	String	Function/operation name
model	String	LLM model used (e.g., gpt-3.5-turbo)
latency_ms	Float	Total request latency in milliseconds
prompt_tokens	Integer	Number of input tokens
completion_tokens	Integer	Number of output tokens
total_cost	Float	Calculated cost in USD
metadata_json	JSON	Additional metadata (TTFT, streaming, etc.)

model_prices Table

Column	Type	Description
model	String (PK)	Model name
input_1k_price	Float	Cost per 1K input tokens (USD)
output_1k_price	Float	Cost per 1K output tokens (USD)

Cost Tracking

The system automatically calculates costs based on:

1. Token Counts: Captured from API responses or calculated via tiktoken
2. Pricing Table: Local model_prices table with current rates
3. Formula: cost = (prompt_tokens/1000 * input_price) + (completion_tokens/1000 * output_price)

Current Pricing (2026 estimates)

Model	Input (per 1K tokens)	Output (per 1K tokens)
gpt-3.5-turbo	$0.0005	$0.0015
gpt-4o	$0.0025	$0.010
gpt-4o-mini	$0.00015	$0.0006

Note: Update prices in database.py seed_prices() method as OpenAI pricing changes.

📚 API Reference

ObsHub

Main orchestration hub for observability.

hub = ObsHub(
    langsmith_api_key: str,  # Your LangSmith API key
    db_manager: DBManager    # Database manager instance
)

Methods:

• observe(name: str) - Decorator to instrument functions
• capture(data: dict) - Manually capture trace data
• worker() - Background worker (run as async task)

DBManager

Database operations manager.

db = DBManager(db_url: str = "sqlite+aiosqlite:///./obs_hub.db")

Methods:

• init_db() - Initialize database schema
• seed_prices() - Seed pricing table
• save_trace(data: dict) - Save trace (simple)
• calculate_and_save(data: dict) - Calculate cost and save

@observe Decorator

Automatically instruments async functions.

@hub.observe(name="function_name")
async def my_function():
    # Your code here
    pass

Captured Metrics:

• Latency (wall-clock time)
• TTFT (for streaming)
• Token counts (prompt + completion)
• Cost (calculated automatically)
• Model name
• Custom metadata

🐛 Troubleshooting

Import Errors

Problem: ModuleNotFoundError: No module named 'ignis_observability'

Solution: Make sure the package is installed:

pip install ignis_observability

Authentication Errors

Problem: AuthenticationError from OpenAI or LangSmith

Solution:

1. Check .env file has correct API keys
2. Verify keys are active (not revoked)
3. Ensure OpenAI account has credits
4. Test keys independently

No Traces in LangSmith

Problem: Traces not appearing in LangSmith dashboard

Solutions:

1. Verify LANGCHAIN_TRACING_V2=true in .env
2. Check LANGCHAIN_PROJECT name matches existing project
3. Wait a few seconds (traces are batched)
4. Check console for error messages
5. Verify API key has write permissions

Database Errors

Problem: SQLAlchemy or aiosqlite errors

Solutions:

1. Delete existing obs_hub.db file
2. Run init_db() again
3. Check file permissions
4. Verify aiosqlite is installed

Streaming Not Working

Problem: Streaming metrics not captured

Solution:

• Ensure you're using async for to consume the stream completely
• The decorator wraps the stream, so all chunks must be consumed
• Check that stream=True is passed to OpenAI API

Contributing

This library is published as a closed-source PyPI package. For bug reports or feature requests, please use the Bug Tracker.

📄 License

MIT License — see LICENSE for details.

References

Built with:

• OpenAI - LLM provider
• Anthropic - Claude LLM provider
• Google Gemini - Gemini LLM provider
• LangSmith - Cloud-based LLM tracing
• LangWatch - LLM observability platform
• Arize Phoenix - ML observability & OpenTelemetry
• SQLAlchemy - Database ORM
• tiktoken - Token counting
• pydantic - Data validation and settings

---

Questions? Open an issue or check the examples directory for more details.

Ready to monitor your LLM applications? Start with examples/01_basic_example.py!