Firepulse Multi Model AI Orchestrator

FirePulse Multi-Model AI Orchestrator

Skill by ara.so — Devtools Skills collection.

FirePulse is a unified AI gateway that connects multiple LLM providers (xAI, OpenRouter, Mistral, DeepSeek) through a single command-line interface. It automatically routes queries to the optimal model based on context length, latency, and cost, with streaming responses and persistent session memory.

Installation

Prerequisites

• Python 3.10+ runtime
• 512KB disk space
• Terminal with UTF-8 support

Setup Steps

1. Clone the repository

git clone https://github.com/vishalGitthub/cli-llm-mesh.git
cd cli-llm-mesh

2. Run bootstrap script

python bootstrap.py

This creates the .firepulse directory structure for configuration and encrypted credentials.

3. Install dependencies (if required)

pip install -r requirements.txt

Configuration

Provider Credentials

FirePulse uses a YAML-based configuration file with encrypted API key storage. Configuration file is typically located at ~/.firepulse/config.yaml.

Example configuration structure:

providers:
  xai:
    api_key: "${XAI_API_KEY}"
    models:
      - grok-1
      - grok-1.5
    priority: high
    timeout: 5000
    
  openrouter:
    api_key: "${OPENROUTER_API_KEY}"
    models:
      - anthropic/claude-2
      - meta-llama/llama-3-70b
    priority: medium
    timeout: 5000
    
  mistral:
    api_key: "${MISTRAL_API_KEY}"
    models:
      - mistral-large
      - mistral-medium
      - mistral-small
    priority: high
    timeout: 5000
    
  deepseek:
    api_key: "${DEEPSEEK_API_KEY}"
    models:
      - deepseek-coder
      - deepseek-v2
    priority: medium
    timeout: 5000

routing:
  strategy: "smart"  # Options: smart, cost, speed, quality
  budget_threshold: 0.01  # USD per query
  max_context_length: 8192
  fallback_enabled: true

session:
  persist_history: true
  max_context_messages: 20
  auto_purge_days: 7

Adding Provider Credentials

# Set environment variables for API keys
export XAI_API_KEY="your-xai-key"
export OPENROUTER_API_KEY="your-openrouter-key"
export MISTRAL_API_KEY="your-mistral-key"
export DEEPSEEK_API_KEY="your-deepseek-key"

# Run credential manager
python firepulse.py --setup

Key Commands

Start Interactive Session

python firepulse.py

Specify Provider Explicitly

python firepulse.py --provider xai
python firepulse.py --provider mistral

Single Query Mode

python firepulse.py --query "Explain quantum computing"

List Available Models

python firepulse.py --list-models

Check Provider Status

python firepulse.py --status

View Performance Metrics

python firepulse.py --metrics

Set Routing Strategy

python firepulse.py --strategy cost  # Optimize for cost
python firepulse.py --strategy speed  # Optimize for latency
python firepulse.py --strategy quality  # Optimize for best model

Code Examples

Python Integration (Library Mode)

If FirePulse exposes a Python API:

from firepulse import Orchestrator, RouterConfig

# Initialize orchestrator
config = RouterConfig(
    strategy="smart",
    budget_threshold=0.01,
    fallback_enabled=True
)

orchestrator = Orchestrator(config_path="~/.firepulse/config.yaml")

# Single query
response = orchestrator.query("Explain machine learning")
print(response.text)
print(f"Provider used: {response.provider}")
print(f"Model: {response.model}")
print(f"Latency: {response.latency_ms}ms")
print(f"Tokens: {response.tokens_used}")

# Streaming response
for chunk in orchestrator.stream("Write a Python script for web scraping"):
    print(chunk.text, end="", flush=True)

# Session-based conversation
session = orchestrator.create_session()
session.add_message("What is recursion?")
response1 = session.query()

session.add_message("Can you give me a code example?")
response2 = session.query()  # Maintains context

# Explicit provider selection
response = orchestrator.query(
    "Generate TypeScript types",
    provider="mistral",
    model="mistral-large"
)

Custom Router Logic

from firepulse import Orchestrator, RouterStrategy

class CostOptimizedRouter(RouterStrategy):
    def select_provider(self, query, context):
        # Filter providers by cost
        providers = self.get_available_providers()
        cheapest = min(providers, key=lambda p: p.cost_per_token)
        
        # Ensure context window fits
        if len(query) + len(context) > cheapest.max_context:
            return self.fallback_provider
            
        return cheapest

orchestrator = Orchestrator(router_strategy=CostOptimizedRouter())

Batch Processing

from firepulse import Orchestrator
import asyncio

orchestrator = Orchestrator()

async def batch_process(queries):
    tasks = [orchestrator.async_query(q) for q in queries]
    responses = await asyncio.gather(*tasks)
    return responses

queries = [
    "Summarize quantum mechanics",
    "Explain neural networks",
    "What is blockchain?"
]

results = asyncio.run(batch_process(queries))
for i, result in enumerate(results):
    print(f"Query {i+1}: {result.provider} - {result.latency_ms}ms")

Go Integration (if project uses Go)

Based on the project topics mentioning go and golang:

package main

import (
    "fmt"
    "github.com/vishalGitthub/cli-llm-mesh/pkg/orchestrator"
)

func main() {
    // Initialize orchestrator
    config := orchestrator.Config{
        ConfigPath: "~/.firepulse/config.yaml",
        Strategy:   orchestrator.SmartRouting,
    }
    
    orch, err := orchestrator.New(config)
    if err != nil {
        panic(err)
    }
    
    // Single query
    response, err := orch.Query("Explain Docker containers")
    if err != nil {
        panic(err)
    }
    
    fmt.Printf("Response: %s\n", response.Text)
    fmt.Printf("Provider: %s\n", response.Provider)
    fmt.Printf("Latency: %dms\n", response.LatencyMs)
    
    // Streaming query
    stream, err := orch.StreamQuery("Write a REST API in Go")
    if err != nil {
        panic(err)
    }
    
    for chunk := range stream {
        fmt.Print(chunk.Text)
    }
    
    // Explicit provider
    response, err = orch.QueryWithProvider(
        "Generate code documentation",
        orchestrator.ProviderMistral,
        "mistral-large",
    )
}

Session Management

package main

import (
    "fmt"
    "github.com/vishalGitthub/cli-llm-mesh/pkg/orchestrator"
)

func main() {
    orch, _ := orchestrator.New(orchestrator.Config{})
    
    // Create persistent session
    session := orch.NewSession()
    
    // First query
    session.AddMessage("What is Kubernetes?")
    resp1, _ := session.Query()
    fmt.Println(resp1.Text)
    
    // Follow-up with context
    session.AddMessage("How does it differ from Docker Swarm?")
    resp2, _ := session.Query()
    fmt.Println(resp2.Text)
    
    // Save session
    session.Save("~/.firepulse/sessions/k8s-discussion.json")
    
    // Load session later
    loadedSession, _ := orch.LoadSession("~/.firepulse/sessions/k8s-discussion.json")
}

Common Patterns

Pattern 1: Cost-Optimized Queries

from firepulse import Orchestrator

orchestrator = Orchestrator()

# Set budget constraint
response = orchestrator.query(
    "Generate marketing copy",
    max_cost=0.005,  # Max $0.005 per query
    fallback_on_budget_exceeded=True
)

Pattern 2: Multi-Provider Validation

# Query multiple providers for consensus
providers = ["xai", "mistral", "deepseek"]
responses = []

for provider in providers:
    resp = orchestrator.query(
        "Is this code correct? print('hello)",
        provider=provider
    )
    responses.append(resp)

# Compare responses
consensus = analyze_consensus(responses)

Pattern 3: Latency-Critical Operations

# Force fastest provider
orchestrator.set_strategy("speed")
response = orchestrator.query(
    "Quick yes/no: Is Python dynamically typed?",
    timeout=1000  # 1 second timeout
)

Pattern 4: Code Generation with Fallback

# Try DeepSeek (code specialist) first, fallback to Mistral
try:
    response = orchestrator.query(
        "Generate a binary search tree in Python",
        provider="deepseek",
        model="deepseek-coder"
    )
except TimeoutError:
    response = orchestrator.query(
        "Generate a binary search tree in Python",
        provider="mistral",
        model="mistral-large"
    )

Pattern 5: Multilingual Support

# Auto-detect language and respond accordingly
response = orchestrator.query("Explique l'apprentissage automatique")
print(response.detected_language)  # "fr"
print(response.text)  # Response in French

# Force output language
response = orchestrator.query(
    "Explain machine learning",
    output_language="es"  # Force Spanish output
)

Troubleshooting

Issue: Provider Timeout

Symptom: Queries hang or fail with timeout errors

Solution:

# Increase timeout in config.yaml
providers:
  xai:
    timeout: 10000  # Increase to 10 seconds

Or programmatically:

orchestrator = Orchestrator(default_timeout=10000)

Issue: Authentication Errors

Symptom: 401 Unauthorized or Invalid API Key

Solution:

# Verify environment variables
echo $XAI_API_KEY
echo $MISTRAL_API_KEY

# Re-run setup
python firepulse.py --setup --force

# Check credential encryption
python firepulse.py --verify-credentials

Issue: Model Not Available

Symptom: Requested model returns 404 or is unavailable

Solution:

# List currently available models
python firepulse.py --list-models --provider xai

# Update config to use available models
python firepulse.py --refresh-models

Issue: High Token Usage / Cost

Symptom: Unexpectedly high API costs

Solution:

# Enable cost monitoring
orchestrator = Orchestrator(
    budget_threshold=0.01,
    warn_on_threshold=True
)

# Check cost estimates before query
estimate = orchestrator.estimate_cost("Your long query...")
print(f"Estimated cost: ${estimate.cost}")

Issue: Slow Response Times

Symptom: Latency exceeds expected 200ms to first token

Solution:

# Check provider status
python firepulse.py --status

# Switch to speed-optimized routing
python firepulse.py --strategy speed

# Enable local query validation to reduce roundtrips

orchestrator = Orchestrator(
    offline_validation=True,  # Validate locally first
    parallel_requests=True     # Query multiple providers in parallel
)

Issue: Context Length Exceeded

Symptom: Context length exceeded errors

Solution:

# Automatically truncate or summarize context
session = orchestrator.create_session(
    max_context_length=4096,
    auto_summarize=True  # Summarize old messages when limit reached
)

Issue: Provider Fallback Not Working

Symptom: Query fails instead of falling back to alternate provider

Solution:

# Ensure fallback is enabled in config.yaml
routing:
  fallback_enabled: true
  fallback_order:
    - mistral
    - openrouter
    - deepseek
    - xai

# Force fallback behavior
orchestrator = Orchestrator(
    strict_fallback=False,  # Allow any provider as fallback
    retry_on_failure=3      # Retry up to 3 times
)

Performance Optimization

Enable Response Caching

from firepulse import Orchestrator, CacheConfig

cache_config = CacheConfig(
    enabled=True,
    ttl=3600,  # 1 hour cache
    max_size_mb=100
)

orchestrator = Orchestrator(cache_config=cache_config)

# Repeated queries use cached responses
response1 = orchestrator.query("What is Python?")  # API call
response2 = orchestrator.query("What is Python?")  # Cached

Parallel Provider Interrogation

# Query multiple providers simultaneously, use fastest response
response = orchestrator.query_parallel(
    "Explain Docker",
    providers=["xai", "mistral", "deepseek"],
    return_first=True  # Return as soon as first provider responds
)

Docker Deployment

FROM python:3.11-slim

WORKDIR /app
COPY . .

RUN pip install -r requirements.txt

# Mount config as volume
VOLUME /root/.firepulse

CMD ["python", "firepulse.py"]

# Build and run
docker build -t firepulse .
docker run -it \
  -e XAI_API_KEY="${XAI_API_KEY}" \
  -e MISTRAL_API_KEY="${MISTRAL_API_KEY}" \
  -v ~/.firepulse:/root/.firepulse \
  firepulse

Environment Variables Reference

# Provider API Keys
export XAI_API_KEY="your-xai-key"
export OPENROUTER_API_KEY="your-openrouter-key"
export MISTRAL_API_KEY="your-mistral-key"
export DEEPSEEK_API_KEY="your-deepseek-key"

# Configuration Overrides
export FIREPULSE_CONFIG_PATH="/custom/path/config.yaml"
export FIREPULSE_ROUTING_STRATEGY="cost"  # smart, cost, speed, quality
export FIREPULSE_BUDGET_THRESHOLD="0.01"
export FIREPULSE_DEFAULT_TIMEOUT="5000"
export FIREPULSE_LOG_LEVEL="info"  # debug, info, warn, error