PWA, one send/stop button, audio in and out, web search as a tool

Four pieces of work.

**Installable.** A manifest carrying the instance name, PWA icons rasterised
from the existing mark at design time, a service worker and a themed offline
page. The worker caches the shell only and bails out on /api/, /auth/, /admin/
and anything accepting text/event-stream -- passing a reply stream through a
worker turns it into one delivery at the end, or nothing. It is served from
GET /sw.js rather than the static mount because a worker's scope is the path it
came from.

**Send and Stop are one button.** They were two, and the hidden one was never
hidden: `.btn` is display: inline-flex, which outranks the browser's own
`[hidden] { display: none }`, so Stop sat permanently beside Send. app.css now
forces the attribute to win -- every control toggled with `hidden` depended on
that -- and the composer renders one button carrying both icons, with ui.js
flipping data-composer-action and the type with it.

**Audio.** Speech to text and text to speech against any OpenAI-shaped
/v1/audio/* endpoint: dictate into the composer, have a reply read out.
Instance settings in Admin, per-reader overrides in Settings, with the voice
list discovered from the server where it offers one. Recorded audio is capped
and never written to disk -- it is not an attachment, it has no owner, and
nothing would ever sweep it.

**Web search, as a tool.** This is the tool loop PLAN.md described as the real
work: one reply is now a bounded sequence of requests rather than one. The model
asks, the tool runs, the result goes back and it is asked again, up to three
rounds. Providers are DuckDuckGo (no setup), SearXNG and Firecrawl.

Two decisions worth stating. Tools are only offered to models flagged `tools`,
because an endpoint without support rejects the whole request rather than
ignoring the array -- the same reason images only reach models flagged
`vision`. And tool results are not replayed as context on the next turn, for the
same reasons reasoning is not: the answer already contains what the model made
of them, and replaying stale results into every later request wastes the window
and reliably sends a small model into a search loop. The sources stay visible in
the transcript instead.

Search results are untrusted third-party text and are treated as such: escaped,
and only http/https URLs rendered as links.

338 tests, ruff clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jaroslav Beneš
2026-07-21 17:56:50 +02:00
parent de178837b8
commit 7456525d19
63 changed files with 4597 additions and 140 deletions
+300
View File
@@ -0,0 +1,300 @@
"""Speech to text and text to speech, against OpenAI-shaped audio endpoints.
The same reasoning as the chat client: plain httpx rather than an SDK, because
the target is not api.openai.com so much as whisper.cpp's server, Speaches,
faster-whisper-server, Kokoro and anything else exposing ``/v1/audio/*``. They
agree on the request and disagree politely about the response, so this is
tolerant about what comes back.
Two endpoints, not one. A local install almost always runs transcription and
speech as separate processes -- they are different models on different
schedules -- and forcing them onto one base URL would mean the common case
could not be configured at all.
"""
from __future__ import annotations
import logging
import time
from collections.abc import AsyncIterator
from typing import Any
import httpx
from lembas.config import settings as env_settings
from lembas.services.crypto import decrypt
from lembas.services.llm.openai_client import (
Endpoint,
LLMError,
describe_http_error,
wrap_transport_error,
)
log = logging.getLogger(__name__)
# api.openai.com has no endpoint that lists voices, so when one is not offered
# these are what a caller can reasonably assume. Anything else -- Kokoro's sixty
# or so -- is discovered.
OPENAI_VOICES = ("alloy", "echo", "fable", "onyx", "nova", "shimmer")
# Formats every player in a browser can decode. opus is deliberately absent:
# some endpoints emit it in an ogg container that Safari will not play.
FORMATS = ("mp3", "wav", "flac", "aac")
# Discovery is cached because the voice list is read every time anyone opens
# their settings, and waking a model server to answer that is rude.
_VOICE_TTL = 300.0
_voice_cache: dict[str, tuple[float, list[str]]] = {}
def endpoint_for(config: dict[str, Any], side: str) -> Endpoint:
"""Build an Endpoint from the stored audio settings.
`side` is "stt" or "tts". Endpoint is a frozen snapshot with the key
already decrypted, so nothing downstream has to know the secret was ever
encrypted -- or hold a database session while it streams.
"""
base_url = (config.get(f"{side}_base_url") or "").strip()
if not base_url:
raise LLMError("No audio endpoint has been configured.")
return Endpoint(
base_url=base_url.rstrip("/"),
api_key=decrypt(config.get(f"{side}_api_key_encrypted") or ""),
extra_headers={},
name=base_url,
)
async def transcribe(
endpoint: Endpoint,
*,
data: bytes,
filename: str,
content_type: str,
model: str = "whisper-1",
language: str = "",
) -> str:
"""Turn recorded audio into text.
`model` is sent even to servers that ignore it: whisper.cpp serves one model
and does not care, while a router in front of several will not dispatch
without it. `language` is omitted when empty, which is what asks the server
to detect it -- sending an empty string instead makes some of them fail.
"""
form: dict[str, Any] = {"model": model, "response_format": "json"}
if language:
form["language"] = language
try:
async with httpx.AsyncClient(timeout=env_settings.request_timeout) as client:
response = await client.post(
endpoint.url("audio/transcriptions"),
headers=_headers_without_content_type(endpoint),
data=form,
files={"file": (filename, data, content_type or "application/octet-stream")},
)
response.raise_for_status()
except httpx.HTTPStatusError as exc:
raise LLMError(describe_http_error(exc), status_code=exc.response.status_code) from exc
except httpx.RequestError as exc:
raise wrap_transport_error(exc, endpoint) from exc
try:
payload = response.json()
except ValueError:
# response_format=text is what some servers give regardless of the ask.
return response.text.strip()
if isinstance(payload, dict):
text = payload.get("text")
if isinstance(text, str):
return text.strip()
error = payload.get("error")
if error:
raise LLMError(str(error))
raise LLMError("The transcription endpoint returned no text.")
async def speak(
endpoint: Endpoint,
text: str,
*,
model: str = "tts-1",
voice: str = "",
fmt: str = "mp3",
speed: float = 1.0,
) -> tuple[str, AsyncIterator[bytes]]:
"""Synthesise speech, returning its content type and a byte stream.
Streamed rather than buffered: a long reply is a lot of audio, and playback
can start on the first chunk instead of after the last.
"""
if not text.strip():
raise LLMError("There is nothing to read out.")
body: dict[str, Any] = {
"model": model,
"input": text,
"response_format": fmt if fmt in FORMATS else "mp3",
}
if voice:
body["voice"] = voice
if speed and speed != 1.0:
body["speed"] = speed
client = httpx.AsyncClient(timeout=env_settings.request_timeout)
try:
request = client.build_request(
"POST", endpoint.url("audio/speech"), headers=endpoint.headers(), json=body
)
response = await client.send(request, stream=True)
if response.status_code >= 400:
# Nothing has been read yet on a streaming response, and the error
# detail is in the body.
await response.aread()
response.raise_for_status()
except httpx.HTTPStatusError as exc:
await client.aclose()
raise LLMError(describe_http_error(exc), status_code=exc.response.status_code) from exc
except httpx.RequestError as exc:
await client.aclose()
raise wrap_transport_error(exc, endpoint) from exc
except Exception:
await client.aclose()
raise
media_type = response.headers.get("content-type", f"audio/{body['response_format']}")
async def stream() -> AsyncIterator[bytes]:
# The client is closed here rather than by the caller: it has to outlive
# this function, and a response abandoned without aclose leaks a socket.
try:
async for chunk in response.aiter_bytes():
yield chunk
finally:
await response.aclose()
await client.aclose()
return media_type, stream()
async def voices(endpoint: Endpoint, *, refresh: bool = False) -> list[str]:
"""Voices the speech endpoint offers, newest answer cached briefly.
Falls back to the OpenAI six on a 404, which is not an error: the official
API simply has no such endpoint, and its voices are a fixed list everyone
already knows.
"""
key = endpoint.base_url
cached = _voice_cache.get(key)
if cached and not refresh and time.monotonic() - cached[0] < _VOICE_TTL:
return cached[1]
try:
async with httpx.AsyncClient(timeout=15.0) as client:
response = await client.get(
endpoint.url("audio/voices"), headers=endpoint.headers()
)
if response.status_code == 404:
found = list(OPENAI_VOICES)
_voice_cache[key] = (time.monotonic(), found)
return found
response.raise_for_status()
payload = response.json()
except httpx.HTTPStatusError as exc:
raise LLMError(describe_http_error(exc), status_code=exc.response.status_code) from exc
except httpx.RequestError as exc:
raise wrap_transport_error(exc, endpoint) from exc
except ValueError as exc:
raise LLMError("The endpoint returned a response that was not JSON.") from exc
found = _parse_voices(payload)
if not found:
found = list(OPENAI_VOICES)
_voice_cache[key] = (time.monotonic(), found)
return found
def _parse_voices(payload: Any) -> list[str]:
"""Pull voice names out of whatever shape the server chose.
Kokoro answers ``{"voices": [{"id": "af_heart", ...}]}``; older builds and
some others answer ``{"voices": ["af_heart", ...]}``; a couple return the
bare list. All three are the same information.
"""
entries = payload
if isinstance(payload, dict):
for field in ("voices", "data"):
if isinstance(payload.get(field), list):
entries = payload[field]
break
if not isinstance(entries, list):
return []
names: list[str] = []
for entry in entries:
if isinstance(entry, str) and entry:
names.append(entry)
elif isinstance(entry, dict):
name = entry.get("id") or entry.get("name") or entry.get("voice")
if isinstance(name, str) and name:
names.append(name)
# Sorted and de-duplicated: sixty voices in the server's arbitrary order is
# not a list anyone can pick from.
return sorted(dict.fromkeys(names))
def _headers_without_content_type(endpoint: Endpoint) -> dict[str, str]:
"""Endpoint headers minus Content-Type.
httpx sets the multipart Content-Type itself, including the boundary.
Leaving the JSON one in place overrides it and the server sees a body it
cannot parse.
"""
return {k: v for k, v in endpoint.headers().items() if k.lower() != "content-type"}
def forget_voices() -> None:
"""Drop the discovery cache. Used when an administrator changes the URL."""
_voice_cache.clear()
def template_flags(db, user) -> dict[str, Any]:
"""What the chat templates need to know about audio.
Lives here rather than in one page module because a message bubble is
rendered from four places -- the chat page, the two message endpoints, and
the SSE stream, which has no request at all -- and each of them needs the
same three booleans. Getting one of them wrong is how a speaker button ends
up on a page that cannot use it.
"""
from lembas.security import permissions
from lembas.services import settings_store
config = settings_store.audio(db)
allowed = permissions.resolve(db, user)
listen = bool(config.get("tts_enabled")) and allowed.get("audio.listen", False)
preferences = (user.settings_json or {}).get("audio") or {} if user else {}
return {
"audio": config,
"user_audio": preferences,
"can_dictate": bool(config.get("stt_enabled"))
and allowed.get("audio.transcribe", False),
"can_listen": listen,
# Only meaningful when can_listen; the template guards on both.
"audio_autoplay": listen
and bool(preferences.get("autoplay", config.get("tts_autoplay"))),
}
__all__ = [
"FORMATS",
"OPENAI_VOICES",
"LLMError",
"endpoint_for",
"forget_voices",
"speak",
"transcribe",
"voices",
]
+21
View File
@@ -19,6 +19,13 @@ from lembas.config import settings
log = logging.getLogger(__name__)
# Rendered in a form in place of a stored secret. If a submitted value still
# equals this, the field was never touched and the stored secret must be kept --
# otherwise saving a name change would silently wipe the credential beside it.
# Lives here rather than in one admin module because every form that edits a
# secret needs the same dance.
UNCHANGED_SENTINEL = "" * 12
@lru_cache
def _fernet() -> Fernet:
@@ -56,3 +63,17 @@ def mask(secret: str) -> str:
if len(secret) <= 8:
return "*" * len(secret)
return f"{secret[:3]}{'*' * 8}{secret[-4:]}"
def keep_or_replace(submitted: str, stored_ciphertext: str) -> str:
"""Resolve a submitted secret field against what is already stored.
Three cases, and the middle one is the reason this exists: the sentinel
means "the form rendered a mask and nobody typed over it", which is not the
same as an empty field. An explicitly emptied field does mean "this endpoint
needs no key", so it clears the stored value.
"""
submitted = submitted.strip()
if submitted == UNCHANGED_SENTINEL:
return stored_ciphertext
return encrypt(submitted)
+103 -28
View File
@@ -22,10 +22,18 @@ import time
from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta
from lembas.db.models import ROLE_ASSISTANT, ROLE_USER, Chat, Message
from lembas.db.models import ROLE_ASSISTANT, ROLE_USER, Chat, Message, User
from lembas.db.session import session_scope
from lembas.services import chat as chat_service
from lembas.services.llm.openai_client import LLMError, delta_reasoning, delta_text, stream_chat
from lembas.services import settings_store
from lembas.services import tools as tools_service
from lembas.services.llm.openai_client import (
LLMError,
delta_reasoning,
delta_text,
delta_tool_calls,
stream_chat,
)
from lembas.services.reasoning import REASONING, ReasoningSplitter
log = logging.getLogger(__name__)
@@ -52,6 +60,10 @@ class Generation:
reasoning: list[str] = field(default_factory=list)
reasoning_ms: int = 0
# One entry per tool call made while producing this reply, in order. Shown
# live as the model works and kept on the message afterwards.
tool_events: list[dict] = field(default_factory=list)
error: str = ""
stopped: bool = False
done: bool = False
@@ -130,7 +142,15 @@ async def shutdown() -> None:
async def _run(generation: Generation) -> None:
"""Produce one reply, then persist it. Never raises into the task."""
"""Produce one reply, then persist it. Never raises into the task.
A reply is not necessarily one request. When tools are offered and the
model asks to use one, the loop below runs it, appends the result to the
conversation and asks again -- up to tools_service.MAX_ROUNDS times, after
which the model has to answer with what it has. Text produced before a tool
call is kept, so a model that narrates what it is about to look up does not
lose that when the results come back.
"""
splitter = ReasoningSplitter()
started = time.monotonic()
reasoning_started: float | None = None
@@ -151,35 +171,89 @@ async def _run(generation: Generation) -> None:
question = _question_from(payload)
needs_title = not chat.title_generated
async for chunk in stream_chat(endpoint, payload):
thought = delta_reasoning(chunk)
if thought:
if reasoning_started is None:
reasoning_started = time.monotonic()
generation.reasoning.append(thought)
generation.touch()
# Read while the session is open: everything below outlives it.
offered = tools_service.enabled_tools(db, chat, db.get(User, chat.user_id))
search_config = settings_store.search(db) if offered else {}
text = delta_text(chunk)
if text:
for kind, piece in splitter.feed(text):
if kind == REASONING:
if reasoning_started is None:
reasoning_started = time.monotonic()
generation.reasoning.append(piece)
else:
if reasoning_started is not None and not generation.reasoning_ms:
generation.reasoning_ms = int(
(time.monotonic() - reasoning_started) * 1000
)
generation.content.append(piece)
generation.touch()
if offered:
payload = {**payload, "tools": offered}
if generation.cancel:
generation.stopped = True
for round_number in range(tools_service.MAX_ROUNDS + 1):
accumulator = tools_service.ToolCallAccumulator()
# Text the model produced in *this* round, needed separately from
# generation.content when echoing the assistant turn back.
round_text: list[str] = []
async for chunk in stream_chat(endpoint, payload):
thought = delta_reasoning(chunk)
if thought:
if reasoning_started is None:
reasoning_started = time.monotonic()
generation.reasoning.append(thought)
generation.touch()
if offered:
fragments = delta_tool_calls(chunk)
if fragments:
accumulator.feed(fragments)
text = delta_text(chunk)
if text:
for kind, piece in splitter.feed(text):
if kind == REASONING:
if reasoning_started is None:
reasoning_started = time.monotonic()
generation.reasoning.append(piece)
else:
if reasoning_started is not None and not generation.reasoning_ms:
generation.reasoning_ms = int(
(time.monotonic() - reasoning_started) * 1000
)
generation.content.append(piece)
round_text.append(piece)
generation.touch()
if generation.cancel:
generation.stopped = True
break
# Let followers and other tasks run between chunks.
await asyncio.sleep(0)
calls = accumulator.calls
if generation.stopped or not calls:
break
# Let followers and other tasks run between chunks.
await asyncio.sleep(0)
if round_number == tools_service.MAX_ROUNDS:
# Out of rounds with the model still asking for tools. Recorded
# rather than silently dropped: an answer that stops here needs
# to be explicable.
generation.tool_events.append(
{
"name": calls[0]["name"],
"status": "error",
"error": (
f"Stopped after {tools_service.MAX_ROUNDS} rounds of tool "
f"calls without an answer."
),
}
)
generation.touch()
break
messages = [
*payload["messages"],
tools_service.assistant_turn(calls, "".join(round_text)),
]
for call in calls:
outcome = await tools_service.run_tool(
search_config, call["name"], call["arguments"]
)
generation.tool_events.append(outcome.event)
generation.touch()
messages.append(tools_service.tool_turn(call, outcome.content))
payload = {**payload, "messages": messages}
for kind, piece in splitter.flush():
(generation.reasoning if kind == REASONING else generation.content).append(piece)
@@ -248,6 +322,7 @@ def _persist(generation: Generation, title: str, elapsed: float) -> None:
message.content = generation.text
message.reasoning = generation.thinking
message.reasoning_ms = generation.reasoning_ms
message.tool_calls_json = generation.tool_events
message.error = generation.error
message.stopped = generation.stopped
message.complete = True
+46 -8
View File
@@ -77,9 +77,13 @@ class Endpoint:
return headers
def _describe_http_error(exc: httpx.HTTPStatusError) -> str:
def describe_http_error(exc: httpx.HTTPStatusError) -> str:
"""Turn an upstream error response into something worth reading.
Public because the audio and search clients talk to the same class of
server and want the same translation; LLMError stays the one thing a
caller has to catch.
Providers put the useful part in wildly different places, so try the common
shapes before falling back to the raw body.
"""
@@ -109,7 +113,7 @@ def _describe_http_error(exc: httpx.HTTPStatusError) -> str:
return friendly or detail or f"The endpoint returned HTTP {status}."
def _wrap_transport_error(exc: httpx.RequestError, endpoint: Endpoint) -> LLMError:
def wrap_transport_error(exc: httpx.RequestError, endpoint: Endpoint) -> LLMError:
if isinstance(exc, httpx.ConnectError):
return LLMError(
f"Could not reach {endpoint.base_url}. Is the endpoint running and "
@@ -131,9 +135,9 @@ async def list_models(endpoint: Endpoint) -> list[dict[str, Any]]:
response.raise_for_status()
payload = response.json()
except httpx.HTTPStatusError as exc:
raise LLMError(_describe_http_error(exc), status_code=exc.response.status_code) from exc
raise LLMError(describe_http_error(exc), status_code=exc.response.status_code) from exc
except httpx.RequestError as exc:
raise _wrap_transport_error(exc, endpoint) from exc
raise wrap_transport_error(exc, endpoint) from exc
except ValueError as exc:
raise LLMError("The endpoint returned a response that was not JSON.") from exc
@@ -195,9 +199,9 @@ async def stream_chat(
continue
except httpx.HTTPStatusError as exc:
raise LLMError(_describe_http_error(exc), status_code=exc.response.status_code) from exc
raise LLMError(describe_http_error(exc), status_code=exc.response.status_code) from exc
except httpx.RequestError as exc:
raise _wrap_transport_error(exc, endpoint) from exc
raise wrap_transport_error(exc, endpoint) from exc
async def complete(endpoint: Endpoint, payload: dict[str, Any]) -> str:
@@ -211,9 +215,9 @@ async def complete(endpoint: Endpoint, payload: dict[str, Any]) -> str:
response.raise_for_status()
data = response.json()
except httpx.HTTPStatusError as exc:
raise LLMError(_describe_http_error(exc), status_code=exc.response.status_code) from exc
raise LLMError(describe_http_error(exc), status_code=exc.response.status_code) from exc
except httpx.RequestError as exc:
raise _wrap_transport_error(exc, endpoint) from exc
raise wrap_transport_error(exc, endpoint) from exc
except ValueError as exc:
raise LLMError("The endpoint returned a response that was not JSON.") from exc
@@ -245,6 +249,40 @@ def delta_reasoning(chunk: dict[str, Any]) -> str:
return ""
def delta_tool_calls(chunk: dict[str, Any]) -> list[dict[str, Any]]:
"""Pull tool-call fragments out of one streamed chunk.
Each entry carries an ``index`` and, across chunks, a name that arrives
once and an ``arguments`` string that arrives in pieces. Reassembling them
is lembas.services.tools.ToolCallAccumulator's job; this only extracts.
"""
try:
choices = chunk.get("choices") or []
if not choices:
return []
calls = (choices[0].get("delta") or {}).get("tool_calls")
return calls if isinstance(calls, list) else []
except (AttributeError, TypeError):
return []
def finish_reason(chunk: dict[str, Any]) -> str:
"""Why the model stopped, when the chunk says so.
``tool_calls`` here is the signal that the reply is not an answer but a
request to run something and come back. Some servers send ``stop`` even
when they emitted tool calls, so the accumulator's contents are the real
authority and this is only a hint.
"""
try:
choices = chunk.get("choices") or []
if not choices:
return ""
return choices[0].get("finish_reason") or ""
except (AttributeError, TypeError):
return ""
def delta_text(chunk: dict[str, Any]) -> str:
"""Pull the text out of one streamed chunk, tolerating provider variation."""
try:
+38
View File
@@ -13,6 +13,7 @@ from __future__ import annotations
import functools
import html
import re
import nh3
from markdown_it import MarkdownIt
@@ -132,3 +133,40 @@ def escape_text(text: str) -> str:
slashes, which triples the size of a streamed token for no benefit.
"""
return html.escape(text, quote=False)
# Code blocks are dropped whole rather than read out. A speech model given a
# code fence pronounces every bracket and underscore, which is unlistenable and
# takes longer than the prose it was buried in.
#
# Matched on <pre> rather than on the .code-block wrapper: the wrapper also
# contains a label div, so a non-greedy match for its closing tag stops at the
# label's and leaves the code behind. <pre> cannot nest, so this is exact.
_CODE_BLOCK = re.compile(r"<pre\b[^>]*>.*?</pre>", re.DOTALL)
_CODE_LABEL = re.compile(r"<div class=\"code-block__label\">.*?</div>", re.DOTALL)
_TAG = re.compile(r"<[^>]+>")
_WHITESPACE = re.compile(r"[ \t]*\n\s*\n\s*")
# Speech endpoints reject or truncate very long inputs, and a reply long enough
# to hit this is not one anybody is listening to in full.
MAX_SPEAKABLE = 8000
def speakable_text(text: str) -> str:
"""Markdown reduced to something worth reading aloud.
Goes through the renderer rather than stripping the Markdown source
directly, so tables, lists and links come out as their text instead of as
punctuation, and there is one definition of what a message *says*.
"""
if not text:
return ""
rendered = _CODE_LABEL.sub(" ", _CODE_BLOCK.sub("\n", render_markdown(text)))
stripped = html.unescape(_TAG.sub(" ", rendered))
# Paragraph breaks survive as a single newline: speech models use them as a
# pause, and a wall of one line is read without any.
stripped = _WHITESPACE.sub("\n", stripped)
lines = [" ".join(line.split()) for line in stripped.splitlines()]
return "\n".join(line for line in lines if line)[:MAX_SPEAKABLE]
+108
View File
@@ -0,0 +1,108 @@
"""Web search providers.
One shape in, one shape out: a query and a limit go in, a list of SearchResult
comes back, and which service answered is a setting rather than a code path any
caller has to know about.
Everything here returns *untrusted third-party text*. A title or snippet from a
search result is exactly as much attacker-controlled as model output, and gets
the same treatment: escaped on the way into a page, and only http/https URLs
rendered as links.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from typing import Any
from lembas.services.search import ddg, firecrawl, searxng
from lembas.services.search.base import SearchError, SearchResult
log = logging.getLogger(__name__)
@dataclass(frozen=True)
class Provider:
key: str
label: str
description: str
# Whether an administrator has to configure something before it works.
needs_setup: bool
PROVIDERS: tuple[Provider, ...] = (
Provider(
"ddgs",
"DuckDuckGo",
"No account, no key, no server to run. Rate limited if used heavily.",
False,
),
Provider(
"searxng",
"SearXNG",
"Your own metasearch instance. Needs its JSON format enabled.",
True,
),
Provider(
"firecrawl",
"Firecrawl",
"Hosted search API. Needs an account and a key.",
True,
),
)
_RUNNERS = {"ddgs": ddg.search, "searxng": searxng.search, "firecrawl": firecrawl.search}
def provider(key: str) -> Provider:
return next((p for p in PROVIDERS if p.key == key), PROVIDERS[0])
def availability(key: str) -> str:
"""Why a provider cannot be used, or "" when it can.
Checked before a search is attempted so the admin screen can say what is
wrong while it is being configured, rather than the first chat to try it
being where the problem surfaces.
"""
if key == "ddgs" and not ddg.is_available():
return (
"The ddgs package is not installed. Install it with: "
'pip install "lembas[search]"'
)
return ""
async def run(
config: dict[str, Any], query: str, *, limit: int | None = None
) -> list[SearchResult]:
"""Search with whichever provider is configured.
Raises SearchError with something worth reading; every provider translates
its own failures rather than letting an httpx exception escape.
"""
query = " ".join(query.split())[:400]
if not query:
raise SearchError("There was nothing to search for.")
key = config.get("provider") or "ddgs"
problem = availability(key)
if problem:
raise SearchError(problem)
runner = _RUNNERS.get(key)
if runner is None:
raise SearchError(f"Unknown search provider '{key}'.")
count = limit or int(config.get("max_results") or 5)
# A model that asks for fifty results is asking for a prompt nobody can
# afford; the administrator's number is the ceiling either way.
count = min(max(count, 1), int(config.get("max_results") or 5))
results = await runner(config, query, count)
log.info("web search (%s) for %r: %d results", key, query[:60], len(results))
return results[:count]
__all__ = ["PROVIDERS", "Provider", "SearchError", "SearchResult", "availability", "run"]
+62
View File
@@ -0,0 +1,62 @@
"""What every search provider produces, and how it fails."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
from urllib.parse import urlparse
# A snippet is context, not an article. Longer than this and a handful of
# results crowds out the conversation they were meant to inform.
MAX_SNIPPET = 400
class SearchError(Exception):
"""A search failure with a message fit to show a user.
Same contract as LLMError in the chat client: one exception type, always
carrying text that can be put on screen without editing.
"""
def __init__(self, message: str) -> None:
super().__init__(message)
self.message = message
@dataclass(frozen=True)
class SearchResult:
title: str
url: str
snippet: str
@property
def host(self) -> str:
try:
return urlparse(self.url).netloc or self.url
except ValueError:
return self.url
@property
def is_linkable(self) -> bool:
"""Whether this result's URL may be rendered as a link.
Only http and https. A search provider is an untrusted source, and a
javascript: or data: URL arriving in a result and being turned into an
anchor is the obvious way this feature would be abused.
"""
try:
return urlparse(self.url).scheme in ("http", "https")
except ValueError:
return False
def clean(title: Any, url: Any, snippet: Any) -> SearchResult | None:
"""Normalise one provider's row, or None if there is nothing usable in it."""
url = str(url or "").strip()
if not url:
return None
return SearchResult(
title=" ".join(str(title or "").split())[:300] or url,
url=url[:2000],
snippet=" ".join(str(snippet or "").split())[:MAX_SNIPPET],
)
+88
View File
@@ -0,0 +1,88 @@
"""DuckDuckGo, via the ddgs package.
The default provider because it is the only one that works with no account, no
key and no server to run: enabling web search should not also be a
configuration exercise.
Optional at install time -- see the `search` extra in pyproject.toml -- so the
import is guarded and its absence is reported as something to install rather
than as a crash.
"""
from __future__ import annotations
import asyncio
from typing import Any
from lembas.services.search.base import SearchError, SearchResult, clean
try: # pragma: no cover - exercised by whether the extra is installed
from ddgs import DDGS
_IMPORT_ERROR = ""
except ImportError as exc: # pragma: no cover
DDGS = None
_IMPORT_ERROR = str(exc)
def is_available() -> bool:
return DDGS is not None
def _blocking_search(query: str, count: int, region: str, safesearch: str) -> list[dict[str, Any]]:
with DDGS() as client:
return list(
client.text(
query,
region=region or "wt-wt",
safesearch=safesearch or "moderate",
max_results=count,
)
)
async def search(config: dict[str, Any], query: str, count: int) -> list[SearchResult]:
if not is_available():
raise SearchError(
'The ddgs package is not installed. Install it with: pip install "lembas[search]"'
)
try:
# ddgs is synchronous. Run it on a thread: blocking the event loop here
# would stall every other chat in the process, including the one that
# asked for the search.
rows = await asyncio.wait_for(
asyncio.to_thread(
_blocking_search,
query,
count,
str(config.get("region") or "wt-wt"),
str(config.get("safesearch") or "moderate"),
),
timeout=float(config.get("timeout") or 20.0),
)
except TimeoutError as exc:
raise SearchError("DuckDuckGo did not answer in time.") from exc
except Exception as exc: # noqa: BLE001 - the library raises its own types
# Rate limiting is the common failure and worth naming, because the fix
# is to wait rather than to change anything.
detail = str(exc)
if "ratelimit" in detail.lower() or "202" in detail:
raise SearchError(
"DuckDuckGo is rate limiting this instance. Try again shortly, "
"or configure SearXNG instead."
) from exc
raise SearchError(f"DuckDuckGo search failed: {detail[:200]}") from exc
results = []
for row in rows:
# ddgs renamed its fields across versions; both spellings are read so
# an upgrade does not silently return empty snippets.
result = clean(
row.get("title"),
row.get("href") or row.get("url") or row.get("link"),
row.get("body") or row.get("description") or row.get("snippet"),
)
if result is not None:
results.append(result)
return results
+75
View File
@@ -0,0 +1,75 @@
"""Firecrawl's hosted search API.
The paid option, and the only one of the three that needs a key. Included
because it answers with cleaned page content rather than a search engine's
snippet, which is materially better material for a model to read.
"""
from __future__ import annotations
from typing import Any
import httpx
from lembas.services.crypto import decrypt
from lembas.services.search.base import SearchError, SearchResult, clean
DEFAULT_BASE_URL = "https://api.firecrawl.dev"
async def search(config: dict[str, Any], query: str, count: int) -> list[SearchResult]:
api_key = decrypt(str(config.get("firecrawl_api_key_encrypted") or ""))
if not api_key:
raise SearchError("No Firecrawl API key has been configured.")
base_url = str(config.get("firecrawl_base_url") or DEFAULT_BASE_URL).strip().rstrip("/")
try:
async with httpx.AsyncClient(timeout=float(config.get("timeout") or 20.0)) as client:
response = await client.post(
f"{base_url}/v1/search",
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
},
json={"query": query, "limit": count},
)
except httpx.RequestError as exc:
raise SearchError(f"Could not reach {base_url}: {exc}") from exc
if response.status_code == 401:
raise SearchError("Firecrawl rejected the API key.")
if response.status_code == 402:
raise SearchError("The Firecrawl account is out of credit.")
try:
payload = response.json()
except ValueError as exc:
raise SearchError(f"Firecrawl returned HTTP {response.status_code}.") from exc
if response.status_code >= 400 or (
isinstance(payload, dict) and payload.get("success") is False
):
detail = payload.get("error") if isinstance(payload, dict) else ""
raise SearchError(str(detail) or f"Firecrawl returned HTTP {response.status_code}.")
rows = payload.get("data") if isinstance(payload, dict) else None
# Newer responses nest the list under data.web; older ones put it directly
# in data. Both are read so an API revision does not empty the results.
if isinstance(rows, dict):
rows = rows.get("web")
if not isinstance(rows, list):
raise SearchError("Firecrawl returned a response in an unexpected shape.")
results = []
for row in rows[:count]:
if not isinstance(row, dict):
continue
result = clean(
row.get("title"),
row.get("url"),
row.get("description") or row.get("markdown") or row.get("content"),
)
if result is not None:
results.append(result)
return results
+72
View File
@@ -0,0 +1,72 @@
"""SearXNG, a self-hosted metasearch instance.
The right answer for anyone already running one: no third party sees the
queries, and it aggregates several engines. It needs one thing switched on
first, which a stock install does not have, so that case is detected and named
rather than reported as "search failed".
"""
from __future__ import annotations
from typing import Any
import httpx
from lembas.services.search.base import SearchError, SearchResult, clean
# What a stock settings.yml is missing. Worth quoting exactly: it is the whole
# fix, and hunting for it in the documentation takes longer than reading it.
JSON_DISABLED = (
"This SearXNG instance will not answer in JSON. Add \"- json\" under "
"search.formats in its settings.yml and restart it."
)
async def search(config: dict[str, Any], query: str, count: int) -> list[SearchResult]:
base_url = str(config.get("searxng_base_url") or "").strip().rstrip("/")
if not base_url:
raise SearchError("No SearXNG instance has been configured.")
params = {
"q": query,
"format": "json",
"categories": "general",
"safesearch": {"off": "0", "moderate": "1", "strict": "2"}.get(
str(config.get("safesearch") or "moderate"), "1"
),
}
try:
async with httpx.AsyncClient(
timeout=float(config.get("timeout") or 20.0), follow_redirects=True
) as client:
response = await client.get(f"{base_url}/search", params=params)
except httpx.RequestError as exc:
raise SearchError(f"Could not reach {base_url}: {exc}") from exc
# 403 on an otherwise working instance means the JSON format is not in the
# allowed list -- SearXNG refuses the format rather than the request.
if response.status_code == 403:
raise SearchError(JSON_DISABLED)
if response.status_code >= 400:
raise SearchError(f"{base_url} returned HTTP {response.status_code}.")
try:
payload = response.json()
except ValueError as exc:
# An HTML page where JSON was asked for is the same misconfiguration
# wearing a different status code.
raise SearchError(JSON_DISABLED) from exc
rows = payload.get("results") if isinstance(payload, dict) else None
if not isinstance(rows, list):
raise SearchError("SearXNG returned a response in an unexpected shape.")
results = []
for row in rows[:count]:
if not isinstance(row, dict):
continue
result = clean(row.get("title"), row.get("url"), row.get("content"))
if result is not None:
results.append(result)
return results
+67 -2
View File
@@ -20,9 +20,11 @@ from lembas.config import settings as env_settings
from lembas.db.models import Setting
GENERAL = "general"
AUDIO = "audio"
SEARCH = "search"
def _defaults() -> dict[str, Any]:
def _general_defaults() -> dict[str, Any]:
return {
"allow_signup": env_settings.allow_signup,
# When on, new accounts land in the `pending` role and cannot sign in
@@ -35,9 +37,64 @@ def _defaults() -> dict[str, Any]:
}
def _audio_defaults() -> dict[str, Any]:
"""Speech-to-text and text-to-speech endpoints.
Two separate endpoints rather than one, because they usually are: a local
install runs whisper.cpp for one and Kokoro for the other. Both speak the
OpenAI audio API, so the shape below is the same on each side.
"""
return {
"stt_enabled": False,
"stt_base_url": "",
"stt_api_key_encrypted": "",
"stt_model": "whisper-1",
# Empty means "let the server detect it", which is what whisper does
# best. A forced language is an override, not a default.
"stt_language": "",
"tts_enabled": False,
"tts_base_url": "",
"tts_api_key_encrypted": "",
"tts_model": "tts-1",
"tts_voice": "",
"tts_format": "mp3",
"tts_speed": 1.0,
# The instance-wide starting point for the per-user toggle, not a
# setting that forces anything on anyone.
"tts_autoplay": False,
}
def _search_defaults() -> dict[str, Any]:
return {
"enabled": False,
"provider": "ddgs",
"max_results": 5,
"region": "wt-wt",
"safesearch": "moderate",
"searxng_base_url": "",
"firecrawl_base_url": "https://api.firecrawl.dev",
"firecrawl_api_key_encrypted": "",
"timeout": 20.0,
}
_DEFAULTS: dict[str, Any] = {
GENERAL: _general_defaults,
AUDIO: _audio_defaults,
SEARCH: _search_defaults,
}
def defaults(key: str = GENERAL) -> dict[str, Any]:
"""The built-in values for a settings group, with nothing stored applied."""
factory = _DEFAULTS.get(key)
return factory() if factory else {}
def get_group(db: DBSession, key: str = GENERAL) -> dict[str, Any]:
"""Stored settings for a group, with defaults filled in for absent keys."""
values = _defaults() if key == GENERAL else {}
values = defaults(key)
row = db.get(Setting, key)
if row is not None and isinstance(row.value, dict):
values.update(row.value)
@@ -64,3 +121,11 @@ def update(db: DBSession, changes: dict[str, Any], *, key: str = GENERAL) -> dic
def signup_allowed(db: DBSession) -> bool:
return bool(get(db, "allow_signup"))
def audio(db: DBSession) -> dict[str, Any]:
return get_group(db, AUDIO)
def search(db: DBSession) -> dict[str, Any]:
return get_group(db, SEARCH)
+261
View File
@@ -0,0 +1,261 @@
"""Tools a model may call while it answers.
One tool so far -- web search -- but the shape is the point: a registry of
named callables with a JSON schema each, offered to the endpoint and executed
here when it asks. Built-in tools, MCP servers and agentic execution all plug
in at the same place.
Two things gate whether a tool is offered at all:
* the administrator has configured and enabled it, and
* the chat's model is marked as supporting tools.
The second is not optional politeness. Sending a ``tools`` array to an endpoint
that does not implement tool calling fails the entire request, exactly the way
sending image parts to a model without vision does -- and for the same reason,
the capability flag on the model is what decides.
"""
from __future__ import annotations
import json
import logging
from collections.abc import Awaitable, Callable
from dataclasses import dataclass, field
from typing import Any
from sqlalchemy.orm import Session as DBSession
from lembas.db.models import Chat, User
from lembas.services import search as search_service
from lembas.services import settings_store
from lembas.services.search.base import SearchError
log = logging.getLogger(__name__)
# How many times a model may call tools before it has to answer with words.
# Not a safety limit so much as a termination one: a small model that has
# decided searching is the answer will otherwise search until the context runs
# out, and each round costs a full request.
MAX_ROUNDS = 3
WEB_SEARCH = "web_search"
WEB_SEARCH_SCHEMA: dict[str, Any] = {
"type": "function",
"function": {
"name": WEB_SEARCH,
"description": (
"Search the web for current information. Use this when the answer "
"depends on recent events, on facts you are unsure of, or on "
"anything that may have changed since your training data. Returns "
"a numbered list of results with titles, URLs and short extracts."
),
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search terms. Keep them short and specific.",
},
"max_results": {
"type": "integer",
"description": "How many results to return. Defaults to the site setting.",
},
},
"required": ["query"],
},
},
}
@dataclass
class ToolOutcome:
"""What running a tool produced, for the model and for the reader.
The two are deliberately different. `content` is the flat text the model
reads back; `event` is what the transcript shows, and keeps the results
structured so they can be rendered as links rather than as a wall of URLs.
"""
content: str
event: dict[str, Any] = field(default_factory=dict)
def enabled_tools(db: DBSession, chat: Chat, user: User | None) -> list[dict[str, Any]]:
"""The tool schemas to offer for this chat, which is usually none."""
from lembas.security import permissions
from lembas.services import chat as chat_service
config = settings_store.search(db)
if not config.get("enabled"):
return []
if not permissions.has(db, user, "tools.web_search"):
return []
if not chat_service.model_supports(db, chat, "tools"):
return []
if search_service.availability(str(config.get("provider") or "ddgs")):
# Configured but unusable -- offering a tool that will fail on every
# call is worse than not offering it.
return []
return [WEB_SEARCH_SCHEMA]
async def run_tool(config: dict[str, Any], name: str, arguments: str) -> ToolOutcome:
"""Execute one tool call.
Never raises. A tool that fails hands the model an explanation and lets it
carry on -- a failed search should produce "I could not look that up"
rather than killing the whole reply.
"""
if name != WEB_SEARCH:
return ToolOutcome(
content=f"There is no tool called {name!r}.",
event={"name": name, "status": "error", "error": "Unknown tool."},
)
try:
parsed = json.loads(arguments) if arguments.strip() else {}
except json.JSONDecodeError:
# Small models emit malformed argument JSON often enough that this is a
# normal path, not an exceptional one. Treat the whole string as the
# query rather than giving up.
parsed = {"query": arguments.strip()}
if not isinstance(parsed, dict):
parsed = {"query": str(parsed)}
query = str(parsed.get("query") or "").strip()
if not query:
return ToolOutcome(
content="No search query was given.",
event={"name": name, "status": "error", "error": "No query was given."},
)
limit = parsed.get("max_results")
try:
limit = int(limit) if limit is not None else None
except (TypeError, ValueError):
limit = None
try:
results = await search_service.run(config, query, limit=limit)
except SearchError as exc:
log.info("web search failed for %r: %s", query[:60], exc.message)
return ToolOutcome(
content=f"The search failed: {exc.message}",
event={"name": name, "query": query, "status": "error", "error": exc.message},
)
event = {
"name": name,
"query": query,
"status": "ok",
"results": [
{"title": r.title, "url": r.url, "snippet": r.snippet, "host": r.host}
for r in results
],
}
if not results:
return ToolOutcome(content=f"No results were found for {query!r}.", event=event)
lines = [f"Search results for {query!r}:"]
for index, result in enumerate(results, start=1):
lines.append(f"\n[{index}] {result.title}\n{result.url}\n{result.snippet}")
return ToolOutcome(content="\n".join(lines), event=event)
class ToolCallAccumulator:
"""Reassembles tool calls arriving as streamed fragments.
An endpoint sends ``delta.tool_calls`` as a list of partial objects: the id
and the function name arrive once, and ``arguments`` arrives as a string
split across however many chunks the tokeniser produced. Entries are keyed
by ``index`` because that is the only field guaranteed on every fragment --
the id is absent from continuations, and matching on name breaks the moment
a model calls the same tool twice in one turn.
"""
def __init__(self) -> None:
self._calls: dict[int, dict[str, Any]] = {}
def feed(self, fragments: list[dict[str, Any]]) -> None:
for fragment in fragments:
if not isinstance(fragment, dict):
continue
index = fragment.get("index")
if not isinstance(index, int):
# Some servers omit index entirely when there is only one call.
index = 0
call = self._calls.setdefault(index, {"id": "", "name": "", "arguments": ""})
if fragment.get("id"):
call["id"] = str(fragment["id"])
function = fragment.get("function") or {}
if isinstance(function, dict):
if function.get("name"):
call["name"] = str(function["name"])
arguments = function.get("arguments")
if isinstance(arguments, str):
call["arguments"] += arguments
@property
def calls(self) -> list[dict[str, Any]]:
"""Completed calls, in the order the endpoint indexed them."""
return [
{
# An id is required when the results are sent back, and not
# every server supplies one.
"id": call["id"] or f"call_{index}",
"name": call["name"],
"arguments": call["arguments"],
}
for index, call in sorted(self._calls.items())
if call["name"]
]
def __bool__(self) -> bool:
return bool(self.calls)
def assistant_turn(calls: list[dict[str, Any]], content: str) -> dict[str, Any]:
"""The assistant message to send back with the tool results.
The endpoint needs its own tool_calls echoed before the tool replies, or it
has nothing to match the tool_call_ids against.
"""
return {
"role": "assistant",
"content": content or None,
"tool_calls": [
{
"id": call["id"],
"type": "function",
"function": {"name": call["name"], "arguments": call["arguments"]},
}
for call in calls
],
}
def tool_turn(call: dict[str, Any], content: str) -> dict[str, Any]:
return {
"role": "tool",
"tool_call_id": call["id"],
"name": call["name"],
"content": content,
}
ToolRunner = Callable[..., Awaitable[ToolOutcome]]
__all__ = [
"MAX_ROUNDS",
"WEB_SEARCH",
"ToolCallAccumulator",
"ToolOutcome",
"assistant_turn",
"enabled_tools",
"run_tool",
"tool_turn",
]