""" Kushya API — the in-process interface the Talmudic Engine calls. Architecture §3.6 / §4 Step 22. Three backends are supported, in priority order: 1. Fine-tuned local model (when available — operator's deployed adapter) 2. External model via Korban Processor (eish-zarah-disciplined) 3. Rule-based fallback that returns a structured "no kushya" indicator The caller (Talmudic Engine via socket, or vav_pipeline.SanhedrinClient) decides which backend to use; this module surfaces all three behind a single function. """ from dataclasses import dataclass from typing import Optional # Import korban_processor at module level so tests can patch # `kushya_api.consult_external_model`. If the package isn't on the # path (e.g., someone only wants the dataset tools), these bindings # are set to None and _try_external returns a no-backend result. try: from korban_processor.external_offering import consult_external_model from korban_processor.korban import OfferingType, Provenance except ImportError: # pragma: no cover consult_external_model = None OfferingType = None Provenance = None @dataclass class KushyaResult: """The kushya generator's reply to an interpretation.""" interpretation: str kushya: Optional[str] backend: str # "fine_tuned", "external", "fallback" torah_refs: list[str] raw_response: Optional[str] = None # the full model output, before extraction notes: Optional[str] = None # backend-specific notes (e.g., warning) def generate_kushya( interpretation: str, backend: str = "auto", fine_tuned_endpoint: Optional[str] = None, ) -> KushyaResult: """Generate a kushya for the given interpretation. Args: interpretation: The proposed Torah interpretation. backend: One of "auto", "fine_tuned", "external", "fallback". "auto" tries fine_tuned, then external, then fallback. fine_tuned_endpoint: HTTP URL of the local fine-tuned model server. When None and backend is "auto" or "fine_tuned", this backend is skipped. Returns: A KushyaResult. The kushya field may be None if no backend produced a substantive challenge — the caller must handle that case (typically by treating the interpretation as un-challenged). """ if backend == "fine_tuned": return _try_fine_tuned(interpretation, fine_tuned_endpoint, allow_fallback=False) if backend == "external": return _try_external(interpretation, allow_fallback=False) if backend == "fallback": return _fallback(interpretation) # auto: try in priority order if fine_tuned_endpoint: result = _try_fine_tuned(interpretation, fine_tuned_endpoint, allow_fallback=True) if result.kushya: return result result = _try_external(interpretation, allow_fallback=True) if result.kushya: return result return _fallback(interpretation) def _try_fine_tuned( interpretation: str, endpoint: Optional[str], *, allow_fallback: bool, ) -> KushyaResult: """Call the fine-tuned local kushya model over HTTP. The serve.py script wraps llama.cpp behind a small HTTP API at POST /generate that accepts {"prompt": "..."} and returns {"completion": "..."}. """ if not endpoint: if allow_fallback: return KushyaResult( interpretation=interpretation, kushya=None, backend="fine_tuned", torah_refs=[], notes="fine-tuned endpoint not configured", ) raise ValueError("fine_tuned backend requires fine_tuned_endpoint") try: import httpx except ImportError: return KushyaResult( interpretation=interpretation, kushya=None, backend="fine_tuned", torah_refs=[], notes="httpx not installed", ) prompt = ( "Generate a kushya (challenge) for this interpretation. " "Cite a specific Torah passage. End with a question mark.\n\n" f"Proposed interpretation: {interpretation}\n\n" "Kushya: " ) try: with httpx.Client(timeout=60.0) as client: response = client.post(endpoint, json={"prompt": prompt}) response.raise_for_status() completion = response.json().get("completion", "").strip() except httpx.RequestError as e: if allow_fallback: return KushyaResult( interpretation=interpretation, kushya=None, backend="fine_tuned", torah_refs=[], notes=f"endpoint unreachable: {e}", ) raise if not completion: return KushyaResult( interpretation=interpretation, kushya=None, backend="fine_tuned", torah_refs=[], raw_response=completion, notes="empty completion", ) return KushyaResult( interpretation=interpretation, kushya=completion, backend="fine_tuned", torah_refs=_extract_refs(completion), raw_response=completion, ) def _try_external(interpretation: str, *, allow_fallback: bool) -> KushyaResult: """Call an external model (Claude API) via Korban Processor. The output flows through the offering protocol so external response is treated as testimony, not law (eish_zarah_reading.md). Provenance is preserved on the resulting consultation record. """ if consult_external_model is None: if allow_fallback: return KushyaResult( interpretation=interpretation, kushya=None, backend="external", torah_refs=[], notes="korban_processor not on sys.path", ) raise RuntimeError("korban_processor not on sys.path") system_prompt = ( "You are a Talmudic challenger. Generate exactly one kushya (challenge) " "to the proposed interpretation. The kushya must:\n" " 1. Reference a specific Torah passage that creates tension\n" " 2. Identify the logical or textual weakness\n" " 3. Be expressed as a question, not a statement\n" " 4. Not be resolvable by trivial clarification\n\n" "Return only the kushya question. No preamble, no closing." ) query = f"Proposed interpretation: {interpretation}" try: consultation = consult_external_model( query=query, source_query=f"kushya_request:{interpretation[:80]}", provenance=Provenance.CLAUDE, offering_type=OfferingType.OLAH, system_prompt=system_prompt, ) except Exception as e: if allow_fallback: return KushyaResult( interpretation=interpretation, kushya=None, backend="external", torah_refs=[], notes=f"external consultation failed: {type(e).__name__}: {e}", ) raise raw = consultation.raw_response.content vital = consultation.processed.vital_content # The vital content already passed through shechitah (preamble strip) # and zerikah (vital extraction). Use it as the kushya. if not vital.strip(): return KushyaResult( interpretation=interpretation, kushya=None, backend="external", torah_refs=[], raw_response=raw, notes="zerikah found no vital content in external response", ) return KushyaResult( interpretation=interpretation, kushya=vital, backend="external", torah_refs=_extract_refs(vital), raw_response=raw, notes=( "External testimony — recorded under offering protocol. " "NOT binding ruling." ), ) def _fallback(interpretation: str) -> KushyaResult: """No external backend available. Return a structured no-kushya result. The Talmudic Engine has its own rule-based kushya generator (challenge.pl). This fallback is used when even external testimony is unavailable — we do NOT invent a kushya, we honestly report that no challenge was generated. """ return KushyaResult( interpretation=interpretation, kushya=None, backend="fallback", torah_refs=[], notes=( "No backend available. Rule-based kushyot from the Talmudic " "Engine should be used; the system does not invent challenges." ), ) def _extract_refs(text: str) -> list[str]: """Extract Torah verse references from kushya text. Returns the full match strings (e.g. "Shemot 35:3", "Devarim 21:1-9"), not just the captured book names. """ import re # Non-capturing group so re.findall returns the full match. pattern = r"\b(?:Bereishit|Shemot|Vayikra|Bamidbar|Devarim)\s+\d+:\d+(?:[-–]\d+)?" return re.findall(pattern, text)