Spaces:
Sleeping
Sleeping
File size: 6,623 Bytes
a28ec65 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 | """Shared `--explain*` CLI plumbing for the four target commands.
Spec v0.4 §3.2 enumerates the explain-specific options; rather than
duplicating the argparse setup in each command, this module provides:
- `add_explain_arguments(parser)` — registers the standard flag set
- `args_to_options(args, pack_name) -> InterpretationOptions`
- `print_envelope(env, format)` — renders to text/json/markdown
- `print_degradation(exc, mode, format)` — handles graceful degradation
per spec §3.7
Each command's `run()` calls these after running its primary work.
"""
from __future__ import annotations
import json as _json
import sys
from typing import Literal
from uofa_cli.interpretation.degrade import make_degradation_notice
from uofa_cli.interpretation.envelope import InterpretationEnvelope
from uofa_cli.interpretation.formatters import render_envelope
from uofa_cli.interpretation.pipeline import InterpretationOptions
from uofa_cli.llm.errors import LLMError
def add_explain_arguments(parser) -> None:
"""Register the `--explain*` flag set on a command's argparse parser.
Identical across rules/check/diff/shacl per spec §3.2 — duplicating
the wiring per command would be a maintenance trap.
"""
parser.add_argument(
"--explain", action="store_true",
help="run the interpretation pipeline after the primary analysis",
)
parser.add_argument(
"--explain-functions", default=None,
help="comma-separated list of interpretation functions to run "
"(values: explain, group, contextualize, cross, narrative). "
"Default: all applicable for the command.",
)
parser.add_argument(
"--explain-format", default=None,
choices=["text", "json", "markdown", "html"],
help="output format for the interpretation block. Default: same as "
"the command's primary --format, falling back to text.",
)
parser.add_argument(
"--explain-backend", default=None,
choices=["ollama", "anthropic", "openai", "openai-compatible", "bundled", "mock"],
help="LLM backend for explain (overrides [llm] backend in uofa.toml)",
)
parser.add_argument(
"--explain-model", default=None,
help="model name on the chosen backend (overrides [llm] model)",
)
parser.add_argument(
"--explain-base-url", default=None,
help="base URL for openai-compatible backends",
)
parser.add_argument(
"--explain-max-items", type=int, default=None,
help="limit interpretation to top N items by severity",
)
parser.add_argument(
"--explain-no-cache", action="store_true",
help="bypass cached interpretation results",
)
def args_to_options(args, *, pack_name: str = "vv40") -> InterpretationOptions:
"""Convert argparse-parsed args into InterpretationOptions.
Resolves the backend from the `--explain-*` flags via the unified
LLM config resolver (spec §3.6 precedence). Callers pass `pack_name`
so the right pack templates are selected.
"""
# Build cli_overrides for the LLM config resolver if the user passed
# any --explain-backend / --explain-model / --explain-base-url flag.
backend = None
if any((args.explain_backend, args.explain_model, args.explain_base_url)):
from uofa_cli.llm import resolve_llm_config, get_backend
cli_overrides: dict = {}
if args.explain_backend:
cli_overrides["backend"] = args.explain_backend
if args.explain_model:
cli_overrides["model"] = args.explain_model
if args.explain_base_url:
cli_overrides["base_url"] = args.explain_base_url
# Convention env var defaults (mirror extract_cmd's logic)
if cli_overrides.get("backend") in ("anthropic", "openai"):
cli_overrides.setdefault(
"api_key_env",
{"anthropic": "ANTHROPIC_API_KEY", "openai": "OPENAI_API_KEY"}[cli_overrides["backend"]],
)
config = resolve_llm_config(cli_overrides=cli_overrides)
backend = get_backend(config)
functions: list[str] = ["all"]
if args.explain_functions:
functions = [name.strip() for name in args.explain_functions.split(",") if name.strip()]
# CLI invocations get the animated TTY spinner around each LLM call
# (no-op on non-TTY); programmatic callers building InterpretationOptions
# directly fall back to the dataclass default (also no-op).
from uofa_cli.output import spinner
return InterpretationOptions(
functions=functions,
max_items=args.explain_max_items,
no_cache=args.explain_no_cache,
backend=backend,
pack_name=pack_name,
spinner_factory=spinner,
)
# ── Rendering ──────────────────────────────────────────────
Format = Literal["text", "json", "markdown", "html"]
def print_envelope(env: InterpretationEnvelope, *, format: Format = "text") -> None:
"""Render `env` to the chosen format and print to stdout.
Thin shell over `formatters.render_envelope`; tests / programmatic
consumers wanting the rendered string call the formatter directly.
"""
rendered = render_envelope(env, format=format)
if rendered:
# Strip trailing newline so print's own newline doesn't double-up
# for formats that already terminate (text/markdown/html do; json
# doesn't). Conditional dropping keeps things tidy in all cases.
print(rendered, end="" if rendered.endswith("\n") else "\n")
def print_degradation(
exc: LLMError,
*,
mode: Literal["explain", "extract"] = "explain",
format: Format = "text",
command: str | None = None,
structured_output=None,
) -> None:
"""Print a graceful-degradation notice (spec §3.7) for an LLM error.
Returns nothing; caller decides exit code (explain → 0, extract → 1).
"""
notice = make_degradation_notice(exc, mode=mode)
if format == "json":
if mode == "extract":
envelope = notice.to_extract_envelope()
else:
envelope = notice.to_explain_envelope(
command=command or "unknown",
structured_output=structured_output if structured_output is not None else {},
)
print(_json.dumps(envelope, indent=2))
return
# text / markdown fall through to the bracket-wrapped notice
print()
print(notice.to_text(), file=sys.stderr if mode == "extract" else sys.stdout)
|