"""Shared plumbing for tool endpoints. One decision worth stating: recoverable failures come back as HTTP 200 with an explanatory body, not as a 4xx. Open WebUI turns any status >= 400 into the string `{"error": "HTTP error 502: ..."}` and marks the call failed in the UI. For a genuine fault that is right. But "that page is behind a login" and "the search returned nothing" are not faults, they are *findings* - and they are exactly the findings that should stop the model inventing an answer. Handed back as prose in a successful call, they read as information the model can act on. Handed back as an error, they read as something to retry. 401 and 429 stay as real status codes; those are about the caller, not the question. """ from __future__ import annotations import functools import logging import httpx from fastapi import HTTPException from .budget import SOURCES_PREFIX, SOURCES_SEPARATOR, cap from .httpc import FetchError, explain from .security import UnsafeURL log = logging.getLogger(__name__) def tool(func): """Wrap a tool endpoint: normalise failures, enforce the size cap.""" @functools.wraps(func) async def wrapper(*args, **kwargs): try: result = await func(*args, **kwargs) except UnsafeURL as exc: return f"CANNOT DO THAT: {exc}" except FetchError as exc: return f"FAILED: {exc}" except httpx.HTTPError as exc: return f"FAILED: {explain(exc, 'the requested source')}" except HTTPException as exc: # The rate limiter raises these carrying a message written for the # model to act on. Letting them reach the generic handler below # replaced "you have used all 30 network probes this hour" with # "internal error" - which reads like our bug, tells the user # nothing, and invites exactly the retry the message forbids. return f"FAILED: {exc.detail}" except Exception: # Never leak a traceback into the model's context - it is a lot # of tokens describing our bug, which the model cannot fix and # may try to explain to the user. log.exception("unhandled error in %s", func.__name__) return ( "FAILED: This tool hit an internal error. Do not retry it with the same " "input. Use a different tool, or tell the user you could not look this up." ) return cap(result) if isinstance(result, str) else result return wrapper NO_RESULTS = ( "NO RESULTS. Nothing matched. Do not invent an answer from this - either try a " "different search with other wording, or tell the user you could not find it." ) def cite_line(first_author: str | None, title: str, venue: str | None = None, year: str | int | None = None, doi: str | None = None, *, et_al: bool = False) -> str: """One citation line the model can lift verbatim. Lifted out of `search_papers`, where it was introduced after the model wrote "the TIME2 trial (NEJM 2022, n=148)" for a study actually titled "Calorie Restriction with or without Time-Restricted Eating in Weight Loss" with 139 participants - correct finding, invented name and count. Identifiers it can copy are the cheapest fix for that. It is shared with `fetch_page` because the same failure recurred one layer down: the model fetched a real paper, transcribed every number from it correctly, and then attributed it to the wrong authors in the wrong journal, because the fetched text began at the abstract and carried no bibliographic data at all. Having the numbers but not the citation is exactly the state in which a plausible one gets invented. """ surname = "" if first_author and first_author.strip(): name = first_author.strip() # Two conventions arrive here. Citation meta tags and most registries # write "Surname, Given"; page bylines and OpenAlex write "Given # Surname". Taking the last token of the first form yields the given # name, which is how arXiv's "Vaswani, Ashish" came out as # 'Ashish et al.' - a citation that looks authoritative and is wrong # in the one field anyone would use to look the paper up. if _is_organisation(name): # "Cleveland Clinic" is not a Dr Clinic. An organisational # author is cited whole. surname = name else: surname = name.split(",")[0].strip() if "," in name else name.split()[-1] head = f'{surname}{" et al." if et_al else ""}, "{title}"' if surname else f'"{title}"' if venue: head += f", {venue}" if year: head += f" ({year})" if doi: d = doi.strip() # OpenAlex hands over a full https://doi.org/... URL; a citation_doi # meta tag hands over the bare 10.x form. Both should read as a DOI. if d.startswith("10."): d = f"doi:{d}" head += f". {d}" return head _ORG_WORDS = frozenset(""" clinic hospital institute institutes university college school foundation association society agency administration organization organisation department ministry council centre center committee commission authority bureau office group inc ltd llc plc corporation company co laboratory laboratories labs lab staff team editors editorial board service services network press news journal magazine times post health medicine medical trust federation union consortium collaboration project program programme initiative alliance partnership """.split()) # A site's own name as author: "ElectricalLicenseRenewal com" is a domain with # the dot lost, not a Dr com. _TLD_WORDS = frozenset("com org net edu gov io co uk us de fr eu info biz".split()) # What a byline carries after the person: "Kelly Johnson-Arbor; MD Medical # Toxicologist" split on the semicolon into two authors, and the second one # made the citation read "Johnson-Arbor et al." for a single-author page. _CREDENTIALS = frozenset( "md phd do rn np pa mph msc ms ma mba dds dmd pharmd dvm frcp frcs faan facp facs " "bsc bs ba llb jd edd dphil mrcp cpa pe rd ld ccc-slp".split() ) def _is_organisation(name: str) -> bool: """A multi-word author with no comma whose last word is an institutional noun.""" if "," in name: return False words = [w.strip(".").lower() for w in name.split()] if len(words) < 2: return False return (words[-1] in _ORG_WORDS or words[-1] in _TLD_WORDS or words[0] in {"the"} or any(w in {"of", "for", "&"} for w in words[1:-1])) def clean_authors(names) -> list[str]: """Drop the entries in an author list that are not people or organisations. Bylines and metadata split into a list on ";" and "," carry credentials ("MD Medical Toxicologist"), degrees ("PhD") and role words as separate items. Each becomes a phantom co-author, and one phantom co-author is enough to turn a single-author citation into "et al.". """ kept: list[str] = [] for raw in names or []: name = " ".join(str(raw or "").split()).strip(" ;,") if not name: continue tokens = [t.strip(".,").lower() for t in name.split()] if tokens and tokens[0] in _CREDENTIALS: continue if all(t in _CREDENTIALS for t in tokens): continue # A trailing degree on a real name is fine to strip: "J. Smith, MD" # arrives as two items and the "MD" is dropped above; "J. Smith MD" # arrives as one and loses the suffix here. while len(tokens) > 1 and tokens[-1] in _CREDENTIALS: tokens.pop() name = " ".join(name.split()[:-1]).rstrip(",") kept.append(name) return kept def sources_line(*entries: str) -> str: """The trailing line a tool-assisted answer should end with, ready to copy. Two graded rounds established that the model will not construct a citation from material it has to assemble - a `# Source:` header field at the top of a fetched page, a `Homepage:` row among five others - but will lift one that is already written in the shape the answer wants. Of forty answers, exactly one carried a URL, and it was the one whose tool had printed `CITE AS:` lines. This is that observation applied to every tool that returns a single identifiable document. Deliberately not accompanied by an instruction to use it. The same rounds showed tool output that *reports* being acted on and tool output that *instructs* being weighed and often declined, so the line is offered as a fact about the document rather than as a request. `cap()` protects it from truncation, which is why it must stay one line. """ kept = [" ".join(e.split()) for e in entries if e and e.strip()] if not kept: return "" return SOURCES_SEPARATOR + SOURCES_PREFIX + " ยท ".join(dict.fromkeys(kept)) from pydantic import BaseModel, ConfigDict, field_validator # noqa: E402 class ToolModel(BaseModel): """Base for every tool's request body. Models hand over `""` for optional parameters they have decided not to use, roughly as often as they omit them. Open WebUI strips empty *query* parameters for exactly this reason but does no such thing for body fields, so the coercion has to happen here or half the optional parameters in this suite would arrive as empty strings and fail validation. """ model_config = ConfigDict(extra="ignore", str_strip_whitespace=True) @field_validator("*", mode="before") @classmethod def _blank_to_none(cls, value): if isinstance(value, str) and not value.strip(): return None return value