Bases: ResultBackendSelector
Gen0 candidate provided by the user.
Source code in peagen/plugins/selectors/input_selector.py
| def __init__(
self, backend: Any, initial_candidate: Dict[str, Any], num_candidates: int = 1
) -> None:
super().__init__(backend, num_candidates)
self.initial_candidate = initial_candidate
self._used_input = False
|
initial_candidate = initial_candidate
num_candidates = num_candidates
Source code in peagen/plugins/selectors/input_selector.py
| async def select(self) -> Dict[str, Any]:
if not self._used_input:
self._used_input = True
return {"leader": None, "candidates": [self.initial_candidate]}
return await super().select()
|
Source code in peagen/plugins/selectors/selector_base.py
| async def get_tasks(self) -> List[Dict[str, Any]]:
return await self.backend.list_tasks()
|
Source code in peagen/plugins/selectors/selector_base.py
| async def get_leader(self, tasks: List[Dict[str, Any]]) -> Dict[str, Any] | None:
successes = [t for t in tasks if t.get("status") == Status.success.value]
if successes:
successes.sort(key=lambda d: d.get("result", {}).get("score", float("inf")))
return successes[0]
return None
|
get_running_candidates(tasks)
Source code in peagen/plugins/selectors/selector_base.py
| async def get_running_candidates(
self, tasks: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
running = [t for t in tasks if t.get("status") == Status.running.value]
running.sort(key=lambda d: str(d.get("date_created") or ""))
return running[: self.num_candidates]
|