1510 lines
71 KiB
Python
1510 lines
71 KiB
Python
"""Web search and page reading.
|
||
|
||
These are the tools the assistant should reach for most, and their
|
||
descriptions are written to make that happen: each one says when to use it,
|
||
when not to, and what to do next with the result.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import html
|
||
import html as html_mod
|
||
import json
|
||
import logging
|
||
import re
|
||
|
||
import httpx
|
||
from typing import Literal
|
||
from urllib.parse import quote, urljoin, urlparse
|
||
|
||
from fastapi import APIRouter, Depends, Header
|
||
from pydantic import Field, field_validator
|
||
from starlette.concurrency import run_in_threadpool
|
||
|
||
from .. import cache
|
||
from ..budget import paginate
|
||
from ..httpc import FetchError, explain, get_bytes, get_client, get_json, get_local_client
|
||
from ..limits import (fetch_history, note_failure, note_fetch, note_success,
|
||
recall_failure)
|
||
from ..security import require_auth, validate_url
|
||
from ..settings import settings
|
||
from ..textwin import search_window
|
||
from ..toolbase import NO_RESULTS, ToolModel, cite_line, clean_authors, sources_line, tool
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
router = APIRouter(prefix="/tools", tags=["web"], dependencies=[Depends(require_auth)])
|
||
|
||
|
||
# ---------------------------------------------------------------- web_search
|
||
|
||
class WebSearchRequest(ToolModel):
|
||
query: str = Field(
|
||
description="Search terms as they would appear on a good result page, not a conversational "
|
||
"question; include model numbers, error strings and versions verbatim. Search "
|
||
"for the thing, not for your answer - 'Bob Dylan mother', not 'Mary Lee "
|
||
"Pfeiffer mother of Bob Dylan': a query containing your guess only returns "
|
||
"pages that agree with it.",
|
||
)
|
||
max_results: int = Field(
|
||
default=6,
|
||
ge=1,
|
||
le=15,
|
||
description="Results to return, default 6. Raise it only to compare many sources.",
|
||
)
|
||
category: Literal["general", "it", "science", "news"] = Field(
|
||
default="general",
|
||
description="'general' (default) for most questions, including consumer health, medicines, "
|
||
"everyday physics, history and people. 'it' for programming, Linux, networking "
|
||
"and sysadmin - Stack Overflow, Ask Ubuntu, Super User, MDN, man pages - far "
|
||
"better than 'general' for those. 'science' searches scholarly indexes only "
|
||
"(arXiv, PubMed, OpenAlex, Crossref): for when you want papers, never for a "
|
||
"practical question. 'news' for current events.",
|
||
)
|
||
time_range: Literal["any", "day", "week", "month", "year"] = Field(
|
||
default="any",
|
||
description="Restrict to recently published pages when the answer changes over time - "
|
||
"current versions, recent events, outages. Default 'any'.",
|
||
)
|
||
site: str | None = Field(
|
||
default=None,
|
||
description="Restrict to one domain, e.g. 'cisco.com'. Good for vendor documentation.",
|
||
)
|
||
language: str | None = Field(
|
||
default=None,
|
||
description="Two-letter language code, e.g. 'en'. Empty means any.",
|
||
)
|
||
|
||
|
||
# A practical question about a medicine, as opposed to a request for research
|
||
# on one. Both the tool hint and the category note below key off this: asked
|
||
# about a dose, the scholarly index returns papers and the model falls back on
|
||
# remembered numbers, which is the one domain where that is least acceptable.
|
||
_PRACTICAL_MEDICINE = re.compile(
|
||
r"\bmg\b|\bmcg\b|\bmilligrams?\b|\bdos(?:e|es|age|ing)\b(?!\s*[-–]?\s*(?:response|dependent|effect|finding))"
|
||
r"|overdose|maximum daily"
|
||
r"|side[\s-]?effects?|drug interactions?|contraindicat|\bnsaid|\bpharmac"
|
||
r"|how much .{0,30}(?:can|should) (?:i|you|they|he|she) take",
|
||
re.I,
|
||
)
|
||
|
||
_TOOL_HINTS: tuple[tuple[re.Pattern[str], str], ...] = (
|
||
# Searching for cable impedance or motor-start limits means the model is about to
|
||
# work a calculation by hand off whatever numbers the results happen to contain.
|
||
# The suite has the tabulated data and the published criteria; say so here, at the
|
||
# moment it is choosing, rather than hoping the system prompt is remembered.
|
||
(
|
||
re.compile(
|
||
r"(?:voltage[\s-]*drop|vdrop|volt\s*drop)[^\n]*"
|
||
r"(?:motor|3[\s-]*phase|three[\s-]*phase|feeder|conductor|cable|awg)"
|
||
r"|(?:motor|3[\s-]*phase|three[\s-]*phase|feeder|conductor|cable|awg)[^\n]*"
|
||
r"(?:voltage[\s-]*drop|vdrop|volt\s*drop)"
|
||
r"|nec\s*(?:chapter\s*9\s*)?table\s*9"
|
||
r"|conductor\s+(?:impedance|reactance)"
|
||
r"|motor[\s-]*start(?:ing)?[^\n]*(?:voltage|volts|drop|sag|dip)"
|
||
r"|(?:ampacity|430\.2[0-9])[^\n]*motor|motor[^\n]*ampacity",
|
||
re.I,
|
||
),
|
||
"[This suite computes this exactly — do not work it out by hand from these results. "
|
||
"Call `wiring_calc` with calculation=three_phase_drop (awg, length, current, voltage, "
|
||
"power_factor, conduit). It uses the tabulated NEC Chapter 9 Table 9 resistance *and* "
|
||
"reactance, applies the full \u221a3 \u00d7 I \u00d7 (R\u00b7cos\u03c6 + X\u00b7sin\u03c6), and returns the "
|
||
"published IEEE 399 Table 9-1 starting-voltage criteria and the NEC motor "
|
||
"conductor-sizing articles alongside the answer. Those citations are verified; a "
|
||
"standard you half-remember from a search snippet is not.]",
|
||
),
|
||
# Dosing questions have primary sources that are free, authoritative and
|
||
# readable, and the model reached for none of them: asked to compare two
|
||
# analgesics it ran two searches, got twelve results containing no label,
|
||
# no dose and no consumer-health page, and answered every figure from
|
||
# memory. The figures happened to be right. Naming the sites is what turns
|
||
# "search again" into something it can actually do.
|
||
(
|
||
_PRACTICAL_MEDICINE,
|
||
"[For medicines and dosing there are primary sources you can read: the official "
|
||
"product label (search again with site:dailymed.nlm.nih.gov, or site:medicines.org.uk "
|
||
"in the UK) or a national health service page (site:medlineplus.gov, site:nhs.uk). "
|
||
"Read one with fetch_page and cite it. A dose taken from a blog post or from a search "
|
||
"snippet is not a number to hand to someone. Labels differ by product and strength - "
|
||
"read the label for the product the question names, not a generic one.]",
|
||
),
|
||
)
|
||
|
||
|
||
def _tool_hint(query: str) -> str | None:
|
||
for pattern, hint in _TOOL_HINTS:
|
||
if pattern.search(query):
|
||
return hint
|
||
return None
|
||
|
||
|
||
def _category_note(category: str, query: str) -> str | None:
|
||
"""Say so when the chosen index cannot answer the question that was asked."""
|
||
if category == "science" and _PRACTICAL_MEDICINE.search(query):
|
||
return ("[Note: 'science' searches scholarly indexes only. This reads as a practical "
|
||
"question about a medicine rather than a request for research papers - "
|
||
"category 'general' finds the product label and the health-service pages "
|
||
"that actually answer it.]")
|
||
return None
|
||
|
||
|
||
# Words carrying no topic. Kept short on purpose: this only has to make an
|
||
# overlap fraction mean something, not to be a real stoplist.
|
||
_STOPWORDS = frozenset("""
|
||
a an and are as at be by for from has have how in is it its of on or that the
|
||
their there these this to was were what when where which who why will with do
|
||
does did can could should would about into over under than then them they you
|
||
your our us not no but so such only just more most some any each while after
|
||
before between during also very much many one two get make used using
|
||
""".split())
|
||
|
||
|
||
def _terms(query: str) -> list[str]:
|
||
raw = re.findall(r"[A-Za-z0-9][A-Za-z0-9._+-]*", query.lower())
|
||
return [w for w in raw if len(w) >= 3 and w not in _STOPWORDS]
|
||
|
||
|
||
def _overlap(terms: list[str], haystack: str) -> float:
|
||
if not terms:
|
||
return 1.0
|
||
hay = haystack.lower()
|
||
return sum(1 for t in terms if t in hay) / len(terms)
|
||
|
||
|
||
def _quoted_phrases(query: str) -> list[str]:
|
||
return [p.strip().lower() for p in re.findall(r'"([^"]{3,})"', query) if p.strip()]
|
||
|
||
|
||
# A term specific enough that its absence from every single result is itself a
|
||
# finding: a surname, a journal, a place, a part number. Deliberately narrow -
|
||
# ordinary words scatter across unrelated pages and prove nothing.
|
||
_ALNUM_ID = re.compile(r"^(?=.*[A-Za-z])(?=.*\d)[A-Za-z0-9._\-]{4,}$")
|
||
|
||
|
||
def _distinctive(query: str) -> list[str]:
|
||
# Quoted spans are dropped: a missing quoted phrase is already the
|
||
# relevance floor's business, and words that are only capitalised because
|
||
# they sit inside a quoted heading ("Address offset", "Reset value") are
|
||
# ordinary words and say nothing when absent. What is left is the model's
|
||
# own unquoted terms, which is exactly where its guesses live.
|
||
unquoted = re.sub(r'"[^"]*"', " ", query)
|
||
found: list[str] = []
|
||
for token in re.findall(r"\S+", unquoted):
|
||
word = token.strip("'“”[].,;:!?")
|
||
# Parentheses come off only when they are punctuation around the
|
||
# word, never out of the middle of it: "RDS(on)" is one identifier,
|
||
# and stripping its closing bracket produced a term - "RDS(on" -
|
||
# that matched nothing and was reported absent from results that
|
||
# plainly carried it.
|
||
while word.startswith("(") and word.count("(") > word.count(")"):
|
||
word = word[1:]
|
||
while word.endswith(")") and word.count(")") > word.count("("):
|
||
word = word[:-1]
|
||
# Sentence openers - "What", "Which", "Does" - are already stopwords,
|
||
# so the first token needs no special case and must not have one: the
|
||
# surname a model puts first in a query is the term whose absence
|
||
# matters most.
|
||
if len(word) < 4 or word.lower() in _STOPWORDS:
|
||
continue
|
||
if word[:1].isupper() or _ALNUM_ID.match(word):
|
||
found.append(word)
|
||
# Uncapped. This used to keep the first six, and a model's guess is the
|
||
# term it appends last: a nine-term query about Australian capitals had
|
||
# its two invented words in positions eight and nine, so they were the
|
||
# only ones never checked.
|
||
return list(dict.fromkeys(found))
|
||
|
||
|
||
def _squash(text: str) -> str:
|
||
"""Lower-case alphanumerics only, so spacing and punctuation cannot hide a match."""
|
||
return re.sub(r"[^a-z0-9]+", "", text.lower())
|
||
|
||
|
||
# Google-style annotations that SearXNG passes through inside a snippet:
|
||
# "Missing: Port Jackson | Show results with: Port Jackson". They put the
|
||
# very words that are absent from the page INTO the snippet, which made the
|
||
# absence check below report them present.
|
||
_ENGINE_NOTE = re.compile(r"\s*(?:Missing|Must include|Show results with)\s*:.*$", re.IGNORECASE)
|
||
|
||
|
||
def _absent_note(query: str, haystacks: list[str]) -> str | None:
|
||
"""Report, as a count, which distinctive query terms appear in no result.
|
||
|
||
The relevance floor above answers "how well did this match overall". This
|
||
answers a different and sharper question: *which* of the things you asked
|
||
for is not here. A model hunting a paper it half-remembers puts its own
|
||
guesses into the query - an author, a journal, a year - and a result set
|
||
that is topically plausible reads as partial confirmation. It was: a
|
||
graded run searched for "... Finland Korhonen Circulation", got six
|
||
on-topic results containing none of those three words, and reported the
|
||
trial as real. Saying "0 of 6" is a fact about the corpus; the model can
|
||
do what it likes with it, which is the point.
|
||
"""
|
||
if not haystacks:
|
||
return None
|
||
squashed = [_squash(hay) for hay in haystacks]
|
||
missing = [f'"{term}"' for term in _distinctive(query)
|
||
if _squash(term) and not any(_squash(term) in hay for hay in squashed)]
|
||
if not missing:
|
||
return None
|
||
if len(missing) > 4:
|
||
missing = missing[:4] + [f"and {len(missing) - 4} more"]
|
||
return (f"[Appears in none of these results: {', '.join(missing)} - 0 of "
|
||
f"{len(haystacks)}. A name, a journal, a place or an identifier present in no "
|
||
"result was not found here; that is a different outcome from ranking low.]")
|
||
|
||
|
||
_READ_THE_PAGE = (
|
||
"Snippets are search-engine excerpts, not the page. Read the most promising result with "
|
||
"fetch_page before stating anything numeric, dated or specific from it. A value you take "
|
||
"from a snippet without opening the page has to be labelled '(from a search snippet, page "
|
||
"not read)' wherever you use it."
|
||
)
|
||
|
||
|
||
def _relevance_note(terms: list[str], best: float,
|
||
phrases: list[str], phrase_hit: bool) -> str:
|
||
"""Say how well these results actually match, instead of always saying "read one".
|
||
|
||
The old trailer told the model to fetch the most promising result whatever
|
||
came back, including when four of five engines were down and the results
|
||
were about something else entirely. Twelve graded runs recorded it being
|
||
either obeyed onto a useless page or ignored with no consequence, which are
|
||
the two ways a fixed instruction fails. A search that found nothing relevant
|
||
and a search that found the answer should not read identically.
|
||
"""
|
||
# An exact-phrase search that matched nothing is a null result whatever the
|
||
# loose word overlap says. The words of a fabricated paper title are
|
||
# ordinary words - "ferrous", "lattice", "copper" - and they scatter across
|
||
# unrelated pages, which is enough overlap to look like a weak hit. The
|
||
# quoted phrase is the signal the model actually asked for.
|
||
phrase_missed = bool(phrases) and not phrase_hit
|
||
|
||
# Under three content words the fraction is too coarse to mean anything.
|
||
if not phrase_missed and (len(terms) < 3 or best >= 0.5):
|
||
return _READ_THE_PAGE
|
||
pct = int(round(best * 100))
|
||
if not phrase_missed and best >= 0.25:
|
||
return (f"[WEAK MATCH: the best of these results carries only {pct}% of your query "
|
||
"terms. Before reading any of them, search again with fewer and more "
|
||
"distinctive terms - and drop any term that is your own guess at the answer "
|
||
"rather than part of the question.]")
|
||
tail = ""
|
||
if phrase_missed:
|
||
tail = (" The phrase you quoted appears in none of them, which for a title, a name or "
|
||
"an exact quotation means it was not found rather than ranked low.")
|
||
return (f"[LOW RELEVANCE: none of these results carries much of your query - the best match "
|
||
f"has {pct}% of your terms. Treat this as NO RESULTS for what you actually asked."
|
||
f"{tail} Search again with different wording, or tell the user you could not find "
|
||
"it. Do not read these pages hoping they help, and do not answer from memory "
|
||
"instead.]")
|
||
|
||
|
||
def _result_citation(item: dict, title: str, published: str) -> str | None:
|
||
"""A liftable citation line, when the engine handed over enough to build one.
|
||
|
||
SearXNG's scholarly engines return `journal`, `authors`, `doi` and
|
||
`pdf_url` on every hit, and all four were being dropped. The model was left
|
||
with a real URL, a real title and no venue, and filled the venue in from
|
||
the domain - calling an npj Flexible Electronics paper "Nature Electronics"
|
||
because the URL said nature.com.
|
||
"""
|
||
authors = [a for a in (item.get("authors") or []) if a]
|
||
journal = (item.get("journal") or item.get("publisher") or "").strip()
|
||
doi = (item.get("doi") or "").strip()
|
||
if not (authors or journal or doi):
|
||
return None
|
||
year = published[:4] if published[:4].isdigit() else None
|
||
return cite_line(authors[0] if authors else None, title, journal or None,
|
||
year, doi or None, et_al=len(authors) > 1)
|
||
|
||
|
||
def _annotate_seen(result: str, read: set[str], failed: dict[str, str]) -> str:
|
||
"""Mark results this conversation has already opened.
|
||
|
||
Applied after the cache, not before it, so the annotation follows the
|
||
conversation rather than being baked into a shared cache entry. A URL the
|
||
model has already read is not a new lead, and one that already failed is
|
||
not worth a second call - but inside a result list both look identical to
|
||
a fresh hit, and both were re-fetched in graded runs.
|
||
"""
|
||
if not read and not failed:
|
||
return result
|
||
out = []
|
||
for line in result.split("\n"):
|
||
target = line.strip()
|
||
if target in read:
|
||
out.append(f"{line} [already read in this conversation]")
|
||
elif target in failed:
|
||
out.append(f"{line} [fetch failed earlier: {failed[target]}]")
|
||
else:
|
||
out.append(line)
|
||
return "\n".join(out)
|
||
|
||
|
||
@router.post("/web_search", operation_id="web_search", response_model=str,
|
||
summary="Search the web")
|
||
@tool
|
||
async def web_search(body: WebSearchRequest,
|
||
x_openwebui_chat_id: str | None = Header(default=None, include_in_schema=False)) -> str:
|
||
"""Search the live web; returns ranked results with titles, URLs and snippets.
|
||
|
||
Use it whenever the answer depends on anything you cannot be certain of from memory:
|
||
versions, prices, releases, recent events, part details, error messages, configuration
|
||
syntax. One search is rarely enough - refine the terms or narrow with `site` and search
|
||
again. Snippets are too thin to answer from: fetch_page the result that holds the
|
||
answer. NO RESULTS or off-topic results mean you say you could not find it, not that
|
||
you fill the gap from memory.
|
||
"""
|
||
read_urls, failed_urls = await fetch_history(x_openwebui_chat_id)
|
||
|
||
cache_key = (body.query, str(body.max_results), body.category, body.time_range,
|
||
body.site or "", body.language or "")
|
||
if hit := await cache.cached("search", *cache_key):
|
||
return _annotate_seen(hit, read_urls, failed_urls)
|
||
|
||
query = body.query.strip()
|
||
if body.site:
|
||
site = body.site.strip().lstrip("*.").replace("site:", "")
|
||
query = f"site:{site} {query}"
|
||
|
||
params = {"q": query, "format": "json", "categories": body.category}
|
||
if body.time_range != "any":
|
||
params["time_range"] = body.time_range
|
||
if body.language:
|
||
params["language"] = body.language
|
||
|
||
# SearXNG lives in this namespace, so it must not go via squid.
|
||
response = await get_local_client().get(
|
||
f"{settings.searxng_url}/search", params=params, timeout=25.0)
|
||
response.raise_for_status()
|
||
payload = response.json()
|
||
|
||
results = payload.get("results") or []
|
||
# SearXNG's `score` fuses each result's position across every engine that
|
||
# returned it, so a page three engines agree on outranks one engine's top
|
||
# hit. That is a better relevance signal than any re-ranking done here.
|
||
results.sort(key=lambda r: r.get("score") or 0, reverse=True)
|
||
|
||
terms = _terms(body.query)
|
||
phrases = _quoted_phrases(body.query)
|
||
best_overlap = 0.0
|
||
phrase_hit = False
|
||
haystacks: list[str] = []
|
||
|
||
lines: list[str] = []
|
||
seen_urls: set[str] = set()
|
||
per_host: dict[str, int] = {}
|
||
|
||
for item in results:
|
||
url = (item.get("url") or "").strip()
|
||
title = (item.get("title") or "").strip()
|
||
if not url or not title or url in seen_urls:
|
||
continue
|
||
|
||
host = urlparse(url).netloc.lower().removeprefix("www.")
|
||
# Two per host at most. Ten pages of the same wiki crowds out the
|
||
# second opinion that would have caught an error.
|
||
if per_host.get(host, 0) >= 2:
|
||
continue
|
||
|
||
seen_urls.add(url)
|
||
per_host[host] = per_host.get(host, 0) + 1
|
||
|
||
snippet = re.sub(r"\s+", " ", (item.get("content") or "").replace(" ", " ")).strip()
|
||
snippet = _ENGINE_NOTE.sub("", snippet).strip()
|
||
# Measured before truncation, so a match deep in a long snippet still
|
||
# counts. This is the whole relevance signal; see _relevance_note.
|
||
haystack = f"{title} {snippet} {url}".lower()
|
||
haystacks.append(haystack)
|
||
best_overlap = max(best_overlap, _overlap(terms, haystack))
|
||
phrase_hit = phrase_hit or any(p in haystack for p in phrases)
|
||
|
||
if len(snippet) > 240:
|
||
snippet = snippet[:240].rsplit(" ", 1)[0] + "…"
|
||
|
||
published = item.get("publishedDate") or ""
|
||
date_note = f" ({published[:10]})" if published[:4].isdigit() else ""
|
||
|
||
block = [f"{len(lines) + 1}. **{title}**{date_note}",
|
||
f" {url}",
|
||
f" {snippet or '(no snippet)'}"]
|
||
if citation := _result_citation(item, title, published):
|
||
block.append(f" CITE AS: {citation}")
|
||
if pdf := (item.get("pdf_url") or "").strip():
|
||
block.append(f" PDF: {pdf}")
|
||
|
||
lines.append("\n".join(block))
|
||
if len(lines) >= body.max_results:
|
||
break
|
||
|
||
if not lines:
|
||
# Distinguishing "nothing matched" from "the engines fell over" is the
|
||
# difference between the model rewording its query and the model
|
||
# quietly deciding the thing does not exist.
|
||
dead = [e[0] for e in (payload.get("unresponsive_engines") or [])]
|
||
# A failed search is exactly when the model reaches for memory, so the
|
||
# pointer to a tool that has the data matters more here, not less.
|
||
tail = f"\n\n{hint}" if (hint := _tool_hint(query)) else ""
|
||
if dead:
|
||
return (
|
||
f"NO RESULTS - but {len(dead)} search engine(s) failed to respond "
|
||
f"({', '.join(sorted(set(dead))[:5])}), so this may be a temporary fault rather "
|
||
"than an absence of information. Try once more with different wording. If it "
|
||
"fails again, tell the user the search is not working rather than answering from memory."
|
||
+ tail
|
||
)
|
||
return NO_RESULTS + tail
|
||
|
||
out = [f"Search results for: {query}"]
|
||
if answers := payload.get("answers"):
|
||
out.append(f"\nDirect answer offered by the search engine: {' '.join(str(a) for a in answers)[:400]}")
|
||
out.append("")
|
||
out.extend(lines)
|
||
|
||
if dead := [e[0] for e in (payload.get("unresponsive_engines") or [])]:
|
||
out.append(f"\n[Note: {', '.join(sorted(set(dead))[:5])} did not respond, so coverage is "
|
||
"narrower than usual. Consider a second search if these results look thin.]")
|
||
|
||
if any("CITE AS:" in line for line in lines):
|
||
out.append("\nThe CITE AS lines are the only citation data here. A journal, an author "
|
||
"or a year that is not on one of them is not known - do not supply one from "
|
||
"memory, and do not read it off the site name in the URL.")
|
||
|
||
if hint := _tool_hint(query):
|
||
out.append("\n" + hint)
|
||
if note := _category_note(body.category, query):
|
||
out.append("\n" + note)
|
||
|
||
if absent := _absent_note(body.query, haystacks):
|
||
out.append("\n" + absent)
|
||
|
||
out.append("\n" + _relevance_note(terms, best_overlap, phrases, phrase_hit))
|
||
|
||
result = "\n".join(out)
|
||
await cache.store("search", result, *cache_key)
|
||
return _annotate_seen(result, read_urls, failed_urls)
|
||
|
||
|
||
# ---------------------------------------------------------------- fetch_page
|
||
|
||
_MAX_WINDOW = 10000
|
||
|
||
|
||
class FetchPageRequest(ToolModel):
|
||
url: str = Field(
|
||
description="Full public URL of the page or PDF, e.g. https://example.com/docs/page.",
|
||
)
|
||
|
||
@field_validator("max_chars", mode="before")
|
||
@classmethod
|
||
def _clamp_window(cls, value):
|
||
# "[MORE AVAILABLE: 10,716 characters remain]" was answered with
|
||
# max_chars=10716, and the schema bounced the whole call as a 422.
|
||
# A window a little too large is a request to read the rest, not a
|
||
# mistake worth a round trip.
|
||
try:
|
||
number = int(float(value))
|
||
except (TypeError, ValueError):
|
||
return value
|
||
return max(500, min(number, _MAX_WINDOW))
|
||
max_chars: int = Field(
|
||
default=6000,
|
||
ge=500,
|
||
# Above ~10000 the global response cap takes over and replaces the
|
||
# continuation hint with a dead-end "this was trimmed" message, so a
|
||
# larger window made the document *harder* to read, not easier.
|
||
le=10000,
|
||
description="Characters returned per call, 500-10000, default 6000. Large values consume "
|
||
"the context window fast.",
|
||
)
|
||
offset: int = Field(
|
||
default=0,
|
||
ge=0,
|
||
description="Character position to continue from, after a call reported more was "
|
||
"available.",
|
||
)
|
||
search: str | None = Field(
|
||
default=None,
|
||
description="Jump to the first occurrence of a term instead of reading from the top - a "
|
||
"register name, a section heading, a person's name, a quoted phrase. One call "
|
||
"with `search` replaces a dozen with `offset`. With `offset` also set, the "
|
||
"search starts there, which finds the next occurrence.",
|
||
)
|
||
|
||
|
||
_PREPRINT_HOSTS = (
|
||
("arxiv.org", "arXiv preprint"),
|
||
("biorxiv.org", "bioRxiv preprint"),
|
||
("medrxiv.org", "medRxiv preprint"),
|
||
("chemrxiv.org", "ChemRxiv preprint"),
|
||
("egusphere.copernicus.org", "EGUsphere preprint (open discussion)"),
|
||
("preprints.org", "Preprints.org preprint"),
|
||
("researchsquare.com", "Research Square preprint"),
|
||
("ssrn.com", "SSRN working paper"),
|
||
("osf.io/preprints", "OSF preprint"),
|
||
("techrxiv.org", "TechRxiv preprint"),
|
||
)
|
||
|
||
_PREPRINT_TEXT = (
|
||
"preprint. discussion started",
|
||
"this preprint has not been peer reviewed",
|
||
"has not been peer-reviewed",
|
||
"has not been peer reviewed",
|
||
"not been certified by peer review",
|
||
"preprint under review",
|
||
)
|
||
|
||
# What a fetch of this host actually delivers, as against what citing it would
|
||
# imply. A PubMed record is a catalogue entry, not the paper; a publisher
|
||
# landing page is not the paper either. Reporting a between-group difference
|
||
# out of one is how you end up stating the opposite of what a study found, and
|
||
# the system prompt's "say what you actually read" rule has no way to be
|
||
# followed unless the tool says which of the two arrived.
|
||
_ACCESS_HOSTS = (
|
||
("pmc.ncbi.nlm.nih.gov/articles",
|
||
"the open-access full text on PMC - this is the paper itself"),
|
||
("ncbi.nlm.nih.gov/pmc/articles",
|
||
"the open-access full text on PMC - this is the paper itself"),
|
||
("pubmed.ncbi.nlm.nih.gov",
|
||
"a PubMed catalogue record, not the paper - what you have is the abstract, so anything "
|
||
"you report from it is 'from the abstract'. If the full text is open it is on PMC, or "
|
||
"at the DOI"),
|
||
("arxiv.org/abs/",
|
||
"an arXiv abstract page; the paper itself is this URL with /abs/ replaced by /pdf/"),
|
||
)
|
||
|
||
_PAYWALL_MARKERS = (
|
||
"access through your institution", "purchase this article", "buy this article",
|
||
"get access to the full", "sign in to view", "subscribe to access",
|
||
"institutional access", "rent this article", "purchase access",
|
||
)
|
||
|
||
_OPINION_MARKERS = (
|
||
("opinion:", "an opinion piece, not a research paper"),
|
||
("comment on", "a comment on another paper, not an independent study"),
|
||
("reply to:", "a reply to a criticism, not an independent study"),
|
||
("matters arising", "a Matters Arising comment, not an independent study"),
|
||
("perspective:", "a perspective piece, not a research paper"),
|
||
("editorial:", "an editorial, not a research paper"),
|
||
)
|
||
|
||
|
||
def _provenance_note(final_url: str, text: str) -> str:
|
||
"""Say when a document is not what a citation of it would imply.
|
||
|
||
A model that reads one unreviewed opinion preprint and adopts its
|
||
bibliography produces an answer that looks like wide reading and is in fact
|
||
one source's framing. Observed on an AMOC question: 26 citations, almost all
|
||
traceable to a single EGUsphere preprint titled "Opinion: ...", presented as
|
||
a survey of the literature.
|
||
|
||
The model cannot see a journal masthead. It can see a line of tool output,
|
||
and it treats that as fact - so the label goes here.
|
||
"""
|
||
host = urlparse(final_url).netloc.lower()
|
||
path = urlparse(final_url).path.lower()
|
||
head = text[:1500].lower()
|
||
notes = []
|
||
|
||
for needle, label in _PREPRINT_HOSTS:
|
||
if needle in host + path:
|
||
notes.append(label)
|
||
break
|
||
else:
|
||
if "/preprints/" in path or any(m in head for m in _PREPRINT_TEXT):
|
||
notes.append("a preprint - not peer reviewed")
|
||
|
||
for marker, label in _OPINION_MARKERS:
|
||
if marker in head[:800]:
|
||
notes.append(label)
|
||
break
|
||
|
||
access = ""
|
||
for needle, label in _ACCESS_HOSTS:
|
||
if needle in host + path:
|
||
access = f"\n> **What this is: {label}.**\n"
|
||
break
|
||
else:
|
||
# A short page from a publisher asking for money is the abstract,
|
||
# whatever its title says.
|
||
if len(text) < 2500 and any(marker in head for marker in _PAYWALL_MARKERS):
|
||
access = ("\n> **What this is: a publisher landing page - the abstract and the "
|
||
"metadata, not the full text.** Do not report anything from inside the "
|
||
"paper: a between-group difference, a subgroup effect, or whether a "
|
||
"result reached significance was never delivered to you. Say plainly "
|
||
"that the full text was not open.\n")
|
||
|
||
if not notes:
|
||
return access
|
||
return access + (
|
||
"\n> **Provenance: this document is " + "; and ".join(notes) + ".** Say so if you "
|
||
"cite it, and do not present its reference list as the literature - a bibliography "
|
||
"assembled to support one position is not a survey. Where a claim here matters, cite "
|
||
"the underlying study you actually read, not this document's citation of it. In "
|
||
"particular, do not carry this document's paraphrase of another paper's numbers over "
|
||
"to that paper: a gloss like 'a weakening from 20 Sv down to 15 Sv (Smith et al.)' is "
|
||
"this author's wording, not Smith's, and attributing it to Smith makes a secondary "
|
||
"reading look like a primary one. If you did not open the underlying paper, say the "
|
||
"figure reaches you via this document.\n"
|
||
)
|
||
|
||
|
||
@router.post("/fetch_page", operation_id="fetch_page", response_model=str,
|
||
summary="Read a web page or PDF")
|
||
@tool
|
||
async def fetch_page(body: FetchPageRequest,
|
||
x_openwebui_chat_id: str | None = Header(default=None, include_in_schema=False)) -> str:
|
||
"""Fetch a URL and return its main text as markdown. Handles HTML and PDFs, including datasheets.
|
||
|
||
Use it whenever a snippet is not enough, when the user pastes a link, or when you need
|
||
exact details - configuration syntax, specification values, version numbers, ratings.
|
||
Long documents come back one window at a time: `search` jumps to a term, `offset`
|
||
continues. Private addresses, the user's local network and anything behind a login are
|
||
unreachable; when a page is inaccessible you are told so - report that rather than
|
||
guessing what it said.
|
||
"""
|
||
url = validate_url(body.url)
|
||
cache_key = (url,)
|
||
|
||
# A URL that already failed in this conversation fails the same way again.
|
||
# Declining costs nothing and is a clearer signal than a second identical
|
||
# error, which is what a stuck model collects instead of answering: one
|
||
# graded run re-fetched a 504ing mirror and a timed-out vendor URL after
|
||
# both had already failed, and stopped only when the budget forced it.
|
||
if reason := await recall_failure(x_openwebui_chat_id, url):
|
||
return (f"FAILED (not retried): {url} already failed earlier in this conversation - "
|
||
f"{reason} Retrying the same address will not change that. Use a different "
|
||
"source, or tell the user this one could not be read.")
|
||
|
||
# Counted per distinct URL rather than per call. The attempt still counts,
|
||
# not the success - the runaway loop this budget exists to stop is made of
|
||
# *failing* fetches, "that 403'd, let me try another route", and counting
|
||
# only the ones that returned text made the budget blind to it. But reading
|
||
# further into a page already open now costs nothing, because charging for
|
||
# that hit STOP FETCHING on page nine of a single article and told the model
|
||
# to stop doing the one thing that page needed.
|
||
budget_note = await note_fetch(x_openwebui_chat_id, url)
|
||
budget_tail = f"\n\n{budget_note}" if budget_note else ""
|
||
|
||
cached_doc = await cache.cached("page", *cache_key)
|
||
if cached_doc is None:
|
||
raw = b""
|
||
served = await _europepmc_fulltext(url)
|
||
pmc_style = any(p.search(url) for p in _PMC_ID_RES + _EPMC_MED_RES)
|
||
if served is not None:
|
||
text, meta, kind, final_url = served
|
||
else:
|
||
try:
|
||
raw, content_type, final_url = await get_bytes(url)
|
||
except FetchError as exc:
|
||
# A failed fetch still spends budget, so the note has to ride along
|
||
# with the failure - that is the moment the model decides whether to
|
||
# try yet another route. Formatted exactly as toolbase would have.
|
||
await note_failure(x_openwebui_chat_id, url, str(exc))
|
||
return f"FAILED: {exc}{budget_tail}"
|
||
except httpx.HTTPError as exc:
|
||
detail = explain(exc, url)
|
||
await note_failure(x_openwebui_chat_id, url, str(detail))
|
||
return f"FAILED: {detail}{budget_tail}"
|
||
if not raw:
|
||
await note_failure(x_openwebui_chat_id, url, "the server returned an empty response.")
|
||
return f"FAILED: {url} returned an empty response.{budget_tail}"
|
||
|
||
# A PDF request answered with something that is not a PDF is a vendor
|
||
# bouncing us to a landing page. This has to be caught before the content
|
||
# dispatch: the landing page extracts perfectly well as HTML, so nothing
|
||
# downstream fails, and the model is handed a plausible page about the
|
||
# wrong subject. TI answers a dead lit/ds/symlink URL with a 144 KB
|
||
# product-category page, which trafilatura renders as clean prose about
|
||
# LDO regulators in general.
|
||
if url.lower().endswith(".pdf") and raw[:5] != b"%PDF-":
|
||
await note_failure(x_openwebui_chat_id, url,
|
||
"the server answered a PDF request with something else.")
|
||
return (
|
||
f"FAILED: {url} did not return a PDF - the server answered with "
|
||
f"{content_type or 'something else'}"
|
||
+ (f" from {final_url}" if final_url != url else "")
|
||
+ ". The vendor has probably moved or renamed the document. Search for it by "
|
||
"part number or literature number rather than guessing another URL. Do not "
|
||
"report this as the document being unreadable: it was never served, and "
|
||
"whatever text this URL does return is a different document."
|
||
+ budget_tail
|
||
)
|
||
|
||
# Trust the bytes, not the URL. Keying on a ".pdf" suffix meant an HTML
|
||
# page served in answer to a PDF request was still handed to the PDF
|
||
# extractor, which failed obscurely instead of reporting the redirect.
|
||
meta: dict = {}
|
||
if raw[:5] == b"%PDF-" or "pdf" in content_type:
|
||
# Bounded, because the gateway will otherwise answer for us
|
||
# with a bare 504 - a 995-page reference manual took longer
|
||
# to extract than nginx was prepared to wait, and the model
|
||
# received an HTML error page as the tool result.
|
||
try:
|
||
text, meta = await asyncio.wait_for(
|
||
run_in_threadpool(_extract_pdf, raw), timeout=_EXTRACT_TIMEOUT_S)
|
||
except asyncio.TimeoutError:
|
||
mb = len(raw) / (1024 * 1024)
|
||
await note_failure(x_openwebui_chat_id, url,
|
||
f"its text could not be extracted within {_EXTRACT_TIMEOUT_S} seconds.")
|
||
return (f"FAILED: {url} is a {mb:.1f} MB PDF whose text could not be extracted "
|
||
f"within {_EXTRACT_TIMEOUT_S} seconds, so none of it was read. Look for "
|
||
"a smaller document - a single chapter, an application note, an HTML "
|
||
"version - or search for the section title you need to find a shorter "
|
||
f"copy.{budget_tail}")
|
||
kind = "PDF"
|
||
elif "json" in content_type or url.lower().endswith(".json"):
|
||
# JSON endpoints are primary sources - vendor security feeds, registry
|
||
# metadata, open data. Sent to the HTML extractor they came back as
|
||
# "could not extract readable text ... may be a scanned image", which
|
||
# is both wrong and discouraging, so the model went looking elsewhere
|
||
# for something it already had.
|
||
try:
|
||
text = json.dumps(json.loads(raw.decode("utf-8", errors="replace")), indent=2)
|
||
kind = "JSON"
|
||
except ValueError:
|
||
text = raw.decode("utf-8", errors="replace")
|
||
kind = "text (malformed JSON)"
|
||
elif content_type.startswith("text/plain"):
|
||
text = raw.decode("utf-8", errors="replace")
|
||
kind = "plain text"
|
||
else:
|
||
text, meta = await run_in_threadpool(_extract_html, raw, final_url)
|
||
kind = "web page"
|
||
|
||
text = _tidy(text)
|
||
if problem := _unreadable(url, raw, text, kind):
|
||
if pmc_style:
|
||
problem += ("\n\nEurope PMC's open-access API was tried first and has no full text "
|
||
"for this article, so it is not in the open-access subset: the paper "
|
||
"itself is not readable from here by any PMC route. The abstract may be "
|
||
"on the PubMed record; the full text only at the publisher or the DOI.")
|
||
await note_failure(x_openwebui_chat_id, url, problem.split(".")[0] + ".")
|
||
return f"FAILED: {problem}{budget_tail}"
|
||
|
||
cached_doc = "\x00".join([kind, final_url, json.dumps(meta, ensure_ascii=False), text])
|
||
await cache.store("page", cached_doc, *cache_key)
|
||
|
||
parts = cached_doc.split("\x00", 3)
|
||
if len(parts) == 4:
|
||
kind, final_url, meta_json, text = parts
|
||
try:
|
||
meta = json.loads(meta_json)
|
||
except ValueError:
|
||
meta = {}
|
||
else:
|
||
# A record cached before this tool captured metadata. They expire
|
||
# within the hour; until then, read it without the citation block.
|
||
kind, final_url, text = parts
|
||
meta = {}
|
||
|
||
await note_success(x_openwebui_chat_id, url)
|
||
|
||
header = _citation_header(meta, kind, final_url, text)
|
||
if budget_note:
|
||
header = f"{header}\n{budget_note}"
|
||
|
||
# The citation, formatted the way the answer should carry it. Built before
|
||
# the window so its length can be reserved out of the page budget.
|
||
cite = ""
|
||
if (title := meta.get("title")) and (meta.get("authors") or meta.get("journal") or meta.get("doi")):
|
||
authors = meta.get("authors") or []
|
||
cite = cite_line(authors[0] if authors else None, title,
|
||
meta.get("journal") or meta.get("publisher") or None,
|
||
_year_of(meta.get("date", "")), meta.get("doi") or None,
|
||
et_al=len(authors) > 1)
|
||
trailer = sources_line(cite, final_url)
|
||
|
||
if body.search:
|
||
window = search_window(
|
||
text, body.search, window=body.max_chars, start_at=body.offset,
|
||
continue_hint=f"call fetch_page again with url='{final_url}' and offset= to read more",
|
||
miss_note=(f"'{body.search}' does not occur in this document ({len(text):,} "
|
||
"characters). Try a shorter or differently worded term. If it is "
|
||
"genuinely not here, do not state it as this document's content."),
|
||
)
|
||
else:
|
||
# Leave room for the header, the continuation hint and the citation
|
||
# line. Above this the global response cap takes over, and it trims
|
||
# from the end - which removes the very line telling the model how to
|
||
# read the rest.
|
||
room = settings.max_response_chars - len(header) - len(trailer) - 900
|
||
window = paginate(
|
||
text, body.offset, max(500, min(body.max_chars, room)),
|
||
how_to_continue=f"To continue, call fetch_page again with url='{final_url}' and "
|
||
f"offset={{offset}} (max_chars is at most {_MAX_WINDOW:,} per call).",
|
||
)
|
||
return f"{header}\n{window}{trailer}"
|
||
|
||
|
||
# PubMed Central, Europe PMC's article pages and PubMed itself all now sit
|
||
# behind a browser check or a cookie wall for a plain HTTP client, and they
|
||
# are where the open-access full text of a paper lives. A graded run tried
|
||
# five routes to one open-access article and got a stub from every one, then
|
||
# found by itself that Europe PMC's REST API serves the same full text as
|
||
# JATS XML with no check at all. That is now the first route for any URL
|
||
# naming a PMC article, and for a PubMed record whose paper is in PMC.
|
||
_EPMC_REST = "https://www.ebi.ac.uk/europepmc/webservices/rest"
|
||
_PMC_ID_RES = (
|
||
re.compile(r"(?:pmc\.ncbi\.nlm\.nih\.gov|ncbi\.nlm\.nih\.gov)/(?:pmc/)?articles/(PMC\d+)", re.I),
|
||
re.compile(r"europepmc\.org/(?:article|abstract|articles)/PMC/(PMC\d+)", re.I),
|
||
)
|
||
_EPMC_MED_RES = (
|
||
re.compile(r"europepmc\.org/(?:article|abstract)/MED/(\d+)", re.I),
|
||
re.compile(r"pubmed\.ncbi\.nlm\.nih\.gov/(\d+)", re.I),
|
||
)
|
||
|
||
|
||
async def _europepmc_fulltext(url: str) -> tuple[str, dict, str, str] | None:
|
||
"""(text, meta, kind, final_url) for an article Europe PMC holds in full, else None."""
|
||
pmcid = None
|
||
for pattern in _PMC_ID_RES:
|
||
if found := pattern.search(url):
|
||
pmcid = found.group(1).upper()
|
||
break
|
||
if pmcid is None:
|
||
for pattern in _EPMC_MED_RES:
|
||
if found := pattern.search(url):
|
||
try:
|
||
payload = await get_json(f"{_EPMC_REST}/search", params={
|
||
"query": f"EXT_ID:{found.group(1)} AND SRC:MED",
|
||
"format": "json", "resultType": "lite",
|
||
})
|
||
hits = (payload.get("resultList") or {}).get("result") or []
|
||
pmcid = (hits[0].get("pmcid") or None) if hits else None
|
||
except Exception as exc:
|
||
log.info("europepmc search failed for %s: %s", url, exc)
|
||
break
|
||
if not pmcid:
|
||
return None
|
||
|
||
rest_url = f"{_EPMC_REST}/{pmcid}/fullTextXML"
|
||
try:
|
||
raw, _, _ = await get_bytes(rest_url)
|
||
except Exception as exc:
|
||
log.info("europepmc full text unavailable for %s: %s", pmcid, exc)
|
||
return None
|
||
if not raw or b"<article" not in raw[:8000]:
|
||
return None
|
||
text, meta = _jats_to_text(raw.decode("utf-8", errors="replace"))
|
||
if len(text.strip()) < 300:
|
||
return None
|
||
meta["via"] = rest_url
|
||
return text, meta, "open-access full text (JATS XML from Europe PMC)", url
|
||
|
||
|
||
def _jats_to_text(xml: str) -> tuple[str, dict]:
|
||
"""Render JATS article XML as readable markdown, with its front matter as metadata."""
|
||
def between(src: str, open_tag: str, close_tag: str) -> str:
|
||
i = src.find(open_tag)
|
||
if i == -1:
|
||
return ""
|
||
j = src.find(close_tag, i)
|
||
return src[i:j] if j != -1 else src[i:]
|
||
|
||
def strip(fragment: str) -> str:
|
||
return html.unescape(re.sub(r"<[^>]+>", "", fragment)).strip()
|
||
|
||
def first(tag: str, src: str, attrs: str = "") -> str:
|
||
found = re.search(rf"<{tag}\b{attrs}[^>]*>(.*?)</{tag}>", src, re.S)
|
||
return strip(found.group(1)) if found else ""
|
||
|
||
meta: dict = {}
|
||
front = between(xml, "<front>", "</front>")
|
||
if title := first("article-title", front):
|
||
meta["title"] = re.sub(r"\s+", " ", title)
|
||
if journal := first("journal-title", front):
|
||
meta["journal"] = journal
|
||
if doi := first("article-id", front, r'[^>]*pub-id-type="doi"'):
|
||
meta["doi"] = doi
|
||
authors = []
|
||
# Some archives label each <contrib contrib-type="author">; others label
|
||
# the <contrib-group content-type="author"> and leave the contribs bare.
|
||
# The second shape rendered a three-author paper with no authors at all.
|
||
contribs = re.findall(r'<contrib\b[^>]*contrib-type="author"[^>]*>(.*?)</contrib>', front, re.S)
|
||
if not contribs:
|
||
for group in re.findall(r'<contrib-group\b[^>]*(?:content|contrib)-type="author"[^>]*>(.*?)</contrib-group>',
|
||
front, re.S):
|
||
contribs += re.findall(r"<contrib\b[^>]*>(.*?)</contrib>", group, re.S)
|
||
if not contribs:
|
||
contribs = [c for c in re.findall(r"<contrib\b[^>]*>(.*?)</contrib>", front, re.S)
|
||
if "<surname" in c or "<collab" in c]
|
||
for contrib in contribs:
|
||
surname, given = first("surname", contrib), first("given-names", contrib)
|
||
if surname:
|
||
authors.append(f"{surname}, {given}" if given else surname)
|
||
elif collab := first("collab", contrib):
|
||
authors.append(collab)
|
||
if authors:
|
||
meta["authors"] = authors[:12]
|
||
if year := re.search(r"<pub-date\b[^>]*>.*?<year>(\d{4})</year>", front, re.S):
|
||
meta["date"] = year.group(1)
|
||
|
||
parts = []
|
||
if abstract := between(xml, "<abstract", "</abstract>"):
|
||
if "<title" not in abstract[:400]:
|
||
abstract = "<title>Abstract</title>" + abstract
|
||
parts.append(abstract)
|
||
parts.append(between(xml, "<body>", "</body>"))
|
||
body = "\n\n".join(parts)
|
||
|
||
body = re.sub(r"<ref-list\b.*?</ref-list>", "", body, flags=re.S)
|
||
body = re.sub(r"<(?:mml:)?math\b.*?</(?:mml:)?math>", " [equation] ", body, flags=re.S)
|
||
body = re.sub(r"<title\b[^>]*>(.*?)</title>", r"\n\n## \1\n", body, flags=re.S)
|
||
body = re.sub(r"<label\b[^>]*>(.*?)</label>", r"\1 ", body, flags=re.S)
|
||
body = re.sub(r"</p>", "\n\n", body)
|
||
body = re.sub(r"<list-item\b[^>]*>", "\n- ", body)
|
||
body = re.sub(r"</(?:td|th)>", " | ", body)
|
||
body = re.sub(r"</tr>", "\n", body)
|
||
body = re.sub(r"<(?:table-wrap|fig)\b[^>]*>", "\n\n", body)
|
||
body = strip(body)
|
||
body = re.sub(r"[ \t]+", " ", body)
|
||
body = re.sub(r"\n{3,}", "\n\n", body)
|
||
return body, meta
|
||
|
||
|
||
# Anti-bot interstitials, consent walls and "you need JavaScript" shells. A
|
||
# page that says one of these and almost nothing else was served but not
|
||
# delivered, and it used to read to the model as a successful fetch.
|
||
_STUB_MARKERS = (
|
||
"enable javascript", "javascript is required", "javascript is disabled",
|
||
"please enable cookies", "just a moment", "checking your browser",
|
||
"verify you are human", "cf-browser-verification", "captcha",
|
||
"accept all cookies", "cookie consent", "access denied",
|
||
"request unsuccessful", "pardon our interruption", "are you a robot",
|
||
# PubMed's cookie wall: 76 characters, served with a 200. It read as a
|
||
# successful fetch, carried a provenance note describing an abstract that
|
||
# was not there, and emitted a Sources: trailer the answer then listed.
|
||
"cookies must be enabled", "enable cookies for",
|
||
)
|
||
|
||
_META_TAG = re.compile(r"<meta\b[^>]*>", re.I)
|
||
_META_ATTR = re.compile(r"""([\w.:-]+)\s*=\s*(?:"([^"]*)"|'([^']*)')""")
|
||
_DOI_RE = re.compile(r"\b10\.\d{4,9}/[^\s\"'<>]+")
|
||
_ARXIV_RE = re.compile(r"\barXiv:\s*(\d{4}\.\d{4,5}(?:v\d+)?)", re.I)
|
||
|
||
|
||
def _unreadable(url: str, raw: bytes, text: str, kind: str) -> str | None:
|
||
"""Say what actually went wrong when a page arrives carrying no content.
|
||
|
||
The old message offered three causes - scanned image, video, JavaScript -
|
||
for every case, and was wrong about all three whenever the real cause was
|
||
something else. Two failures in a graded run came from that. A 13MB manual
|
||
was reported as possibly a scan. And the CISA known-exploited-vulnerability
|
||
catalogue returned 1,293 characters of navigation out of 150KB of HTML,
|
||
with not one CVE row in it; nothing said so, the model read it as a
|
||
successful fetch of the catalogue, and answered from memory instead.
|
||
"""
|
||
stripped = (text or "").strip()
|
||
|
||
if kind == "PDF" and len(stripped) < 40:
|
||
return (f"{url} is a PDF with no extractable text layer - it is a scan or an "
|
||
"image-only document, so none of it was read. Look for an HTML version, or a "
|
||
"different source; do not describe its contents.")
|
||
if kind == "PDF" and (ratio := _glyph_ratio(stripped)) >= _GLYPH_FAIL:
|
||
# Plenty of characters, none of them readable. A datasheet whose
|
||
# fonts carry no character map extracts as "/G73/G68/G32" for every
|
||
# cell of every table; the tool reported 28,716 characters of
|
||
# content, and the model spent a turn reverse-engineering the glyph
|
||
# numbering and got the one figure the question turned on wrong.
|
||
return (f"{url} is a PDF whose text layer is not decodable: {ratio:.0%} of the extracted "
|
||
"text is glyph codes (/G123, (cid:123)) rather than characters, because the file "
|
||
"carries no character map. The words - and above all the numbers - cannot be "
|
||
"read from this copy, and nothing decoded from it by guesswork is trustworthy. "
|
||
"Get the same document from another source: a distributor's copy of the PDF, the "
|
||
"manufacturer's HTML product page, or a different revision of the file.")
|
||
if len(stripped) < 40:
|
||
return (f"Could not extract readable text from {url}. Try a different source rather "
|
||
"than guessing its contents.")
|
||
# Structured formats are whatever length they are; a short JSON feed is a
|
||
# complete answer, not a stub.
|
||
if kind in {"JSON", "text (malformed JSON)", "plain text", "PDF"}:
|
||
return None
|
||
if len(stripped) >= 1500:
|
||
return None
|
||
|
||
lowered = raw[:200_000].decode("utf-8", errors="replace").lower()
|
||
if not any(marker in lowered for marker in _STUB_MARKERS) and len(raw) <= 30_000:
|
||
return None
|
||
|
||
host = urlparse(url).netloc
|
||
return (
|
||
f"{url} is client-rendered or gated. The server sent {len(raw) // 1024} KB of page "
|
||
f"scaffolding but only {len(stripped)} characters of readable text, and none of the "
|
||
"data this page shows in a browser. Do not answer from this fragment. Try, in order: "
|
||
"the same data as a file the page links (an export is often a CSV or JSON URL), a "
|
||
f"static sub-page found by searching site:{host}, or a different source that "
|
||
"reproduces the data.\n\nWhat was extracted:\n> " + stripped[:400]
|
||
)
|
||
|
||
|
||
def _meta_tags(html: str) -> list[tuple[str, str]]:
|
||
"""(name, content) for every meta tag in the head, both attribute orders."""
|
||
found: list[tuple[str, str]] = []
|
||
for tag in _META_TAG.findall(html[:65536]):
|
||
attrs: dict[str, str] = {}
|
||
for match in _META_ATTR.finditer(tag):
|
||
value = match.group(2) if match.group(2) is not None else match.group(3)
|
||
attrs[match.group(1).lower()] = value or ""
|
||
key = attrs.get("name") or attrs.get("property")
|
||
content = attrs.get("content")
|
||
if key and content:
|
||
found.append((key.strip().lower(), content.strip()))
|
||
return found
|
||
|
||
|
||
def _html_meta(html: str) -> dict:
|
||
"""Who wrote this, where it was published, and under what identifier.
|
||
|
||
requirements.txt says trafilatura was chosen partly because "it hands back
|
||
metadata (title, date, sitename) in the same call, which is what the
|
||
citation lines are built from" - and then nothing ever asked it for any.
|
||
The model fetched a real paper, transcribed every number correctly, and
|
||
attributed it to authors and a journal that appear nowhere in its tools,
|
||
because the extracted text began at the abstract.
|
||
|
||
The journal is taken only from tags that name the container explicitly,
|
||
never from og:site_name and never from the domain: on an npj Flexible
|
||
Electronics article og:site_name reads "Nature", which is precisely the
|
||
mislabel that cost a graded run.
|
||
"""
|
||
import trafilatura
|
||
|
||
got: dict[str, list[str]] = {}
|
||
for key, value in _meta_tags(html):
|
||
got.setdefault(key, []).append(value)
|
||
|
||
def first(*names: str) -> str:
|
||
for name in names:
|
||
for value in got.get(name, []):
|
||
if value:
|
||
return value
|
||
return ""
|
||
|
||
meta: dict = {}
|
||
if title := first("citation_title", "dc.title", "og:title"):
|
||
meta["title"] = title
|
||
if authors := clean_authors(got.get("citation_author") or got.get("dc.creator") or []):
|
||
meta["authors"] = authors[:12]
|
||
if journal := first("citation_journal_title", "prism.publicationname",
|
||
"citation_conference_title", "dc.source"):
|
||
meta["journal"] = journal
|
||
if publisher := first("citation_publisher", "dc.publisher"):
|
||
meta["publisher"] = publisher
|
||
if date := first("citation_publication_date", "citation_date", "citation_online_date",
|
||
"article:published_time", "dc.date"):
|
||
meta["date"] = date
|
||
doi = first("citation_doi", "prism.doi")
|
||
if not doi:
|
||
candidate = first("dc.identifier")
|
||
if candidate.lower().startswith(("10.", "doi:")):
|
||
doi = candidate
|
||
if doi:
|
||
meta["doi"] = doi
|
||
if pdf := first("citation_pdf_url"):
|
||
meta["pdf_url"] = pdf
|
||
|
||
# Ordinary pages carry none of the scholarly tags but do have a title, an
|
||
# author and a date worth citing.
|
||
if not (meta.get("title") and meta.get("date") and meta.get("authors")):
|
||
try:
|
||
document = trafilatura.extract_metadata(html)
|
||
except Exception:
|
||
document = None
|
||
if document is not None:
|
||
if not meta.get("title"):
|
||
meta["title"] = (getattr(document, "title", "") or "").strip()
|
||
if not meta.get("date"):
|
||
meta["date"] = (getattr(document, "date", "") or "").strip()
|
||
if not meta.get("authors"):
|
||
author = (getattr(document, "author", "") or "").strip()
|
||
names = clean_authors(re.split(r"[;,]", author))
|
||
if names:
|
||
meta["authors"] = names[:8]
|
||
return {key: value for key, value in meta.items() if value}
|
||
|
||
|
||
def _year_of(date: str) -> str | None:
|
||
found = re.search(r"\b(1[6-9]\d{2}|20\d{2})\b", date or "")
|
||
return found.group(1) if found else None
|
||
|
||
|
||
def _citation_header(meta: dict, kind: str, final_url: str, text: str) -> str:
|
||
"""The document's identity, before any of its contents.
|
||
|
||
Placed at the top because the global response cap trims from the end: a
|
||
citation that can be trimmed off is a citation the model will reconstruct
|
||
from memory.
|
||
"""
|
||
title = meta.get("title") or ""
|
||
authors = meta.get("authors") or []
|
||
journal = meta.get("journal") or ""
|
||
publisher = meta.get("publisher") or ""
|
||
doi = meta.get("doi") or ""
|
||
|
||
lines = [f"# Source: {final_url}"]
|
||
if via := meta.get("via"):
|
||
lines.append(f"Retrieved via: {via} - the article page itself sits behind a browser check, "
|
||
"so this is the same full text from the Europe PMC open-access API.")
|
||
if (ratio := meta.get("glyph_ratio")) and ratio < _GLYPH_FAIL:
|
||
lines.append(f"Warning: {ratio:.0%} of this PDF's text layer is undecodable glyph codes "
|
||
"(/G123 ...), typically the parameter tables. A value that looks garbled or "
|
||
"is missing here cannot be read from this copy - do not fill it in from memory.")
|
||
if title:
|
||
lines.append(f"Title: {title}")
|
||
if authors:
|
||
shown = "; ".join(authors[:6]) + (" et al." if len(authors) > 6 else "")
|
||
lines.append(f"Authors: {shown}")
|
||
|
||
facts = []
|
||
if date := meta.get("date"):
|
||
facts.append(f"Published: {date}")
|
||
if journal:
|
||
facts.append(f"Journal: {journal}")
|
||
if publisher and publisher != journal:
|
||
facts.append(f"Publisher: {publisher}")
|
||
if doi:
|
||
facts.append(f"DOI: {doi}")
|
||
if facts:
|
||
lines.append(" · ".join(facts))
|
||
|
||
if title and (authors or journal or doi):
|
||
lines.append("CITE AS: " + cite_line(
|
||
authors[0] if authors else None, title, journal or publisher or None,
|
||
_year_of(meta.get("date", "")), doi or None, et_al=len(authors) > 1))
|
||
elif not (authors or journal or doi):
|
||
lines.append("No author, journal or DOI metadata is published on this page. Cite it by "
|
||
"title and URL only - do not supply authors or a venue from memory, and do "
|
||
"not infer the venue from the site name in the URL.")
|
||
|
||
lines.append(f"({kind}, {len(text):,} characters)")
|
||
if note := _provenance_note(final_url, text):
|
||
lines.append(note)
|
||
return "\n".join(lines)
|
||
|
||
|
||
_ANCHOR = re.compile(r'<a\b([^>]*\bhref\s*=\s*["\']([^"\']+)["\'][^>]*)>(.*?)</a>', re.I | re.S)
|
||
_DOCUMENT_HREF = re.compile(
|
||
r"\.(?:pdf|docx?|xlsx?|pptx?|csv|json|xml|zip)(?:[?#]|$)"
|
||
r"|/(?:document|download|attachment|file)s?(?:/|$|\?)",
|
||
re.I,
|
||
)
|
||
_DOCUMENT_TEXT = re.compile(r"\b(?:download|pdf|document|presentation|webcast|full text|data)\b", re.I)
|
||
|
||
|
||
def _keep_document_links(html: str, base: str) -> str:
|
||
"""Turn links to documents into plain text carrying their URL.
|
||
|
||
A results index listed every report as a title, a date and a "Download"
|
||
link, and the extraction kept the titles and dates and dropped every
|
||
link: the anchors sat in a link-dense grid that boilerplate removal
|
||
discards whole. The model read the index and could not get from "Interim
|
||
results 2025" to the PDF it needed, on a page whose only purpose is to
|
||
lead there. A link whose target is a document (a PDF, a /document or
|
||
/download path) or whose text says so becomes "Download (https://...)",
|
||
a span the extractor has no reason to drop, in the place the link was.
|
||
Ordinary navigation links are left alone.
|
||
"""
|
||
def rewrite(match: re.Match) -> str:
|
||
href, inner = match.group(2).strip(), match.group(3)
|
||
if href.startswith(("#", "javascript:", "mailto:", "tel:")):
|
||
return match.group(0)
|
||
text = " ".join(re.sub(r"<[^>]+>", " ", inner).split())
|
||
if not _DOCUMENT_HREF.search(href) and not _DOCUMENT_TEXT.search(text):
|
||
return match.group(0)
|
||
label = html_mod.escape(text or "link")
|
||
return f"<span>{label} ({html_mod.escape(urljoin(base, href))})</span>"
|
||
|
||
return _ANCHOR.sub(rewrite, html[:2_000_000]) + html[2_000_000:]
|
||
|
||
|
||
def _extract_html(raw: bytes, url: str) -> tuple[str, dict]:
|
||
import trafilatura
|
||
|
||
html = _keep_document_links(raw.decode("utf-8", errors="replace"), url)
|
||
extracted = trafilatura.extract(
|
||
html, url=url, output_format="markdown",
|
||
include_links=True, include_tables=True, include_comments=False,
|
||
favor_precision=True,
|
||
)
|
||
if not extracted:
|
||
# Precision mode drops thin pages entirely; a recall pass is better than
|
||
# telling the model the page was unreadable when it simply was terse.
|
||
extracted = trafilatura.extract(html, url=url, output_format="markdown",
|
||
favor_recall=True, include_comments=False) or ""
|
||
return extracted, _html_meta(html)
|
||
|
||
|
||
def _extract_pdf(raw: bytes) -> tuple[str, dict]:
|
||
"""Two-tier PDF extraction; see requirements.txt for why not PyMuPDF."""
|
||
import io
|
||
|
||
text = ""
|
||
meta: dict = {}
|
||
try:
|
||
from pypdf import PdfReader
|
||
|
||
reader = PdfReader(io.BytesIO(raw))
|
||
text = "\n\n".join((page.extract_text() or "") for page in reader.pages)
|
||
try:
|
||
info = reader.metadata or {}
|
||
title = _PDF_TITLE_PREFIX.sub("", str(info.get("/Title") or "").strip())
|
||
# Producers routinely put the filename here, or a template name.
|
||
if _looks_like_title(title):
|
||
meta["title"] = title
|
||
if author := str(info.get("/Author") or "").strip():
|
||
# The Author field is whoever was logged in when the file was
|
||
# made: "csmith2" produced the citation 'csmith2,
|
||
# "IRLZ44NPbF.pmd"' - a username and a layout file, presented
|
||
# as a bibliographic reference.
|
||
names = [a for a in clean_authors(re.split(r"[;,]| and ", author))
|
||
if not _looks_like_username(a)]
|
||
if names:
|
||
meta["authors"] = names[:8]
|
||
except Exception:
|
||
pass
|
||
except Exception as exc:
|
||
log.info("pypdf failed, falling back to pdfminer: %s", exc)
|
||
|
||
# pypdf interleaves columns on multi-column layouts, which is the normal
|
||
# shape of a datasheet parameter table - the result is text that looks
|
||
# fine and states the wrong numbers. A low alphanumeric ratio is the
|
||
# cheapest reliable tell.
|
||
# pdfminer is minutes-slow on a large file, and the gateway's clock is
|
||
# running; above 8 MB pypdf's text is what there is.
|
||
if _looks_garbled(text) and len(raw) <= 8 * 1024 * 1024:
|
||
try:
|
||
from pdfminer.high_level import extract_text as pdfminer_extract
|
||
|
||
better = pdfminer_extract(io.BytesIO(raw))
|
||
if len(better.strip()) > len(text.strip()):
|
||
text = better
|
||
except Exception as exc:
|
||
log.info("pdfminer fallback failed: %s", exc)
|
||
|
||
# A paper's own front matter is the only citation data a PDF carries, and
|
||
# it is worth more than anything reconstructed from the filename.
|
||
head = text[:3000]
|
||
if "doi" not in meta:
|
||
if found := _DOI_RE.search(head):
|
||
meta["doi"] = found.group(0).rstrip(".,);]")
|
||
elif arxiv := _ARXIV_RE.search(head):
|
||
meta["doi"] = f"arXiv:{arxiv.group(1)}"
|
||
|
||
if (ratio := _glyph_ratio(text)) >= _GLYPH_WARN:
|
||
meta["glyph_ratio"] = round(ratio, 3)
|
||
|
||
return text, meta
|
||
|
||
|
||
_GLYPH_RE = re.compile(r"/G\d{1,4}|\(cid:\d{1,5}\)")
|
||
# Below the first, the document is usable with a warning (a few table cells
|
||
# from one embedded font); at the second it is not a text layer at all.
|
||
_GLYPH_WARN = 0.03
|
||
_GLYPH_FAIL = 0.35
|
||
_EXTRACT_TIMEOUT_S = 120
|
||
|
||
_PDF_TITLE_PREFIX = re.compile(r"^(?:Microsoft (?:Word|PowerPoint|Excel)|PowerPoint Presentation)\s*[-–:]\s*", re.I)
|
||
_PDF_TEMPLATE_TITLE = re.compile(r"^(?:untitled(?:[ -]?\d+)?|layout ?\d*|slide ?\d*|document ?\d*|book ?\d*|"
|
||
r"presentation ?\d*|doc ?\d*|new document|template)$", re.I)
|
||
|
||
|
||
def _glyph_ratio(text: str) -> float:
|
||
"""Share of the text that is undecoded glyph names rather than characters."""
|
||
if not text:
|
||
return 0.0
|
||
sample = text[:200_000]
|
||
covered = sum(len(m) for m in _GLYPH_RE.findall(sample))
|
||
return covered / len(sample)
|
||
|
||
|
||
def _looks_like_title(title: str) -> bool:
|
||
if len(title) <= 8 or _PDF_TEMPLATE_TITLE.match(title):
|
||
return False
|
||
# "IRLZ44NPbF.pmd", "ds1234.indd", "report_final.docx": a filename.
|
||
if re.fullmatch(r"\S+\.[A-Za-z0-9]{1,5}", title):
|
||
return False
|
||
return True
|
||
|
||
|
||
def _looks_like_username(name: str) -> bool:
|
||
if len(name) < 3:
|
||
return True
|
||
if " " in name or "," in name:
|
||
return False
|
||
# One token with a digit in it, or all lower case: an account, not a person.
|
||
return bool(re.search(r"\d", name)) or name.islower()
|
||
|
||
|
||
def _looks_garbled(text: str) -> bool:
|
||
stripped = text.strip()
|
||
if len(stripped) < 200:
|
||
return True
|
||
alnum = sum(c.isalnum() or c.isspace() for c in stripped[:4000])
|
||
return alnum / min(len(stripped), 4000) < 0.75
|
||
|
||
|
||
def _tidy(text: str) -> str:
|
||
text = text.replace("\f", "\n").replace(" ", " ")
|
||
text = re.sub(r"[ \t]+", " ", text)
|
||
text = re.sub(r"\n{3,}", "\n\n", text)
|
||
return text.strip()
|
||
|
||
|
||
# ----------------------------------------------------------- wikipedia_lookup
|
||
|
||
class WikipediaRequest(ToolModel):
|
||
topic: str = Field(description="Article title or subject, e.g. 'OSI model'.")
|
||
full_article: bool = Field(
|
||
default=False,
|
||
description="Default false returns the introduction, which answers most questions; true "
|
||
"returns the whole article.",
|
||
)
|
||
search: str | None = Field(
|
||
default=None,
|
||
description="Jump to a term inside the article - a date, a name, a section heading, a "
|
||
"quoted line. Implies the full article.",
|
||
)
|
||
offset: int = Field(
|
||
default=0, ge=0,
|
||
description="Character position to continue a long article from.",
|
||
)
|
||
language: str = Field(default="en", description="Wikipedia language edition, default 'en'.")
|
||
|
||
|
||
def _title_overlap(topic: str, title: str) -> float:
|
||
want = {w for w in re.findall(r"[a-z0-9]+", topic.lower()) if len(w) > 1}
|
||
got = {w for w in re.findall(r"[a-z0-9]+", title.lower()) if len(w) > 1}
|
||
return len(want & got) / len(want) if want else 1.0
|
||
|
||
|
||
async def _disambiguation_options(api: str, title: str) -> str:
|
||
try:
|
||
payload = await get_json(api, params={
|
||
"action": "query", "titles": title, "prop": "links", "plnamespace": "0",
|
||
"pllimit": "20", "format": "json", "formatversion": "2",
|
||
})
|
||
pages = (payload.get("query") or {}).get("pages") or []
|
||
links = [link["title"] for link in (pages[0].get("links") or [])] if pages else []
|
||
except Exception:
|
||
links = []
|
||
return "\n".join(f"- {name}" for name in links[:15]) or "- (the page lists no articles)"
|
||
|
||
|
||
@router.post("/wikipedia_lookup", operation_id="wikipedia_lookup", response_model=str,
|
||
summary="Look up a Wikipedia article")
|
||
@tool
|
||
async def wikipedia_lookup(body: WikipediaRequest) -> str:
|
||
"""Get a Wikipedia article on a topic, person, place, event, work, organism or standard.
|
||
|
||
Use it for any "everyone knows this" fact - dates, who held an office and when, who is
|
||
related to whom, what a text actually says, how many of something there are. Settled
|
||
facts are where confident memory is wrong by one detail; one call checks them. Not for
|
||
anything current - releases, prices, news - use web_search for those. The introduction
|
||
rarely holds a specific date, result, list or quotation: pass `search` to jump to it.
|
||
If the topic has no article of its own you are told so, with the nearest matches; do
|
||
not infer facts about it from articles that merely mention the words.
|
||
"""
|
||
lang = (body.language or "en").strip().lower()[:8]
|
||
want_full = body.full_article or bool(body.search)
|
||
|
||
cached_doc = await cache.cached("wikipedia", lang, body.topic, str(want_full))
|
||
if cached_doc is None:
|
||
api = f"https://{lang}.wikipedia.org/w/api.php"
|
||
|
||
# Exact title first. This used to go straight to full-text search and
|
||
# take its top hit, so a person with no article of their own resolved
|
||
# to whatever article merely contained the words: "Mary Lee Pfeiffer"
|
||
# returned a Tom Clancy novel, and the model - which had already
|
||
# guessed an answer - read the miss as confirmation and kept the guess.
|
||
# The Action API's plaintext extract is by a wide margin the cleanest
|
||
# source here: no markup, no reference clutter, sections as "== Foo ==".
|
||
params = {
|
||
"action": "query", "titles": body.topic, "redirects": "1",
|
||
"prop": "extracts|pageprops", "ppprop": "disambiguation",
|
||
"explaintext": "1", "format": "json", "formatversion": "2",
|
||
}
|
||
if not want_full:
|
||
params["exintro"] = "1"
|
||
|
||
payload = await get_json(api, params=params)
|
||
pages = (payload.get("query") or {}).get("pages") or []
|
||
page = pages[0] if pages else {}
|
||
title = ""
|
||
extract = ""
|
||
|
||
if page and not page.get("missing"):
|
||
if "disambiguation" in (page.get("pageprops") or {}):
|
||
options = await _disambiguation_options(api, page.get("title") or body.topic)
|
||
return (f"'{body.topic}' is a disambiguation page - several articles share that "
|
||
f"name:\n{options}\n\nPick the one the question is about and call again "
|
||
"with its exact title. Do not answer from the name alone.")
|
||
title = page.get("title") or body.topic
|
||
extract = page.get("extract") or ""
|
||
|
||
if not extract.strip():
|
||
search = await get_json(api, params={
|
||
"action": "query", "list": "search", "srsearch": body.topic,
|
||
"srlimit": 5, "format": "json", "formatversion": "2",
|
||
})
|
||
hits = (search.get("query") or {}).get("search") or []
|
||
if not hits:
|
||
return f"NO RESULTS. Wikipedia has no article matching '{body.topic}'. {NO_RESULTS}"
|
||
|
||
# A full-text hit is not the same thing as an article about the
|
||
# topic, and saying so is the whole point of this branch.
|
||
if _title_overlap(body.topic, hits[0]["title"]) < 0.5:
|
||
near = ", ".join(h["title"] for h in hits[:3])
|
||
return (
|
||
f"NO ARTICLE. Wikipedia has no article titled '{body.topic}'. The closest "
|
||
f"full-text matches are: {near} - they contain the words but are about "
|
||
"something else. If this is a person, a place or a thing with no article "
|
||
"of its own, do not infer facts about it from those: search the web for "
|
||
"the name on its own, with no guessed relation or answer attached to it, "
|
||
"and read a page that states the fact independently. If it is a variant "
|
||
"spelling of a real title, call again with that title."
|
||
)
|
||
|
||
title = hits[0]["title"]
|
||
params["titles"] = title
|
||
payload = await get_json(api, params=params)
|
||
pages = (payload.get("query") or {}).get("pages") or []
|
||
extract = (pages[0].get("extract") if pages else "") or ""
|
||
if not extract.strip():
|
||
return f"NO RESULTS. The Wikipedia article '{title}' has no readable extract."
|
||
|
||
url = f"https://{lang}.wikipedia.org/wiki/{quote(title.replace(' ', '_'))}"
|
||
cached_doc = "\x00".join([title, url, _tidy(extract)])
|
||
await cache.store("wikipedia", cached_doc, lang, body.topic, str(want_full))
|
||
|
||
title, url, extract = cached_doc.split("\x00", 2)
|
||
|
||
scope = "full article" if want_full else "introduction only"
|
||
header = f"# {title}\nSource: {url}\n(Wikipedia, {scope}, {len(extract):,} characters)"
|
||
if not want_full:
|
||
header += ("\n[Introduction only. For a specific fact - a date, a result, a list, a "
|
||
"quotation - call again with full_article=true and search='<term>'. The "
|
||
"lead section summarises; it rarely carries the detail.]")
|
||
|
||
trailer = sources_line(url)
|
||
|
||
if body.search:
|
||
window = search_window(
|
||
extract, body.search, window=5000, start_at=body.offset,
|
||
continue_hint=(f"call wikipedia_lookup with topic='{title}', full_article=true "
|
||
"and offset= to read more"),
|
||
miss_note=(f"'{body.search}' does not occur in this article ({len(extract):,} "
|
||
"characters). Try a different term. If it is genuinely not here, do not "
|
||
"state it as this article's content."),
|
||
)
|
||
else:
|
||
room = settings.max_response_chars - len(header) - len(trailer) - 900
|
||
window = paginate(
|
||
extract, body.offset, max(500, min(7000, room)),
|
||
how_to_continue=(f"To continue, call wikipedia_lookup with topic='{title}', "
|
||
"full_article=true and offset={offset}."),
|
||
)
|
||
return f"{header}\n\n{window}{trailer}"
|