"""Dataset loading (file, interactive).""" import json import sys from pathlib import Path from typing import Dict, List, Optional, Tuple import questionary from cli_logging.cli_logger import emit_structured_log from cli_logging.logging_utils import Operation from common import RunConfig from schema_handler import DocumentUpgrader def _validate_captured_responses(items: List[Dict]) -> None: """Require a captured response for every item or turn.""" for item_index, item in enumerate(items): turns = item.get("turns") if isinstance(turns, list): for turn_index, turn in enumerate(turns): if "response" not in turn: raise ValueError( f"Missing response for item {item_index}, turn {turn_index}" ) elif "response" not in item: raise ValueError(f"Missing response for item {item_index}") def load_prompts_from_file( file_path: str, *, eval_document_only: bool = False, require_responses: bool = False, ) -> Tuple[List[Dict], Optional[Dict], Optional[Dict]]: """Load prompts and expected responses from a JSON file. Supports three formats: 1. Eval document: {"schemaVersion": "1.0.0", "items": [{"prompt": "..."}]} 2. Array format: [{"prompt": "...", "expected_response": "..."}] 3. Dict format: {"prompts": [...], "expected_responses": [...]} For eval documents (format 1) and array format (format 2), schema validation and auto-upgrade are applied via DocumentUpgrader. Returns: Tuple of (eval_items, default_evaluators, document_extensions). Items are dicts with prompt, expected_response, and optional evaluators/evaluators_mode fields. ``document_extensions`` is the verbatim ``metadata.extensions`` object from the input document (eval-document format only), or ``None``. """ try: with open(file_path, 'r', encoding='utf-8') as f: data = json.load(f) if eval_document_only and not ( isinstance(data, dict) and isinstance(data.get("schemaVersion"), str) and isinstance(data.get("items"), list) ): raise ValueError( "--evaluate-only requires a v1 eval document with schemaVersion and items" ) # Detect if this is an eval document (has "items" key) or could be upgraded is_eval_document = ( isinstance(data, dict) and "items" in data ) or isinstance(data, list) # Run schema validation and auto-upgrade for eval documents if is_eval_document: try: upgrader = DocumentUpgrader() except Exception as e: # Schema infrastructure not available (missing files, etc.) — skip emit_structured_log("warning", f"Unable to initialize document upgrader: {e}", operation=Operation.LOAD_PROMPTS) upgrader = None if upgrader is not None: result = upgrader.upgrade(Path(file_path)) if result.error: emit_structured_log("error", f"Schema validation error: {result.error}", operation=Operation.LOAD_PROMPTS) sys.exit(1) if result.upgraded and result.message: emit_structured_log("info", result.message, operation=Operation.LOAD_PROMPTS) # Use the parsed document from the upgrade result if result.document is not None: data = result.document if isinstance(data, list): # Format: [{"prompt": "...", "expected_response": "..."}, ...] return data, None, None elif isinstance(data, dict): if "items" in data: # Eval document format: {"schemaVersion": "...", "items": [...]} if require_responses: _validate_captured_responses(data["items"]) document_extensions = None metadata = data.get("metadata") if isinstance(metadata, dict): ext = metadata.get("extensions") if isinstance(ext, dict): document_extensions = ext return data["items"], data.get("default_evaluators"), document_extensions else: # Format: {"prompts": [...], "expected_responses": [...]} prompts = data.get("prompts", []) expected_responses = data.get("expected_responses", []) eval_items = [ {"prompt": p, "expected_response": e} for p, e in zip(prompts, expected_responses) ] return eval_items, None, None else: raise ValueError("Invalid file format") except SystemExit: raise except Exception as e: emit_structured_log("error", f"Error loading prompts from file: {e}", operation=Operation.LOAD_PROMPTS) sys.exit(1) def get_interactive_prompts() -> Tuple[List[str], List[str]]: """Get prompts and expected responses interactively.""" prompts = [] expected_responses = [] print("Interactive mode: Enter your prompts and expected responses.") print("Press Enter with empty prompt to finish.") while True: prompt = input(f"\nPrompt {len(prompts) + 1}: ").strip() if not prompt: break expected = input(f"Expected response {len(expected_responses) + 1}: ").strip() prompts.append(prompt) expected_responses.append(expected) if not prompts: print("No prompts entered. Exiting.") sys.exit(1) return prompts, expected_responses def get_prompt_datasets(config: RunConfig) -> Tuple[List[Dict], Optional[Dict], Optional[Dict]]: """Get prompts and expected responses based on command line arguments. Returns: Tuple of (eval_items, default_evaluators, document_extensions). ``document_extensions`` is populated only when loading an eval-document file that carries ``metadata.extensions``; otherwise ``None``. """ if config.prompts: if config.expected and len(config.prompts) != len(config.expected): emit_structured_log( "error", "Number of prompts must match number of expected responses. " "Update --expected values to match the prompt count.", ) sys.exit(1) expected_responses = config.expected or [""] * len(config.prompts) eval_items = [ {"prompt": p, "expected_response": e} for p, e in zip(config.prompts, expected_responses) ] return eval_items, None, None elif config.prompts_file: return load_prompts_from_file(config.prompts_file) elif config.evaluate_only: return load_prompts_from_file( config.evaluate_only, eval_document_only=True, require_responses=True, ) elif config.interactive: prompts, expected_responses = get_interactive_prompts() eval_items = [ {"prompt": p, "expected_response": e} for p, e in zip(prompts, expected_responses) ] return eval_items, None, None else: emit_structured_log( "error", "No prompts provided. Use --prompts, --prompts-file, " "--evaluate-only, or --interactive.", operation=Operation.SETUP, ) sys.exit(1)