tahamajs/Sysmem2_in_AI/ComputerAssignments/CA10_knowledge_graphs/integrations - Storage Bucket

Integrations

Overview

This directory contains integrations with external services and APIs.

Available Integrations

1. Google Gemini API

File: gemini_experimental.py

Integration with Google's Gemini API for advanced reasoning:

• Multi-modal understanding
• Advanced reasoning capabilities
• Natural language processing
• Code generation

Setup:

# Set API key in .env
GOOGLE_API_KEY=your_gemini_api_key_here

Usage:

from integrations.gemini_experimental import GeminiReasoner

reasoner = GeminiReasoner(api_key="your_key")
result = reasoner.reason_about_query("What is the relationship between X and Y?")

2. Advanced Execution

File: main_execution_advanced.py

Advanced execution script with all integrations enabled:

• LangGraph reasoning
• RAG system
• Multi-agent coordination
• Gemini API integration

Run:

python integrations/main_execution_advanced.py

Adding New Integrations

Template Structure

#!/usr/bin/env python3
"""
Integration with [Service Name]
"""

import os
from dotenv import load_dotenv

class ServiceIntegration:
    def __init__(self, api_key=None):
        load_dotenv()
        self.api_key = api_key or os.getenv('SERVICE_API_KEY')
        if not self.api_key:
            raise ValueError("API key not found")

    def connect(self):
        """Establish connection to service"""
        pass

    def query(self, query_text):
        """Query the service"""
        pass

    def disconnect(self):
        """Close connection"""
        pass

# Example usage
if __name__ == "__main__":
    integration = ServiceIntegration()
    result = integration.query("test query")
    print(result)

Supported Services

AI/ML Services

• Google Gemini: Advanced reasoning and generation
• OpenAI GPT: Language understanding and generation
• Anthropic Claude: Reasoning and analysis
• Hugging Face: Model hosting and inference

Vector Databases

• ChromaDB: Vector storage and retrieval
• FAISS: Similarity search
• Pinecone: Managed vector database
• Weaviate: Vector search engine

Graph Databases

• Neo4j: Graph database for knowledge graphs
• ArangoDB: Multi-model database
• JanusGraph: Distributed graph database

Monitoring & Logging

• Weights & Biases: Experiment tracking
• MLflow: Model versioning and tracking
• TensorBoard: Visualization and monitoring

Configuration

Environment Variables

Create a .env file in the project root:

# AI Services
GOOGLE_API_KEY=your_gemini_key
OPENAI_API_KEY=your_openai_key
ANTHROPIC_API_KEY=your_anthropic_key
HUGGINGFACE_API_KEY=your_hf_key

# Vector Databases
CHROMA_PERSIST_DIRECTORY=./chroma_db
PINECONE_API_KEY=your_pinecone_key
PINECONE_ENVIRONMENT=us-west1-gcp

# Graph Databases
NEO4J_URI=bolt://localhost:7687
NEO4J_USER=neo4j
NEO4J_PASSWORD=your_password

# Monitoring
WANDB_API_KEY=your_wandb_key
MLFLOW_TRACKING_URI=http://localhost:5000

Integration Testing

Test integrations before use:

# Test Gemini integration
python -c "from integrations.gemini_experimental import GeminiReasoner; print('Gemini OK')"

# Test all integrations
python integrations/test_integrations.py

Error Handling

Handle integration errors gracefully:

try:
    from integrations.gemini_experimental import GeminiReasoner
    reasoner = GeminiReasoner()
except ImportError:
    print("Gemini integration not available")
    reasoner = None
except ValueError as e:
    print(f"Configuration error: {e}")
    reasoner = None

Rate Limiting

Be aware of API rate limits:

import time
from functools import wraps

def rate_limit(calls_per_minute=60):
    min_interval = 60.0 / calls_per_minute
    last_called = [0.0]

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_called[0]
            left_to_wait = min_interval - elapsed
            if left_to_wait > 0:
                time.sleep(left_to_wait)
            result = func(*args, **kwargs)
            last_called[0] = time.time()
            return result
        return wrapper
    return decorator

@rate_limit(calls_per_minute=10)
def call_api(query):
    # API call here
    pass

Caching

Cache API responses to reduce costs:

from functools import lru_cache
import hashlib

@lru_cache(maxsize=1000)
def cached_api_call(query):
    # API call here
    return result

Best Practices

1. API Keys: Never commit API keys to version control
2. Error Handling: Always handle API errors gracefully
3. Rate Limiting: Respect API rate limits
4. Caching: Cache responses when possible
5. Logging: Log all API calls for debugging
6. Timeouts: Set appropriate timeouts for API calls
7. Retries: Implement retry logic for failed requests

Troubleshooting

Issue: API Key Not Found

Solution: Ensure .env file exists and contains the required keys

Issue: Connection Timeout

Solution: Check network connectivity and increase timeout values

Issue: Rate Limit Exceeded

Solution: Implement rate limiting and backoff strategies

Documentation

For detailed integration documentation, see:

• Gemini API Docs
• OpenAI API Docs
• Anthropic API Docs
• ChromaDB Docs
• Neo4j Docs

---

Note: Some integrations may require additional setup or paid subscriptions.