LLM Services

{
  "version": "v1.9.0",
  "providers_configured": ["openai", "anthropic", "deepseek"],
  "providers_status": {
    "openai": {"configured": true},
    "anthropic": {"configured": true},
    "ollama": {
      "configured": false,
      "missing_keys": ["OLLAMA_CHAT_ENDPOINT"]
    }
  },
  "environment": {
    "LLM_API_MODE": false,
    "LLM_LOG_LEVEL": "INFO",
    "LLM_DEBUG": false,
    "config_loaded": true
  },
  "clients_initialized": {
    "openai": true,
    "anthropic": true,
    "gemini": false
  },
  "circuit_breakers": {
    "openai": {
      "state": "CLOSED",
      "failure_count": 0,
      "failure_threshold": 5
    }
  }
}

from llmservices import LLMService, LLMRequest

req = LLMRequest(
    provider="openai",
    model="gpt-4o",
    prompt="Hello"
)

# Access the auto-generated request ID
print(f"Request ID: {req.request_id}")

response = LLMService.call(req)

# Response includes debugging information
print(f"Request ID: {response.request_id}")
print(f"Finish Reason: {response.finish_reason}")
print(f"Usage: {response.usage}")
if response.timing:
    print(f"Duration: {response.timing.total_duration_ms}ms")
if response.metadata:
    print(f"Metadata: {response.metadata.to_dict()}")

from llmservices import LLMService, LLMRequest, LLMError

try:
    req = LLMRequest(provider="openai", model="gpt-4o", prompt="Hello")
    response = LLMService.call(req)
except LLMError as e:
    # Access structured error information
    print(f"Error: {e.message}")
    print(f"Provider: {e.provider}")
    print(f"Model: {e.model}")
    print(f"Status Code: {e.status_code}")
    print(f"Request ID: {e.request_id}")
    print(f"Error Code: {e.error_code}")

    # Convert to dictionary for logging
    error_dict = e.to_dict()
    import json
    print(json.dumps(error_dict, indent=2))

# .env file for development
LLM_LOG_LEVEL=DEBUG
LLM_DEBUG=true

# .env file for production
LLM_LOG_LEVEL=WARNING
LLM_DEBUG=false

"""
Example of using llmservices in Library Mode with Streaming Enabled

This example demonstrates how to use LLMService as a Python library
(without Flask) to stream responses from an LLM provider.

Prerequisites:
1. Install required packages:
   pip install python-dotenv openai anthropic google-genai requests urllib3

2. Set up your .env file with API keys:
   The .env file can be located in either:
   - This directory (llmbase_demo/.env) - for local configuration
   - Parent llmbase directory (llmbase/.env) - shared configuration

   Example .env file:
   OPENAI_API_KEY=sk-...
   ANTHROPIC_API_KEY=sk-ant-...
   GEMINI_API_KEY=...
   LLM_CONFIG_FILE=llm_config.json

3. (Optional) Create llm_config.json with custom model configurations
   - Default config: llmbase/llm_config.json (already exists with sensible defaults)
   - Local override: Create llmbase_demo/llm_config.json for project-specific settings
   - Useful for overriding default settings like max_tokens, temperature, etc.
"""

import os
import json
from dotenv import load_dotenv
import sys
from pathlib import Path

# ============================================================================
# PATH CONFIGURATION
# ============================================================================
# Add parent llmbase directory to Python path for imports
# This allows importing llmservices from a sibling directory
# Example structure:
#   /Users/cio/gai/
#     ├── llmbase/           (contains llmservices.py)
#     └── llmbase_demo/      (contains this file)
sys.path.insert(0, str(Path(__file__).parent.parent / "llmbase"))


# ============================================================================
# LLM SERVICE IMPORTS
# ============================================================================
from llmservices import LLMService, LLMRequest, CircuitBreakerOpenException
# LLMService: Main class for making LLM calls
# LLMRequest: Data class for request parameters
# CircuitBreakerOpenException: Raised when provider is blocked due to failures

# ============================================================================
# ENVIRONMENT SETUP
# ============================================================================
# Load environment variables from .env file
# Required: OPENAI_API_KEY, ANTHROPIC_API_KEY, GEMINI_API_KEY, etc.
# Optional: LLM_CONFIG_FILE (defaults to llm_config.json)
load_dotenv()


def parse_stream_chunk(chunk: str) -> str:
    """
    Parse SSE (Server-Sent Events) format and extract the text content.

    Handles formats like:
    - data: {"chunk": "text"}
    - data: {"delta": {"content": "text"}}
    - data: [DONE] (stream end marker)
    - raw text

    Returns:
        str: The extracted text content, or empty string for [DONE] markers
    """
    chunk = chunk.strip()

    # Handle SSE format: data: {...}
    if chunk.startswith('data: '):
        data_part = chunk[6:]  # Remove "data: " prefix

        # Check for stream end marker
        if data_part.strip() == '[DONE]':
            return ''  # Signal end of stream

        try:
            data = json.loads(data_part)

            # Try different possible keys for the content
            if 'chunk' in data:
                return data['chunk']
            elif 'delta' in data and 'content' in data['delta']:
                return data['delta']['content']
            elif 'content' in data:
                return data['content']
            elif 'text' in data:
                return data['text']
        except json.JSONDecodeError:
            pass

    # Return as-is if not SSE format or parsing failed
    return chunk


def stream_basic_example():
    """Basic streaming example with default settings."""
    llm_request = LLMRequest(
        provider='openai',
        model='gpt-4o',
        prompt='Write a short haiku about artificial intelligence.',
        stream=True  # Enable streaming
    )

    try:
        for raw_chunk in LLMService.stream(llm_request):
            text = parse_stream_chunk(raw_chunk)
            print(text, end='', flush=True)
        print()
    except CircuitBreakerOpenException as e:
        print(f"Service Unavailable (Circuit Breaker): {e}")
    except Exception as e:
        print(f"Error: {e}")


# Run the example
if __name__ == "__main__":
    stream_basic_example()