Class peagen.plugins.selectors.result_backend_selector.ResultBackendSelector

peagen.plugins.selectors.result_backend_selector.ResultBackendSelector

ResultBackendSelector(backend, num_candidates=1)

Bases: SelectorBase

Select candidates based on the results backend.

Source code in peagen/plugins/selectors/result_backend_selector.py

def __init__(self, backend: Any, num_candidates: int = 1) -> None:
    super().__init__(backend, num_candidates)

backend instance-attribute

backend = backend

num_candidates instance-attribute

num_candidates = num_candidates

get_tasks async

get_tasks()

Source code in peagen/plugins/selectors/selector_base.py

async def get_tasks(self) -> List[Dict[str, Any]]:
    return await self.backend.list_tasks()

get_leader async

get_leader(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 async

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]

select async

select()

Source code in peagen/plugins/selectors/selector_base.py

async def select(self) -> Dict[str, Any]:
    tasks = await self.get_tasks()
    leader = await self.get_leader(tasks)
    candidates = await self.get_running_candidates(tasks)
    return {"leader": leader, "candidates": candidates}