"""The starting tool set: local, deterministic, no egress, no filesystem. **Every tool here answers in one call.** That is the selection criterion, not a coincidence. ``TOOL_CATEGORIES_REVIEW.md`` §1 shows that call depth is the filter that decides which categories of tool are viable at all below 7B: BFCL multi-turn accuracy is 16.88 per cent at 1.7B and 1.38 per cent at 0.6B, so any benefit needing three coordinated, stateful calls is unreachable however good the idea is. Single-call, self-contained tools are what survive that filter. **The categories represented, and why.** * *Computation and exactness* (``calculate``, ``convert_units``, ``calculate_date``, ``extract_from_text``). The best-evidenced category: Toolformer took a 6.7B model from 5.2 to 29.4 per cent on SVAMP, past a 175B model without a calculator, while the same paper's search tool failed to close the gap on factual questions. A tool that *computes an exact answer* to a subproblem buys a great deal at this scale; a tool that supplies *more text to reason over* buys much less. * *Local retrieval and grounding* (``search_document``). Lexical, because BM25 beats the popular dense retrievers for supplied-document retrieval, which means no embedding model, no vector store and no extra weights. Returns short spans rather than chunks: flooding a small model's context is what makes RAG a net negative below 7B. * *Deterministic verification* (``verify_quote``). Amplifying a weak model needs an external soundness signal; a model reviewing its own work is documented to make reasoning worse. **What is deliberately absent.** No todo list, scratchpad, notes or plan tool. The premise is sound and the mechanism is not: those need coordinated calls across turns, and the one direct small-model measurement found has Llama-1B performing *worse* with its own plans (23.2 per cent) than with no plan at all (25.2 per cent). That state belongs to the harness, which can hold it at zero tool calls. See ``TOOL_CATEGORIES_REVIEW.md`` §3. Nothing here opens a socket, reads a file, spawns a process or consults a model. Every one is a pure function of its arguments. **Naming is a design decision, not a detail.** Schema misalignment, where the model emits a plausible but non-existent name, is the predominant tool-use failure below 10B (arXiv:2510.07248), and renaming schema components to match pretraining conventions cut that error class by 80 per cent. So the tools are called ``calculate``, ``convert_units``, ``calculate_date`` and ``extract_from_text``, not ``distinct.arith`` or ``text.scan.v2``. Every description states **when not to call** the tool, because the most common cause of a repeated-tool loop is a description that never says to stop. **Nothing here registers itself.** ``register_local_tools`` must be called explicitly, and the operator's :mod:`distinct_tools.approval` policy still has to approve each one before it is advertised. """ from __future__ import annotations import datetime as _datetime import decimal import difflib import math import re import unicodedata from collections.abc import Mapping from decimal import Decimal from .core import Registry, ToolContext, ToolInputError, ToolSpec # -------------------------------------------------------------------------- # Shared bounds # -------------------------------------------------------------------------- MAX_EXPRESSION_CHARACTERS = 500 MAX_TOKENS = 200 MAX_PARSE_DEPTH = 64 #: Longest document a text tool accepts. #: #: Chosen to sit *inside* the broker's ``max_input_bytes`` (16,384 by default, #: and the worker-owned ceiling in ``distinct_agent.tools.HARD_TOOL_LIMITS``), #: with room for JSON escaping and the other arguments. A schema advertising a #: longer limit than the broker will actually accept is a schema that lies to #: the model: the call would be refused as invalid input for a reason the #: schema said was fine. Non-ASCII text still reaches the byte cap sooner, #: because the broker measures UTF-8 bytes and this measures characters; that #: refusal is the broker's and is reported cleanly. MAX_TEXT_CHARACTERS = 12_000 MAX_MATCHES = 50 #: Exponents are the one operator that turns a short input into unbounded work, #: so they are bounded to integers in this range and nothing else. MAX_EXPONENT = 64 #: Enough precision to be exact for money and ordinary measurement, small #: enough that no single operation is expensive. CALCULATION_PRECISION = 34 def _calculation_context() -> decimal.Context: """A bounded arithmetic context. ``Emax``/``Emin`` matter as much as precision: they turn an overflow into a catchable signal rather than an enormous allocation. """ return decimal.Context( prec=CALCULATION_PRECISION, Emax=9_999, Emin=-9_999, traps=[ decimal.InvalidOperation, decimal.DivisionByZero, decimal.Overflow, ], ) def _require_text(arguments: Mapping[str, object], key: str, limit: int) -> str: value = arguments.get(key) if not isinstance(value, str) or not value.strip(): raise ToolInputError(f"{key} must be a non-empty string") if len(value) > limit: raise ToolInputError(f"{key} must be at most {limit:,} characters") return value def _require_choice(arguments: Mapping[str, object], key: str, allowed: Mapping[str, object]) -> str: value = arguments.get(key) if not isinstance(value, str): raise ToolInputError(f"{key} must be a string") lowered = value.strip().lower() if lowered not in allowed: names = ", ".join(sorted(allowed)) raise ToolInputError(f"{key} must be one of: {names}") return lowered def _reject_unknown(arguments: Mapping[str, object], known: frozenset[str]) -> None: """Refuse arguments the tool does not understand. Ignoring an unrecognised argument is always the permissive outcome and hides a model's misreading of the schema, which is exactly the failure this tool set exists to make visible. """ unknown = sorted(set(map(str, arguments)) - known) if unknown: raise ToolInputError(f"unsupported arguments: {', '.join(unknown)}") # -------------------------------------------------------------------------- # calculate # -------------------------------------------------------------------------- _NUMBER_RE = re.compile(r"(?:\d+(?:\.\d+)?|\.\d+)(?:[eE][+-]?\d{1,4})?") _FUNCTION_RE = re.compile(r"[A-Za-z]+") _FUNCTIONS = frozenset({"abs", "sqrt", "round", "floor", "ceil", "min", "max"}) _OPERATORS = ("**", "//", "+", "-", "*", "/", "%", "(", ")", ",", "^") class _Tokeniser: """Turn an expression into a bounded token list, or refuse it. There is deliberately no ``eval`` anywhere in this module. The parser below accepts a fixed grammar and nothing else, so a model cannot reach Python semantics through an arithmetic tool however it shapes the string. """ def __init__(self, text: str) -> None: # An underscore *between digits* is a separator and is removed before # tokenising, so 1_000 is one number rather than two. A comma is # deliberately not treated this way: it separates min/max arguments, # and a token meaning both "thousands" and "next argument" would make # min(1,000) ambiguous. A model writing 1,000 gets a clear parse error # it can correct, which is better than a quietly different number. self.text = re.sub(r"(?<=\d)_(?=\d)", "", text) self.position = 0 self.tokens: list[tuple[str, str]] = [] def run(self) -> list[tuple[str, str]]: while self.position < len(self.text): character = self.text[self.position] if character.isspace(): self.position += 1 continue if len(self.tokens) >= MAX_TOKENS: raise ToolInputError(f"expression has more than {MAX_TOKENS} tokens") number = _NUMBER_RE.match(self.text, self.position) if number: self.tokens.append(("num", number.group())) self.position = number.end() continue word = _FUNCTION_RE.match(self.text, self.position) if word: # Case-folded so SQRT(16) is a recognised function rather than # an "unexpected character" three tokens from the real problem. name = word.group().lower() if name not in _FUNCTIONS: allowed = ", ".join(sorted(_FUNCTIONS)) raise ToolInputError( f"unknown function {name!r}; this tool supports only: {allowed}" ) self.tokens.append(("func", name)) self.position = word.end() continue for operator in _OPERATORS: if self.text.startswith(operator, self.position): self.tokens.append(("op", "**" if operator == "^" else operator)) self.position += len(operator) break else: raise ToolInputError( f"unexpected character {character!r} at position {self.position}" ) return self.tokens class _Parser: """Recursive descent over the token list, evaluating as it goes.""" def __init__(self, tokens: list[tuple[str, str]], context: decimal.Context) -> None: self.tokens = tokens self.index = 0 self.context = context def peek(self) -> tuple[str, str] | None: return self.tokens[self.index] if self.index < len(self.tokens) else None def take(self) -> tuple[str, str]: token = self.peek() if token is None: raise ToolInputError("expression ended unexpectedly") self.index += 1 return token def expect(self, value: str) -> None: token = self.take() if token != ("op", value): raise ToolInputError(f"expected {value!r} in the expression") def parse(self) -> Decimal: value = self.expression(0) if self.peek() is not None: raise ToolInputError("unexpected trailing text in the expression") return value def expression(self, depth: int) -> Decimal: if depth > MAX_PARSE_DEPTH: raise ToolInputError("expression is nested too deeply") value = self.term(depth + 1) while True: token = self.peek() if token == ("op", "+"): self.take() value = self.context.add(value, self.term(depth + 1)) elif token == ("op", "-"): self.take() value = self.context.subtract(value, self.term(depth + 1)) else: return value def term(self, depth: int) -> Decimal: if depth > MAX_PARSE_DEPTH: raise ToolInputError("expression is nested too deeply") value = self.unary(depth + 1) while True: token = self.peek() if token == ("op", "*"): self.take() value = self.context.multiply(value, self.unary(depth + 1)) elif token == ("op", "/"): self.take() value = self._divide(value, self.unary(depth + 1)) elif token == ("op", "//"): self.take() value = self._floor_divide(value, self.unary(depth + 1)) elif token == ("op", "%"): self.take() value = self._modulo(value, self.unary(depth + 1)) else: return value def unary(self, depth: int) -> Decimal: token = self.peek() if token == ("op", "-"): self.take() return self.context.minus(self.unary(depth + 1)) if token == ("op", "+"): self.take() return self.unary(depth + 1) return self.power(depth + 1) def power(self, depth: int) -> Decimal: base = self.primary(depth + 1) if self.peek() == ("op", "**"): self.take() exponent = self.unary(depth + 1) return self._power(base, exponent) return base def primary(self, depth: int) -> Decimal: if depth > MAX_PARSE_DEPTH: raise ToolInputError("expression is nested too deeply") kind, value = self.take() if kind == "num": try: return self.context.create_decimal(value) except decimal.DecimalException as exc: raise ToolInputError(f"{value!r} is not a usable number") from exc if kind == "func": self.expect("(") arguments = [self.expression(depth + 1)] while self.peek() == ("op", ","): self.take() arguments.append(self.expression(depth + 1)) self.expect(")") return self._apply(value, arguments) if (kind, value) == ("op", "("): inner = self.expression(depth + 1) self.expect(")") return inner raise ToolInputError(f"unexpected {value!r} in the expression") def _divide(self, left: Decimal, right: Decimal) -> Decimal: if right == 0: raise ToolInputError("division by zero") return self.context.divide(left, right) def _floor_divide(self, left: Decimal, right: Decimal) -> Decimal: if right == 0: raise ToolInputError("division by zero") return self.context.divide_int(left, right) def _modulo(self, left: Decimal, right: Decimal) -> Decimal: if right == 0: raise ToolInputError("division by zero") return self.context.remainder(left, right) def _power(self, base: Decimal, exponent: Decimal) -> Decimal: if exponent != exponent.to_integral_value(): raise ToolInputError("exponents must be whole numbers") try: whole = int(exponent) except (ValueError, OverflowError) as exc: raise ToolInputError("exponent is out of range") from exc if abs(whole) > MAX_EXPONENT: raise ToolInputError( f"exponents are limited to plus or minus {MAX_EXPONENT} so a short " "expression cannot request unbounded work" ) if base == 0 and whole < 0: raise ToolInputError("division by zero") try: return self.context.power(base, self.context.create_decimal(whole)) except decimal.DecimalException as exc: raise ToolInputError("the result is too large to represent") from exc def _apply(self, name: str, arguments: list[Decimal]) -> Decimal: if name in ("min", "max"): if len(arguments) < 2: raise ToolInputError(f"{name} needs at least two arguments") return min(arguments) if name == "min" else max(arguments) if len(arguments) != 1: raise ToolInputError(f"{name} takes exactly one argument") value = arguments[0] try: if name == "abs": return self.context.abs(value) if name == "sqrt": if value < 0: raise ToolInputError("sqrt of a negative number is not defined here") return self.context.sqrt(value) if name == "round": return value.quantize(Decimal(1), rounding=decimal.ROUND_HALF_UP) if name == "floor": return value.to_integral_value(rounding=decimal.ROUND_FLOOR) if name == "ceil": return value.to_integral_value(rounding=decimal.ROUND_CEILING) except decimal.DecimalException as exc: raise ToolInputError(f"{name} could not be computed for this value") from exc raise ToolInputError(f"unknown function {name!r}") def _format_decimal(value: Decimal) -> str: """Render without an exponent where that is reasonable, and never as a float, so 0.1 + 0.2 reads as 0.3 rather than 0.30000000000000004.""" if value == value.to_integral_value() and abs(value) < Decimal(10) ** 18: return str(value.quantize(Decimal(1))) normalised = value.normalize() exponent = normalised.as_tuple().exponent if isinstance(exponent, int) and -18 < exponent <= 0: return f"{normalised:f}" return str(normalised) CALCULATE_SPEC = ToolSpec( tool_id="calculate", version="1", description=( "Evaluate one arithmetic expression exactly and return the number. " "Use it for any sum, product, percentage, ratio or comparison of numbers, " "including money, where an exact answer matters. " "Supports + - * / // % and ** with brackets, and the functions " "abs, sqrt, round, floor, ceil, min and max. " "Do not use it for dates or times (use calculate_date), for converting between " "units (use convert_units), or for anything that is not a self-contained " "arithmetic expression. Call it once per expression; the answer is exact, " "so there is never a reason to call it again with the same expression." ), required_hosts=frozenset(), input_schema={ "type": "object", "additionalProperties": False, "required": ["expression"], "properties": { "expression": { "type": "string", "minLength": 1, "maxLength": MAX_EXPRESSION_CHARACTERS, "description": "Arithmetic only, for example (128 * 0.175) + 12.5", } }, }, ) def calculate_handler(arguments: Mapping[str, object], context: ToolContext) -> dict[str, object]: """Evaluate a bounded arithmetic expression with exact decimal semantics.""" _reject_unknown(arguments, frozenset({"expression"})) expression = _require_text(arguments, "expression", MAX_EXPRESSION_CHARACTERS) tokens = _Tokeniser(expression).run() if not tokens: raise ToolInputError("expression contains nothing to evaluate") calculation_context = _calculation_context() try: with decimal.localcontext(calculation_context): value = _Parser(tokens, calculation_context).parse() # Formatted inside the bounded context so a large result cannot # trip the caller's ambient decimal settings instead of ours. rendered = _format_decimal(value) except decimal.DecimalException as exc: raise ToolInputError("the expression could not be evaluated exactly") from exc return { "expression": expression, "result": rendered, "exact": True, } # -------------------------------------------------------------------------- # convert_units # -------------------------------------------------------------------------- #: Multiplicative units, grouped by dimension, expressed in a base unit. #: Conversions between different dimensions are refused rather than coerced. _UNITS: dict[str, tuple[str, str]] = {} _SCALES: dict[str, dict[str, str]] = { "length": { "mm": "0.001", "cm": "0.01", "m": "1", "km": "1000", "in": "0.0254", "ft": "0.3048", "yd": "0.9144", "mi": "1609.344", "nmi": "1852", }, "mass": { "mg": "0.000001", "g": "0.001", "kg": "1", "t": "1000", "oz": "0.028349523125", "lb": "0.45359237", "st": "6.35029318", }, "time": { "ms": "0.001", "s": "1", "min": "60", "h": "3600", "day": "86400", "week": "604800", }, "data": { "bit": "0.125", "byte": "1", "kb": "1000", "mb": "1000000", "gb": "1000000000", "tb": "1000000000000", "kib": "1024", "mib": "1048576", "gib": "1073741824", "tib": "1099511627776", }, "speed": { "mps": "1", "kph": "0.277777777777777778", "mph": "0.44704", "knot": "0.514444444444444444", }, } for _dimension, _members in _SCALES.items(): for _unit, _factor in _members.items(): _UNITS[_unit] = (_dimension, _factor) #: Temperature is affine, not multiplicative, so it gets its own path rather #: than a fudged scale factor. _TEMPERATURES = frozenset({"c", "f", "k"}) _ALL_UNITS = tuple(sorted(set(_UNITS) | _TEMPERATURES)) def _to_kelvin(value: Decimal, unit: str, context: decimal.Context) -> Decimal: if unit == "k": return value if unit == "c": return context.add(value, Decimal("273.15")) return context.add( context.multiply(context.subtract(value, Decimal(32)), Decimal(5) / Decimal(9)), Decimal("273.15"), ) def _from_kelvin(value: Decimal, unit: str, context: decimal.Context) -> Decimal: if unit == "k": return value if unit == "c": return context.subtract(value, Decimal("273.15")) return context.add( context.multiply(context.subtract(value, Decimal("273.15")), Decimal(9) / Decimal(5)), Decimal(32), ) CONVERT_UNITS_SPEC = ToolSpec( tool_id="convert_units", version="1", description=( "Convert a number from one unit to another and return the converted value. " "Handles length, mass, time, data size, speed and temperature. " "Both units must measure the same kind of thing; converting kilograms to metres " "is refused rather than guessed. " "Do not use it for plain arithmetic (use calculate) or for date arithmetic " "(use calculate_date). One call gives the exact answer, so do not repeat it. " "To convert several quantities, put them all in `conversions` in ONE call " "rather than calling this tool once per quantity." ), required_hosts=frozenset(), input_schema={ "type": "object", "additionalProperties": False, "properties": { "value": {"type": "number", "description": "The quantity to convert."}, "from_unit": {"type": "string", "enum": list(_ALL_UNITS)}, "to_unit": {"type": "string", "enum": list(_ALL_UNITS)}, "conversions": { "type": "array", "description": "Several conversions at once, instead of value/from_unit/to_unit.", "items": { "type": "object", "additionalProperties": False, "required": ["value", "from_unit", "to_unit"], "properties": { "value": {"type": "number"}, "from_unit": {"type": "string", "enum": list(_ALL_UNITS)}, "to_unit": {"type": "string", "enum": list(_ALL_UNITS)}, }, }, }, }, }, ) #: How many conversions one call may carry. Generous for any real request and #: still a bound, because a list is a way to ask for arbitrary work in one go. MAX_CONVERSIONS = 24 def convert_units_handler( arguments: Mapping[str, object], context: ToolContext ) -> dict[str, object]: """Convert between units of the same dimension, exactly where possible. SIX CONVERSIONS ASKED FOR AT ONCE WERE SIX CALLS, AND THE MODEL STOPPED AFTER THREE. "Convert these six quantities" is one request, and making it six calls put the burden of remembering how many were left on the part of the system least able to carry it. A benchmark workload failed that way every time: three conversions done correctly, the other three simply never attempted, and an answer confidently presenting half the list. Accepting the list in one call moves the counting to code, which can count. The single form is untouched, so nothing that worked before changes. """ _reject_unknown(arguments, frozenset({"value", "from_unit", "to_unit", "conversions"})) listed = arguments.get("conversions") if listed is not None: return _convert_many(listed, context) raw = arguments.get("value") if isinstance(raw, bool) or not isinstance(raw, int | float): raise ToolInputError("value must be a number") if isinstance(raw, float) and (raw != raw or raw in (float("inf"), float("-inf"))): raise ToolInputError("value must be a finite number") if abs(raw) > 10**30: raise ToolInputError("value is too large to convert") source = _require_choice(arguments, "from_unit", dict.fromkeys(_ALL_UNITS)) target = _require_choice(arguments, "to_unit", dict.fromkeys(_ALL_UNITS)) calculation_context = _calculation_context() try: with decimal.localcontext(calculation_context): # int goes through Decimal directly; float goes through repr so # 0.1 stays 0.1 rather than becoming its binary expansion. quantity = ( calculation_context.create_decimal(raw) if isinstance(raw, int) else calculation_context.create_decimal(repr(raw)) ) source_temperature = source in _TEMPERATURES target_temperature = target in _TEMPERATURES if source_temperature != target_temperature: raise ToolInputError( f"cannot convert {source} to {target}: they measure different things" ) if source_temperature: kelvin = _to_kelvin(quantity, source, calculation_context) if kelvin < 0: raise ToolInputError("that temperature is below absolute zero") converted = _from_kelvin(kelvin, target, calculation_context) dimension = "temperature" else: source_dimension, source_factor = _UNITS[source] target_dimension, target_factor = _UNITS[target] if source_dimension != target_dimension: raise ToolInputError( f"cannot convert {source} ({source_dimension}) to " f"{target} ({target_dimension}): they measure different things" ) base = calculation_context.multiply(quantity, Decimal(source_factor)) converted = calculation_context.divide(base, Decimal(target_factor)) dimension = source_dimension rendered_input = _format_decimal(quantity) rendered_result = _format_decimal(converted) except decimal.DecimalException as exc: raise ToolInputError("the conversion could not be computed") from exc return { "value": rendered_input, "from_unit": source, "to_unit": target, "dimension": dimension, "result": rendered_result, } def _convert_many(listed: object, context: ToolContext) -> dict[str, object]: """Every conversion in the list, or a refusal naming the one that failed. One bad entry does not lose the others: each carries its own answer or its own reason, in the order they were asked for, so a model reading the result can fix the one it got wrong rather than starting again. """ if not isinstance(listed, list | tuple) or not listed: raise ToolInputError("conversions must be a non-empty list") if len(listed) > MAX_CONVERSIONS: raise ToolInputError(f"conversions may contain at most {MAX_CONVERSIONS} entries") results: list[dict[str, object]] = [] for index, entry in enumerate(listed, start=1): if not isinstance(entry, Mapping): results.append({"position": index, "error": "each conversion must be an object"}) continue try: answer = convert_units_handler(dict(entry), context) except ToolInputError as exc: results.append({"position": index, "error": str(exc)}) continue answer["position"] = index results.append(answer) return { "conversions": results, "converted": sum(1 for item in results if "result" in item), "refused": sum(1 for item in results if "error" in item), } # -------------------------------------------------------------------------- # calculate_date # -------------------------------------------------------------------------- _DATE_UNITS = { "days": 1, "weeks": 7, } def _parse_iso_date(value: object, label: str) -> _datetime.date: if not isinstance(value, str) or not value.strip(): raise ToolInputError(f"{label} must be a date in YYYY-MM-DD form") text = value.strip() try: return _datetime.date.fromisoformat(text) except ValueError as exc: raise ToolInputError( f"{label} must be a date in YYYY-MM-DD form, for example 2026-08-16" ) from exc CALCULATE_DATE_SPEC = ToolSpec( tool_id="calculate_date", version="1", description=( "Do exact calendar arithmetic on ISO dates written as YYYY-MM-DD. " "operation='difference' returns how far apart two dates are and needs date and " "other_date. operation='add' shifts a date and needs date, amount and unit; " "amount may be negative to go backwards. operation='weekday' returns the day name " "and needs only date. " "It has no clock: it cannot tell you today's date, so do not call it to find out. " "Do not use it for plain arithmetic (use calculate). One call is exact." ), required_hosts=frozenset(), input_schema={ "type": "object", "additionalProperties": False, "required": ["operation", "date"], "properties": { "operation": {"type": "string", "enum": ["difference", "add", "weekday"]}, "date": {"type": "string", "description": "ISO date, YYYY-MM-DD."}, "other_date": { "type": "string", "description": "Second ISO date. Required for operation='difference'.", }, "amount": { "type": "integer", "minimum": -400_000, "maximum": 400_000, "description": "Whole units to add. Required for operation='add'.", }, "unit": {"type": "string", "enum": list(_DATE_UNITS)}, }, }, ) def calculate_date_handler( arguments: Mapping[str, object], context: ToolContext ) -> dict[str, object]: """Exact calendar arithmetic. No clock, no timezone, no locale.""" _reject_unknown( arguments, frozenset({"operation", "date", "other_date", "amount", "unit"}) ) operation = _require_choice( arguments, "operation", {"difference": 1, "add": 1, "weekday": 1} ) date = _parse_iso_date(arguments.get("date"), "date") if operation == "weekday": return { "operation": operation, "date": date.isoformat(), "weekday": date.strftime("%A"), "iso_weekday": date.isoweekday(), } if operation == "difference": other = _parse_iso_date(arguments.get("other_date"), "other_date") days = (other - date).days return { "operation": operation, "date": date.isoformat(), "other_date": other.isoformat(), "days": days, "weeks": round(days / 7, 4), "direction": "after" if days > 0 else ("before" if days < 0 else "same day"), } amount = arguments.get("amount") if isinstance(amount, bool) or not isinstance(amount, int): raise ToolInputError("amount must be a whole number for operation='add'") if not -400_000 <= amount <= 400_000: raise ToolInputError("amount must be between -400000 and 400000") # AN OPTIONAL ARGUMENT THAT IS REQUIRED IN PRACTICE IS A TRAP. # # ``unit`` is not in the schema's ``required`` list, so a model reading the # schema leaves it out -- and then "add 45 to this date" was refused with # "unit must be a string", which reads like the model sent the wrong type # rather than nothing at all. Days is what "add 45" means to everybody, so # it is the default, and the answer says which unit it used either way. unit = ( _require_choice(arguments, "unit", _DATE_UNITS) if arguments.get("unit") is not None else "days" ) try: result = date + _datetime.timedelta(days=amount * _DATE_UNITS[unit]) except (OverflowError, ValueError) as exc: raise ToolInputError("the resulting date is outside the supported range") from exc return { "operation": operation, "date": date.isoformat(), "amount": amount, "unit": unit, "result": result.isoformat(), "weekday": result.strftime("%A"), } # -------------------------------------------------------------------------- # extract_from_text # -------------------------------------------------------------------------- _EXTRACTORS: Mapping[str, re.Pattern[str]] = { "number": re.compile(r"[-+]?\d{1,15}(?:[.,]\d+)?"), "integer": re.compile(r"[-+]?\d{1,15}(?![\d.,])"), "date_iso": re.compile(r"\d{4}-\d{2}-\d{2}"), "email": re.compile(r"[A-Za-z0-9._%+-]{1,64}@[A-Za-z0-9.-]{1,190}\.[A-Za-z]{2,24}"), "url": re.compile(r"https?://[^\s<>\"')]{1,500}"), "percentage": re.compile(r"[-+]?\d{1,12}(?:\.\d+)?\s?%"), "currency_amount": re.compile( r"(?:[$£€¥]\s?[-+]?\d{1,15}(?:[.,]\d+)*" r"|[-+]?\d{1,15}(?:[.,]\d+)*\s?(?:GBP|USD|EUR|JPY))" ), } EXTRACT_FROM_TEXT_SPEC = ToolSpec( tool_id="extract_from_text", version="1", description=( "Scan a piece of text and return every value of one kind, in the order they appear. " "kind='number', 'integer', 'date_iso', 'email', 'url', 'percentage' or " "'currency_amount'. Use it when you need to be certain you have found all of " "something in a long passage, which is easy to get wrong by reading. " "It matches patterns literally and does not interpret meaning, so it cannot answer " "questions about the text or summarise it. To find the passages about a topic use " "search_document instead. To check whether an exact sentence appears use " "verify_quote instead. Do not call it more than once for the same kind and text: " "the result is complete the first time." ), required_hosts=frozenset(), input_schema={ "type": "object", "additionalProperties": False, "required": ["text", "kind"], "properties": { "text": {"type": "string", "minLength": 1, "maxLength": MAX_TEXT_CHARACTERS}, "kind": {"type": "string", "enum": sorted(_EXTRACTORS)}, }, }, ) def extract_from_text_handler( arguments: Mapping[str, object], context: ToolContext ) -> dict[str, object]: """Return every literal match of one kind, bounded and in document order.""" _reject_unknown(arguments, frozenset({"text", "kind"})) text = _require_text(arguments, "text", MAX_TEXT_CHARACTERS) kind = _require_choice(arguments, "kind", _EXTRACTORS) # Normalising to NFKC first means a full-width digit or a non-breaking # space cannot hide a match from the pattern. prepared = unicodedata.normalize("NFKC", text) matches: list[str] = [] truncated = False for found in _EXTRACTORS[kind].finditer(prepared): if len(matches) >= MAX_MATCHES: truncated = True break matches.append(found.group().strip()) return { "kind": kind, "count": len(matches), "matches": matches, # Stated rather than implied: a caller must be able to tell a complete # answer from a capped one, and a silent cap is a wrong answer. "complete": not truncated, } # -------------------------------------------------------------------------- # search_document # -------------------------------------------------------------------------- #: Ranked passages returned per call. Small on purpose. The sub-7B failure #: mode for retrieval is context utilisation, not retrieval quality: flooding a #: small model with several long chunks is what makes RAG a net negative at that #: scale. Short spans are a different proposition from chunks. MAX_PASSAGES = 5 DEFAULT_PASSAGES = 3 MAX_QUERY_CHARACTERS = 200 MAX_PASSAGE_CHARACTERS = 400 MAX_SENTENCES = 800 _SENTENCE_END_RE = re.compile(r"(?<=[.!?])\s+|\n{2,}") _WORD_RE = re.compile(r"[a-z0-9][a-z0-9'_-]*") #: Terms too common to discriminate. Deliberately short: an aggressive stop #: list throws away the exact-token matches lexical search exists to catch. _STOP_WORDS = frozenset( { "a", "an", "and", "are", "as", "at", "be", "by", "for", "from", "has", "he", "in", "is", "it", "its", "of", "on", "or", "that", "the", "to", "was", "were", "will", "with", } ) # BM25 parameters. The standard defaults; nothing here is tuned, because tuning # on no data would be inventing a number. _BM25_K1 = 1.5 _BM25_B = 0.75 def _tokenise(text: str) -> list[str]: """Fold to comparable word tokens. NFKC is applied here rather than to the whole document, so reported offsets always point into the text the caller supplied. Normalising first and then reporting positions into the normalised copy would hand back offsets that do not line up with the caller's own string. """ return _WORD_RE.findall(unicodedata.normalize("NFKC", text).lower()) def _split_sentences(text: str) -> list[tuple[int, str]]: """Return ``(start_offset, sentence)`` pairs, bounded in number. Offsets are kept so a caller can point at the source rather than only quoting it, which is what makes a retrieved span checkable. """ spans: list[tuple[int, str]] = [] cursor = 0 for piece in _SENTENCE_END_RE.split(text): if not piece: continue start = text.find(piece, cursor) if start < 0: start = cursor cursor = start + len(piece) leading = len(piece) - len(piece.lstrip()) stripped = piece.strip() if stripped: spans.append((start + leading, stripped)) if len(spans) >= MAX_SENTENCES: break return spans SEARCH_DOCUMENT_SPEC = ToolSpec( tool_id="search_document", version="1", description=( "Find the passages in a supplied document that are about a topic, ranked by " "relevance, and return them with their positions. " "Use it when the document is long and you need the parts that matter, rather than " "reading all of it: models reliably miss material in the middle of a long passage. " "It matches words, not meaning, so use the words you expect the document to use. " "For finding every value of a kind, such as all dates or all amounts, use " "extract_from_text instead. To check whether an exact sentence appears, use " "verify_quote instead. One call returns the ranked passages; do not repeat it with " "the same query." ), required_hosts=frozenset(), input_schema={ "type": "object", "additionalProperties": False, "required": ["text", "query"], "properties": { "text": {"type": "string", "minLength": 1, "maxLength": MAX_TEXT_CHARACTERS}, "query": {"type": "string", "minLength": 1, "maxLength": MAX_QUERY_CHARACTERS}, "max_results": { "type": "integer", "minimum": 1, "maximum": MAX_PASSAGES, "default": DEFAULT_PASSAGES, }, }, }, ) def search_document_handler( arguments: Mapping[str, object], context: ToolContext ) -> dict[str, object]: """Rank supplied passages against a query with BM25, and return short spans. Lexical rather than dense on purpose. For supplied-document retrieval BM25 beats the popular dense retrievers, and it needs no embedding model, no vector store and no extra weights, which keeps the agent binary self-contained and the runtime free of outbound requests. """ _reject_unknown(arguments, frozenset({"text", "query", "max_results"})) text = _require_text(arguments, "text", MAX_TEXT_CHARACTERS) query = _require_text(arguments, "query", MAX_QUERY_CHARACTERS) limit = arguments.get("max_results", DEFAULT_PASSAGES) if isinstance(limit, bool) or not isinstance(limit, int): raise ToolInputError("max_results must be an integer") if not 1 <= limit <= MAX_PASSAGES: raise ToolInputError(f"max_results must be between 1 and {MAX_PASSAGES}") sentences = _split_sentences(text) if not sentences: return {"query": query, "count": 0, "matches": [], "passages_searched": 0} terms = [term for term in _tokenise(query) if term not in _STOP_WORDS] if not terms: raise ToolInputError( "the query is only common words; use the specific words you expect in the text" ) tokenised = [_tokenise(sentence) for _, sentence in sentences] lengths = [len(tokens) for tokens in tokenised] total = len(tokenised) average_length = (sum(lengths) / total) if total else 0.0 document_frequency: dict[str, int] = {} for tokens in tokenised: for term in set(tokens): if term in terms: document_frequency[term] = document_frequency.get(term, 0) + 1 scored: list[tuple[float, int]] = [] for index, tokens in enumerate(tokenised): if not tokens: continue score = 0.0 counts: dict[str, int] = {} for token in tokens: counts[token] = counts.get(token, 0) + 1 for term in terms: frequency = counts.get(term, 0) if not frequency: continue appearances = document_frequency.get(term, 0) idf = math.log(1 + (total - appearances + 0.5) / (appearances + 0.5)) denominator = frequency + _BM25_K1 * ( 1 - _BM25_B + _BM25_B * (lengths[index] / average_length if average_length else 1) ) score += idf * (frequency * (_BM25_K1 + 1)) / denominator if score > 0: scored.append((score, index)) # Sorted by score, then by position, so an identical query always returns # an identical answer. A retrieval tool that reorders equal results between # runs makes every downstream failure unreproducible. scored.sort(key=lambda item: (-item[0], item[1])) matches = [] for score, index in scored[:limit]: start, sentence = sentences[index] excerpt = sentence[:MAX_PASSAGE_CHARACTERS] matches.append( { "text": excerpt + ("..." if len(sentence) > MAX_PASSAGE_CHARACTERS else ""), "start": start, "score": round(score, 4), } ) return { "query": query, "count": len(matches), "matches": matches, "passages_searched": total, } # -------------------------------------------------------------------------- # verify_quote # -------------------------------------------------------------------------- MAX_QUOTE_CHARACTERS = 500 #: Above this the near-match is close enough to be worth showing as a probable #: paraphrase; below it, reporting a "closest" match would be noise. NEAR_MATCH_THRESHOLD = 0.6 #: Candidate sentences put through the expensive comparison. Bounded so a long #: document cannot turn one call into a quadratic scan. MAX_NEAR_MATCH_CANDIDATES = 8 def _normalise_for_comparison(value: str) -> str: """Fold the differences a model introduces when re-typing a quotation. Case, run-together whitespace and the typographic quotation marks a model substitutes for the plain ones in the source. Not a similarity measure: this is still an exact-containment test, just one that does not fail on a change of curly apostrophe. """ folded = unicodedata.normalize("NFKC", value).casefold() for fancy, plain in (("‘", "'"), ("’", "'"), ("“", '"'), ("”", '"')): folded = folded.replace(fancy, plain) return " ".join(folded.split()) VERIFY_QUOTE_SPEC = ToolSpec( tool_id="verify_quote", version="1", description=( "Check whether an exact quotation really appears in a supplied document, and say " "where. Use it before attributing a quotation or a specific claim to the text, " "because a quotation that is nearly right is still wrong. " "It returns found=true with a position, or found=false with the closest wording it " "did find, so you can correct the quotation rather than repeat it. " "Differences of case, spacing and curly quotation marks are ignored; differences of " "wording are not. To find passages about a topic use search_document instead. " "One call settles it." ), required_hosts=frozenset(), input_schema={ "type": "object", "additionalProperties": False, "required": ["text", "quote"], "properties": { "text": {"type": "string", "minLength": 1, "maxLength": MAX_TEXT_CHARACTERS}, "quote": {"type": "string", "minLength": 1, "maxLength": MAX_QUOTE_CHARACTERS}, }, }, ) def verify_quote_handler( arguments: Mapping[str, object], context: ToolContext ) -> dict[str, object]: """Exact-containment check with a diagnostic near match. Deterministic. This is the one verification shape the evidence supports for a small model: an external, deterministic soundness signal. A tool that asked the model to review its own answer would be the shape that is documented to make reasoning worse. """ _reject_unknown(arguments, frozenset({"text", "quote"})) text = _require_text(arguments, "text", MAX_TEXT_CHARACTERS) quote = _require_text(arguments, "quote", MAX_QUOTE_CHARACTERS) haystack = _normalise_for_comparison(text) needle = _normalise_for_comparison(quote) if not needle: raise ToolInputError("quote contains no comparable text") position = haystack.find(needle) if position >= 0: return { "found": True, "quote": quote, "normalised_position": position, "occurrences": haystack.count(needle), "closest": None, "similarity": 1.0, } # Not present. Report the nearest wording so the caller can correct the # quotation instead of asserting it again. Candidates are prefiltered by # cheap token overlap; only a handful reach the expensive comparison. sentences = _split_sentences(text) quote_tokens = set(_tokenise(quote)) ranked = sorted( sentences, key=lambda item: -len(quote_tokens.intersection(_tokenise(item[1]))), )[:MAX_NEAR_MATCH_CANDIDATES] best_ratio = 0.0 best_text = None for _, sentence in ranked: ratio = difflib.SequenceMatcher( None, needle, _normalise_for_comparison(sentence) ).ratio() if ratio > best_ratio: best_ratio = ratio best_text = sentence[:MAX_PASSAGE_CHARACTERS] return { "found": False, "quote": quote, "normalised_position": None, "occurrences": 0, "closest": best_text if best_ratio >= NEAR_MATCH_THRESHOLD else None, "similarity": round(best_ratio, 4), } # -------------------------------------------------------------------------- # Registration # -------------------------------------------------------------------------- LOCAL_SPECS: tuple[ToolSpec, ...] = ( CALCULATE_SPEC, CALCULATE_DATE_SPEC, CONVERT_UNITS_SPEC, EXTRACT_FROM_TEXT_SPEC, SEARCH_DOCUMENT_SPEC, VERIFY_QUOTE_SPEC, ) _HANDLERS = { CALCULATE_SPEC.ref: calculate_handler, CALCULATE_DATE_SPEC.ref: calculate_date_handler, CONVERT_UNITS_SPEC.ref: convert_units_handler, EXTRACT_FROM_TEXT_SPEC.ref: extract_from_text_handler, SEARCH_DOCUMENT_SPEC.ref: search_document_handler, VERIFY_QUOTE_SPEC.ref: verify_quote_handler, } def register_local_tools( registry: Registry, *, only: frozenset[object] | None = None ) -> tuple[ToolSpec, ...]: """Add the local, no-egress tools to ``registry`` and return what was added. Registration is not approval. These become *installable*; the operator's :class:`~distinct_tools.approval.OperatorPolicy` still decides whether any of them is advertised or callable. ``only`` restricts registration to a set of :class:`~distinct_tools.core.ToolRef`, which is how the approved subset is built without registering the rest and filtering afterwards. """ added: list[ToolSpec] = [] for spec in LOCAL_SPECS: if only is not None and spec.ref not in only: continue registry.register(spec, _HANDLERS[spec.ref]) added.append(spec) return tuple(added) def local_registry() -> Registry: """A registry containing exactly the local tool set. Convenience only.""" registry = Registry() register_local_tools(registry) return registry __all__ = [ "CALCULATE_DATE_SPEC", "CALCULATE_SPEC", "CONVERT_UNITS_SPEC", "DEFAULT_PASSAGES", "EXTRACT_FROM_TEXT_SPEC", "LOCAL_SPECS", "MAX_EXPONENT", "MAX_EXPRESSION_CHARACTERS", "MAX_MATCHES", "MAX_PASSAGES", "MAX_PASSAGE_CHARACTERS", "MAX_QUOTE_CHARACTERS", "MAX_TEXT_CHARACTERS", "NEAR_MATCH_THRESHOLD", "SEARCH_DOCUMENT_SPEC", "VERIFY_QUOTE_SPEC", "calculate_date_handler", "calculate_handler", "convert_units_handler", "extract_from_text_handler", "local_registry", "register_local_tools", "search_document_handler", "verify_quote_handler", ]