مرجع Python SDK والأوامر

أنشئ عميلًا بأمان

يقبل TrainingClient مفتاح API وعنوان HTTPS في base_url ومهلة موجبة محدودة وmax_get_retries من 0 إلى 5. المهلة الافتراضية 60 ثانية وعدد محاولات GET الافتراضي 2. تُرفض عمليات إعادة التوجيه. يقتصر allow_loopback على الاختبار المحلي الصريح، ولا يبرر إرسال مفتاح حقيقي عبر اتصال بعيد غير آمن.

• أبقِ أسماء الشفرة ومتغيرات البيئة بالإنجليزية في جميع اللغات.
• يظل TrainingClient متاحًا لتكاملات الجلسات الحالية.
• استخدم 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.

استخدم الأساليب العامة

ينفّذ open_session وcreate_session العملية نفسها. ينفّذ prepare_session إعدادًا محدودًا. يفحص get_session الحالة. يتحقق download_dataset من DatasetManifest. يرسل post_event تقدم بيئة التشغيل الفعلي. يشغّل train محرك تدريب الجلسة. يبلّغ complete عن ملف ناتج محلي، ويلغي cancel حالة جلسة الخدمة. تعرض القائمة أدناه التواقيع الدقيقة لـTrainingClient المتاحة للتكاملات القائمة.

• يُمرّر reference_model_id كوسيط مسمّى فقط، ويجب أن يكون عددًا صحيحًا موجبًا لا قيمة منطقية.
• يحدّ max_requests من طلبات الإعداد؛ وليس عدد دورات التدريب للنموذج.
• يقبل TrainingClient.train معرّف جلسة ومجلد مخرجات جديدًا؛ أما TrainingProject فيحمّل مسار النموذج من إعداد المشروع.

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.

الاستثناءات والنواتج المتوقعة

يرفع TrainingClient في عمليات الجلسات APIError وIntegrityError. ويعرض TrainingProject أخطاء ConfigurationError وCompatibilityError وAuthenticationError وBudgetExceededError وEvaluationFailedError وCheckpointError وCancellationError وExportError بوصفها فئات فرعية من TrainingError. لا تطبع البيئة كاملة أو ترويسات التفويض عند التصحيح. وللعميل القديم وواجهة المشروع الجديدة عقدان منفصلان.

• احتفظ بإصدارات الحزمة وتبعياتها مع سجل كل تشغيل.
• تتضمن أخطاء TrainingProject تشخيصات منقحة وتقارير متى توفرت.
• اقرأ الحالة قبل إعادة أي عملية تغيّرها.

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