orka.orchestrator_error_wrapper module

Error handling wrapper for OrKa Orchestrator. Provides comprehensive error tracking and telemetry without modifying the core orchestrator logic.

class orka.orchestrator_error_wrapper.OrkaErrorHandler(orchestrator)

Bases: object

Comprehensive error handling system for OrKa orchestrator. Tracks errors, retries, status codes, and provides detailed debugging reports.

record_error(error_type: str, agent_id: str, error_msg: str, exception: Exception = None, step: int = None, status_code: int = None, recovery_action: str = None)

Record an error in the error telemetry system.

record_silent_degradation(agent_id: str, degradation_type: str, details: str)

Record silent degradations like JSON parsing failures.

save_comprehensive_error_report(logs: List[Dict], final_error: Exception = None)

Save comprehensive error report with all logged data up to the failure point.

async run_with_error_handling(input_data)

Run the orchestrator with comprehensive error handling. Always returns a JSON report, even on failure, for debugging purposes.

async orka.orchestrator_error_wrapper.run_orchestrator_with_error_handling(orchestrator, input_data)

Enhanced wrapper function to run any orchestrator with comprehensive error handling.

Usage:

from orka.orchestrator_error_wrapper import run_orchestrator_with_error_handling

orchestrator = Orchestrator(“config.yml”) result = await run_orchestrator_with_error_handling(orchestrator, input_data)