agentic_controller#

AgenticController: runs a CopilotCLIAgent for a fixed number of iterations.

Classes

AgenticController(run_config, job_id, task_id)

Runs a CopilotCLIAgent for a fixed number of iterations.
class kernelfoundry.algorithm.agentic_controller.AgenticController(run_config, job_id: int | None, task_id: int | None, copilot_exe: str = 'copilot', env_overrides: dict[str, str] | None = None, agent_timeout: float = 3600.0, logging_level: int | str | None = None, extra_mcp_servers: dict[str, dict] | None = None)[source]#

Runs a CopilotCLIAgent for a fixed number of iterations.

Parameters:

  • run_config – Hydra configuration object (must have logdir and max_iters).

  • job_id – Job identifier passed through to the agent.

  • task_id – Task identifier (stored for reference).

  • copilot_exe – Path or name of the Copilot CLI binary.

  • env_overrides – Extra environment variables forwarded to the Copilot subprocess.

  • agent_timeout – Per-iteration timeout in seconds.

  • logging_level – Logging level for the file handler added by this controller.

  • extra_mcp_servers – Optional additional MCP server definitions merged into the agent configuration.

__init__(run_config, job_id: int | None, task_id: int | None, copilot_exe: str = 'copilot', env_overrides: dict[str, str] | None = None, agent_timeout: float = 3600.0, logging_level: int | str | None = None, extra_mcp_servers: dict[str, dict] | None = None)[source]#
static build_container_image_for_task(task: Task) → Task[source]#

Build the container image for the given Task if use_container is True.

Updates task.config["container_image"] in-place and returns the task.

sample_evolve_programs()[source]#

Sample parent code, inspirations, top program from the database

create_program0(task: Task) → Program[source]#

Create initial Program object for a Task.

Parameters:

task – Task instance containing problem details.

Returns:

Initial Program object built from the EVOLVE block.

evolve_prompt(problem_name: str, ref_arch_src: str, trial: int, task: Task | None = None) → tuple[str, Program | None][source]#

Construct a fresh prompt for a single branch and report the sampled parent.

Parameters:

  • problem_name – Operation name

  • ref_arch_src – Reference source code

  • trial – Trial number (to pass to LLM server)

  • task – Custom task object that can provide the initial implementation

Returns:

the prompt used for generation, and the parent program sampled from the database (None if no parent was available). The parent’s id identifies the agent to fork in order to continue the search from it.

Return type:

A (prompt, parent) pair

evaluate_reference(task: Task) → tuple[Task, dict[str, EvalResult]][source]#
run(task: Task) → list[tuple[Program, EvalResult]][source]#

Create an agent for task and run it for config.max_iters iterations.

Parameters:

task – The task to optimize.

Returns:

All (Program, EvalResult) pairs collected across all iterations, in the order they were produced.