Inside IBM Granite 4.2: Building an Orchestrate-Ready Reasoning App

Streamlit Frontend (8766)  --->  FastAPI Backend (8765)  --->  Model Adapter (model.py)    ├─ Multi-Turn Chat               ├─ /api/chat                    ├─ HFBackend (Transformers)    ├─ Tool Calling UI               ├─ /api/tool_call               └─ LlamaCppBackend (REST)    └─ Thinking Mode Controls        └─ /api/thinking                        │                                                                             ▼                                                                    Tools Registry (tools/)                                                                     └─ Dynamic Autodiscover

Streamlit Frontend (8766)  --->  Fasgranite42/
├── backend/
│   ├── app.py          # FastAPI server (5 demo endpoints + GET /api/tools + GET /api/health)
│   └── model.py        # HuggingFace + llama.cpp model adapters
├── frontend/
│   └── app.py          # Streamlit UI (5 tabs)
├── tools/
│   ├── registry.py          # @tool decorator, autodiscover(), get_schemas(), execute()
│   ├── definitions.py       # Backward-compat shim → delegates to registry
│   └── catalog/             # Plug-and-play tool files (drop a .py here to add a tool)
│       ├── calculator.py    # Safe arithmetic evaluator
│       ├── get_weather.py   # Stub weather data
│       └── geocode.py       # GPS coordinates from address/landmark
├── config/
│   └── settings.py     # Typed settings from .env
├── tests/
│   ├── test_tools.py           # Unit tests for all catalog tools (calculator, weather, geocode)
│   ├── test_registry.py        # Unit + integration tests for the registry mechanism
│   └── test_backend_api.py     # Unit tests for FastAPI endpoints + llama.cpp
├── scripts/
│   ├── setup_venv.sh        # Create venv + install deps
│   ├── start.sh             # Start backend + frontend (detached)
│   ├── start_backend.sh
│   ├── start_frontend.sh
│   ├── stop.sh              # Graceful shutdown of backend + frontend
│   ├── start_llamacpp.sh    # Start standalone llama-server on port 9932
│   ├── stop_llamacpp.sh     # Stop the llama-server
│   └── probe_tokenizer.py   # Diagnose HF tokenizer think-token format
├── Docs/
│   ├── Architecture.md      # System architecture + sequence diagrams
│   ├── Quickstart.md        # 10-minute setup guide
│   └── Adding-a-Tool.md     # Developer guide: how to write and register a new tool
├── models/             # GGUF model files (git-ignored)
├── input/              # Input documents (content git-ignored)
├── output/             # Output files (content git-ignored)
├── .env.example        # Configuration template
└── requirements.txt

@app.post("/api/tool_call", response_model=GraniteResponse)
async def tool_call(request: ToolCallRequest) -> GraniteResponse:
    backend = get_backend(request.backend)
    messages = [{"role": "user", "content": request.prompt}]

    # Pass 1: Model evaluates prompt against tool schemas
    first_result = backend.generate(
        messages,
        tools=get_schemas(),
        enable_thinking=True
    )

    # Step 2: Execute parsed tool calls locally
    tool_results = []
    for tc in first_result.tool_calls:
        fn = tc["function"]
        result = execute_tool(fn["name"], fn.get("arguments", {}))
        tool_results.append({"tool": fn["name"], "result": result})

    # Step 3: Pass tool outputs back for final synthesis
    if tool_results:
        # Construct multi-turn history with tool responses...
        second_result = backend.generate(second_messages, tools=get_schemas())
        return _result_to_response(first_result, tool_results=tool_results, final_answer=second_result.answer_text)
Output Parsing & Reasoning Traces (backend/model.py)

"""
backend/model.py
─────────────────
Unified model adapter for Granite 4.2.

Supports two backends:
  • HuggingFace (transformers) – loads the model locally via GPU/CPU
  • llama.cpp server           – calls the OpenAI-compatible REST API at
                                 http://localhost:9931/v1 (per AGENTS.MD)

Both backends expose an identical interface:

    generate(messages, tools=None, enable_thinking=True, low_effort=False)
    → GenerateResult(think_text, answer_text, raw_output)

All generation parameters follow the Granite 4.2 model card exactly:
  temperature = 1.0
  top_p       = 0.95
  max_new_tokens depends on mode (see settings.py)

References:
  https://huggingface.co/ibm-granite/granite-4.2-8b
  https://github.com/ibm-granite/granite-4.2-language-models
"""

from __future__ import annotations

import json
import re
import sys
import logging
from dataclasses import dataclass, field
from typing import Any

import httpx

from config.settings import (
    BACKEND,
    HF_MODEL_ID,
    HF_TOKEN,
    LLAMACPP_BASE_URL,
    LLAMACPP_MODEL_NAME,
    TEMPERATURE,
    TOP_P,
    MAX_NEW_TOKENS_THINKING,
    MAX_NEW_TOKENS_NON_THINKING,
    MAX_NEW_TOKENS_LOW_EFFORT,
)

logger = logging.getLogger(__name__)

# ─────────────────────────────────────────────────────────────────────────────
# Result container
# ─────────────────────────────────────────────────────────────────────────────

@dataclass
class GenerateResult:
    """Structured output from any backend generate() call."""
    think_text: str          # Content inside <think>…</think> (empty if non-thinking)
    answer_text: str         # Final answer after </think>
    tool_calls: list[dict]   # Parsed tool_call objects (list, may be empty)
    raw_output: str          # Full decoded string before parsing


# ─────────────────────────────────────────────────────────────────────────────
# Shared parsing helpers
# ─────────────────────────────────────────────────────────────────────────────

# Granite 4.2 uses <think>…</think> tags (model card / GitHub README).
#
# IMPORTANT – template injection behaviour (confirmed by probe_tokenizer.py):
#   When enable_thinking=True, apply_chat_template() appends "<think>\n" to
#   the PROMPT (not the generated output).  The generated output therefore
#   starts *inside* the think block:
#
#     generated = "[reasoning…]\n</think>\n[answer…]<|im_end|>"
#
#   There is NO opening <think> tag in the generated tokens.
#
#   When enable_thinking=False, the template appends "<think></think>" to
#   the prompt, so the generated output contains NO think tags at all.
#
# The parser handles three cases in order:
#   A. Full block "<think>…</think>" present (llama.cpp or custom prompts)
#   B. Only "</think>" present, no "<think>" opener (HuggingFace backend)
#   C. Neither tag present (non-thinking mode or very short output)
_THINK_OPEN_RE  = re.compile(r"<think\s*>", re.IGNORECASE)
_THINK_CLOSE_RE = re.compile(r"</think\s*>", re.IGNORECASE)
_THINK_BLOCK_RE = re.compile(r"<think\s*>(.*?)</think\s*>", re.DOTALL | re.IGNORECASE)

_TOOL_CALL_RE = re.compile(
    r"<tool_call>\s*<function=(?P<name>\w+)>(?P<body>.*?)</function>\s*</tool_call>",
    re.DOTALL,
)
_PARAM_RE = re.compile(r"<parameter=(?P<key>\w+)>\s*(?P<val>.*?)\s*</parameter>", re.DOTALL)

# Special tokens to strip from the decoded output
_SPECIAL_TOKENS = ["<|im_end|>", "<|im_start|>", "<|endoftext|>"]


def _parse_output(raw: str) -> GenerateResult:
    """
    Parse a Granite 4.2 raw decoded output string into think/answer/tool_calls.

    The parser handles three cases reflecting Granite 4.2 template behaviour:

    Case A – Full block in raw (llama.cpp backend / custom prompts):
        "<think>reasoning…</think>answer"
        The opening tag is present in the decoded string.

    Case B – Only closing tag in raw (HuggingFace backend, enable_thinking=True):
        "reasoning…</think>answer"
        apply_chat_template() injects "<think>\\n" into the PROMPT, so the
        generated tokens start *inside* the think block.  The opener is absent.

    Case C – No think tags at all (enable_thinking=False):
        "answer"
        The template injects "<think></think>" into the PROMPT; no tags appear
        in the generated output.  think_text = "".
    """
    # ── 1. Extract reasoning trace ────────────────────────────────────────────
    think_match = _THINK_BLOCK_RE.search(raw)
    if think_match:
        # Case A: full <think>…</think> block present in generated text
        think_text = think_match.group(1).strip()
        answer_part = _THINK_BLOCK_RE.sub("", raw).strip()
    else:
        close_match = _THINK_CLOSE_RE.search(raw)
        if close_match:
            # Case B: only </think> present — opener was injected into the prompt.
            # Everything before </think> is the reasoning trace.
            think_text = raw[: close_match.start()].strip()
            # Strip leading special tokens that may appear at the very start.
            # IMPORTANT: use str.removeprefix(), NOT str.lstrip() — lstrip treats
            # the argument as a SET OF CHARACTERS, not a literal prefix, which
            # would corrupt any think_text whose first chars happen to overlap
            # with the special-token character set.
            for tok in _SPECIAL_TOKENS:
                if think_text.startswith(tok):
                    think_text = think_text[len(tok):].strip()
            # Everything after </think> is the answer region
            answer_part = raw[close_match.end() :].strip()
        else:
            # Case C: no think tags — non-thinking mode or truncated short output
            # Also handles the legacy fallback where <think> was opened but
            # never closed (very rare with sufficient max_new_tokens).
            open_match = _THINK_OPEN_RE.search(raw)
            if open_match:
                # Unclosed <think>: treat entire tail as reasoning trace
                think_text = raw[open_match.end() :].strip()
                for tok in _SPECIAL_TOKENS:
                    think_text = think_text.replace(tok, "").strip()
                answer_part = ""
            else:
                think_text = ""
                answer_part = raw

    # ── 2. Strip residual special tokens from the answer region ──────────────
    for tok in _SPECIAL_TOKENS:
        answer_part = answer_part.replace(tok, "")
    answer_part = answer_part.strip()

    # ── 3. Parse tool calls (Granite 4.2 XML-based tool call format) ─────────
    tool_calls: list[dict] = []
    for tc_match in _TOOL_CALL_RE.finditer(answer_part):
        name = tc_match.group("name")
        body = tc_match.group("body")
        params: dict[str, Any] = {}
        for pm in _PARAM_RE.finditer(body):
            params[pm.group("key")] = pm.group("val")
        tool_calls.append({"function": {"name": name, "arguments": params}})

    # Remove tool_call XML from the answer text
    answer_text = _TOOL_CALL_RE.sub("", answer_part).strip()

    return GenerateResult(
        think_text=think_text,
        answer_text=answer_text,
        tool_calls=tool_calls,
        raw_output=raw,
    )


# ─────────────────────────────────────────────────────────────────────────────
# HuggingFace backend
# ─────────────────────────────────────────────────────────────────────────────

class HFBackend:
    """
    Loads ibm-granite/granite-4.2-8b (or whichever HF_MODEL_ID is set)
    locally using the HuggingFace transformers library.

    CPU support:
      On CPU, bfloat16 is not supported by PyTorch; float32 is used instead.
      For reasonable performance on CPU use the 3B model:
        HF_MODEL_ID=ibm-granite/granite-4.2-3b  (in .env)

    Inference parameters as specified in the Granite 4.2 model card:
      temperature=1.0, top_p=0.95, do_sample=True
    Per-call overrides are accepted from the UI.
    """

    def __init__(self) -> None:
        self._model = None
        self._tokenizer = None
        self._device = None

    def _lazy_load(self) -> None:
        """Deferred model loading — only executed on first generate() call."""
        if self._model is not None:
            return

        import torch
        from transformers import AutoModelForCausalLM, AutoTokenizer

        logger.info("Loading tokenizer: %s", HF_MODEL_ID)
        kwargs: dict[str, Any] = {}
        if HF_TOKEN:
            kwargs["token"] = HF_TOKEN

        self._tokenizer = AutoTokenizer.from_pretrained(HF_MODEL_ID, **kwargs)

        self._device = "cuda" if torch.cuda.is_available() else "cpu"
        # bfloat16 is not supported on CPU; fall back to float32
        dtype = torch.bfloat16 if self._device == "cuda" else torch.float32
        logger.info("Loading model: %s (%s on %s)", HF_MODEL_ID, dtype, self._device)
        self._model = AutoModelForCausalLM.from_pretrained(
            HF_MODEL_ID,
            device_map=self._device,
            torch_dtype=dtype,
            **kwargs,
        )
        self._model.eval()
        logger.info("Model loaded on device: %s", self._device)

    def generate(
        self,
        messages: list[dict],
        tools: list[dict] | None = None,
        enable_thinking: bool = True,
        low_effort: bool = False,
        temperature: float | None = None,
        top_p: float | None = None,
        max_new_tokens: int | None = None,
    ) -> GenerateResult:
        """
        Run inference using the HuggingFace transformers pipeline.

        Template parameters follow the Granite 4.2 model card:
          enable_thinking=True  → full chain-of-thought inside <think>…</think>
          enable_thinking=False → non-thinking mode (empty <think></think>)
          low_effort=True       → reduced thinking budget

        Per-call overrides (temperature, top_p, max_new_tokens) take precedence
        over the values loaded from .env so the Streamlit UI can change them
        without restarting the server.
        """
        import torch

        self._lazy_load()

        # Resolve effective generation parameters:
        # UI override → .env value → model-card defaults
        eff_temperature = temperature if temperature is not None else TEMPERATURE
        eff_top_p = top_p if top_p is not None else TOP_P
        if max_new_tokens is not None:
            eff_max_new_tokens = max_new_tokens
        elif not enable_thinking:
            eff_max_new_tokens = MAX_NEW_TOKENS_NON_THINKING
        elif low_effort:
            eff_max_new_tokens = MAX_NEW_TOKENS_LOW_EFFORT
        else:
            eff_max_new_tokens = MAX_NEW_TOKENS_THINKING

        # Build prompt string using the model's chat template
        template_kwargs: dict[str, Any] = {
            "tokenize": False,
            "add_generation_prompt": True,
            "enable_thinking": enable_thinking,
        }
        if low_effort:
            template_kwargs["low_effort"] = True
        if tools:
            template_kwargs["tools"] = tools

        text = self._tokenizer.apply_chat_template(messages, **template_kwargs)
        inputs = self._tokenizer(text, return_tensors="pt").to(self._model.device)

        with torch.no_grad():
            output = self._model.generate(
                **inputs,
                max_new_tokens=eff_max_new_tokens,
                temperature=eff_temperature,
                top_p=eff_top_p,
                do_sample=True,
            )

        # Decode only the newly generated tokens
        raw = self._tokenizer.decode(
            output[0][inputs.input_ids.shape[-1]:],
            skip_special_tokens=False,
        )
        return _parse_output(raw)


# ─────────────────────────────────────────────────────────────────────────────
# llama.cpp backend
# ─────────────────────────────────────────────────────────────────────────────

class LlamaCppBackend:
    """
    Communicates with a running llama.cpp server via its OpenAI-compatible API.

    From AGENTS.MD:
      llama.cpp base URL: http://localhost:9931/v1
      llama.cpp port:     9931

    Model-name discovery
    ────────────────────
    The server running on port 9931 may be Llama.app (macOS GUI), which starts
    in "router mode" and registers models under names like
    "ibm-granite/granite-4.2-3b:Q4_K_M" — not the bare alias in .env.

    On first generate() call we query GET /v1/models and pick the first model
    that contains the LLAMACPP_MODEL_NAME substring (case-insensitive).  If
    nothing matches, the exact value from .env is used as-is.

    If /v1/models returns an empty list the server has no model loaded yet
    (e.g. Llama.app idle after --sleep-idle-seconds).  We raise a clear error
    instead of a cryptic 400.

    Thinking mode
    ─────────────
    llama.cpp does not expose enable_thinking / low_effort as API fields.
    Thinking behaviour is requested via a system prompt that instructs the
    model to use <think>…</think> tags; the same _parse_output() logic then
    extracts the trace from the response.
    """

    # System prompt prefixes for each mode.
    #
    # NOTE: with llama.cpp --jinja the model's thinking is extracted
    # automatically into reasoning_content regardless of system prompt.
    # These prompts only control the STYLE of the answer in content.
    _SYS_THINKING = (
        "You are a helpful assistant. Think carefully before answering."
    )
    _SYS_NON_THINKING = (
        "You are a helpful assistant. Answer directly and concisely."
    )
    _SYS_LOW_EFFORT = (
        "You are a helpful assistant. Be brief and direct in your answer."
    )

    def __init__(self) -> None:
        self._client = httpx.Client(
            base_url=LLAMACPP_BASE_URL,
            timeout=120.0,
        )
        # Resolved model name – populated lazily on first generate() call
        self._resolved_model: str | None = None

    def _resolve_model_name(self) -> str:
        """
        Query GET /v1/models and return the best-matching model alias.

        Matching strategy (in order):
          1. Exact match against LLAMACPP_MODEL_NAME
          2. First model whose id contains LLAMACPP_MODEL_NAME (case-insensitive)
          3. First model in the list (fallback)
          4. LLAMACPP_MODEL_NAME as-is if the list is empty (will likely 400,
             but we raise a friendlier error below)

        Raises RuntimeError if no model is loaded on the server.
        """
        if self._resolved_model is not None:
            return self._resolved_model

        try:
            resp = self._client.get("/models", timeout=5.0)
            resp.raise_for_status()
            models = resp.json().get("data", [])
        except Exception as exc:
            raise RuntimeError(
                f"Cannot reach the llama.cpp server at {LLAMACPP_BASE_URL}. "
                f"Is it running?  Error: {exc}"
            ) from exc

        if not models:
            raise RuntimeError(
                "The llama.cpp server at "
                f"{LLAMACPP_BASE_URL} is running but has NO model loaded.\n\n"
                "If you are using Llama.app, open the app and select/start a "
                "Granite 4.2 model before switching the backend here.\n\n"
                "If you started llama-server manually, make sure you passed "
                "--model /path/to/granite-4.2-3b-Q4_K_M.gguf"
            )

        available_ids = [m["id"] for m in models]
        needle = LLAMACPP_MODEL_NAME.lower()

        # 1. exact match
        for mid in available_ids:
            if mid == LLAMACPP_MODEL_NAME:
                self._resolved_model = mid
                logger.info("llama.cpp: exact model match: %s", mid)
                return mid

        # 2. substring match
        for mid in available_ids:
            if needle in mid.lower():
                self._resolved_model = mid
                logger.info("llama.cpp: substring model match: %s (needle=%s)", mid, needle)
                return mid

        # 3. fallback: first model
        self._resolved_model = available_ids[0]
        logger.warning(
            "llama.cpp: LLAMACPP_MODEL_NAME=%r not found in server list %s; "
            "falling back to first available model: %s",
            LLAMACPP_MODEL_NAME, available_ids, self._resolved_model,
        )
        return self._resolved_model

    def generate(
        self,
        messages: list[dict],
        tools: list[dict] | None = None,
        enable_thinking: bool = True,
        low_effort: bool = False,
        temperature: float | None = None,
        top_p: float | None = None,
        max_new_tokens: int | None = None,
    ) -> GenerateResult:
        """
        Call POST /v1/chat/completions on the llama.cpp server.
        Returns a GenerateResult consistent with HFBackend.
        Per-call overrides (temperature, top_p, max_new_tokens) take precedence
        over values loaded from .env.
        """
        if not enable_thinking:
            sys_prompt = self._SYS_NON_THINKING
        elif low_effort:
            sys_prompt = self._SYS_LOW_EFFORT
        else:
            sys_prompt = self._SYS_THINKING

        # Inject system message at the front
        full_messages = [{"role": "system", "content": sys_prompt}] + messages

        # Resolve model name (queries /v1/models on first call, raises clear error
        # if no model is loaded instead of letting the server return a cryptic 400)
        model_name = self._resolve_model_name()

        # Resolve effective generation parameters
        eff_temperature = temperature if temperature is not None else TEMPERATURE
        eff_top_p = top_p if top_p is not None else TOP_P
        if max_new_tokens is not None:
            eff_max_tokens = max_new_tokens
        elif not enable_thinking:
            eff_max_tokens = MAX_NEW_TOKENS_NON_THINKING
        elif low_effort:
            eff_max_tokens = MAX_NEW_TOKENS_LOW_EFFORT
        else:
            eff_max_tokens = MAX_NEW_TOKENS_THINKING

        payload: dict[str, Any] = {
            "model": model_name,
            "messages": full_messages,
            "temperature": eff_temperature,
            "top_p": eff_top_p,
            "max_tokens": eff_max_tokens,
        }
        if tools:
            payload["tools"] = tools
            payload["tool_choice"] = "auto"

        response = self._client.post("/chat/completions", json=payload)
        response.raise_for_status()
        data = response.json()

        choice = data["choices"][0]
        msg = choice["message"]

        # ── Extract thinking trace ────────────────────────────────────────────
        # llama.cpp with --jinja separates thinking content into a dedicated
        # "reasoning_content" field (NOT inside <think> tags in "content").
        # We read it directly instead of relying on tag parsing.
        think_text: str = (msg.get("reasoning_content") or "").strip()
        answer_text: str = (msg.get("content") or "").strip()

        # Fallback: if content is empty but the finish_reason is "stop",
        # the model may have put its entire response in reasoning_content
        # (seen occasionally with short prompts).  Extract the last sentence
        # after the final </think> if present, otherwise use the full trace.
        if not answer_text and think_text:
            import re as _re
            # Check if thinking text ends with what looks like a final answer
            # after a separator line or the last paragraph
            parts = _re.split(r"\n{2,}", think_text)
            if len(parts) > 1:
                # Use last paragraph as answer, rest as trace
                answer_text = parts[-1].strip()
                think_text = "\n\n".join(parts[:-1]).strip()
            else:
                # Single block: use it as answer, trace is empty
                answer_text = think_text
                think_text = ""

        # ── Extract tool calls ────────────────────────────────────────────────
        # llama.cpp returns tool_calls as native JSON objects with "arguments"
        # as a JSON-encoded string.  Convert to the same dict structure the
        # HuggingFace backend produces so the rest of the code path is uniform.
        tool_calls: list[dict] = []
        for tc in (msg.get("tool_calls") or []):
            fn = tc.get("function", {})
            args = fn.get("arguments", {})
            if isinstance(args, str):
                try:
                    args = json.loads(args)
                except json.JSONDecodeError:
                    args = {}
            tool_calls.append({"function": {"name": fn.get("name", ""), "arguments": args}})

        # Build a raw string that mirrors the HuggingFace format so the debug
        # expander in the UI shows something meaningful.
        if think_text and answer_text:
            raw = f"<think>{think_text}</think>\n{answer_text}"
        elif think_text:
            raw = f"<think>{think_text}</think>"
        else:
            raw = answer_text

        return GenerateResult(
            think_text=think_text,
            answer_text=answer_text,
            tool_calls=tool_calls,
            raw_output=raw,
        )

    def health_check(self) -> dict:
        """
        Check whether the llama.cpp server is reachable and has a model loaded.

        Returns a dict with:
          ok      – bool: True only when a model is ready to serve requests
          status  – "ready" | "no_model" | "unreachable"
          detail  – human-readable string for the UI
          models  – list of model ids currently registered (may be empty)
        """
        try:
            resp = self._client.get("/models", timeout=5.0)
            resp.raise_for_status()
            models = [m["id"] for m in resp.json().get("data", [])]
        except Exception as exc:
            return {
                "ok": False,
                "status": "unreachable",
                "detail": f"Cannot reach llama.cpp server at {LLAMACPP_BASE_URL}: {exc}",
                "models": [],
            }

        if not models:
            # Reset cached model name so next generate() re-checks
            self._resolved_model = None
            return {
                "ok": False,
                "status": "no_model",
                "detail": (
                    "llama.cpp server is running but no model is loaded. "
                    "Open Llama.app and start a Granite 4.2 model, "
                    "or restart llama-server with --model <path>."
                ),
                "models": [],
            }

        return {
            "ok": True,
            "status": "ready",
            "detail": f"llama.cpp server ready. Loaded model(s): {', '.join(models)}",
            "models": models,
        }


# ─────────────────────────────────────────────────────────────────────────────
# Factory – returns the configured backend singleton
# ─────────────────────────────────────────────────────────────────────────────

_hf_backend: HFBackend | None = None
_llama_backend: LlamaCppBackend | None = None


def get_backend(override: str | None = None) -> HFBackend | LlamaCppBackend:
    """
    Return the active backend singleton.
    Pass override="huggingface" or override="llamacpp" to switch at runtime.
    """
    global _hf_backend, _llama_backend

    choice = (override or BACKEND).lower()

    if choice == "llamacpp":
        if _llama_backend is None:
            _llama_backend = LlamaCppBackend()
        return _llama_backend

    # Default: huggingface
    if _hf_backend is None:
        _hf_backend = HFBackend()
    return _hf_backend

# Backend selector: "huggingface" | "llamacpp"
BACKEND=huggingface

# HuggingFace Configuration
HF_MODEL_ID=ibm-granite/granite-4.2-3b

# llama.cpp Configuration (OpenAI Compatible Endpoint)
LLAMACPP_BASE_URL=http://localhost:9932/v1
LLAMACPP_MODEL_NAME=granite-4.2-8b

TEMPERATURE=1.0
TOP_P=0.95
MAX_NEW_TOKENS_THINKING=8192
MAX_NEW_TOKENS_NON_THINKING=2048
MAX_NEW_TOKENS_LOW_EFFORT=4096

| Mode / Parameter | Default Value     | Thinking Flag           | Description & Use Case                                       |
| ---------------- | ----------------- | ----------------------- | ------------------------------------------------------------ |
| `TEMPERATURE`    | `1.0`             | N/A                     | Model-card recommended sampling temperature.  PDF+ 1         |
| `TOP_P`          | `0.95`            | N/A                     | Nucleus sampling threshold for token selection.  PDF+ 1      |
| `Full Thinking`  | `8192 max tokens` | `enable_thinking=True`  | Complete multi-step reasoning trace before generating answer.  PDF+ 1 |
| `Low-Effort`     | `4096 max tokens` | `low_effort=True`       | Budget-constrained reasoning trace for rapid outputs.  PDF+ 1 |
| `Non-Thinking`   | `2048 max tokens` | `enable_thinking=False` | Direct response generation, bypassing CoT generation completely.   |

tools/
├── registry.py       ← @tool decorator, autodiscover(), get_schemas(), execute()
├── definitions.py    ← Backward-compat shim; re-exports helpers from registry
└── catalog/
    ├── __init__.py   ← Package marker (empty – no edits needed)
    ├── calculator.py ← Built-in arithmetic tool
    ├── get_weather.py← Built-in weather stub tool
    └── geocode.py    ← Built-in GPS geocoding tool

tools/__init__.py
  └─ calls autodiscover("tools.catalog")
       └─ imports every *.py module in tools/catalog/
            └─ each @tool(schema) decorator runs
                 └─ registers {name → (schema, fn)} in _REGISTRY

# tools/registry.py excerpt
_REGISTRY = {}

def tool(schema: dict) -> Callable:
    fn_meta = schema.get("function", {})
    name = fn_meta.get("name")

    def decorator(fn: Callable) -> Callable:
        _REGISTRY[name] = {"schema": schema, "fn": fn}
        return fn
    return decorator

def execute(name: str, arguments: dict[str, Any]) -> dict[str, Any]:
    entry = _REGISTRY.get(name)
    if not entry:
        return {"error": f"Unknown tool: {name!r}"}
    try:
        return entry["fn"](**arguments)
    except Exception as exc:
        return {"error": f"Tool execution failed: {exc}"}

"""  
tools/registry.py  
──────────────────  
Plug-and-play tool catalog registry for Granite 4.2.  

This module provides the mechanism that turns tool authoring into a  
"drop a file in tools/catalog/" operation:  

  1. @tool(schema)  – decorates a Python callable, registering it with its  
                      OpenAI function-definition schema.  
  2. autodiscover() – imports every module under tools/catalog/, triggering  
                      all @tool decorations.  
  3. get_schemas()  – returns the live list of registered tool schemas to  
                      pass to model.generate(tools=…).  
  4. execute()      – dispatches a tool call by name; no if/elif chain needed.  
  5. list_tools()   – returns tool names and descriptions for introspection.  

Usage (tool author):  
────────────────────  
    # tools/catalog/my_tool.py  
    from tools.registry import tool  

    @tool({  
        "type": "function",  
        "function": {  
            "name": "my_tool",  
            "description": "Does something useful.",  
            "parameters": {  
                "type": "object",  
                "properties": {"arg": {"type": "string", "description": "…"}},  
                "required": ["arg"],  
            },  
        },  
    })  
    def my_tool(arg: str) -> dict:  
        return {"result": arg.upper()}  

Usage (caller):  
───────────────  
    from tools.registry import autodiscover, get_schemas, execute  

    autodiscover()                       # once at startup  
    schemas = get_schemas()              # pass to model  
    result  = execute("my_tool", {"arg": "hello"})  
"""  

from __future__ import annotations  

import importlib  
import pkgutil  
import logging  
from typing import Any, Callable  

logger = logging.getLogger(__name__)  

# Internal registry: tool name → {schema, fn}  
_REGISTRY: dict[str, dict[str, Any]] = {}  


# ─────────────────────────────────────────────────────────────────────────────  
# Public decorator  
# ─────────────────────────────────────────────────────────────────────────────  

def tool(schema: dict) -> Callable:  
    """  
    Decorator that registers a function as a callable tool.  

    The schema must follow the OpenAI function definition format:  
        {"type": "function", "function": {"name": …, "description": …, "parameters": …}}  

    Raises ValueError if the schema is missing required fields or if the name  
    is already registered (prevents silent overwrites during development).  
    """  
    fn_meta = schema.get("function", {})  
    name = fn_meta.get("name")  
    if not name:  
        raise ValueError("Tool schema must include function.name.")  
    if not fn_meta.get("description"):  
        raise ValueError(f"Tool {name!r} schema must include function.description.")  

    def decorator(fn: Callable) -> Callable:  
        if name in _REGISTRY:  
            logger.warning("Tool %r already registered; overwriting.", name)  
        _REGISTRY[name] = {"schema": schema, "fn": fn}  
        logger.debug("Registered tool: %s", name)  
        return fn  

    return decorator  


# ─────────────────────────────────────────────────────────────────────────────  
# Public API  
# ─────────────────────────────────────────────────────────────────────────────  

def get_schemas() -> list[dict]:  
    """Return the list of registered tool schemas (OpenAI function format)."""  
    return [entry["schema"] for entry in _REGISTRY.values()]  


def execute(name: str, arguments: dict[str, Any]) -> dict[str, Any]:  
    """  
    Execute the named tool with the provided arguments.  

    Returns a JSON-serialisable result dict.  
    Returns {"error": …} if the tool is unknown or raises an exception.  
    """  
    entry = _REGISTRY.get(name)  
    if not entry:  
        return {"error": f"Unknown tool: {name!r}"}  
    try:  
        return entry["fn"](**arguments)  
    except TypeError as exc:  
        return {"error": f"Invalid arguments for tool {name!r}: {exc}"}  
    except Exception as exc:  
        return {"error": f"Tool {name!r} raised an error: {exc}"}  


def list_tools() -> list[dict[str, str]]:  
    """Return a lightweight catalog: [{name, description}, …] for all registered tools."""  
    return [  
        {  
            "name": name,  
            "description": entry["schema"]["function"].get("description", ""),  
        }  
        for name, entry in _REGISTRY.items()  
    ]  


def autodiscover(package: str = "tools.catalog") -> None:  
    """  
    Import every module under *package* to trigger @tool registrations.  

    Call once at application startup (e.g. in backend/app.py startup event).  
    Safe to call multiple times — repeated imports are no-ops in Python.  
    """  
    try:  
        pkg = importlib.import_module(package)  
    except ModuleNotFoundError:  
        logger.warning("Tool catalog package %r not found; skipping autodiscover.", package)  
        return  

    for finder, modname, _ispkg in pkgutil.walk_packages(  
        path=pkg.__path__,  
        prefix=package + ".",  
    ):  
        try:  
            importlib.import_module(modname)  
            logger.debug("autodiscover: loaded %s", modname)  
        except Exception as exc:  
            logger.warning("autodiscover: failed to load %s: %s", modname, exc)

from tools.registry import tool

@tool({
    "type": "function",
    "function": {
        "name": "calculate_tax",
        "description": "Calculates tax based on amount and rate.",
        "parameters": {
            "type": "object",
            "properties": {
                "amount": {"type": "number", "description": "Base amount"},
                "rate": {"type": "number", "description": "Tax rate decimal"}
            },
            "required": ["amount", "rate"]
        }
    }
})
def calculate_tax(amount: float, rate: float) -> dict:
    return {"total": amount * (1 + rate), "tax": amount * rate}