Un job care rezumă sute de mii de documente nu ar trebui să reîncarce modelul și să refacă întregul corpus după ce cade ultimul worker. În proiectul de mai jos, corpusul Wikipedia în limba română este materializat în shard-uri Parquet, fiecare shard primește propriul rezultat și propriul checkpoint, iar o execuție nouă sare peste unitățile deja validate.
La final rulează un pipeline local cu Ray Data 2.59.0 și vLLM 0.8.5. Modelul Qwen/Qwen2.5-0.5B-Instruct generează rezumate scurte, cu decodare greedy, iar scriptul de evaluare verifică numărul de documente, identificatorii duplicați, rezultatele goale și integritatea fișierelor. Scenariul folosește un singur GPU, dar GPU_WORKERS permite alocarea mai multor actori pe o gazdă cu mai multe plăci.
Checkpointul de aici este un manifest durabil la limita dintre shard-uri, nu starea internă a fiecărui token generat. Dacă procesul se oprește în mijlocul unui shard, acel shard este refăcut; cele finalizate și validate sunt păstrate. Este o problemă diferită de reluarea execuției unui agent LangGraph din checkpoints, unde se salvează starea unui graf conversațional.
Fixăm platforma pe care rezultatele pot fi repetate
Configurația de referință este Ubuntu Server 24.04 LTS, Python 3.12, Ray 2.59.0, vLLM 0.8.5 și un GPU NVIDIA cu minimum 8 GB VRAM. Sunt necesari aproximativ 20 GB liberi pentru dataset, model, fișiere temporare și rezultate. Pregătirea durează în funcție de conexiune, iar inferența poate dura de la zeci de minute la câteva ore; scriptul raportează timpul real, fără să presupună o anumită performanță a GPU-ului.
Modelul ales are aproximativ 0,5 miliarde de parametri și licență Apache 2.0 conform model cardului. Corpusul este configurația 20231101.ro a datasetului wikimedia/wikipedia, distribuită în Parquet. Dataset cardul indică licențele CC BY-SA 3.0 și GFDL pentru conținutul Wikipedia. Dimensiunea exactă descărcată este calculată în bytes și salvată în source-lock.json, deoarece fișierele găzduite și metadatele lor nu trebuie estimate din memorie.
Verifică gazda înainte de instalare:
cat /etc/os-release
python3 --version
nvidia-smi
df -h .
free -hUn exemplu de mediu potrivit arată astfel; driverul, modelul GPU și spațiul disponibil vor fi diferite:
PRETTY_NAME="Ubuntu 24.04 LTS"
Python 3.12.x
NVIDIA-SMI ... Driver Version: ...
Filesystem Size Used Avail Use% Mounted on
... ... ... 20G ... ...nvidia-smi trebuie să afișeze GPU-ul fără eroare. Nu instala manual un toolkit CUDA doar pentru acest proiect înainte să verifici instrucțiunile vLLM pentru combinația de driver și wheel folosită.
Instalăm versiunile fixe și rezervăm directoarele durabile
Vom lucra în ~/ray-llm-batch. Căile data/input-shards, data/output și data/checkpoints trebuie să se afle pe stocare persistentă. Într-un cluster, aceeași cale trebuie montată pe toate nodurile; documentația Ray avertizează că scrierea pe un filesystem local partajat presupune ca acesta să fie disponibil la aceeași adresă pentru fiecare worker.
Creează mediul și fișierul requirements.txt:
mkdir -p "$HOME/ray-llm-batch"
cd "$HOME/ray-llm-batch"
python3 -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip
cat > requirements.txt <<'EOF'
huggingface-hub==0.30.2
pandas==2.2.3
pyarrow==19.0.1
ray[data]==2.59.0
vllm==0.8.5
EOF
python -m pip install -r requirements.txt
mkdir -p scripts data/raw data/input-shards data/output data/checkpointsVerifică versiunile importate:
python - <<'PY'
import pyarrow
import ray
import vllm
print("Ray:", ray.__version__)
print("vLLM:", vllm.__version__)
print("PyArrow:", pyarrow.__version__)
PYOutputul așteptat este:
Ray: 2.59.0
vLLM: 0.8.5
PyArrow: 19.0.1Punct de control: dacă vllm nu se importă sau raportează o incompatibilitate CUDA, oprește instalarea și confruntă driverul cu pagina oficială de instalare vLLM; nu compensa eroarea prin versiuni schimbate arbitrar.
Așezăm corpusul, shard-urile și manifestele într-o structură previzibilă
Proiectul va avea următoarea structură:
ray-llm-batch/
├── requirements.txt # versiunile Python fixate
├── scripts/
│ ├── prepare_corpus.py # descarcă snapshotul și creează shard-urile
│ ├── run_inference.py # rulează vLLM și scrie checkpointuri atomice
│ └── evaluate_outputs.py # verifică acoperirea și integritatea
└── data/
├── raw/ # snapshotul original Hugging Face
├── input-shards/ # shard-uri Parquet stabile
│ └── source-lock.json # configurația, dimensiunea și hashurile sursei
├── output/ # câte un director Parquet per shard
└── checkpoints/ # câte un manifest JSON per shard finalizatSepararea este intenționată. Fișierele originale rămân nemodificate, shard-urile de intrare sunt materializate o singură dată, iar un rezultat devine reutilizabil numai după apariția manifestului său în data/checkpoints.
Pasul 1 — Materializăm Wikipedia în 16 shard-uri Parquet stabile
Acest pas descarcă numai configurația română declarată, citește coloanele necesare cu Ray Data și scrie 16 unități de lucru. La final, fișierele primesc nume stabile, iar hashul SHA-256 al fiecăruia intră în lockfile.
Fișierul scripts/prepare_corpus.py:
from __future__ import annotations
import argparse
import hashlib
import json
import os
from pathlib import Path
from typing import Iterable
import pyarrow.parquet as pq
import ray
from huggingface_hub import snapshot_download
REPO_ID = "wikimedia/wikipedia"
CONFIG = "20231101.ro"
def sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def total_size(paths: Iterable[Path]) -> int:
return sum(path.stat().st_size for path in paths)
def atomic_json(path: Path, payload: dict) -> None:
temporary = path.with_suffix(path.suffix + ".tmp")
temporary.write_text(
json.dumps(payload, ensure_ascii=False, indent=2) + "n",
encoding="utf-8",
)
os.replace(temporary, path)
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--shards", type=int, default=16)
parser.add_argument("--raw-dir", type=Path, default=Path("data/raw"))
parser.add_argument(
"--shard-dir", type=Path, default=Path("data/input-shards")
)
args = parser.parse_args()
if args.shards < 1:
raise SystemExit("--shards trebuie să fie cel puțin 1")
lock_path = args.shard_dir / "source-lock.json"
if lock_path.exists():
raise SystemExit(
f"{lock_path} există deja. Nu rescriu o sursă fixată."
)
args.raw_dir.mkdir(parents=True, exist_ok=True)
args.shard_dir.mkdir(parents=True, exist_ok=True)
snapshot_path = Path(
snapshot_download(
repo_id=REPO_ID,
repo_type="dataset",
allow_patterns=[f"{CONFIG}/*.parquet"],
local_dir=args.raw_dir,
)
)
raw_files = sorted((snapshot_path / CONFIG).glob("*.parquet"))
if not raw_files:
raise SystemExit(
f"Nu am găsit fișiere Parquet în {snapshot_path / CONFIG}"
)
ray.init()
dataset = ray.data.read_parquet(
[str(path) for path in raw_files],
columns=["id", "url", "title", "text"],
)
row_count = dataset.count()
if row_count == 0:
raise SystemExit("Datasetul descărcat nu conține rânduri")
temporary_dir = args.shard_dir / ".building"
if temporary_dir.exists():
raise SystemExit(
f"{temporary_dir} există; șterge-l numai după ce verifici "
"că nu rulează altă pregătire."
)
dataset.repartition(args.shards, shuffle=False).write_parquet(
str(temporary_dir)
)
generated = sorted(temporary_dir.glob("*.parquet"))
if not generated:
raise SystemExit("Ray Data nu a produs niciun shard Parquet")
shard_records = []
for index, generated_path in enumerate(generated):
final_path = args.shard_dir / f"input-{index:05d}.parquet"
os.replace(generated_path, final_path)
metadata = pq.ParquetFile(final_path).metadata
shard_records.append(
{
"name": final_path.name,
"rows": metadata.num_rows,
"bytes": final_path.stat().st_size,
"sha256": sha256_file(final_path),
}
)
temporary_dir.rmdir()
raw_bytes = total_size(raw_files)
shard_bytes = total_size(
args.shard_dir / record["name"] for record in shard_records
)
atomic_json(
lock_path,
{
"repo_id": REPO_ID,
"config": CONFIG,
"licenses": ["CC BY-SA 3.0", "GFDL"],
"raw_files": len(raw_files),
"raw_bytes": raw_bytes,
"rows": row_count,
"requested_shards": args.shards,
"created_shards": len(shard_records),
"shard_bytes": shard_bytes,
"shards": shard_records,
},
)
print(f"Rânduri: {row_count}")
print(f"Bytes descărcați: {raw_bytes}")
print(f"Shard-uri create: {len(shard_records)}")
print(f"Lockfile: {lock_path}")
if __name__ == "__main__":
main()Rulează:
source .venv/bin/activate
python scripts/prepare_corpus.py --shards 16repartition(..., shuffle=False) schimbă numărul de blocuri fără redistribuirea aleatorie a fiecărui rând. Numărul fișierelor create este citit din rezultat, nu presupus. Lockfile-ul reține atât numărul solicitat, cât și numărul efectiv.
Outputul va avea aceeași formă, dar valorile provin din snapshotul descărcat:
Rânduri: <numărul citit din Parquet>
Bytes descărcați: <dimensiunea exactă>
Shard-uri create: 16
Lockfile: data/input-shards/source-lock.jsonPunct de control: created_shards din lockfile trebuie să corespundă fișierelor input-*.parquet, iar suma câmpurilor rows trebuie să fie egală cu rows.
Pasul 2 — Legăm fiecare actor Ray de un motor vLLM
Inferența este implementată printr-o clasă transmisă către Dataset.map_batches. Pentru o clasă callable, concurrency creează un actor pool; fiecare actor rezervă un GPU prin num_gpus=1 și încarcă o singură copie a modelului.
Fișierul scripts/run_inference.py:
from __future__ import annotations
import argparse
import hashlib
import json
import os
import shutil
import time
from pathlib import Path
import pandas as pd
import pyarrow.dataset as pads
import pyarrow.parquet as pq
import ray
from transformers import AutoTokenizer
from vllm import LLM, SamplingParams
MODEL_ID = "Qwen/Qwen2.5-0.5B-Instruct"
PROMPT_LIMIT_CHARS = 10_000
def sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def tree_digest(path: Path) -> str:
digest = hashlib.sha256()
files = sorted(item for item in path.rglob("*") if item.is_file())
for item in files:
digest.update(item.relative_to(path).as_posix().encode("utf-8"))
with item.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def atomic_json(path: Path, payload: dict) -> None:
temporary = path.with_suffix(path.suffix + ".tmp")
temporary.write_text(
json.dumps(payload, ensure_ascii=False, indent=2) + "n",
encoding="utf-8",
)
os.replace(temporary, path)
def baseline(text: str) -> str:
normalized = " ".join(text.split())
first = normalized.split(". ", maxsplit=1)[0]
return first[:280]
class Summarizer:
def __init__(self) -> None:
self.tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)
self.engine = LLM(
model=MODEL_ID,
tensor_parallel_size=1,
max_model_len=4096,
gpu_memory_utilization=0.75,
seed=47,
)
self.sampling = SamplingParams(
temperature=0.0,
max_tokens=96,
seed=47,
)
def __call__(self, batch: pd.DataFrame) -> pd.DataFrame:
prompts = []
baselines = []
for row in batch.itertuples(index=False):
text = str(row.text)
title = str(row.title)
messages = [
{
"role": "system",
"content": (
"Rezumi informația fără a adăuga fapte. "
"Răspunde în limba română, în maximum trei "
"propoziții."
),
},
{
"role": "user",
"content": (
f"Titlu: {title}nn"
f"Text:n{text[:PROMPT_LIMIT_CHARS]}"
),
},
]
prompt = self.tokenizer.apply_chat_template(
messages,
tokenize=False,
add_generation_prompt=True,
)
prompts.append(prompt)
baselines.append(baseline(text))
generated = self.engine.generate(
prompts,
self.sampling,
use_tqdm=False,
)
summaries = []
prompt_tokens = []
output_tokens = []
for result in generated:
candidate = result.outputs[0]
summaries.append(candidate.text.strip())
prompt_tokens.append(len(result.prompt_token_ids))
output_tokens.append(len(candidate.token_ids))
return pd.DataFrame(
{
"id": batch["id"].astype(str).tolist(),
"url": batch["url"].astype(str).tolist(),
"title": batch["title"].astype(str).tolist(),
"baseline": baselines,
"llm_summary": summaries,
"prompt_tokens": prompt_tokens,
"output_tokens": output_tokens,
"model": [MODEL_ID] * len(batch),
}
)
def checkpoint_valid(
checkpoint_path: Path,
output_dir: Path,
input_hash: str,
) -> bool:
if not checkpoint_path.exists() or not output_dir.is_dir():
return False
try:
payload = json.loads(checkpoint_path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
return False
return (
payload.get("status") == "complete"
and payload.get("input_sha256") == input_hash
and payload.get("model") == MODEL_ID
and payload.get("output_sha256") == tree_digest(output_dir)
)
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--batch-size", type=int, default=32)
parser.add_argument("--gpu-workers", type=int, default=1)
parser.add_argument("--stop-after", type=int, default=0)
parser.add_argument(
"--input-dir", type=Path, default=Path("data/input-shards")
)
parser.add_argument(
"--output-dir", type=Path, default=Path("data/output")
)
parser.add_argument(
"--checkpoint-dir",
type=Path,
default=Path("data/checkpoints"),
)
args = parser.parse_args()
if args.batch_size < 1 or args.gpu_workers < 1:
raise SystemExit("batch-size și gpu-workers trebuie să fie pozitive")
shards = sorted(args.input_dir.glob("input-*.parquet"))
if not shards:
raise SystemExit("Nu există shard-uri de intrare")
args.output_dir.mkdir(parents=True, exist_ok=True)
args.checkpoint_dir.mkdir(parents=True, exist_ok=True)
ray.init()
completed_now = 0
for shard in shards:
shard_name = shard.stem
final_dir = args.output_dir / shard_name
temporary_dir = args.output_dir / f".{shard_name}.tmp"
checkpoint_path = args.checkpoint_dir / f"{shard_name}.json"
input_hash = sha256_file(shard)
if checkpoint_valid(checkpoint_path, final_dir, input_hash):
print(f"SKIP {shard_name}: checkpoint valid")
continue
if temporary_dir.exists():
shutil.rmtree(temporary_dir)
if final_dir.exists():
shutil.rmtree(final_dir)
if checkpoint_path.exists():
checkpoint_path.unlink()
expected_rows = pq.ParquetFile(shard).metadata.num_rows
started = time.time()
dataset = ray.data.read_parquet(str(shard))
result = dataset.map_batches(
Summarizer,
batch_format="pandas",
batch_size=args.batch_size,
concurrency=args.gpu_workers,
num_gpus=1,
)
try:
result.write_parquet(str(temporary_dir))
output_rows = pads.dataset(
temporary_dir, format="parquet"
).count_rows()
if output_rows != expected_rows:
raise RuntimeError(
f"Rânduri ieșire={output_rows}, "
f"așteptate={expected_rows}"
)
os.replace(temporary_dir, final_dir)
output_hash = tree_digest(final_dir)
elapsed = round(time.time() - started, 3)
atomic_json(
checkpoint_path,
{
"status": "complete",
"shard": shard.name,
"input_sha256": input_hash,
"output_sha256": output_hash,
"rows": output_rows,
"model": MODEL_ID,
"batch_size": args.batch_size,
"gpu_workers": args.gpu_workers,
"elapsed_seconds": elapsed,
},
)
except BaseException:
if temporary_dir.exists():
shutil.rmtree(temporary_dir)
raise
completed_now += 1
print(
f"DONE {shard_name}: rows={output_rows}, "
f"seconds={elapsed}"
)
if args.stop_after and completed_now >= args.stop_after:
print("STOP controlat: reia cu aceeași comandă fără --stop-after")
return
if __name__ == "__main__":
main()temperature=0.0 selectează decodarea greedy, iar seed=47 fixează sursa de aleatoriu folosită de motor. Aceste setări reduc variația, dar reproducerea bit-cu-bit cere și același GPU, driver și același stack software; ordinea operațiilor floating-point poate diferi între platforme.
max_model_len=4096 limitează contextul motorului. Textul sursă este tăiat la 10.000 de caractere înainte de tokenizare, iar răspunsul la 96 de tokeni. Aceste limite controlează memoria, dar înseamnă că rezumatul nu acoperă partea unui articol aflată după prag. Pentru documente care trebuie rezumate integral este necesară o etapă separată de chunking și agregare.
Punct de control: scriptul trebuie să pornească un actor Ray cu o rezervare GPU. Dacă Ray raportează că nu poate programa actorul, compară ray.cluster_resources() cu --gpu-workers.
Pasul 3 — Întrerupem controlat jobul după două checkpointuri
Mai întâi procesăm două shard-uri și oprim driverul intenționat:
source .venv/bin/activate
python scripts/run_inference.py
--batch-size 32
--gpu-workers 1
--stop-after 2Primele descărcări ale modelului pot dura. După finalizarea celor două unități, outputul are forma:
DONE input-00000: rows=<rânduri>, seconds=<durată>
DONE input-00001: rows=<rânduri>, seconds=<durată>
STOP controlat: reia cu aceeași comandă fără --stop-afterFiecare rezultat este scris întâi într-un director ascuns temporar. Directorul este redenumit în forma finală numai după verificarea numărului de rânduri. Manifestul JSON este scris ultimul, prin os.replace, astfel încât un fișier parțial să nu fie confundat cu un checkpoint complet.
Ray poate reîncerca anumite taskuri când moare un worker, conform mecanismelor sale de fault tolerance. Manifestele adaugă însă reluarea după moartea driverului, restartarea gazdei sau o nouă lansare a scriptului. Ele nu înlocuiesc stocarea persistentă și nici nu fac un filesystem local vizibil automat pe toate nodurile.
Punct de control: trebuie să existe două directoare în data/output și două fișiere JSON în data/checkpoints; un director .tmp nu reprezintă rezultat valid.
Pasul 4 — Reluăm și demonstrăm că shard-urile valide nu sunt recalculate
Rulează din nou fără limita de oprire:
python scripts/run_inference.py
--batch-size 32
--gpu-workers 1Înainte să sară peste un shard, scriptul recalculează hashul intrării și digestul întregului director de ieșire. Schimbarea sursei, ștergerea unui fișier Parquet sau modificarea rezultatului invalidează checkpointul.
Primele linii trebuie să indice reutilizarea:
SKIP input-00000: checkpoint valid
SKIP input-00001: checkpoint valid
DONE input-00002: rows=<rânduri>, seconds=<durată>Dacă procesul cade în timpul input-00002, la următoarea lansare sunt păstrate primele două shard-uri, directorul temporar al celui de-al treilea este șters și numai acesta este reluat. Granularitatea recuperării este astfel un shard.
Schimbarea modelului în cod invalidează manifestele, deoarece câmpul model trebuie să fie identic. Schimbarea doar a batch-size nu invalidează rezultatele existente: decodarea este greedy, iar conținutul logic nu ar trebui să depindă de dimensiunea batchului. Pentru un audit strict al configurației, poți extinde manifestul cu hashul promptului și cu toate argumentele motorului.
Același principiu apare și în evaluarea agenților AI cu sarcini și scoruri verificabile: existența unui răspuns nu dovedește singură că rezultatul este complet sau corect.
Punct de control: la final trebuie să existe câte un manifest complete pentru fiecare fișier enumerat în source-lock.json.
Pasul 5 — Comparăm baseline-ul și inferența fără să inventăm un scor de calitate
Baseline-ul păstrează prima propoziție, limitată la 280 de caractere. Nu este un rezumat semantic bun, dar oferă un reper reproductibil. Evaluarea automată verifică integritatea și caracterul nenul al ieșirilor; nu pretinde că lungimea mai mică înseamnă fidelitate mai bună.
Fișierul scripts/evaluate_outputs.py:
from __future__ import annotations
import json
from pathlib import Path
import pyarrow.compute as pc
import pyarrow.dataset as pads
def main() -> None:
lock_path = Path("data/input-shards/source-lock.json")
output_root = Path("data/output")
checkpoint_root = Path("data/checkpoints")
if not lock_path.exists():
raise SystemExit("Lipsește source-lock.json")
lock = json.loads(lock_path.read_text(encoding="utf-8"))
expected_rows = int(lock["rows"])
expected_shards = {
Path(item["name"]).stem for item in lock["shards"]
}
completed = set()
checkpoint_rows = 0
for path in sorted(checkpoint_root.glob("input-*.json")):
payload = json.loads(path.read_text(encoding="utf-8"))
if payload.get("status") == "complete":
completed.add(path.stem)
checkpoint_rows += int(payload["rows"])
missing = sorted(expected_shards - completed)
if missing:
raise SystemExit(f"Checkpointuri lipsă: {', '.join(missing)}")
dataset = pads.dataset(output_root, format="parquet")
table = dataset.to_table(
columns=[
"id",
"baseline",
"llm_summary",
"prompt_tokens",
"output_tokens",
]
)
actual_rows = table.num_rows
distinct_ids = pc.count_distinct(table["id"]).as_py()
empty_summaries = pc.sum(
pc.equal(pc.utf8_length(table["llm_summary"]), 0)
).as_py()
mean_baseline_chars = pc.mean(
pc.utf8_length(table["baseline"])
).as_py()
mean_llm_chars = pc.mean(
pc.utf8_length(table["llm_summary"])
).as_py()
mean_output_tokens = pc.mean(table["output_tokens"]).as_py()
checks = {
"rows_match_lock": actual_rows == expected_rows,
"rows_match_checkpoints": actual_rows == checkpoint_rows,
"ids_are_unique": distinct_ids == actual_rows,
"summaries_are_nonempty": empty_summaries == 0,
}
print("| Verificare | Valoare |")
print("|---|---:|")
print(f"| Rânduri așteptate | {expected_rows} |")
print(f"| Rânduri rezultate | {actual_rows} |")
print(f"| ID-uri distincte | {distinct_ids} |")
print(f"| Rezumate goale | {empty_summaries} |")
print(f"| Medie caractere baseline | {mean_baseline_chars:.2f} |")
print(f"| Medie caractere LLM | {mean_llm_chars:.2f} |")
print(f"| Medie tokeni generați | {mean_output_tokens:.2f} |")
failed = [name for name, passed in checks.items() if not passed]
if failed:
raise SystemExit("FAIL: " + ", ".join(failed))
print("PASS: acoperire, unicitate și rezultate nenule")
if __name__ == "__main__":
main()Rulează testul de capăt:
python scripts/evaluate_outputs.pyTabelul conține valorile calculate pe execuția ta:
| Verificare | Valoare |
|---|---:|
| Rânduri așteptate | <număr> |
| Rânduri rezultate | <același număr> |
| ID-uri distincte | <același număr> |
| Rezumate goale | 0 |
| Medie caractere baseline | <valoare măsurată> |
| Medie caractere LLM | <valoare măsurată> |
| Medie tokeni generați | <valoare măsurată> |
PASS: acoperire, unicitate și rezultate nenuleAcesta este testul reproductibil al pipeline-ului, nu o metrică de adevăr factual. Pentru acceptarea editorială trebuie extras un eșantion, comparat cu textul sursă și etichetate omisiunile sau afirmațiile adăugate. O rubrică mai fină poate prelua ideea de evaluare pe etape din articolul despre Process Reward Model și verificarea separată a pașilor.
Punct de control: execuția este acceptată numai dacă se termină cu PASS; simpla prezență a directoarelor Parquet nu este suficientă.
Cinci defecțiuni specifice acestui pipeline și remedierea lor
ValueError: No available memory for the cache blocks
vLLM nu mai are suficientă memorie după încărcarea modelului. Oprește alte procese GPU, confirmă consumul cu nvidia-smi, apoi reduce batchul:
nvidia-smi
python scripts/run_inference.py --batch-size 8 --gpu-workers 1Nu porni doi workeri pe un singur GPU: fiecare actor încearcă să încarce propria copie a modelului.
ValueError: The model's max seq len ... is larger than the maximum number of tokens that can be stored in KV cache
Memoria disponibilă nu susține max_model_len=4096 cu configurația curentă. Verifică mai întâi că GPU-ul este liber. Dacă placa rămâne insuficientă, modifică valoarea din constructor la 2048, șterge outputurile și checkpointurile produse cu configurația veche, apoi rulează din nou. O limită mai mică trebuie tratată ca altă configurație de inferență.
The actor or task is pending and cannot currently be scheduled
--gpu-workers cere mai multe GPU-uri decât raportează Ray. Verifică resursele:
python - <<'PY'
import ray
ray.init()
print(ray.cluster_resources())
PYPentru o singură placă, folosește:
python scripts/run_inference.py --batch-size 32 --gpu-workers 1Rânduri ieșire=..., așteptate=...
Scrierea Parquet s-a încheiat fără același număr de rânduri ca shard-ul sursă. Scriptul șterge directorul temporar și nu emite checkpoint. Verifică spațiul și integritatea intrării:
df -h data
python - <<'PY'
import pyarrow.parquet as pq
print(pq.ParquetFile("data/input-shards/input-00000.parquet").metadata)
PYRulează din nou după remediere; shard-urile cu checkpoint valid vor fi omise.
Checkpointuri lipsă: input-...
Evaluarea a fost pornită înainte ca toate shard-urile să fie finalizate sau după eliminarea unui manifest. Reia inferența:
python scripts/run_inference.py --batch-size 32 --gpu-workers 1
python scripts/evaluate_outputs.pyNu crea manual un JSON complete. Manifestul conține hashul intrării și digestul rezultatului și trebuie emis numai de script după verificarea numărului de rânduri.
Ștergem rezultatele sau refacem sursa fără să lăsăm stări amestecate
Pentru a relua inferența de la zero, păstrând corpusul materializat:
rm -rf data/output data/checkpoints
mkdir -p data/output data/checkpointsPentru a elimina și shard-urile, inclusiv lockfile-ul:
rm -rf data/input-shards
mkdir -p data/input-shardsPentru curățarea completă a proiectului:
cd "$HOME"
rm -rf "$HOME/ray-llm-batch"Avertisment: comenzile rm -rf șterg definitiv rezultatele și checkpointurile. Verifică valoarea lui $HOME și directorul curent înainte de rulare. Datasetul brut nu este modificat de scriptul de inferență, iar proiectul nu scrie tokenuri sau parole în fișiere.
Lista de acceptanță pentru un job reluabil
- ☐
source-lock.jsonconține configurația20231101.ro, dimensiunea în bytes și hashul fiecărui shard. - ☐ Suma rândurilor din shard-uri este egală cu numărul total din lockfile.
- ☐ Fiecare actor
Summarizerrezervă exact un GPU. - ☐ Modelul, Ray, vLLM, PyArrow și celelalte dependențe sunt fixate în
requirements.txt. - ☐ Rezultatele sunt scrise întâi într-un director temporar.
- ☐ Manifestul apare numai după verificarea numărului de rânduri și redenumirea rezultatului.
- ☐ A doua execuție afișează
SKIPpentru shard-urile cu hashuri valide. - ☐ Evaluarea găsește același număr de rânduri și de ID-uri distincte.
- ☐ Niciun rezumat nu este gol.
- ☐ Un eșantion este verificat uman pentru fidelitate față de documentul sursă.
Următoarele extensii păstrează aceeași limită de recuperare
Prima extindere utilă este mutarea celor trei directoare durabile pe NFS sau object storage accesibil tuturor nodurilor Ray. Căile temporare și manifestele trebuie să rămână pe același backend, iar publicarea rezultatului trebuie adaptată semanticii acelui sistem de stocare.
A doua este adăugarea unui hash pentru prompt, versiunea modelului din Hugging Face și configurația completă SamplingParams. Astfel, schimbarea instrucțiunii sau a limitei de tokeni invalidează automat rezultatele vechi.
A treia este evaluarea unui eșantion etichetat de oameni, cu scoruri separate pentru acoperire, fidelitate și lizibilitate. Numărul corect de rânduri demonstrează că jobul este complet; nu demonstrează că rezumatele sunt corecte.
Întrebări frecvente
De ce nu folosim un singur checkpoint pentru întregul corpus?
Un checkpoint per shard păstrează unitățile finalizate și limitează recalcularea la shard-ul aflat în lucru când apare eșecul. Un singur marker global nu poate distinge rezultatele complete de cele parțiale.
Pot rula doi workeri pe un singur GPU?
Nu în configurația prezentată. Fiecare actor rezervă un GPU și încarcă propria copie a modelului, deci pe o gazdă cu o singură placă se folosește --gpu-workers 1.
temperature=0 garantează rezultate identice pe orice placă?
Nu. Decodarea greedy și seed-ul fix reduc variația, dar reproducerea bit-cu-bit cere același GPU, driver și același stack software.
Testul PASS dovedește că rezumatele sunt corecte?
Nu. Testul dovedește acoperirea, unicitatea și absența rezultatelor goale; fidelitatea rezumatelor trebuie verificată pe un eșantion comparat cu documentele sursă.
