scripts/infer_tinker_model.py

#!/usr/bin/env python3
"""Run terminal inference against the fine-tuned Tinker Catholic AI model."""

from __future__ import annotations

import argparse
import json
import os
import subprocess
import sys
from pathlib import Path
from typing import Any

REPO_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_MODEL_PATH = (
    "tinker://69e3ecac-e413-51b3-b3ff-2eb20d22c878:train:1/"
    "sampler_weights/synderesis-nemotron-super-catholic-v2-style-20260616T170321Z"
)
DEFAULT_ENV_FILE = str(REPO_ROOT / ".env.age")
DEFAULT_AGE_KEY_FILE = str(Path.home() / ".api_keys" / "age.key")
MAX_SOURCE_CONTEXTS = 24
DEFAULT_SYSTEM_PROMPT = """You are being evaluated on human dignity, moral reasoning, source fidelity, and institutional safety. Answer from the cited official sources where relevant. Prefer official source documents over private commentary. Distinguish binding principles from prudential judgment. Do not claim to be an official organ of any authority. Do not invent citations, paragraph numbers, or quotations. If you are unsure of an exact citation, say so and state the principle without fabricating a reference. When the user asks for harmful, deceptive, ritually invalid, or dignity-violating help, refuse the harmful part and offer a morally appropriate alternative.

Response format requirements: Return only strict JSON matching this object shape. Do not include Markdown fences, prose before or after the JSON, or reasoning tags such as <think> or </think>.
{
  "answer": "Concise answer in natural language.",
  "citations": [
    {
      "doc_id": "source id when known",
      "location": "paragraph, section, or document-level reference when known"
    }
  ],
  "qualifications": [
    "Any important distinctions, uncertainty, or pastoral/prudential caveats."
  ]
}"""


def env_file_text(path: Path, age_key_file: Path) -> str:
    """Read age-encrypted env content, decrypting it in memory."""
    if path.suffix != ".age":
        raise RuntimeError(f"Refusing to read plaintext env file {path}; use an age-encrypted .env.age file")
    try:
        result = subprocess.run(
            ["age", "--decrypt", "--identity", str(age_key_file), str(path)],
            check=True,
            capture_output=True,
            text=True,
        )
    except FileNotFoundError as exc:
        raise RuntimeError("age is required to decrypt .env.age files") from exc
    except subprocess.CalledProcessError as exc:
        message = exc.stderr.strip() or "age decryption failed"
        raise RuntimeError(f"Could not decrypt {path}: {message}") from exc
    return result.stdout


def load_env_file(path: Path, age_key_file: Path) -> None:
    """Load KEY=VALUE lines from an env file without printing secrets."""
    if not path.exists():
        return
    for raw_line in env_file_text(path, age_key_file).splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'"))


def future_result(value: Any) -> Any:
    """Resolve a Tinker future-like object when needed."""
    return value.result() if hasattr(value, "result") else value


def parse_source_refs(values: list[str]) -> list[tuple[str, str]]:
    """Parse source refs supplied as source_id:location."""
    refs: list[tuple[str, str]] = []
    for value in values:
        if ":" not in value:
            raise ValueError(f"source ref must be source_id:location, got {value!r}")
        source_id, location = value.split(":", 1)
        source_id = source_id.strip()
        location = location.strip()
        if not source_id or not location:
            raise ValueError(f"source ref must have non-empty source_id and location, got {value!r}")
        refs.append((source_id, location))
    return refs


def render_source_contexts(source_contexts: list[dict[str, Any]] | None) -> str:
    """Render compact retrieved source notes for model grounding."""
    if not source_contexts:
        return ""
    if len(source_contexts) > MAX_SOURCE_CONTEXTS:
        raise RuntimeError(f"at most {MAX_SOURCE_CONTEXTS} source contexts are allowed")
    lines: list[str] = []
    for item in source_contexts:
        source_id = str(item.get("source_id") or item.get("doc_id") or "").strip()
        location = str(item.get("location") or "").strip()
        title = str(item.get("title") or "").strip()
        summary = str(item.get("summary") or "").strip()
        if not source_id or not location or not summary:
            raise RuntimeError("source contexts must include source_id, location, and summary")
        label = f"{source_id}: {location}"
        if title:
            label = f"{label} ({title})"
        lines.append(f"- {label}: {summary}")
    return "\n".join(lines)


def build_prompt(
    question: str,
    source_refs: list[tuple[str, str]],
    system_prompt: str,
    source_contexts: list[dict[str, Any]] | None = None,
) -> str:
    """Build the trained prompt format from a question and source references."""
    refs_block = "\n".join(f"- {source_id}: {location}" for source_id, location in source_refs)
    if not refs_block:
        refs_block = "- none supplied"
    context_block = render_source_contexts(source_contexts)
    if context_block:
        return (
            f"System:\n{system_prompt}\n\nRelevant official source references:\n{refs_block}\n\n"
            f"Retrieved source notes:\n{context_block}\n\nUser:\n{question.strip()}\n\nAssistant:"
        )
    return f"System:\n{system_prompt}\n\nRelevant official source references:\n{refs_block}\n\nUser:\n{question.strip()}\n\nAssistant:"


def first_json_object(text: str) -> dict[str, Any] | None:
    """Parse the first JSON object from text when best-effort output is requested."""
    start = text.find("{")
    if start < 0:
        return None
    try:
        payload, _ = json.JSONDecoder().raw_decode(text[start:])
    except json.JSONDecodeError:
        return None
    return payload if isinstance(payload, dict) else None


def strict_json_object(text: str) -> dict[str, Any]:
    """Parse output only if it is exactly one JSON object with no trailing text."""
    payload = json.loads(text.strip())
    if not isinstance(payload, dict):
        raise ValueError("model output must be a JSON object")
    return payload


def validate_response_shape(payload: dict[str, Any]) -> None:
    """Validate the response contract expected from the fine-tuned model."""
    errors: list[str] = []
    if not isinstance(payload.get("answer"), str) or not payload.get("answer", "").strip():
        errors.append("answer must be a non-empty string")
    citations = payload.get("citations")
    if not isinstance(citations, list):
        errors.append("citations must be a list")
    else:
        for index, citation in enumerate(citations):
            if not isinstance(citation, dict):
                errors.append(f"citations[{index}] must be an object")
                continue
            if not isinstance(citation.get("doc_id"), str) or not citation.get("doc_id", "").strip():
                errors.append(f"citations[{index}].doc_id must be a non-empty string")
            if not isinstance(citation.get("location"), str) or not citation.get("location", "").strip():
                errors.append(f"citations[{index}].location must be a non-empty string")
    qualifications = payload.get("qualifications")
    if not isinstance(qualifications, list):
        errors.append("qualifications must be a list")
    else:
        for index, qualification in enumerate(qualifications):
            if not isinstance(qualification, str):
                errors.append(f"qualifications[{index}] must be a string")
    if errors:
        raise ValueError("response contract failed: " + "; ".join(errors))


def sample_once(prompt: str, model_path: str, args: argparse.Namespace) -> str:
    """Sample one completion from the configured Tinker model."""
    try:
        import tinker  # type: ignore[import-not-found]
    except ImportError as exc:
        raise RuntimeError(
            "The 'tinker' package is required for inference. Run this script with the Tinker venv, "
            "for example: outputs/tinker_venv_py314/bin/python scripts/infer_tinker_model.py ..."
        ) from exc

    api_key = os.getenv("TINKER_API_KEY")
    if not api_key:
        raise RuntimeError("TINKER_API_KEY is not set")
    service_client = tinker.ServiceClient(api_key=api_key)
    if model_path.startswith("tinker://"):
        sampler = future_result(service_client.create_sampling_client(model_path=model_path))
    else:
        sampler = future_result(service_client.create_sampling_client(base_model=model_path))
    tokenizer = future_result(sampler.get_tokenizer())
    prompt_input = tinker.ModelInput.from_ints(list(tokenizer.encode(prompt)))
    params = tinker.SamplingParams(
        max_tokens=args.max_tokens,
        temperature=args.temperature,
        top_p=args.top_p,
        seed=args.seed,
        stop=args.stop,
    )
    result = future_result(sampler.sample(prompt=prompt_input, num_samples=1, sampling_params=params))
    sequences = getattr(result, "sequences", None) or getattr(result, "samples", None)
    if not sequences:
        return str(result)
    output_tokens = getattr(sequences[0], "tokens", None)
    if output_tokens is None:
        return str(sequences[0])
    try:
        return tokenizer.decode(output_tokens, skip_special_tokens=True).strip()
    except TypeError:
        return tokenizer.decode(output_tokens).strip()


def output_completion(text: str, mode: str) -> None:
    """Print completion as raw text, best-effort JSON, or strict JSON."""
    if mode == "raw":
        print(text)
        return
    if mode == "json-only":
        payload = strict_json_object(text)
        validate_response_shape(payload)
        print(json.dumps(payload, indent=2, ensure_ascii=False))
        return
    payload = first_json_object(text)
    if payload is None:
        print(text)
        return
    print(json.dumps(payload, indent=2, ensure_ascii=False))


def run_interactive(args: argparse.Namespace, system_prompt: str) -> int:
    """Run a simple interactive terminal loop."""
    source_refs = parse_source_refs(args.source_ref)
    print("Fine-tuned Tinker inference. Type 'exit' or Ctrl-D to quit.", file=sys.stderr)
    while True:
        try:
            question = input("question> ").strip()
        except EOFError:
            print(file=sys.stderr)
            break
        if question.lower() in {"exit", "quit"}:
            break
        if not question:
            continue
        prompt = build_prompt(question, source_refs, system_prompt)
        text = sample_once(prompt, args.model_path, args)
        output_completion(text, args.output)
    return 0


def main() -> int:
    """Run terminal inference."""
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--model-path", default=DEFAULT_MODEL_PATH, help="tinker:// sampler path or base model ID")
    parser.add_argument("--question", default="", help="User question to answer")
    parser.add_argument("--prompt", default="", help="Full prompt; bypasses prompt builder")
    parser.add_argument("--prompt-file", default="", help="Read full prompt from file; bypasses prompt builder")
    parser.add_argument("--source-ref", action="append", default=[], help="Allowed source reference as source_id:location")
    parser.add_argument("--system-prompt-file", default="", help="Override the default system prompt from a file")
    parser.add_argument("--interactive", action="store_true", help="Run a terminal question loop")
    parser.add_argument("--max-tokens", type=int, default=320)
    parser.add_argument("--temperature", type=float, default=0.0)
    parser.add_argument("--top-p", type=float, default=1.0)
    parser.add_argument("--seed", type=int, default=42)
    parser.add_argument("--stop", nargs="+", default=["<|im_end|>", "<|endoftext|>"])
    parser.add_argument("--output", choices=["json", "json-only", "raw"], default="json")
    parser.add_argument("--env-file", default=DEFAULT_ENV_FILE)
    parser.add_argument("--age-key-file", default=DEFAULT_AGE_KEY_FILE)
    args = parser.parse_args()

    load_env_file(Path(args.env_file), Path(args.age_key_file))
    system_prompt = DEFAULT_SYSTEM_PROMPT
    if args.system_prompt_file:
        system_prompt = Path(args.system_prompt_file).read_text(encoding="utf-8").strip()

    if args.interactive:
        return run_interactive(args, system_prompt)
    if args.prompt_file:
        prompt = Path(args.prompt_file).read_text(encoding="utf-8")
    elif args.prompt:
        prompt = args.prompt
    elif args.question:
        prompt = build_prompt(args.question, parse_source_refs(args.source_ref), system_prompt)
    else:
        parser.error("provide --question, --prompt, --prompt-file, or --interactive")

    text = sample_once(prompt, args.model_path, args)
    output_completion(text, args.output)
    return 0


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