Spaces:

sajitkun125
/

inkference-app

Running

File size: 7,694 Bytes

26d4279

"""The JSON action protocol the agent uses instead of provider-native tool calling.

WHY NOT NATIVE TOOL CALLING: RAGConfig.attempts() falls back PER CALL (groq ->
gemini -> extractive). An agent run makes several sequential calls carrying an
accumulating transcript. With native tools that transcript is a provider-specific
object graph — OpenAI `tool_calls`, Gemini `functionCall`, Anthropic `tool_use`
blocks — so a mid-run fallback would mean translating an in-flight tool history
between three schemas, on exactly the rate-limit path that is hardest to test.
Here the transcript is just text, so any provider can resume at any step.

The cost is that we must parse model prose robustly. Everything below exists to
make that never raise: a lenient extractor, then a validator, then a coercion to
`{"action": "answer"}` so a bad reply ends the turn gracefully instead of 500ing.
"""
from __future__ import annotations

import json
import logging
import re
from typing import Any

logger = logging.getLogger("inkference.agent")

# Keep this tiny. Every extra action is another thing a small model gets wrong.
ACTIONS: dict[str, dict[str, Any]] = {
    "search": {
        "args": {"query": str, "k": int},
        "required": ("query",),
        "doc": 'Semantic search over the whole journal. {"action":"search","query":"arrival at Plymouth","k":6}',
    },
    "read_page": {
        "args": {"page": int},
        "required": ("page",),
        "doc": 'Read one full page by its corpus page number. {"action":"read_page","page":118}',
    },
    "read_range": {
        "args": {"start": int, "end": int},
        "required": ("start", "end"),
        "doc": (
            'Read consecutive pages in order — use this for "what happened next" '
            'questions. {"action":"read_range","start":118,"end":123}'
        ),
    },
    "overview": {
        "args": {},
        "required": (),
        "doc": 'Document title, page count, and which page range each book covers. {"action":"overview"}',
    },
    "answer": {
        "args": {"why": str},
        "required": (),
        "doc": 'Stop searching and write the answer from the evidence gathered. {"action":"answer"}',
    },
}

TERMINAL = "answer"

REPAIR_PROMPT = (
    "Your last reply was not a valid action. Reply with ONLY a single JSON object "
    "and nothing else — no explanation, no markdown fence."
)


def render_tool_catalog(actions: list[str] | None = None) -> str:
    """The action menu injected into the plan system prompt."""
    names = actions or list(ACTIONS)
    return "\n".join(f"- {ACTIONS[n]['doc']}" for n in names if n in ACTIONS)


# --------------------------------------------------------------------------- #
# extraction
# --------------------------------------------------------------------------- #
_FENCE_RE = re.compile(r"```(?:json)?\s*(.+?)\s*```", re.DOTALL | re.IGNORECASE)


def _first_balanced_object(text: str) -> str | None:
    """First {...} with balanced braces, ignoring braces inside strings.

    Cheaper and more predictable than a regex for nested objects, and it survives
    the common failure of a model wrapping JSON in a sentence.
    """
    start = text.find("{")
    while start != -1:
        depth = 0
        in_str = False
        esc = False
        for i in range(start, len(text)):
            ch = text[i]
            if esc:
                esc = False
                continue
            if ch == "\\" and in_str:
                esc = True
                continue
            if ch == '"':
                in_str = not in_str
                continue
            if in_str:
                continue
            if ch == "{":
                depth += 1
            elif ch == "}":
                depth -= 1
                if depth == 0:
                    return text[start : i + 1]
        start = text.find("{", start + 1)
    return None


# Repairs for the JSON errors small models actually make, applied in order and
# cumulatively. Each one was added in response to a real observed failure — resist
# adding speculative ones, since a repair that "fixes" valid JSON is worse than a
# parse failure (which costs one retry, not a wrong action).
_REPAIRS = (
    # Trailing comma before a closer: {"a":1,}
    (re.compile(r",\s*([}\]])"), r"\1"),
    # Stray quote after a number: {"end":26"}  — observed from gpt-oss-120b.
    (re.compile(r'(:\s*-?\d+(?:\.\d+)?)"+\s*(?=[,}\]])'), r"\1"),
)


def _loads_lenient(blob: str) -> dict | None:
    candidate = blob
    try:
        obj = json.loads(candidate)
    except json.JSONDecodeError:
        obj = None
        for pattern, repl in _REPAIRS:
            candidate = pattern.sub(repl, candidate)
            try:
                obj = json.loads(candidate)
                break
            except json.JSONDecodeError:
                continue
        if obj is None:
            return None
    return obj if isinstance(obj, dict) else None


def extract_json(text: str) -> dict | None:
    """Pull one JSON object out of a model reply. None if there isn't one."""
    if not text:
        return None
    fenced = _FENCE_RE.search(text)
    for candidate in (fenced.group(1) if fenced else None, text.strip(),
                      _first_balanced_object(text)):
        if candidate and (obj := _loads_lenient(candidate)) is not None:
            return obj
    return None


# --------------------------------------------------------------------------- #
# validation
# --------------------------------------------------------------------------- #
def _coerce_int(value: Any) -> int | None:
    if isinstance(value, bool):
        return None
    if isinstance(value, int):
        return value
    if isinstance(value, float) and value.is_integer():
        return int(value)
    if isinstance(value, str):
        m = re.search(r"-?\d+", value)
        if m:
            return int(m.group())
    return None


def validate(obj: dict) -> dict:
    """Normalise a parsed object into a known action.

    Never raises: an unknown action or missing required arg degrades to `answer`,
    which lets the graph compose from whatever evidence it already holds.
    """
    name = str(obj.get("action", "")).strip().lower()
    if name not in ACTIONS:
        logger.debug("agent: unknown action %r -> answer", name)
        return {"action": TERMINAL, "why": f"unrecognised action {name!r}"}

    spec = ACTIONS[name]
    out: dict[str, Any] = {"action": name}
    for arg, typ in spec["args"].items():
        if arg not in obj or obj[arg] is None:
            continue
        if typ is int:
            coerced = _coerce_int(obj[arg])
            if coerced is not None:
                out[arg] = coerced
        else:
            out[arg] = str(obj[arg]).strip()

    missing = [a for a in spec["required"] if a not in out]
    if missing:
        logger.debug("agent: action %r missing %s -> answer", name, missing)
        return {"action": TERMINAL, "why": f"action {name!r} missing {missing}"}
    return out


def parse_action(text: str) -> tuple[dict | None, dict | None]:
    """Parse a plan reply.

    Returns (action, extras). `action` is None only when nothing JSON-shaped was
    found — the caller may then spend one repair retry. `extras` carries non-action
    keys the plan prompt asks for on the first turn (notably standalone_question).
    """
    obj = extract_json(text)
    if obj is None:
        return None, None
    action = validate(obj)
    extras = {}
    if isinstance(obj.get("standalone_question"), str):
        sq = obj["standalone_question"].strip()
        if sq:
            extras["standalone_question"] = sq
    return action, extras