Metadata-Version: 2.4
Name: ignis-observability
Version: 0.3.0
Summary: Non-Blocking AI Observability Hub for multi-model LLM applications
Author: Infogain-GenAI
License: MIT
Project-URL: Homepage, https://github.com/Infogain-GenAI/ignis_observability
Project-URL: Repository, https://github.com/Infogain-GenAI/ignis_observability.git
Project-URL: Documentation, https://github.com/Infogain-GenAI/ignis_observability#readme
Project-URL: Bug Tracker, https://github.com/Infogain-GenAI/ignis_observability/issues
Keywords: observability,llm,tracing,langsmith,openai,anthropic,gemini,monitoring,async
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries
Classifier: Topic :: System :: Monitoring
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: openai>=1.0.0
Requires-Dist: langsmith>=0.1.0
Requires-Dist: sqlalchemy>=2.0.0
Requires-Dist: aiosqlite>=0.19.0
Requires-Dist: tiktoken>=0.5.0
Requires-Dist: python-dotenv>=1.0.0
Requires-Dist: pydantic-settings>=2.0.0
Requires-Dist: pydantic>=2.0.0
Requires-Dist: wheel
Requires-Dist: langwatch<0.3.0,>=0.2.11
Requires-Dist: arize-phoenix-otel<1.0.0,>=0.6.1
Provides-Extra: dev
Requires-Dist: pytest>=7.0.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.21.0; extra == "dev"
Requires-Dist: pytest-cov>=4.0.0; extra == "dev"
Requires-Dist: black>=23.0.0; extra == "dev"
Requires-Dist: ruff>=0.1.0; extra == "dev"
Requires-Dist: mypy>=1.0.0; extra == "dev"
Provides-Extra: anthropic
Requires-Dist: anthropic>=0.7.0; extra == "anthropic"
Provides-Extra: gemini
Requires-Dist: google-generativeai>=0.3.0; extra == "gemini"
Dynamic: license-file

# 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](#architecture)
- [Quick Start](#quick-start)
- [Configuration](#configuration)
- [Usage Examples](#usage-examples)
- [Running Tests](#running-tests)
- [Database Schema](#database-schema)
- [Cost Tracking](#cost-tracking)
- [API Reference](#api-reference)
- [Troubleshooting](#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 |
| **External Tools** | LangSmith SDK | Cloud-based tracing platform |
| **Tokenization** | tiktoken | Accurate token counting |
| **Configuration** | pydantic-settings | Type-safe environment config |

## Quick Start

### 1. Installation

```powershell
# 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

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

# Edit .env with your favorite editor
notepad .env
```

Update the following values in `.env`:

```ini
# 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

```powershell
# 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
```powershell
# Query your local data
python
```

```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

| Variable | Description | Example | Where to Get |
|----------|-------------|---------|--------------|
| `OPENAI_API_KEY` | OpenAI API key for LLM calls | `sk-proj-abc123...` | [OpenAI Platform](https://platform.openai.com/api-keys) |
| `LANGCHAIN_API_KEY` | LangSmith API key for tracing | `lsv2_pt_xyz789...` | [LangSmith Settings](https://smith.langchain.com/settings) |
| `LANGCHAIN_PROJECT` | LangSmith project name | `my-production-app` | [LangSmith Projects](https://smith.langchain.com/) |

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

### Install as a Python package

#### Syntax for tagged release version deployment with pip

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

#### Syntax for main branch install as module

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

### Example usage in Python

```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.


```python
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

```python
@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/](./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

```powershell
# 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

| 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.

```python
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.

```python
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.

```python
@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:
```powershell
pip install -e .
```

### Package Name with Hyphen

**Problem**: Python doesn't allow hyphens in package names

**Solution**: Imports use underscores while directory uses hyphens:
```python
# 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](https://smith.langchain.com) - Cloud-based LLM tracing
- [OpenAI](https://openai.com) - LLM provider
- [SQLAlchemy](https://www.sqlalchemy.org/) - Database ORM
- [tiktoken](https://github.com/openai/tiktoken) - Token counting
- [PyPI Publish Guide](./docs/pypi_publish_guide.md) - Step-by-step release and publishing checklist

---

## Release Notes

- **[Sync pipeline, unified analytics API, security remediation](docs/dev-to-main-merge.md)** — dev → main merge adding multi-provider sync ingestion, unified analytics endpoints, and DB credential remediation.

---

**Questions?** Open an issue or check the [examples](./examples/) directory for more details.

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

