132 lines
4.1 KiB
Python
132 lines
4.1 KiB
Python
"""Static compilation entry: ensure a model Profile, then compile Skills."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
|
|
from tqdm import tqdm
|
|
|
|
from scripts.provider_router import parse_model_reference
|
|
|
|
from .compiler.compiler import compile_input
|
|
from .profile_generation.pipeline import ensure_profile
|
|
|
|
|
|
class ConsoleProgress:
|
|
def __init__(self, enabled: bool):
|
|
self.enabled = enabled
|
|
self.percent = 0
|
|
self.bar = tqdm(
|
|
total=100,
|
|
desc="starting",
|
|
unit="%",
|
|
dynamic_ncols=True,
|
|
file=sys.stderr,
|
|
disable=not enabled,
|
|
)
|
|
|
|
def update(self, percent: int, message: str) -> None:
|
|
if not self.enabled:
|
|
return
|
|
target = max(self.percent, min(100, percent))
|
|
self.bar.set_description_str(message, refresh=False)
|
|
self.bar.update(target - self.percent)
|
|
self.bar.refresh()
|
|
self.percent = target
|
|
|
|
def close(self) -> None:
|
|
self.bar.close()
|
|
|
|
|
|
def _provider_model(value: str) -> str:
|
|
try:
|
|
return parse_model_reference(value).value
|
|
except ValueError as exc:
|
|
raise argparse.ArgumentTypeError(str(exc)) from exc
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(
|
|
prog="static-compile",
|
|
description=(
|
|
"静态编译入口：复用或生成目标模型画像，然后将输入 Skill 编译为模型适配产物。"
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
"--model",
|
|
required=True,
|
|
type=_provider_model,
|
|
help="Target model whose profile the Skill is compiled for.",
|
|
)
|
|
parser.add_argument(
|
|
"--external-model",
|
|
type=_provider_model,
|
|
help="External semantic-planning model required by hybrid mode.",
|
|
)
|
|
parser.add_argument("--input", required=True, type=Path)
|
|
parser.add_argument("--out-root", required=True, type=Path)
|
|
parser.add_argument(
|
|
"--mode", choices=("deterministic", "hybrid"), default="deterministic"
|
|
)
|
|
parser.add_argument("--allow-deterministic-fallback", action="store_true")
|
|
parser.add_argument("--refresh-profile", action="store_true")
|
|
parser.add_argument("--force", action="store_true")
|
|
parser.add_argument("--dry-run", action="store_true")
|
|
parser.add_argument("--no-progress", action="store_true")
|
|
return parser
|
|
|
|
|
|
def static_compile(args: argparse.Namespace) -> int:
|
|
if args.mode == "hybrid" and not args.dry_run and not args.external_model:
|
|
raise ValueError("--external-model is required when --mode hybrid")
|
|
|
|
profile_path, generated = ensure_profile(
|
|
args.model,
|
|
refresh=args.refresh_profile,
|
|
)
|
|
action = "Generated" if generated else "Reusing"
|
|
print(f"{action} model profile: {profile_path}")
|
|
|
|
progress = ConsoleProgress(not args.no_progress and not args.dry_run)
|
|
try:
|
|
results = compile_input(
|
|
args.input,
|
|
profile_path,
|
|
args.out_root,
|
|
mode=args.mode,
|
|
annotator_model=args.external_model,
|
|
allow_deterministic_fallback=args.allow_deterministic_fallback,
|
|
dry_run=args.dry_run,
|
|
force=args.force,
|
|
progress=progress.update,
|
|
)
|
|
finally:
|
|
progress.close()
|
|
|
|
if args.dry_run:
|
|
print(json.dumps([item.report for item in results], ensure_ascii=False, indent=2))
|
|
return 0
|
|
|
|
failed = False
|
|
for result in results:
|
|
status = result.report["status"]
|
|
if status == "failed":
|
|
print(f"failed: {result.skill_name}: {result.report.get('error', 'unknown error')}")
|
|
else:
|
|
print(f"{status}: {result.skill_name} -> {result.output_dir}")
|
|
failed |= status in {"rolled_back", "failed"}
|
|
return 1 if failed else 0
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = build_parser()
|
|
args = parser.parse_args(argv)
|
|
try:
|
|
return static_compile(args)
|
|
except (OSError, RuntimeError, ValueError, subprocess.CalledProcessError) as error:
|
|
parser.exit(1, f"error: {error}\n")
|