Python API reference¶
Auto-generated from docstrings (via mkdocstrings)
for the public programmatic surface exposed by ksi and its registry
subpackages. For a task-oriented walkthrough, see the
programmatic API guide and the
extension seams.
Running a generation¶
run ¶
run(
config: GenerationConfig,
tasks: list[TaskSpec],
*,
runtime: "RuntimeExecutor",
evaluator: "Evaluator",
llm: "LLMCaller",
persistence: "PersistenceObserver | None" = None,
working_dir: str = ".",
) -> list[TaskTrace]
Run a generational self-improvement loop and return the task traces.
This is the programmatic equivalent of invoking the CLI: it constructs a
GenerationalOrchestrator with the same arguments and executes tasks.
The orchestrator construction and run are identical to the CLI path;
CLI-only conveniences are NOT applied — provider
profile / environment loading and signal handling are the caller's
responsibility, so set any such environment yourself before calling run.
.. important::
Unlike the CLI, run does NOT derive a default knowledge DB path.
config.knowledge_db_path defaults to "" (see
GenerationConfig), and the orchestrator SILENTLY disables the entire
knowledge substrate when it is empty — task attempts are not recorded,
and retrieval, distillation, and cross-generation seeding never run
(only raw per-generation attempts happen). The CLI never hits this
because it always resolves a per-experiment default path
(cli._prepare_knowledge_db_path). To get the knowledge loop, set
config.knowledge_db_path to a writable *_knowledge.sqlite path
(one per concurrent experiment). run emits a UserWarning when
the path is empty so the degrade is visible; pass a path to silence it.
Two further CLI-only conveniences are the caller's responsibility (no code
is run for them here): loading a provider profile / auth environment (the
container host validates provider auth and raises if it is missing — set
MODEL_PROVIDER / *_API_KEY / etc. yourself), and any task-source
runtime delegation the CLI wires up (e.g. TB2). See docs/programmatic_api.md.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
GenerationConfig
|
The run configuration (generations, agents, paths, ...). |
required |
tasks
|
list[TaskSpec]
|
The tasks to attempt. |
required |
runtime
|
'RuntimeExecutor'
|
How agents execute (e.g. |
required |
evaluator
|
'Evaluator'
|
How attempts are scored (e.g. |
required |
llm
|
'LLMCaller'
|
The forum/distillation LLM caller (see
|
required |
persistence
|
'PersistenceObserver | None'
|
Optional |
None
|
working_dir
|
str
|
Project root used to resolve runtime/runner paths. Defaults to |
'.'
|
Returns:
| Type | Description |
|---|---|
list[TaskTrace]
|
One trace per executed task attempt. |
Extension registration¶
Register new seams without editing core dispatch. See extending.md for the surrounding guide.
register_evaluator ¶
register_evaluator(
spec: EvaluatorSpec, *, replace: bool = False
) -> EvaluatorSpec
Register spec (and aliases). Raises ValueError on duplicate unless
replace=True. Returns the spec for convenience.
register_runtime ¶
register_runtime(
spec: RuntimeSpec, *, replace: bool = False
) -> RuntimeSpec
Register spec (and aliases). Raises ValueError on duplicate unless
replace=True. Returns the spec for convenience.
register_task_source ¶
register_task_source(
spec: TaskSourceSpec, *, replace: bool = False
) -> TaskSourceSpec
Register spec (and its aliases) into the global REGISTRY.
Raises ValueError on a duplicate name/alias unless replace=True.
Returns the spec for convenience.
register_strategy ¶
register_strategy(
spec: StrategySpec, *, replace: bool = False
) -> StrategySpec
Register spec (and aliases). Raises ValueError on duplicate unless
replace=True. Returns the spec for convenience.
Specification types¶
EvaluatorSpec
dataclass
¶
How to construct one evaluator. factory is called as factory(args)
where args is a config object exposing the attributes the factory reads
(the CLI passes its argparse Namespace). Non-CLI callers should use
:func:build_evaluator, which supplies the defaults and applies overrides.
__init__ ¶
__init__(
name: str,
factory: Callable[..., Evaluator],
aliases: tuple[str, ...] = (),
description: str = "",
) -> None
RuntimeSpec
dataclass
¶
How to construct one runtime. factory is called as
factory(args, provider_env) -> RuntimeExecutor, where args is a
config object exposing the attributes the factory reads (the CLI passes its
argparse Namespace). Non-CLI callers should use :func:build_runtime.
__init__ ¶
__init__(
name: str,
factory: Callable[..., "RuntimeExecutor"],
aliases: tuple[str, ...] = (),
description: str = "",
) -> None
TaskSourceSpec
dataclass
¶
Per-source variation points that describe how a benchmark plugs into KSI.
Fields fall into three groups:
name/aliases— the canonical id and accepted synonyms a caller may pass as--task-source.default_evaluator/prompt_kind— string keys that select this source's default evaluator and its fallback prompt behavior when no callable override is attached.- Optional callables (
execution_prompt_builder,task_markdown_builder,distill_domain_hint,loader,approach_diagnosis,score_from_eval,trace_condensed,attempt_meta_builder,validate_tasks_path) — per-source hooks a plugin attaches to customize behavior without editing shared dispatch code. Each is documented at its attribute definition below.
Boolean/flag fields (supports_mcp_arc, is_offline,
uses_repo_snapshots, supports_classification,
needs_eval_records, delegates_runtime, arc_task_reference)
declare capabilities the source has or needs; the default for every flag
is the conservative "off" behavior, so a new source only sets the flags
it actually requires.
See adding_a_benchmark.md for a worked example of registering a new source, and the attribute-level docstrings below for what each optional callable receives and returns.
execution_prompt_builder
class-attribute
instance-attribute
¶
execution_prompt_builder: Optional[Callable[..., str]] = (
field(default=None, compare=False)
)
task_markdown_builder
class-attribute
instance-attribute
¶
task_markdown_builder: Optional[Callable[..., str]] = field(
default=None, compare=False
)
distill_domain_hint
class-attribute
instance-attribute
¶
distill_domain_hint: "Optional[str | Callable[[], str]]" = (
field(default=None, compare=False)
)
loader
class-attribute
instance-attribute
¶
loader: Optional[Callable[..., object]] = field(
default=None, compare=False
)
approach_diagnosis
class-attribute
instance-attribute
¶
approach_diagnosis: Optional[Callable[..., list[str]]] = (
field(default=None, compare=False)
)
score_from_eval
class-attribute
instance-attribute
¶
score_from_eval: Optional[Callable[..., "float | None"]] = (
field(default=None, compare=False)
)
trace_condensed
class-attribute
instance-attribute
¶
trace_condensed: Optional[Callable[..., str]] = field(
default=None, compare=False
)
validate_tasks_path
class-attribute
instance-attribute
¶
validate_tasks_path: Optional[
Callable[..., "str | None"]
] = field(default=None, compare=False)
attempt_meta_builder
class-attribute
instance-attribute
¶
attempt_meta_builder: Optional[
Callable[..., "dict[str, Any] | None"]
] = field(default=None, compare=False)
all_names ¶
all_names() -> tuple[str, ...]
Canonical name plus every alias (deduped, canonical first).
__init__ ¶
__init__(
name: str,
aliases: tuple[str, ...] = (),
default_evaluator: str = "none",
prompt_kind: str = "generic",
execution_prompt_builder: Optional[
Callable[..., str]
] = None,
task_markdown_builder: Optional[
Callable[..., str]
] = None,
distill_domain_hint: "Optional[str | Callable[[], str]]" = None,
loader: Optional[Callable[..., object]] = None,
supports_mcp_arc: bool = False,
is_offline: bool = False,
uses_repo_snapshots: bool = False,
supports_classification: bool = False,
needs_eval_records: bool = False,
delegates_runtime: bool = False,
arc_task_reference: bool = False,
upstream_strict: bool = False,
approach_diagnosis: Optional[
Callable[..., list[str]]
] = None,
score_from_eval: Optional[
Callable[..., "float | None"]
] = None,
trace_condensed: Optional[Callable[..., str]] = None,
validate_tasks_path: Optional[
Callable[..., "str | None"]
] = None,
attempt_meta_builder: Optional[
Callable[..., "dict[str, Any] | None"]
] = None,
) -> None
StrategySpec
dataclass
¶
How to construct one improvement strategy. factory is called with no
arguments and returns a fresh ImprovementStrategy.
__init__ ¶
__init__(
name: str,
factory: Callable[[], ImprovementStrategy],
aliases: tuple[str, ...] = (),
description: str = "",
) -> None
Core config & data types¶
GenerationConfig
dataclass
¶
Configuration for a generational orchestrator run.
native_memory_max_chars_per_file
class-attribute
instance-attribute
¶
native_memory_max_chars_per_file: int = 60000
seed_per_task_bundles_path
class-attribute
instance-attribute
¶
seed_per_task_bundles_path: str = ''
cross_task_forum_timeout_sec
class-attribute
instance-attribute
¶
cross_task_forum_timeout_sec: int = 900
cross_task_shared_container
class-attribute
instance-attribute
¶
cross_task_shared_container: bool = False
cross_task_distill_target_conditioning
class-attribute
instance-attribute
¶
cross_task_distill_target_conditioning: bool = True
cross_task_distill_per_target_selection
class-attribute
instance-attribute
¶
cross_task_distill_per_target_selection: bool = False
distill_per_task_model
class-attribute
instance-attribute
¶
distill_per_task_model: str | None = None
distill_cross_task_model
class-attribute
instance-attribute
¶
distill_cross_task_model: str | None = None
forum_early_exit_poll_sec
class-attribute
instance-attribute
¶
forum_early_exit_poll_sec: float = 3.0
forum_early_exit_quorum_pct
class-attribute
instance-attribute
¶
forum_early_exit_quorum_pct: float = 100.0
forum_early_exit_quorum_grace_sec
class-attribute
instance-attribute
¶
forum_early_exit_quorum_grace_sec: float = 0.0
holdout_task_ids
class-attribute
instance-attribute
¶
holdout_task_ids: list[str] = field(default_factory=list)
__init__ ¶
__init__(
num_generations: int,
num_agents: int,
drop_solved: bool = True,
solved_threshold: float = 1.0,
max_concurrent_tasks: int = 25,
max_concurrent_forum_tasks: int = 0,
max_task_retries: int = 3,
knowledge_db_path: str = "",
runtime_db_path: str = "",
experiment_name: str = "ksi",
forum_timeout_sec: int = 900,
native_memory_max_chars: int = 240000,
native_memory_max_files: int = 8,
native_memory_max_chars_per_file: int = 60000,
seed_bundle_path: str = "",
seed_per_task_bundles_path: str = "",
disable_memory_mcp: bool = False,
resume: bool = False,
start_generation: int = 1,
model: str = "",
no_memory: bool = False,
code_commit: str = "",
model_provider: str = "",
scoring_mode: str = "",
per_task_forum_rounds: int = 1,
cross_task_forum_rounds: int = 2,
cross_task_forum_timeout_sec: int = 900,
cross_task_shared_container: bool = False,
distill_enabled: bool = True,
cross_task_distill_target_conditioning: bool = True,
cross_task_distill_per_target_selection: bool = False,
abort_on_distill_stall: int = 0,
distill_per_task_model: str | None = None,
distill_cross_task_model: str | None = None,
forum_early_exit: bool = False,
forum_early_exit_poll_sec: float = 3.0,
forum_early_exit_quorum_pct: float = 100.0,
forum_early_exit_quorum_grace_sec: float = 0.0,
require_vector: bool = False,
holdout_task_ids: list[str] = list(),
config_json: str = "",
) -> None
TaskSpec
dataclass
¶
TaskTrace
dataclass
¶
Result of an agent executing a single task.
eval_result
class-attribute
instance-attribute
¶
eval_result: EvalResult = field(default_factory=EvalResult)
tool_trace
class-attribute
instance-attribute
¶
tool_trace: list[dict[str, Any]] = field(
default_factory=list
)
runtime_meta
class-attribute
instance-attribute
¶
runtime_meta: dict[str, Any] = field(default_factory=dict)
token_usage
class-attribute
instance-attribute
¶
token_usage: TokenUsage = field(default_factory=TokenUsage)
repo
class-attribute
instance-attribute
¶
repo: str = ''
Upstream repo (e.g. owner/name) for the task, propagated from
TaskSpec.repo. Persistence writers use this to populate tasks.repo
even on silent-failure traces where the downstream
insert_task_summary path would otherwise be skipped.
__init__ ¶
__init__(
generation: int,
agent_id: str,
task_id: str,
model_output: str | None = None,
eval_result: EvalResult = EvalResult(),
native_score: float | None = None,
tool_trace: list[dict[str, Any]] = list(),
runtime_meta: dict[str, Any] = dict(),
token_usage: TokenUsage = TokenUsage(),
error: str | None = None,
repo: str = "",
) -> None
EvalResult ¶
Bases: TypedDict
Structured result returned by Evaluator.evaluate and stored on
TaskTrace.eval_result.
total=False so every key is optional: evaluators emit partial dicts
(e.g. an error-only dict before scoring, or a setup-failure dict), and this
type captures the known keys without forcing any to be present. This keeps
three-state fields meaningful at the type level — a key being absent is a
distinct, type-checkable state from False.
This declares the cross-evaluator keys read by generic consumers (e.g.
orchestrator.scoring, orchestrator.engine, approach_diagnosis).
Evaluator-specific scalar keys (terminal_bench_2 reward/exit codes,
swebench patch_source) are intentionally not
enumerated here — they are read only inside their own evaluator modules,
not through this generic type.