vllm_support/src/kalm_t5gemma2_vllm_plugin/cli.py

from __future__ import annotations

import argparse
import json
import sys
from collections import OrderedDict
from pathlib import Path
from typing import Any, Iterable, TextIO

from .constants import (
    DEFAULT_DOCUMENT_MAX_LENGTH,
    DEFAULT_ENCODER_CHUNK_SIZE,
    DEFAULT_MAX_MODEL_LEN,
    DEFAULT_QUERY_MAX_LENGTH,
    MODEL_ID,
    SAMPLE_DOCUMENTS,
    SAMPLE_QUERY,
    SUPPORTED_ENCODER_CHUNK_SIZES,
    parse_encoder_chunk_size,
)
from .reranker import KaLMVLLMReranker


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Offline KaLM-Reranker-V1-Nano scoring with vLLM 0.19.1.",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    parser.add_argument("--input-jsonl", type=Path)
    parser.add_argument("--output-jsonl", type=Path)
    parser.add_argument("--model", default=MODEL_ID)
    parser.add_argument("--query-max-length", type=int, default=DEFAULT_QUERY_MAX_LENGTH)
    parser.add_argument(
        "--document-max-length", type=int, default=DEFAULT_DOCUMENT_MAX_LENGTH
    )
    parser.add_argument(
        "--encoder-chunk-size",
        default=str(DEFAULT_ENCODER_CHUNK_SIZE),
        help=f"One of {sorted(SUPPORTED_ENCODER_CHUNK_SIZES)}.",
    )
    parser.add_argument("--max-model-len", type=int, default=DEFAULT_MAX_MODEL_LEN)
    parser.add_argument("--batch-size", type=int, default=32)
    parser.add_argument("--return-margin", action="store_true")
    parser.add_argument("--top-k", type=int)
    parser.add_argument("--dtype", default="bfloat16")
    parser.add_argument("--gpu-memory-utilization", type=float, default=0.85)
    parser.add_argument("--tensor-parallel-size", type=int, default=1)
    return parser


def _read_jsonl(path: Path) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    with path.open("r", encoding="utf-8") as handle:
        for line_no, line in enumerate(handle, start=1):
            if not line.strip():
                continue
            try:
                row = json.loads(line)
            except json.JSONDecodeError as error:
                raise ValueError(f"{path}:{line_no}: invalid JSON.") from error
            if not isinstance(row, dict):
                raise ValueError(f"{path}:{line_no}: expected a JSON object.")
            if not isinstance(row.get("query"), str):
                raise ValueError(f"{path}:{line_no}: 'query' must be a string.")
            if not isinstance(row.get("document"), str):
                raise ValueError(f"{path}:{line_no}: 'document' must be a string.")
            if "instruction" in row and not isinstance(row["instruction"], str):
                raise ValueError(f"{path}:{line_no}: 'instruction' must be a string.")
            rows.append(row)
    return rows


def _write_jsonl(rows: Iterable[dict[str, Any]], output: TextIO) -> None:
    for row in rows:
        output.write(json.dumps(row, ensure_ascii=False) + "\n")
    output.flush()


def _score_rows(
    reranker: KaLMVLLMReranker,
    rows: list[dict[str, Any]],
    *,
    return_margin: bool,
    top_k: int | None,
) -> list[dict[str, Any]]:
    grouped: OrderedDict[str | None, list[tuple[int, dict[str, Any]]]] = OrderedDict()
    for index, row in enumerate(rows):
        grouped.setdefault(row.get("instruction"), []).append((index, row))

    scored: dict[int, dict[str, Any]] = {}
    for instruction, items in grouped.items():
        predictions = reranker.predict(
            [(row["query"], row["document"]) for _, row in items],
            instruction=instruction,
            return_margin=True,
        )
        for (index, row), prediction in zip(items, predictions):
            assert isinstance(prediction, dict)
            result = {
                "id": row.get("id", index),
                "query": row["query"],
                "document": row["document"],
                "score": prediction["score"],
            }
            if "instruction" in row:
                result["instruction"] = row["instruction"]
            if return_margin:
                result["margin"] = prediction["margin"]
            scored[index] = result

    ordered = [scored[index] for index in range(len(rows))]
    if top_k is None:
        return ordered
    if top_k < 0:
        raise ValueError("--top-k must be non-negative.")
    by_query: OrderedDict[str, list[dict[str, Any]]] = OrderedDict()
    for row in ordered:
        by_query.setdefault(str(row["query"]), []).append(row)
    output: list[dict[str, Any]] = []
    for group in by_query.values():
        group.sort(key=lambda item: float(item["score"]), reverse=True)
        output.extend(group[:top_k])
    return output


def main() -> int:
    args = build_parser().parse_args()
    with KaLMVLLMReranker(
        args.model,
        query_max_length=args.query_max_length,
        document_max_length=args.document_max_length,
        encoder_chunk_size=parse_encoder_chunk_size(args.encoder_chunk_size),
        max_model_len=args.max_model_len,
        batch_size=args.batch_size,
        dtype=args.dtype,
        gpu_memory_utilization=args.gpu_memory_utilization,
        tensor_parallel_size=args.tensor_parallel_size,
    ) as reranker:
        if args.input_jsonl is None:
            rankings = reranker.rank(
                SAMPLE_QUERY,
                SAMPLE_DOCUMENTS,
                top_k=args.top_k,
                return_margin=args.return_margin,
            )
            rows = [
                {
                    "query": SAMPLE_QUERY,
                    "document": SAMPLE_DOCUMENTS[int(item["corpus_id"])],
                    **item,
                }
                for item in rankings
            ]
        else:
            rows = _score_rows(
                reranker,
                _read_jsonl(args.input_jsonl),
                return_margin=args.return_margin,
                top_k=args.top_k,
            )

    if args.output_jsonl is None:
        _write_jsonl(rows, sys.stdout)
    else:
        args.output_jsonl.parent.mkdir(parents=True, exist_ok=True)
        with args.output_jsonl.open("w", encoding="utf-8") as handle:
            _write_jsonl(rows, handle)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())