API reference¶
Generated from the apogee-ai-training source with mkdocstrings. Every symbol below is exported from apogee_ai_training, so it is part of the supported public surface.
Application · DTOs¶
dataclass
¶
dataclass
¶
dataclass
¶
dataclass
¶
Application · Use cases¶
Synthetic SFT prep: N examples + 10% duplicates, measure throughput.
async
¶
Source code in apogee_ai_training/application/use_cases/bench_prep_use_case.py
Python
async def execute(self, examples: int) -> dict[str, float]:
if examples <= 0:
raise ValueError("examples must be positive")
items: list[TrainingExample] = []
for i in range(examples):
base = i // 2 if i % 10 == 0 else i # forces ~10% duplication
items.append(TrainingExample(messages=(
{"role": "user", "content": f"pergunta {base}"},
{"role": "assistant", "content": f"resposta {base}"},
)))
start = time.perf_counter()
dataset = await PrepareDatasetUseCase().execute(items, name="bench")
elapsed = (time.perf_counter() - start) * 1000.0
return {
"examples": float(examples),
"deduped": float(len(dataset)),
"elapsed_ms": elapsed,
"examples_per_second": (examples / elapsed * 1000.0) if elapsed > 0 else 0.0,
}
Round-trips a dataset between supported formats via in-memory lines.
async
¶
Python
execute(dataset: TrainingDataset, target_format: str) -> list[str]
Source code in apogee_ai_training/application/use_cases/convert_format_use_case.py
Source code in apogee_ai_training/application/use_cases/get_job_use_case.py
async
¶
Python
execute(job_id: str) -> TrainingJob
Wraps a list of TrainingExample into a TrainingDataset.
Drops obvious duplicates (same JSON serialisation of messages or prompt/chosen/rejected).
async
¶
Python
execute(examples: Iterable[TrainingExample], name: str = 'dataset') -> TrainingDataset
Source code in apogee_ai_training/application/use_cases/prepare_dataset_use_case.py
Python
async def execute(
self,
examples: Iterable[TrainingExample],
name: str = "dataset",
) -> TrainingDataset:
seen: set[str] = set()
out: list[TrainingExample] = []
for ex in examples:
key = (
f"M::{tuple(tuple(sorted(m.items())) for m in ex.messages)}"
if ex.messages
else f"D::{ex.prompt}::{ex.chosen}::{ex.rejected}"
)
if key in seen:
continue
seen.add(key)
out.append(ex)
return TrainingDataset(name=name, examples=tuple(out))
Source code in apogee_ai_training/application/use_cases/start_job_use_case.py
async
¶
Python
execute(config: TrainingConfig, dataset: TrainingDataset) -> TrainingJob
Walk over examples to surface common shape mistakes.
async
¶
Python
execute(dataset: TrainingDataset) -> dict[str, int]
Source code in apogee_ai_training/application/use_cases/validate_dataset_use_case.py
Python
async def execute(self, dataset: TrainingDataset) -> dict[str, int]:
if len(dataset) == 0:
raise DatasetValidationError("dataset is empty")
sft_count = 0
dpo_count = 0
bad = 0
for i, ex in enumerate(dataset.examples, 1):
if ex.is_dpo:
dpo_count += 1
if not ex.prompt or not (ex.chosen and ex.rejected):
bad += 1
else:
sft_count += 1
if not ex.messages:
raise DatasetValidationError(
"SFT example without messages", line=i,
)
roles = {m.get("role") for m in ex.messages}
if "user" not in roles and "assistant" not in roles:
raise DatasetValidationError(
"SFT example missing user/assistant roles", line=i,
)
return {
"examples": len(dataset),
"sft_count": sft_count,
"dpo_count": dpo_count,
"warnings": bad,
}
Domain¶
Bases: str, Enum
Bases: str, Enum
dataclass
¶
Python
TrainingConfig(base_model: str, kind: TrainingKind = SFT, epochs: int = 3, batch_size: int = 8, learning_rate: float = 5e-05, seed: int = 42, suffix: str = '', hyperparameters: dict[str, str] = dict())
class-attribute
instance-attribute
¶
dataclass
¶
Python
TrainingDataset(name: str, examples: tuple[TrainingExample, ...] = (), metadata: dict[str, str] = dict())
class-attribute
instance-attribute
¶
dataclass
¶
Python
TrainingExample(messages: tuple[dict[str, str], ...] = (), prompt: str = '', chosen: str = '', rejected: str = '', metadata: dict[str, str] = dict())
A single SFT/DPO example.
For SFT: messages only.
For DPO: prompt plus chosen and rejected responses.
class-attribute
instance-attribute
¶
dataclass
¶
Python
TrainingJob(base_model: str, dataset_name: str, status: JobStatus = PENDING, fine_tuned_model: str = '', error_message: str = '', progress: float = 0.0, id: str = (lambda: f'job-{hex[:10]}')(), created_at_s: float = time(), metadata: dict[str, str] = dict())
Domain · Enums¶
Domain · Exceptions¶
Bases: TrainingError
Bases: Exception
Base for apogee-ai-training errors.
Domain · Protocols (ports)¶
Bases: Protocol
Python
to_lines(dataset: TrainingDataset) -> list[str]
Python
from_lines(lines: list[str], dataset_name: str) -> TrainingDataset
Bases: Protocol
async
¶
Python
save(job: TrainingJob) -> None
async
¶
Python
get(job_id: str) -> TrainingJob
async
¶
Python
list() -> Iterable[TrainingJob]
Bases: Protocol
async
¶
Python
start(config: TrainingConfig, dataset: TrainingDataset) -> TrainingJob
async
¶
Python
status(job_id: str) -> TrainingJob
async
¶
Infrastructure¶
Alpaca JSONL: {"instruction", "input", "output"} per line.
Python
to_lines(dataset: TrainingDataset) -> list[str]
Source code in apogee_ai_training/infrastructure/formats/alpaca_converter.py
Python
def to_lines(self, dataset: TrainingDataset) -> list[str]:
out: list[str] = []
for ex in dataset.examples:
if not ex.messages or len(ex.messages) < 2:
raise DatasetValidationError(
"Alpaca export requires user → assistant pair",
)
user = next((m for m in ex.messages if m.get("role") == "user"), None)
assistant = next(
(m for m in ex.messages if m.get("role") == "assistant"), None
)
if user is None or assistant is None:
raise DatasetValidationError(
"missing user or assistant message",
)
out.append(json.dumps({
"instruction": user.get("content", ""),
"input": "",
"output": assistant.get("content", ""),
}))
return out
Python
from_lines(lines: list[str], dataset_name: str) -> TrainingDataset
Source code in apogee_ai_training/infrastructure/formats/alpaca_converter.py
Python
def from_lines(self, lines: list[str], dataset_name: str) -> TrainingDataset:
examples: list[TrainingExample] = []
for i, line in enumerate(lines, 1):
line = line.strip()
if not line:
continue
try:
payload = json.loads(line)
except json.JSONDecodeError as exc:
raise DatasetValidationError(f"invalid json: {exc.msg}", line=i) from exc
instruction = payload.get("instruction", "")
inp = payload.get("input", "")
output = payload.get("output", "")
if not instruction or not output:
raise DatasetValidationError(
"instruction and output are required", line=i,
)
user_content = f"{instruction}\n\n{inp}".strip()
examples.append(TrainingExample(messages=(
{"role": "user", "content": user_content},
{"role": "assistant", "content": output},
)))
return TrainingDataset(name=dataset_name, examples=tuple(examples))
ChatML JSONL: each line {"messages": [{"role": ..., "content": ...}]}.
Python
to_lines(dataset: TrainingDataset) -> list[str]
Python
from_lines(lines: list[str], dataset_name: str) -> TrainingDataset
Source code in apogee_ai_training/infrastructure/formats/chatml_converter.py
Python
def from_lines(self, lines: list[str], dataset_name: str) -> TrainingDataset:
examples: list[TrainingExample] = []
for i, line in enumerate(lines, 1):
line = line.strip()
if not line:
continue
try:
payload = json.loads(line)
except json.JSONDecodeError as exc:
raise DatasetValidationError(f"invalid json: {exc.msg}", line=i) from exc
messages = payload.get("messages")
if not isinstance(messages, list) or not messages:
raise DatasetValidationError("missing or empty messages", line=i)
examples.append(TrainingExample(messages=tuple(messages)))
return TrainingDataset(name=dataset_name, examples=tuple(examples))
In-memory trainer for CI/tests. Marks job SUCCEEDED immediately.
Source code in apogee_ai_training/infrastructure/trainers/dry_run_trainer.py
async
¶
Python
start(config: TrainingConfig, dataset: TrainingDataset) -> TrainingJob
Source code in apogee_ai_training/infrastructure/trainers/dry_run_trainer.py
Python
async def start(
self, config: TrainingConfig, dataset: TrainingDataset
) -> TrainingJob:
suffix = f"-{config.suffix}" if config.suffix else ""
fine_tuned_model = f"{config.base_model}-ft{suffix}"
job = TrainingJob(
base_model=config.base_model,
dataset_name=dataset.name,
status=JobStatus.SUCCEEDED,
fine_tuned_model=fine_tuned_model,
progress=1.0,
metadata={
"kind": config.kind.value,
"epochs": str(config.epochs),
"batch_size": str(config.batch_size),
"examples": str(len(dataset)),
},
)
self._jobs[job.id] = job
return job
async
¶
Python
status(job_id: str) -> TrainingJob
async
¶
HuggingFace DPO JSONL: {"prompt", "chosen", "rejected"} per line.
Python
to_lines(dataset: TrainingDataset) -> list[str]
Source code in apogee_ai_training/infrastructure/formats/dpo_converter.py
Python
def to_lines(self, dataset: TrainingDataset) -> list[str]:
out: list[str] = []
for ex in dataset.examples:
if not ex.is_dpo:
raise DatasetValidationError(
"example missing prompt/chosen/rejected for HF DPO export",
)
out.append(json.dumps({
"prompt": ex.prompt,
"chosen": ex.chosen,
"rejected": ex.rejected,
}))
return out
Python
from_lines(lines: list[str], dataset_name: str) -> TrainingDataset
Source code in apogee_ai_training/infrastructure/formats/dpo_converter.py
Python
def from_lines(self, lines: list[str], dataset_name: str) -> TrainingDataset:
examples: list[TrainingExample] = []
for i, line in enumerate(lines, 1):
line = line.strip()
if not line:
continue
try:
payload = json.loads(line)
except json.JSONDecodeError as exc:
raise DatasetValidationError(f"invalid json: {exc.msg}", line=i) from exc
for required in ("prompt", "chosen", "rejected"):
if required not in payload:
raise DatasetValidationError(
f"missing required field {required!r}", line=i,
)
examples.append(TrainingExample(
prompt=str(payload["prompt"]),
chosen=str(payload["chosen"]),
rejected=str(payload["rejected"]),
))
return TrainingDataset(name=dataset_name, examples=tuple(examples))
Source code in apogee_ai_training/infrastructure/registries/in_memory_job_repository.py
async
¶
Python
save(job: TrainingJob) -> None
async
¶
Python
get(job_id: str) -> TrainingJob
async
¶
Python
list() -> Iterable[TrainingJob]
Loads a JSONL file into a TrainingDataset using the chosen format.
Python
load(path: str | Path, fmt: str = 'chatml', dataset_name: str = '') -> TrainingDataset
Source code in apogee_ai_training/infrastructure/datasets/jsonl_loader.py
Python
def load(self, path: str | Path, fmt: str = "chatml", dataset_name: str = "") -> TrainingDataset:
if fmt not in _CONVERTERS:
raise DatasetValidationError(f"unknown format: {fmt!r}")
path = Path(path)
if not path.is_file():
raise DatasetValidationError(f"file not found: {path}")
lines = path.read_text(encoding="utf-8").splitlines()
converter = _CONVERTERS[fmt]()
return converter.from_lines(lines, dataset_name or path.stem)
OpenAI fine-tune SFT: same shape as ChatML.
Python
to_lines(dataset: TrainingDataset) -> list[str]
Python
from_lines(lines: list[str], dataset_name: str) -> TrainingDataset
Source code in apogee_ai_training/infrastructure/formats/openai_sft_converter.py
Python
def from_lines(self, lines: list[str], dataset_name: str) -> TrainingDataset:
examples: list[TrainingExample] = []
for i, line in enumerate(lines, 1):
line = line.strip()
if not line:
continue
try:
payload = json.loads(line)
except json.JSONDecodeError as exc:
raise DatasetValidationError(f"invalid json: {exc.msg}", line=i) from exc
messages = payload.get("messages")
if not isinstance(messages, list) or not messages:
raise DatasetValidationError("missing or empty messages", line=i)
for j, m in enumerate(messages):
if not isinstance(m, dict) or "role" not in m or "content" not in m:
raise DatasetValidationError(
f"message[{j}] requires role + content", line=i,
)
examples.append(TrainingExample(messages=tuple(messages)))
return TrainingDataset(name=dataset_name, examples=tuple(examples))
Lazy OpenAI fine-tunes adapter — install via extras=openai.
Source code in apogee_ai_training/infrastructure/trainers/openai_trainer.py
async
¶
Python
start(config: TrainingConfig, dataset: TrainingDataset) -> TrainingJob
Source code in apogee_ai_training/infrastructure/trainers/openai_trainer.py
Python
async def start(
self, config: TrainingConfig, dataset: TrainingDataset
) -> TrainingJob: # pragma: no cover
self._ensure_client()
# Real implementation would upload the JSONL file then call
# client.fine_tuning.jobs.create(...). Network-dependent, skipped in CI.
raise TrainerNotAvailableException(
"OpenAI training requires network access; not exercised in CI"
)
async
¶
Python
status(job_id: str) -> TrainingJob
Source code in apogee_ai_training/infrastructure/trainers/openai_trainer.py
Python
async def status(self, job_id: str) -> TrainingJob: # pragma: no cover
self._ensure_client()
try:
payload = await self._client.fine_tuning.jobs.retrieve(job_id) # type: ignore[union-attr]
except Exception as exc:
raise JobNotFoundException(job_id) from exc
return TrainingJob(
base_model=payload.model,
dataset_name=str(payload.training_file or ""),
status=_OPENAI_STATUS_MAP.get(payload.status, JobStatus.PENDING),
fine_tuned_model=payload.fine_tuned_model or "",
id=payload.id,
)
async
¶
Python
write_jsonl(dataset: TrainingDataset, path: str | Path, fmt: str = 'chatml') -> Path
Source code in apogee_ai_training/infrastructure/datasets/jsonl_loader.py
Python
def write_jsonl(dataset: TrainingDataset, path: str | Path, fmt: str = "chatml") -> Path:
if fmt not in _CONVERTERS:
raise DatasetValidationError(f"unknown format: {fmt!r}")
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
converter = _CONVERTERS[fmt]()
lines = converter.to_lines(dataset)
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
return path