Python SDK reference ও command

Client নিরাপদে তৈরি করুন

TrainingClient API key, HTTPS base_url, ধনাত্মক সসীম timeout এবং ০ থেকে ৫-এর max_get_retries গ্রহণ করে। ডিফল্ট timeout 60 সেকেন্ড এবং GET retry-এর ডিফল্ট সংখ্যা 2। Redirect প্রত্যাখ্যাত হয়। allow_loopback শুধু স্পষ্ট স্থানীয় পরীক্ষার জন্য; অনিরাপদ remote connection-এ আসল key পাঠানোর কারণ নয়।

• Locale বদলালেও code ও environment variable-এর নাম ইংরেজিতে রাখুন।
• বিদ্যমান session integration-এর জন্য TrainingClient থাকছে।
• Schema-ভিত্তিক স্থানীয় project run-এর জন্য TrainingProject ও fpai-training ব্যবহার করুন।

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.

Public method ব্যবহার করুন

open_session ও create_session একই কাজ করে। prepare_session সীমিত preparation চালায়। get_session state দেখে। download_dataset একটি DatasetManifest যাচাই করে। post_event প্রকৃত runtime অগ্রগতি জানায়। train session training engine চালায়। complete স্থানীয়ভাবে তৈরি artifact জানায়; cancel service session state বাতিল করে। নিচের method তালিকায় বিদ্যমান integration-এর জন্য রাখা হুবহু TrainingClient signature দেখানো হয়েছে।

• reference_model_id keyword-only এবং ধনাত্মক integer হতে হবে, boolean নয়।
• max_requests preparation call সীমিত করে; এটি model-training epoch-এর সংখ্যা নয়।
• TrainingClient.train-এ session ID ও নতুন output directory লাগে; TrainingProject model path configuration থেকে নেয়।

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.

Exception ও প্রত্যাশিত output

Session operation-এ TrainingClient APIError ও IntegrityError তোলে। TrainingProject-এর ConfigurationError, CompatibilityError, AuthenticationError, BudgetExceededError, EvaluationFailedError, CheckpointError, CancellationError এবং ExportError হলো TrainingError subclass। Debug করার সময় environment বা authorization header print করবেন না। পুরোনো client method ও নতুন project API-র contract আলাদা।

• প্রতিটি run-এর সঙ্গে package ও dependency version রাখুন।
• পাওয়া গেলে TrainingProject error-এ গোপন তথ্য-বর্জিত diagnostic ও report থাকে।
• State না পড়ে state পরিবর্তনকারী operation আবার দেবেন না।

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.

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