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 to both cloud platforms (LangSmith) and a local database.

Key Features

• Zero-Latency Observability: Fire-and-forget pattern ensures LLM responses reach users instantly
• Dual Persistence: Send traces to LangSmith while maintaining a local SQLite database
• 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
• Quick Start
• Configuration
• Usage Examples
• Running Tests
• 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

Quick Start

1. Installation

# Clone or navigate to the repository
cd c:\Aiden\Obsrv\ignis-observability

# Create and activate virtual environment (if not already done)
python -m venv .venv
.venv\Scripts\Activate.ps1

# Install the package with dependencies
pip install -e .

# For development (includes testing tools)
pip install -e ".[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

# Copy the example environment file
Copy-Item .env.example .env

# Edit .env with your favorite editor
notepad .env

Update the following values in .env:

# 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 Example

# Run the basic example
python examples/01_basic_example.py

You should see output like:

Starting Basic Observability Example

Initializing database...
Database ready

Initializing ObsHub...
ObsHub ready

Starting background worker...
Worker running

Calling LLM...
Response:
    Why do database administrators always carry a ladder?
    Because they're always climbing up the SQL!

Waiting for observability data to be processed...
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 src.ignis_observability.database 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

Optional Variables

Usage Examples

Basic Usage

Install as a Python package

Syntax for tagged release version deployment with pip

pip install git+https://github.com/Infogain-GenAI/ignis_observability.git@v1.0.0#egg=ignis_observability

Syntax for main branch install as module

pip install git+https://github.com/Infogain-GenAI/ignis_observability.git@main#egg=ignis_observability

Example usage in Python

 

import asyncio

from dotenv import load_dotenv
from ignis_observability.config import settings
from ignis_observability.database import DBManager
from ignis_observability.engine import ObsHub
from openai import AsyncOpenAI

# Load environment variables
load_dotenv()


async def main():
    """Run a simple example with non-streaming LLM call."""

    print("Starting Basic Observability Example\n")

    # Step 1: Initialize Database
    print("Initializing database...")
    db = DBManager(settings.DATABASE_URL)
    await db.init_db()
    await db.seed_prices()
    print("Database ready\n")

    # Step 2: Initialize Observability Hub
    print("Initializing ObsHub...")
    hub = ObsHub(
        langsmith_api_key=settings.LANGCHAIN_API_KEY,
        db_manager=db,
        project_name=settings.LANGCHAIN_PROJECT,
    )
    print("ObsHub ready\n")

    # Step 3: Start background worker
    print("Starting background worker...")
    worker_task = asyncio.create_task(hub.worker())
    print("Worker running\n")

    # Step 4: Create OpenAI client
    client = AsyncOpenAI(api_key=settings.OPENAI_API_KEY)

    # Step 5: Define and decorate your LLM function
    @hub.observe(name="get_joke")
    async def get_joke(prompt: str):
        """Get a joke from GPT."""
        return await client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": prompt}],
        )

    # Step 6: Call your function - observability happens automatically!
    print("Calling LLM...")
    response = await get_joke("Tell me a short joke about ai and llm.")

    print("Response:")
    print(f"   {response.choices[0].message.content}\n")

    # Step 7: Wait for background processing to complete
    print("Waiting for observability data to be processed...")
    await asyncio.sleep(2)

    print("Done! Check your LangSmith dashboard and local database.\n")
    print(f"   LangSmith Project: {settings.LANGCHAIN_PROJECT}")
    print(f"   Database: {settings.DATABASE_URL}")

    # Cleanup
    worker_task.cancel()
    try:
        await worker_task
    except asyncio.CancelledError:
        pass


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

You can place the above code in a standalone example file and run it after installing the package.

import asyncio
from dotenv import load_dotenv
from openai import AsyncOpenAI
from ignis_observability.database import DBManager
from ignis_observability.engine import ObsHub
from ignis_observability.config import settings

load_dotenv()

async def main():
    # Initialize
    db = DBManager(settings.DATABASE_URL)
    await db.init_db()
    await db.seed_prices()
    
    hub = ObsHub(
        langsmith_api_key=settings.LANGCHAIN_API_KEY,
        db_manager=db
    )
    
    worker_task = asyncio.create_task(hub.worker())
    client = AsyncOpenAI(api_key=settings.OPENAI_API_KEY)
    
    # Decorate your function
    @hub.observe(name="my_llm_call")
    async def ask_llm(question: str):
        return await client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": question}],
        )
    
    # Use it normally - observability happens automatically!
    response = await ask_llm("What is observability?")
    print(response.choices[0].message.content)
    
    # Cleanup
    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)

More Examples

See the examples/ directory for complete working examples:

• 01_basic_example.py - Simple non-streaming call
• 02_streaming_example.py - Streaming with TTFT metrics
• 03_multi_call_example.py - Multiple calls with cost tracking
• 04_complete_example.py - Production-ready setup

Running Tests

# Run all tests
pytest

# Run with coverage
pytest --cov=src --cov-report=html

# Run specific test file
pytest tests/test_database.py

# Run with verbose output
pytest -v

# View coverage report
start htmlcov/index.html

🗄️ Database Schema

llm_traces Table

model_prices Table

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)

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 'src'

Solution: Install package in editable mode:

pip install -e .

Package Name with Hyphen

Problem: Python doesn't allow hyphens in package names

Solution: Imports use underscores while directory uses hyphens:

# Directory: src/ignis-observability/
# Import: from src.ignis_observability import ...

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

Contributions welcome! Areas for improvement:

• Add more observability providers (LangWatch, Weights & Biases, etc.)
• Implement the sync service (pull data from external tools)
• Add dashboard visualization
• Support for other LLM providers (Anthropic, etc.)
• Distributed tracing support
• Performance optimizations

📄 License

[Your License Here]

Refrences

Built with:

• LangSmith - Cloud-based LLM tracing
• OpenAI - LLM provider
• SQLAlchemy - Database ORM
• tiktoken - Token counting
• PyPI Publish Guide - Step-by-step release and publishing checklist

Release Notes

• Sync pipeline, unified analytics API, security remediation — dev → main merge adding multi-provider sync ingestion, unified analytics endpoints, and DB credential remediation.

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!