DSPy

import dspy
import requests
from datetime import datetime
from typing import Dict, Any
import uuid
import time

class ParseableDSPyLogger:
    def __init__(self, parseable_url: str, dataset: str, username: str, password: str):
        self.parseable_url = parseable_url
        self.dataset = dataset
        self.auth = (username, password)
        self.program_id = str(uuid.uuid4())
    
    def _log(self, entry: Dict[str, Any]):
        entry["timestamp"] = datetime.utcnow().isoformat() + "Z"
        entry["program_id"] = self.program_id
        try:
            requests.post(
                f"{self.parseable_url}/api/v1/ingest",
                json=[entry],
                auth=self.auth,
                headers={"X-P-Stream": self.dataset},
                timeout=5
            )
        except Exception as e:
            print(f"Logging failed: {e}")
    
    def log_module_call(self, module_name: str, inputs: Dict, outputs: Dict, duration_ms: float):
        self._log({
            "event": "module_call",
            "module_name": module_name,
            "inputs": {k: str(v)[:100] for k, v in inputs.items()},
            "outputs": {k: str(v)[:200] for k, v in outputs.items()},
            "duration_ms": duration_ms
        })
    
    def log_optimization_trial(self, trial_num: int, score: float):
        self._log({
            "event": "optimization_trial",
            "trial_num": trial_num,
            "score": score
        })

-- Module performance
SELECT module_name, AVG(duration_ms) as avg_duration, COUNT(*) as calls
FROM "dspy-logs"
WHERE event = 'module_call'
GROUP BY module_name

-- Optimization progress
SELECT trial_num, score
FROM "dspy-logs"
WHERE event = 'optimization_trial'
ORDER BY trial_num