Sources API Overview

Authorization: Bearer grlm_your_api_token_here

{
  "detail": "Descriptive error message explaining what went wrong"
}

import time
import requests
from typing import Optional

def api_call_with_retry(
    method: str, 
    url: str, 
    headers: dict, 
    max_retries: int = 3,
    **kwargs
) -> Optional[requests.Response]:
    """Make API call with exponential backoff retry logic."""
    
    for attempt in range(max_retries):
        try:
            response = requests.request(method, url, headers=headers, **kwargs)
            
            # Success cases
            if response.status_code < 400:
                return response
            
            # Don't retry client errors (4xx)
            if 400 <= response.status_code < 500:
                response.raise_for_status()
            
            # Retry server errors (5xx)
            if response.status_code >= 500 and attempt < max_retries - 1:
                wait_time = 2 ** attempt  # Exponential backoff
                print(f"Server error {response.status_code}, retrying in {wait_time}s...")
                time.sleep(wait_time)
                continue
            
            response.raise_for_status()
            
        except requests.exceptions.RequestException as e:
            if attempt < max_retries - 1:
                wait_time = 2 ** attempt
                print(f"Request failed: {e}, retrying in {wait_time}s...")
                time.sleep(wait_time)
            else:
                raise
    
    return None

import requests
from typing import List, Dict, Optional
import time

class GraphorLMSourcesClient:
    def __init__(self, api_token: str):
        self.api_token = api_token
        self.base_url = "https://sources.graphorlm.com"
        self.headers = {
            "Authorization": f"Bearer {api_token}",
            "Content-Type": "application/json"
        }
    
    def upload_document(self, file_path: str) -> Dict:
        """Upload a document to GraphorLM."""
        upload_headers = {"Authorization": f"Bearer {self.api_token}"}
        
        with open(file_path, "rb") as f:
            files = {"file": (file_path, f)}
            response = requests.post(
                f"{self.base_url}/upload",
                headers=upload_headers,
                files=files,
                timeout=300
            )
        
        response.raise_for_status()
        return response.json()
    
    def process_document(self, file_name: str, method: str = "yolox") -> Dict:
        """Reprocess a document with specified method."""
        payload = {
            "file_name": file_name,
            "partition_method": method
        }
        
        response = requests.post(
            f"{self.base_url}/process",
            headers=self.headers,
            json=payload,
            timeout=300
        )
        
        response.raise_for_status()
        return response.json()
    
    def list_documents(self) -> List[Dict]:
        """Get all documents in the project."""
        response = requests.get(
            self.base_url,
            headers=self.headers,
            timeout=30
        )
        
        response.raise_for_status()
        return response.json()
    
    def delete_document(self, file_name: str) -> Dict:
        """Delete a document from the project."""
        payload = {"file_name": file_name}
        
        response = requests.delete(
            f"{self.base_url}/delete",
            headers=self.headers,
            json=payload,
            timeout=60
        )
        
        response.raise_for_status()
        return response.json()
    
    def wait_for_processing(self, file_name: str, timeout: int = 300) -> str:
        """Wait for document processing to complete."""
        start_time = time.time()
        
        while time.time() - start_time < timeout:
            documents = self.list_documents()
            doc = next((d for d in documents if d['file_name'] == file_name), None)
            
            if not doc:
                raise ValueError(f"Document {file_name} not found")
            
            status = doc['status'].lower()
            if status == 'completed':
                return 'completed'
            elif status == 'failed':
                return 'failed'
            
            time.sleep(5)  # Check every 5 seconds
        
        raise TimeoutError(f"Processing timeout for {file_name}")

# Usage Example
client = GraphorLMSourcesClient("grlm_your_api_token")

# Upload and process a document
upload_result = client.upload_document("./document.pdf")
print(f"Uploaded: {upload_result['file_name']}")

# Wait for initial processing
status = client.wait_for_processing(upload_result['file_name'])
print(f"Initial processing: {status}")

# Upgrade to better processing method
if status == 'completed':
    process_result = client.process_document(
        upload_result['file_name'], 
        "yolox"
    )
    print(f"Reprocessed with YOLOX: {process_result['partition_method']}")

class DocumentQualityPipeline {
  constructor(apiToken) {
    this.apiToken = apiToken;
    this.baseUrl = 'https://sources.graphorlm.com';
  }

  async assessAndOptimize(fileName) {
    const methods = ['basic', 'ocr', 'yolox', 'advanced'];
    let bestResult = null;
    let bestScore = 0;

    for (const method of methods) {
      try {
        console.log(`Testing ${method} method for ${fileName}...`);
        
        const result = await this.processDocument(fileName, method);
        const score = this.calculateQualityScore(result);
        
        console.log(`${method}: score ${score}`);
        
        if (score > bestScore) {
          bestScore = score;
          bestResult = { method, result, score };
        }
        
        // If score is good enough, don't try more expensive methods
        if (score >= 0.8 && method !== 'advanced') {
          break;
        }
        
      } catch (error) {
        console.error(`Failed with ${method}:`, error.message);
      }
    }

    return bestResult;
  }

  calculateQualityScore(result) {
    // Implement your quality assessment logic
    // This is a simple example based on available data
    let score = 0;
    
    // Base score for successful processing
    if (result.status === 'success') score += 0.5;
    
    // Higher score for more advanced methods
    const methodScores = {
      'basic': 0.1,
      'ocr': 0.2,
      'yolox': 0.3,
      'advanced': 0.4
    };
    score += methodScores[result.partition_method] || 0;
    
    // File size and type considerations
    if (result.file_size > 1000000) score += 0.1; // Larger files
    if (result.file_type === 'pdf') score += 0.1;  // PDFs often need better processing
    
    return Math.min(score, 1.0);
  }

  async processDocument(fileName, method) {
    const response = await fetch(`${this.baseUrl}/process`, {
      method: 'POST',
      headers: {
        'Authorization': `Bearer ${this.apiToken}`,
        'Content-Type': 'application/json'
      },
      body: JSON.stringify({
        file_name: fileName,
        partition_method: method
      })
    });

    if (!response.ok) {
      throw new Error(`HTTP ${response.status}: ${await response.text()}`);
    }

    return response.json();
  }
}

// Usage
const pipeline = new DocumentQualityPipeline('grlm_your_token');
pipeline.assessAndOptimize('complex_document.pdf')
  .then(result => {
    if (result) {
      console.log(`Best method: ${result.method} (score: ${result.score})`);
    } else {
      console.log('No processing method succeeded');
    }
  });