Référence du SDK Python et commandes

Créer un client en toute sécurité

TrainingClient accepte une clé API, une base_url HTTPS, un délai d’attente positif fini et max_get_retries entre 0 et 5. Le délai par défaut est de 60 secondes et le nombre de tentatives GET de 2. Les redirections sont refusées. allow_loopback est réservé aux tests locaux explicites ; il ne justifie jamais l’envoi d’une vraie clé sur une connexion distante non sécurisée.

• Gardez le code et les noms de variables d’environnement en anglais dans toutes les langues.
• TrainingClient conserve les méthodes de session ; TrainingProject fournit les flux de projet.
• Utilisez fpai-training pour les commandes CLI documentées et les méthodes TrainingProject en Python.

python3 -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip
# Install the GPU-compatible PyTorch build for your runtime first.
python -m pip install 'fusionpointai-training[train] @ https://fusionpointai.com/downloads/fusionpointai-training.zip'
# API-only use without local training dependencies:
# python -m pip install 'fusionpointai-training @ https://fusionpointai.com/downloads/fusionpointai-training.zip'

import os
from fusionpointai_training import TrainingClient

client = TrainingClient(
    api_key=os.environ["FUSIONPOINTAI_API_KEY"],
    base_url="https://fusionpointai.com/api/v1/training",
    timeout=60.0,
    max_get_retries=2,
)
# transport optionally accepts an httpx transport for controlled integrations.
# Never enable insecure remote HTTP to work around a TLS error.

from importlib.metadata import PackageNotFoundError, version
from importlib.util import find_spec

for package in ("fusionpointai-training", "httpx", "torch", "transformers", "peft"):
    try:
        print(f"{package}=={version(package)}")
    except PackageNotFoundError:
        print(f"{package}: not installed")

for module in ("torch", "transformers", "peft", "safetensors"):
    print(f"{module} importable:", find_spec(module) is not None)
# The train extra includes optional model-training dependencies; inspect before use.

Utiliser les méthodes publiques

open_session et create_session réalisent la même opération. prepare_session effectue une préparation limitée. get_session consulte l’état. download_dataset valide un DatasetManifest. post_event signale la progression réelle. train exécute le moteur d’entraînement local. complete signale un artefact local et cancel annule l’état du service. TrainingProject ajoute les méthodes de projet décrites ci-dessous.

• reference_model_id doit être un argument nommé et un entier positif, pas un booléen.
• max_requests limite les requêtes de préparation ; ce n’est pas le nombre d’époques d’entraînement.
• train accepte un ID de session et un nouveau répertoire de sortie, pas un chemin de modèle arbitraire.

from fusionpointai_training.client import DatasetManifest

# Continue with an authenticated client and an owned session.
ready = client.prepare_session(session["id"])
for item in ready["datasets"]:
    manifest = DatasetManifest(
        split=item["split"], sha256=item["sha256"],
        bytes=item["bytes"], rows=item["rows"],
    )
    rows = client.download_dataset(ready["id"], manifest)
    print(manifest.split, len(rows), manifest.sha256)
    # Each row has a text field. Do not log private training content.

# For a genuine external trainer integration, not simulated progress.
def report_progress(client, session_id, *, step, total_steps, loss):
    return client.post_event(session_id, {
        "state": "running",
        "step": step,
        "total_steps": total_steps,
        "metrics": {"loss": loss},
    })
# Supply values observed from your real trainer.
# The built-in train() already reports its progress; do not duplicate it.

import json
from pathlib import Path

# Continue with client/session. Inspect state before retrying reporting.
report = json.loads(Path("customer-model-run-001/training-report.json").read_text())
state = client.get_session(session["id"])
if state["status"] == "evaluating":
    state = client.complete(session["id"], report["completion_report"])
print(state["status"])
# Do not restart weight training just to reconcile reporting.

import os
from fusionpointai_training import TrainingClient

client = TrainingClient(api_key=os.environ["FUSIONPOINTAI_API_KEY"])
session = client.get_session(os.environ["FUSIONPOINTAI_SESSION_ID"])
report = session.get("report")
last_event = report.get("last_event", {}) if isinstance(report, dict) else {}
print("State:", session["status"], "Step:", session["step"])
print("Last reported error code:", last_event.get("error_code"))
# A missing error_code does not prove success. Read state and your local report.
# Do not dump full provider responses, private data or authorization headers.

Exceptions et résultats attendus

Les erreurs du projet héritent de TrainingError et correspondent aux codes de sortie documentés de la CLI. Les erreurs de configuration et de compatibilité surviennent avant l’entraînement ; les échecs de budget ou d’acceptation conservent rapports et critères non satisfaits. Un échec lors de l’arrêt des processus ou de l’enregistrement d’un point de contrôle reste une erreur tant que le nettoyage n’est pas confirmé.

• Une méthode réussie renvoie le résultat documenté. Si l’objectif n’est pas atteint, l’erreur correspondante est levée après l’enregistrement du rapport.
• Consignez la version du SDK et de ses dépendances pour chaque exécution.
• Les exemples de code exportés ne contiennent pas d’identifiants réels de compte.

import os
from fusionpointai_training import TrainingClient
from fusionpointai_training.errors import APIError, IntegrityError, TrainingError

client = TrainingClient(api_key=os.environ["FUSIONPOINTAI_API_KEY"])
try:
    state = client.get_session(os.environ["FUSIONPOINTAI_SESSION_ID"])
    print("Current state:", state["status"])
except IntegrityError:
    print("Integrity check failed; preserve the report and investigate.")
except TrainingError:
    print("Training failed; inspect the local report before retrying.")
except APIError:
    print("Request failed; inspect access and session state before another write.")
except ValueError:
    print("Invalid local configuration; correct the input before retrying.")
# Never print the key or authorization headers.

import os
from fusionpointai_training import TrainingClient
from fusionpointai_training.errors import APIError

client = TrainingClient(
    api_key=os.environ["FUSIONPOINTAI_API_KEY"],
    timeout=30.0,
    max_get_retries=2,
)
try:
    session = client.get_session(os.environ["FUSIONPOINTAI_SESSION_ID"])
except APIError:
    # GET retries are bounded. Do not blindly retry a write after an error.
    print("Session inspection failed; check access and current server state.")
else:
    print("Status:", session["status"])
# Never print the API key, authorization headers, or response secrets.

Interface Python

• TrainingClient(api_key: str, *, base_url: str = "https://fusionpointai.com/api/v1/training", allow_loopback: bool = False, timeout: float = 60.0, max_get_retries: int = 2, transport: httpx.BaseTransport | None = None)
• create_session(job: str, runtime: str, *, reference_model_id: int | None = None) -> dict
• open_session(job: str, runtime: str, *, reference_model_id: int | None = None) -> dict
• get_session(session_id: str) -> dict
• prepare_session(session_id: str, *, max_requests: int | None = None) -> dict
• DatasetManifest(split: str, sha256: str, bytes: int, rows: int)
• download_dataset(session_id: str, manifest: DatasetManifest) -> list[dict[str, str]]
• post_event(session_id: str, event: dict) -> dict
• train(session_id: str, output_dir: str | os.PathLike[str]) -> dict
• complete(session_id: str, report: dict) -> dict
• cancel(session_id: str) -> dict
• TrainingProject.from_file(path: str | os.PathLike[str], *, api_key: str) -> TrainingProject
• TrainingProject.plan(*, dry_run: bool = True) -> dict
• TrainingProject.run() -> dict
• TrainingProject.status() -> dict
• TrainingProject.pause() -> dict
• TrainingProject.resume() -> dict
• TrainingProject.cancel() -> dict
• TrainingProject.evaluate(*, split: str = "test") -> dict
• TrainingProject.export(*, format: str = "safetensors") -> dict
• register_evaluator(name: str, scorer: Callable, *, version: str) -> None