"""SearchQA environment adapter for ReflACT.""" from __future__ import annotations import json import os from skillopt.datasets.base import BatchSpec from skillopt.envs.base import EnvAdapter from skillopt.envs.searchqa.dataloader import SearchQADataLoader from skillopt.envs.searchqa.rollout import run_batch from skillopt.gradient.reflect import run_minibatch_reflect from skillopt.model import get_target_backend class SearchQAAdapter(EnvAdapter): """SearchQA environment adapter.""" def __init__( self, split_dir: str = "", data_path: str = "", split_mode: str = "ratio", split_ratio: str = "2:1:7", split_seed: int = 42, split_output_dir: str = "", max_turns: int = 1, exec_timeout: int = 120, workers: int = 64, analyst_workers: int = 16, failure_only: bool = False, minibatch_size: int = 8, edit_budget: int = 4, seed: int = 42, limit: int = 0, max_completion_tokens: int = 16384, ) -> None: self.max_turns = max_turns self.exec_timeout = exec_timeout self.workers = workers self.max_completion_tokens = int(max_completion_tokens) self.analyst_workers = analyst_workers self.failure_only = failure_only self.minibatch_size = minibatch_size self.edit_budget = edit_budget self.dataloader = SearchQADataLoader( split_dir=split_dir, data_path=data_path, split_mode=split_mode, split_ratio=split_ratio, split_seed=split_seed, split_output_dir=split_output_dir, seed=seed, limit=limit, ) def setup(self, cfg: dict) -> None: super().setup(cfg) self.dataloader.setup(cfg) def get_dataloader(self): return self.dataloader def build_env_from_batch(self, batch: BatchSpec, **kwargs): return list(batch.payload or []) def build_train_env(self, batch_size: int, seed: int, **kwargs): batch = self.dataloader.build_train_batch(batch_size=batch_size, seed=seed, **kwargs) return self.build_env_from_batch(batch, **kwargs) def build_eval_env(self, env_num: int, split: str, seed: int, **kwargs): batch = self.dataloader.build_eval_batch(env_num=env_num, split=split, seed=seed, **kwargs) return self.build_env_from_batch(batch, **kwargs) def rollout( self, env_manager, # actually list[dict] for SearchQA skill_content: str, out_dir: str, **kwargs, ) -> list[dict]: """Run QA agent on items. Resume-aware.""" items: list[dict] = env_manager # type alias for clarity return run_batch( items=items, out_root=out_dir, skill_content=skill_content, max_turns=self.max_turns, exec_timeout=self.exec_timeout, workers=self.workers, max_completion_tokens=self.max_completion_tokens, diagnostic_mode=kwargs.get("diagnostic_mode", False), diagnostic_instruction=kwargs.get("diagnostic_instruction", ""), diagnostic_trace_context_by_id=kwargs.get("diagnostic_trace_context_by_id"), task_timeout=self.exec_timeout, ) def reflect( self, results: list[dict], skill_content: str, out_dir: str, **kwargs, ) -> list[dict | None]: prediction_dir = kwargs.get("prediction_dir", os.path.join(out_dir, "predictions")) patches_dir = kwargs.get("patches_dir", os.path.join(out_dir, "patches")) random_seed = kwargs.get("random_seed") step_buffer_context = kwargs.get("step_buffer_context", "") meta_skill_context = kwargs.get("meta_skill_context", "") return run_minibatch_reflect( results=results, skill_content=skill_content, prediction_dir=prediction_dir, patches_dir=patches_dir, workers=self.analyst_workers, failure_only=self.failure_only, minibatch_size=self.minibatch_size, edit_budget=self.edit_budget, random_seed=random_seed, error_system=self.get_error_minibatch_prompt(), success_system=self.get_success_minibatch_prompt(), step_buffer_context=step_buffer_context, meta_skill_context=meta_skill_context, update_mode=getattr(self, "_cfg", {}).get("skill_update_mode", "patch"), ) def get_task_types(self) -> list[str]: return ["qa"]