From 27eba24c8e90bce923d0c14a8b7e877880d91423 Mon Sep 17 00:00:00 2001 From: sal-uva Date: Tue, 15 Sep 2026 21:22:32 -0400 Subject: [PATCH 01/24] feat: enable embedding models, add text embedding generator processor, similarity finder processor, and embedding mapper processor --- VERSION | 2 +- common/lib/exceptions.py | 6 + common/lib/llm/clients/ollama_client.py | 112 ++++- common/lib/llm/llm_client.py | 37 ++ common/lib/llm/models.py | 58 +++ helper-scripts/migrate/migrate-1.57-1.58.py | 40 ++ .../machine_learning/embedding_similarity.py | 220 ++++++++++ .../generate_text_embeddings.py | 292 +++++++++++++ processors/machine_learning/llm_prompter.py | 29 +- processors/machine_learning/prompt_compass.py | 5 +- processors/visualisation/embedding_map.py | 387 ++++++++++++++++++ setup.py | 1 + tests/test_llm_models.py | 179 ++++++++ 13 files changed, 1346 insertions(+), 22 deletions(-) create mode 100644 common/lib/llm/models.py create mode 100644 helper-scripts/migrate/migrate-1.57-1.58.py create mode 100644 processors/machine_learning/embedding_similarity.py create mode 100644 processors/machine_learning/generate_text_embeddings.py create mode 100644 processors/visualisation/embedding_map.py create mode 100644 tests/test_llm_models.py diff --git a/VERSION b/VERSION index abf39ad47..ac7c5e2a2 100644 --- a/VERSION +++ b/VERSION @@ -1,4 +1,4 @@ -1.57 +1.58 This file should not be modified. It is used by 4CAT to determine whether it needs to run migration scripts to e.g. update the database structure to a more diff --git a/common/lib/exceptions.py b/common/lib/exceptions.py index b0b798a4c..17b5208ae 100644 --- a/common/lib/exceptions.py +++ b/common/lib/exceptions.py @@ -162,3 +162,9 @@ class MediaSignatureException(FourcatException): Raise in media processors when the media cannot be read """ pass + +class LLMServerException(FourcatException): + """ + Raise when an LLM server request fails or returns something unusable + """ + pass diff --git a/common/lib/llm/clients/ollama_client.py b/common/lib/llm/clients/ollama_client.py index 6484e1095..501634177 100644 --- a/common/lib/llm/clients/ollama_client.py +++ b/common/lib/llm/clients/ollama_client.py @@ -3,6 +3,7 @@ """ import requests +from common.lib.exceptions import LLMServerException from common.lib.llm.llm_client import LLMServerClient @@ -46,7 +47,7 @@ def parse_supported_media_types(self, meta: dict) -> list[str]: **Primary path**: reads `meta["capabilities"]`: - `"completion"` → `"text"` - `"vision"` → `"image"` - - `"embedding"` → `"embedding"` + - `"embedding"` → `"text"` (an embedder still consumes text) **Fallback path** (used when capabilities are absent or only yield `"text"`): inspects GGUF `model_info` / `details` for vision signals and adds @@ -57,16 +58,16 @@ def parse_supported_media_types(self, meta: dict) -> list[str]: Returns `[]` when `meta` is `None` (unknown — callers should include the model, not block it). """ - if meta is None or not meta.get("metadata"): + if meta is None or not (meta.get("metadata") or meta.get("capabilities")): return [] - capabilities = meta["metadata"].get("capabilities", []) + capabilities = self._get_capabilities(meta) media_types: list[str] = [] _cap_map = { "completion": "text", + "embedding": "text", "vision": "image", - "embedding": "embedding", } for cap in capabilities: mapped = _cap_map.get(cap) @@ -88,6 +89,109 @@ def parse_supported_media_types(self, meta: dict) -> list[str]: return media_types + @staticmethod + def _get_capabilities(meta: dict) -> list[str]: + """ + Read a model's capability list. + + Ollama reports capabilities in two places: nested under `metadata` for + the `/api/show` response `list_models()` attaches, and inline on the + `/api/tags` entry itself. Prefer the former and fall back to the latter, + so a model whose `/api/show` request failed is still classified. + + :param dict meta: Model metadata, or `None`. + :returns list[str]: Ollama capability strings, or `[]` if unknown. + """ + if not meta: + return [] + + metadata = meta.get("metadata") or {} + return metadata.get("capabilities") or meta.get("capabilities") or [] + + def parse_supported_tasks(self, meta: dict) -> list[str]: + """ + Derive the tasks a model supports from its Ollama capabilities. + + :param dict meta: `/api/show` response dict, or `None`. + :returns list[str]: Supported tasks - `"generate"`, `"embed"`, or both. + """ + capabilities = self._get_capabilities(meta) + + tasks = [] + if "completion" in capabilities: + tasks.append("generate") + if "embedding" in capabilities: + tasks.append("embed") + + # An unreachable or partial /api/show leaves nothing to go on. Assume + # generation, matching the base class, rather than hiding the model + # from every processor at once. + return tasks if tasks else ["generate"] + + def embed(self, model_id: str, inputs: list, media: list | None = None, timeout: int = 300) -> list[list[float]]: + """ + Embed inputs via Ollama's `/api/embed` endpoint. + + The endpoint accepts a list and returns one vector per item, so a caller + that batches its inputs gets one round-trip per batch rather than per + item. + + Text only. `/api/embed` takes `model`, `input`, `truncate`, `options`, + `keep_alive` and `dimensions` - there is no image parameter, and Ollama + *silently discards* an `images` field rather than rejecting it, handing + back a text-only vector that looks entirely valid (verified against + Ollama 0.34.1; see ollama/ollama#5304 and #7677, both still open). So + refuse media here explicitly: a quiet wrong answer is worse than a loud + failure, and this is the only place that can tell the difference. + + :param str model_id: Ollama model name, e.g. `"mxbai-embed-large:latest"`. + :param list inputs: Strings to embed. + :param list media: Not supported; passing any raises. + :param int timeout: Request timeout in seconds. + :returns list[list[float]]: One vector per input, in input order. + :raises LLMServerException: If media is passed, if the server errors, + or if it returns a number of vectors that does not match the number of + inputs (which would silently misalign vectors with their items). + """ + if media: + raise LLMServerException( + "Ollama cannot produce multimodal embeddings: its /api/embed endpoint has no image parameter and " + "ignores one without reporting an error, which would yield text-only vectors that look valid. Use a " + "server that supports multimodal embedding instead.") + + if not inputs: + return [] + + try: + response = self._session.post( + f"{self.base_url}/api/embed", + headers=self._headers, + json={"model": model_id, "input": inputs}, + timeout=timeout, + ) + except requests.RequestException as e: + raise LLMServerException(f"Could not reach Ollama server at {self.base_url}: {e}") + + if response.status_code != 200: + raise LLMServerException( + f"Ollama server returned status {response.status_code} while embedding with model " + f"'{model_id}': {response.text}") + + try: + embeddings = response.json().get("embeddings") + except ValueError as e: + raise LLMServerException(f"Ollama server returned invalid JSON while embedding: {e}") + + if not embeddings: + raise LLMServerException(f"Ollama server returned no embeddings for model '{model_id}'") + + if len(embeddings) != len(inputs): + raise LLMServerException( + f"Ollama server returned {len(embeddings)} embeddings for {len(inputs)} inputs; cannot map " + f"vectors back to items") + + return embeddings + def format_display_name(self, meta: dict) -> str: """ Build a human-readable display name for a model. diff --git a/common/lib/llm/llm_client.py b/common/lib/llm/llm_client.py index 60783294a..1508e6e0a 100644 --- a/common/lib/llm/llm_client.py +++ b/common/lib/llm/llm_client.py @@ -129,9 +129,46 @@ def build_model_entry(self, meta: dict) -> dict: "server": self.server_config["_id"], "wrapper": self.server_config["type"], "supported_media_types": self.parse_supported_media_types(meta), + "supported_tasks": self.parse_supported_tasks(meta), "metadata": meta, } + def parse_supported_tasks(self, meta: dict) -> list[str]: + """ + Derive the tasks a model can be used for from its metadata. + + The default is `["generate"]`: every server 4CAT talks to serves chat + completions, while embedding endpoints are the exception. + + :param dict meta: Model metadata, or `None` if unavailable. + :returns list[str]: Supported tasks - `"generate"`, `"embed"`, or both. + """ + return ["generate"] + + def embed(self, model_id: str, inputs: list, media: list | None = None, timeout: int = 300) -> list[list[float]]: + """ + Embed one or more inputs, returning one vector per input. + + Deliberately *not* routed through `LLMAdapter`/LangChain. LangChain's + embedding interface is text-only (`embed_documents(list[str])`), so it + cannot handle multimodal embeddings. + + A client that cannot embed media **must raise** when `media` is given + rather than embedding the text alone. Returning a text-only vector for + a request that asked for media is silent data corruption: the caller + gets a plausible vector back and has no way to tell it does not describe + what it asked about. + + :param str model_id: Model ID *within this server's context* (i.e. a + `local_id`, not a global model ID). + :param list inputs: Inputs to embed. + :param list media: Media to embed alongside `inputs`, for servers that + support multimodal embedding. The payload shape is server-specific. + :param int timeout: Request timeout in seconds. + :returns list[list[float]]: One vector per input, in input order. + """ + raise NotImplementedError(f"{self.__class__.__name__} does not support embedding") + def get_model_card_url(self, meta: dict) -> str: """ Get a URL for a model card for a given model diff --git a/common/lib/llm/models.py b/common/lib/llm/models.py new file mode 100644 index 000000000..dd8692a66 --- /dev/null +++ b/common/lib/llm/models.py @@ -0,0 +1,58 @@ +""" +Present the configured LLM model inventory to processors. + +Kept separate from `common.lib.llm.adapter` on purpose: that module pulls in the +whole LangChain stack, while this one only reads 4CAT settings. A processor that +merely needs to list models or a test that needs to check the filtering can +import this without the heavy dependencies. +""" + + +def supports_task(model: dict, task: str = "generate") -> bool: + """ + Check whether a model entry can be used for a given task. + + Entries built before `supported_tasks` existed carry no task info. They all + predate embedding support, so they read as generative rather than vanishing + from every processor until the inventory is next refreshed. + + :param dict model: A single `llm.available_models` entry. + :param str task: Task to check for - `"generate"` or `"embed"`. + :return bool: Whether the model supports the task. + """ + return task in model.get("supported_tasks", ["generate"]) + + +def get_model_library(config, task: str = "generate") -> dict: + """ + Get the LLM models available for a given task, grouped by server. + + Only models that are both enabled and capable of `task` are returned, so a + prompting processor never offers an embedding model and an embedding + processor never offers a generative one. Without this filter either choice + produces a dataset that only fails once the request reaches the server. + + :param config: 4CAT config reader (context-aware, so per-user `llm.access` + is respected) + :param str task: Task the model must support - `"generate"` or `"embed"`. + :return dict: `{server name: {model ID: model display name}}`, shaped for + a `UserInput.OPTION_CHOICE` option. + """ + available_models = config.get("llm.available_models", {}) + enabled_model_ids = config.get("llm.enabled_models", []) + servers = config.get("llm.servers", {}) + if not config.get("llm.access"): + enabled_model_ids = [_ for _ in enabled_model_ids if _.startswith("thirdparty")] + + models_option = {} + for key, value in {k: v for k, v in available_models.items() if k in enabled_model_ids}.items(): + if not supports_task(value, task): + continue + + server = servers[value["server"]] + if server["name"] not in models_option: + models_option[server["name"]] = {} + + models_option[server["name"]][key] = value["name"] + + return models_option diff --git a/helper-scripts/migrate/migrate-1.57-1.58.py b/helper-scripts/migrate/migrate-1.57-1.58.py new file mode 100644 index 000000000..88feb2ba0 --- /dev/null +++ b/helper-scripts/migrate/migrate-1.57-1.58.py @@ -0,0 +1,40 @@ +# Model entries in `llm.available_models` gained a `supported_tasks` field, so +# that processors can tell generative models and embedding models apart. +# +# Clearing the inventory forces it to be rebuilt. +import sys +import os + +from pathlib import Path + +sys.path.insert(0, os.path.join(os.path.abspath(os.path.dirname(__file__)), "../..")) +from common.lib.database import Database +from common.lib.logger import Logger + +import configparser # noqa: E402 + +log = Logger(output=True) +ini = configparser.ConfigParser() +ini.read(Path(__file__).parent.parent.parent.resolve().joinpath("config/config.ini")) +db_config = ini["DATABASE"] + +db = Database( + logger=log, + dbname=db_config["db_name"], + user=db_config["db_user"], + password=db_config["db_password"], + host=db_config["db_host"], + port=db_config["db_port"], + appname="4cat-migrate", +) + +print(" Clearing known LLM models so they are re-indexed with task support...") +has_setting = db.fetchone("SELECT COUNT(*) AS num FROM settings WHERE name = 'llm.available_models'") + +if has_setting["num"] > 0: + db.upsert("settings", {"name": "llm.available_models", "value": "{}", "tag": ""}, constraints=["name", "tag"]) + print(" ...cleared; models will be re-indexed on 4CAT restart") +else: + print(" ...no models indexed yet, nothing to do") + +print(" - done!") diff --git a/processors/machine_learning/embedding_similarity.py b/processors/machine_learning/embedding_similarity.py new file mode 100644 index 000000000..92d32fc75 --- /dev/null +++ b/processors/machine_learning/embedding_similarity.py @@ -0,0 +1,220 @@ +""" +Rank dataset items by how close their embedding is to a given text. +""" +import csv + +import numpy as np + +from backend.lib.processor import BasicProcessor +from common.lib.compatibility import Compatibility +from common.lib.exceptions import LLMServerException, ProcessorInterruptedException, QueryParametersException +from common.lib.llm.llm_client import LLMServerClient +from common.lib.user_input import UserInput + +__author__ = "Sal Hagen" +__credits__ = ["Sal Hagen"] +__maintainer__ = "Sal Hagen" +__email__ = "4cat@oilab.eu" + + +class EmbeddingSimilarity(BasicProcessor): + """ + Embed a user-supplied text with the same model the parent dataset used, then + score every item by the cosine similarity between the two vectors. + """ + type = "embedding-similarity" # job type ID + category = "Text analysis" # category + title = "Compare items to a text" # title displayed in UI + description = ("Write a text, and score each item by how similar its meaning is to it, using the embeddings " + "generated for the parent dataset. Similarity is measured as cosine similarity, from -1 " + "(opposite) through 0 (unrelated) to 1 (identical). Because this compares meaning rather than " + "wording, items can score highly without sharing any words with your text.") + extension = "csv" # extension of result file, used internally and in UI + + compatibility = Compatibility(types={"text-embeddings"}) + + references = [ + "[Cosine similarity](https://en.wikipedia.org/wiki/Cosine_similarity)", + ] + + @classmethod + def get_queue_id(cls, remote_id, details, dataset) -> str: + """ + Shared queue for locally hosted models + + :param str remote_id: Job item ID + :param dict details: Job details + :param DataSet dataset: Dataset to run job for + :return str: Queue ID + """ + # Unique queue for locally hosted models; used by other local model processors as well + return "local_models" + + @classmethod + def get_options(cls, parent_dataset=None, config=None) -> dict: + """ + Get processor options + + :param parent_dataset DataSet: An object representing the dataset that + the processor would be or was run on. + :param config ConfigManager|None config: Configuration reader (context-aware) + :return dict: Options for this processor + """ + return { + "query_info": { + "type": UserInput.OPTION_INFO, + "help": "Your text is embedded with the same model the parent dataset used, since vectors from " + "different models cannot be compared.", + }, + "query_text": { + "type": UserInput.OPTION_TEXT_LARGE, + "help": "Text to compare against", + "default": "", + "tooltip": "A word, sentence, or paragraph. Longer texts describing a theme usually work better " + "than single words.", + }, + "sort": { + "type": UserInput.OPTION_TOGGLE, + "help": "Sort by similarity", + "default": True, + "tooltip": "Put the most similar items first. Disable to keep the original order of the dataset.", + }, + "save_annotations": { + "type": UserInput.OPTION_ANNOTATION, + "label": "cosine_similarity", + "default": False, + "tooltip": "Write the similarity score back to the original dataset as an annotation.", + }, + "annotation_label": { + "type": UserInput.OPTION_TEXT, + "help": "Annotation label", + "default": "", + "tooltip": "[optional] Name for the annotation. Defaults to 'similarity_to_'.", + "requires": "save_annotations==true", + }, + } + + @classmethod + def validate_query(cls, query, request, config): + """ + Validate input + + :param dict query: Query parameters, as submitted by the user via the web interface + :param request: Flask request + :param ConfigManager config: Configuration reader + :return dict: Validated query parameters + """ + if not query.get("query_text", "").strip(): + raise QueryParametersException("You need to provide a text to compare against.") + + return query + + def process(self): + """ + Score each item by cosine similarity to the embedded query text. + """ + query_text = self.parameters.get("query_text", "").strip() + + # The query has to be embedded by the same model that produced the + # parent's vectors - vectors from different models live in different + # spaces, so comparing them yields a number that means nothing. + model_id = self.source_dataset.parameters.get("model") + available_models = { + k: v for k, v in self.config.get("llm.available_models", {}).items() + if k in self.config.get("llm.enabled_models", []) + } + + if model_id not in available_models: + self.dataset.finish_with_error( + f"The embedding model used for the parent dataset ('{model_id}') is no longer available, so your text " + f"cannot be embedded in the same way. Ask an admin to re-enable it.") + return + + model = available_models[model_id] + server = self.config.get("llm.servers", {}).get(model["server"]) + if not server: + self.dataset.finish_with_error(f"The LLM server for model '{model_id}' is no longer configured.") + return + + self.dataset.update_status(f"Embedding your text with {model['local_id']}") + try: + client = LLMServerClient.get_client(self.config, server, self.log) + query_vector = np.asarray(client.embed(model["local_id"], [query_text])[0], dtype=np.float64) + except (LLMServerException, ValueError, IndexError) as e: + self.dataset.finish_with_error(f"Could not embed your text: {e}") + return + + query_norm = np.linalg.norm(query_vector) + if not query_norm: + self.dataset.finish_with_error("Your text produced an empty embedding; try a longer or different text.") + return + + self.dataset.update_status("Comparing items to your text") + results = [] + skipped = 0 + + for i, item in enumerate(self.source_dataset.iterate_items(self)): + if self.interrupted: + raise ProcessorInterruptedException("Interrupted while calculating similarities") + + vector = item.get("embedding") + # map_item flattens the vector to a space-separated string; the raw + # NDJSON keeps it as a list. Accept either. + if isinstance(vector, str): + vector = [float(value) for value in vector.split() if value] + + if not vector or len(vector) != len(query_vector): + # a vector of a different width cannot be compared; scoring it + # anyway would silently produce a meaningless number + skipped += 1 + continue + + vector = np.asarray(vector, dtype=np.float64) + norm = np.linalg.norm(vector) + if not norm: + skipped += 1 + continue + + results.append({ + "id": item.get("id"), + "text": item.get("text"), + "similarity": float(np.dot(query_vector, vector) / (query_norm * norm)), + "compared_to": query_text, + "model": item.get("model", model["local_id"]), + }) + + if i % 250 == 0: + self.dataset.update_progress(i / self.source_dataset.num_rows) + + if not results: + self.dataset.finish_with_error("No comparable embeddings were found in the parent dataset.") + return + + if self.parameters.get("sort", True): + results.sort(key=lambda row: row["similarity"], reverse=True) + + with self.dataset.get_results_path().open("w", encoding="utf-8", newline="") as outfile: + writer = csv.DictWriter(outfile, fieldnames=["id", "text", "similarity", "compared_to", "model"]) + writer.writeheader() + for row in results: + writer.writerow(row) + + if self.parameters.get("save_annotations", False): + label = self.parameters.get("annotation_label", "").strip() + if not label: + # keep the label readable when the query is a long paragraph + shortened = query_text if len(query_text) <= 20 else query_text[:20].rstrip() + "…" + label = f"cosine_similarity_{shortened}" + + self.save_annotations([{ + "item_id": row["id"], + "label": label, + "value": round(row["similarity"], 4), + "type": "text", + } for row in results]) + + status = f"Compared {len(results):,} items to your text" + if skipped: + status += f" (skipped {skipped:,} items without a usable embedding)" + self.dataset.update_status(status, is_final=True) + self.dataset.finish(len(results)) diff --git a/processors/machine_learning/generate_text_embeddings.py b/processors/machine_learning/generate_text_embeddings.py new file mode 100644 index 000000000..4b8f20307 --- /dev/null +++ b/processors/machine_learning/generate_text_embeddings.py @@ -0,0 +1,292 @@ +""" +Generate text embeddings via a configured LLM server. +""" +import json +import time + +from datetime import datetime + +from backend.lib.processor import BasicProcessor +from common.lib.compatibility import Compatibility +from common.lib.exceptions import LLMServerException, ProcessorInterruptedException, QueryParametersException +from common.lib.item_mapping import MappedItem +from common.lib.llm.llm_client import LLMServerClient +from common.lib.llm.models import get_model_library +from common.lib.user_input import UserInput + +__author__ = "Sal Hagen" +__credits__ = ["Sal Hagen"] +__maintainer__ = "Sal Hagen" +__email__ = "4cat@oilab.eu" + + +class GenerateTextEmbeddings(BasicProcessor): + """ + Generate an embedding vector per dataset item using an embedding model + hosted on a configured LLM server. + """ + type = "text-embeddings" # job type ID + category = "Text analysis" # category + title = "Generate text embeddings via LLM" # title displayed in UI + description = ("Generate a numerical representation (an 'embedding') of the text of each item using an LLM embedding " + "model. Embeddings can be used to e.g. retrieve similar kinds of texts. Requires an embedding model " + "be installed on this server.") + extension = "ndjson" # extension of result file, used internally and in UI + + compatibility = Compatibility(extensions={"csv", "ndjson"}) + + references = [ + "[Ollama embedding models](https://ollama.com/blog/embedding-models)", + ] + + @classmethod + def get_queue_id(cls, remote_id, details, dataset) -> str: + """ + Shared queue for locally hosted models + + :param str remote_id: Job item ID + :param dict details: Job details + :param DataSet dataset: Dataset to run job for + :return str: Queue ID + """ + # Unique queue for locally hosted models; used by other local model processors as well + return "local_models" + + @classmethod + def is_compatible_with(cls, module=None, config=None): + """ + Text datasets, but only where an embedding model is actually available. + + :param module: Dataset or processor to check against + :param ConfigManager|None config: Context-aware configuration reader + :return bool: + """ + if config is None or not cls.compatibility.is_compatible_with(module, config=config): + return False + + return bool(get_model_library(config, task="embed")) + + @classmethod + def get_options(cls, parent_dataset=None, config=None) -> dict: + """ + Get processor options + + :param parent_dataset DataSet: An object representing the dataset that + the processor would be or was run on. + :param config ConfigManager|None config: Configuration reader (context-aware) + :return dict: Options for this processor + """ + options = { + "model": { + "type": UserInput.OPTION_CHOICE, + "help": "Embedding model", + # embedding models only - a generative model cannot return a vector + "options": get_model_library(config, task="embed") if config else {}, + "default": "", + "tooltip": "Only models that can produce embeddings are listed. Ask an admin to enable more via the " + "control panel.", + }, + "columns": { + "type": UserInput.OPTION_TEXT, + "help": "Column(s) to embed", + "default": "body", + "tooltip": "Values of multiple columns are joined with a space before being embedded.", + }, + "amount": { + "type": UserInput.OPTION_TEXT, + "help": "No. of items", + "default": 100, + "min": 0, + "coerce_type": int, + "tooltip": "Use '0' to embed all items. Embedding is slower than it looks for large datasets; test on " + "a sample first.", + }, + "batch_size": { + "type": UserInput.OPTION_TEXT, + "help": "Items per request", + "default": 50, + "min": 1, + "max": 500, + "coerce_type": int, + "tooltip": "How many items to send to the server at once. Higher is faster but uses more memory on the " + "server; lower this if requests time out.", + }, + } + + if parent_dataset and parent_dataset.get_columns(): + columns = parent_dataset.get_columns() + options["columns"]["type"] = UserInput.OPTION_MULTI + options["columns"]["options"] = {v: v for v in columns} + options["columns"]["default"] = ["body"] if "body" in columns else [columns[0]] + + return options + + @classmethod + def validate_query(cls, query, request, config): + """ + Validate input + + :param dict query: Query parameters, as submitted by the user via the web interface + :param request: Flask request + :param ConfigManager config: Configuration reader + :return dict: Validated query parameters + """ + allowed_models = { + model_id + for server_models in get_model_library(config, task="embed").values() + for model_id in server_models + } + + if query.get("model") not in allowed_models: + raise QueryParametersException(f"The '{query.get('model')}' embedding model is not currently available.") + + if not query.get("columns"): + raise QueryParametersException("You need to select at least one column to embed.") + + return query + + def process(self): + """ + Embed the text of each item and write one vector per item. + """ + columns = self.parameters.get("columns", []) + if isinstance(columns, str): + columns = [columns] + + model_id = self.parameters.get("model") + available_models = { + k: v for k, v in self.config.get("llm.available_models", {}).items() + if k in self.config.get("llm.enabled_models", []) + } + + if model_id not in available_models: + self.dataset.finish_with_error(f"Embedding model '{model_id}' is no longer available. Ask an admin to " + f"enable it, or pick another model.") + return + + model = available_models[model_id] + server = self.config.get("llm.servers", {}).get(model["server"]) + if not server: + self.dataset.finish_with_error(f"The LLM server for model '{model_id}' is no longer configured.") + return + + try: + client = LLMServerClient.get_client(self.config, server, self.log) + except ValueError as e: + self.dataset.finish_with_error(str(e)) + return + + limit = self.parameters.get("amount", 100) + max_processed = min(limit, self.source_dataset.num_rows) if limit else self.source_dataset.num_rows + batch_size = self.parameters.get("batch_size", 50) + + self.dataset.update_status(f"Connecting to LLM server '{server['_id']}' with model '{model['local_id']}'") + self.dataset.log(f"Embedding with model '{model['local_id']}' on server '{server['_id']}'") + + embedded = 0 + skipped = 0 + processed = 0 + batch = [] # (item ID, text) tuples awaiting a request + + try: + with self.dataset.get_results_path().open("w", encoding="utf-8") as outfile: + for item in self.source_dataset.iterate_items(self): + if self.interrupted: + raise ProcessorInterruptedException("Interrupted while generating embeddings") + + if processed >= max_processed: + break + + processed += 1 + + text = " ".join([str(item.get(column, "")) for column in columns]).strip() + if not text: + # nothing to embed; an all-zero vector would be + # indistinguishable from a real one, so leave the item out + skipped += 1 + continue + + batch.append((item.get("id"), text)) + if len(batch) < batch_size: + continue + + embedded += self.embed_batch(client, model, batch, outfile) + batch = [] + self.dataset.update_status(f"Embedded {embedded:,} of {max_processed:,} items with " + f"{model['local_id']}") + self.dataset.update_progress(processed / max_processed) + + # Whatever is left over after the loop: the last batch is rarely + # full, and the final item may have been skipped for having no + # text, which would otherwise strand the batch it sits behind. + embedded += self.embed_batch(client, model, batch, outfile) + + except LLMServerException as e: + if embedded: + self.dataset.finish_with_warning(embedded, f"Not all items were embedded: {e}") + else: + self.dataset.finish_with_error(str(e)) + return + + if not embedded: + self.dataset.finish_with_error("No items could be embedded; check whether the selected columns contain " + "text.") + return + + status = f"Generated embeddings for {embedded:,} items" + if skipped: + status += f" (skipped {skipped:,} items with no text in the selected columns)" + self.dataset.update_status(status, is_final=True) + self.dataset.finish(embedded) + + def embed_batch(self, client, model, batch, outfile) -> int: + """ + Embed one batch of items and write the vectors to the result file. + + :param LLMServerClient client: Client for the server hosting the model + :param dict model: Model metadata, as in `llm.available_models` + :param list batch: List of `(item ID, text)` tuples + :param outfile: Open file handle to write NDJSON lines to + :return int: Number of items written + """ + if not batch: + return 0 + + vectors = client.embed(model["local_id"], [text for _, text in batch]) + + time_created = int(time.time()) + for (item_id, text), vector in zip(batch, vectors): + outfile.write(json.dumps({ + "id": item_id, + "text": text, + "embedding": vector, + "dimensions": len(vector), + "model": model["local_id"], + "time_created": datetime.fromtimestamp(time_created).strftime("%Y-%m-%d %H:%M:%S"), + "time_created_utc": time_created, + }) + "\n") + + return len(batch) + + @staticmethod + def map_item(item): + """ + Map an embedding record to a flat row. + + The vector is written as space-separated values, the same shape word2vec + and GloVe use in their text format, so the CSV export stays readable by + other tooling. + + :param item: Item to map + :return MappedItem: Mapped item + """ + embedding = item.get("embedding", []) + + return MappedItem({ + "id": item.get("id"), + "text": item.get("text"), + "model": item.get("model"), + "dimensions": item.get("dimensions"), + "embedding": " ".join([str(value) for value in embedding]), + "timestamp": item.get("time_created"), + }) diff --git a/processors/machine_learning/llm_prompter.py b/processors/machine_learning/llm_prompter.py index b318bb056..f0d9c8b5a 100644 --- a/processors/machine_learning/llm_prompter.py +++ b/processors/machine_learning/llm_prompter.py @@ -19,6 +19,7 @@ from common.lib.exceptions import ProcessorInterruptedException, QueryParametersException, QueryNeedsExplicitConfirmationException from common.lib.helpers import UserInput, nthify, andify, remove_nuls, flatten_dict from common.lib.llm.adapter import LLMAdapter +from common.lib.llm.models import get_model_library from backend.lib.processor import BasicProcessor from common.lib.compatibility import Compatibility @@ -77,22 +78,18 @@ def get_queue_id(cls, remote_id, details, dataset) -> str: return f"llm-local[{dataset.parameters.get('model').split('://')[1].split('/')[0]}]" @classmethod - def get_model_library(cls, config): - available_models = config.get("llm.available_models", {}) - enabled_model_ids = config.get("llm.enabled_models", []) - servers = config.get("llm.servers", {}) - if not config.get("llm.access"): - enabled_model_ids = [_ for _ in enabled_model_ids if _.startswith("thirdparty")] - - models_option = {} - for key, value in {k: v for k, v in available_models.items() if k in enabled_model_ids}.items(): - server = servers[value["server"]] - if server["name"] not in models_option: - models_option[server["name"]] = {} - - models_option[server["name"]][key] = value["name"] - - return models_option + def get_model_library(cls, config, task="generate"): + """ + Get the models this processor can offer, grouped by server. + + Generative models only: an embedding model returns a vector, not text, + so offering one here would only produce a run that fails at invoke time. + + :param config: 4CAT config reader + :param str task: Task the model must support. + :return dict: `{server name: {model ID: model display name}}` + """ + return get_model_library(config, task=task) @classmethod def get_options(cls, parent_dataset=None, config=None) -> dict: diff --git a/processors/machine_learning/prompt_compass.py b/processors/machine_learning/prompt_compass.py index 8794f8802..54f090f54 100644 --- a/processors/machine_learning/prompt_compass.py +++ b/processors/machine_learning/prompt_compass.py @@ -13,6 +13,7 @@ QueryNeedsExplicitConfirmationException, ) +from common.lib.llm.models import supports_task from processors.machine_learning.llm_prompter import LLMPrompter class PromptCompassRunner(ProcessorPreset): @@ -89,7 +90,9 @@ def get_options(cls, parent_dataset=None, config=None): if not config.get("llm.access"): enabled_model_ids = [_ for _ in enabled_model_ids if _.startswith("thirdparty")] - enabled_models = {k: v for k, v in available_models.items() if k in enabled_model_ids} + # generative models only + enabled_models = {k: v for k, v in available_models.items() + if k in enabled_model_ids and supports_task(v, "generate")} options = { "model": { diff --git a/processors/visualisation/embedding_map.py b/processors/visualisation/embedding_map.py new file mode 100644 index 000000000..364827d60 --- /dev/null +++ b/processors/visualisation/embedding_map.py @@ -0,0 +1,387 @@ +""" +Project embeddings to two dimensions and render them as an interactive map. +""" +import json + +import numpy as np + +from backend.lib.processor import BasicProcessor +from common.lib.compatibility import Compatibility +from common.lib.exceptions import ProcessorInterruptedException +from common.lib.user_input import UserInput + +__author__ = "Sal Hagen" +__credits__ = ["Sal Hagen"] +__maintainer__ = "Sal Hagen" +__email__ = "4cat@oilab.eu" + + +class EmbeddingMap(BasicProcessor): + """ + Reduce item embeddings to two dimensions and write a self-contained, + interactive HTML scatter plot. + """ + type = "embedding-map" # job type ID + category = "Visual" # category + title = "Embedding map" # title displayed in UI + description = ("Reduce the embeddings of each item to two dimensions and plot them as an interactive map, so " + "items with similar meanings sit near each other. Scroll to zoom, drag to pan, and hover over a " + "point to read its text. Note that distances in the map are a compressed approximation of the " + "original embedding space: clusters are meaningful, exact distances less so.") + extension = "html" # extension of result file, used internally and in UI + + compatibility = Compatibility(types={"text-embeddings"}) + + references = [ + "[McInnes, Leland, John Healy, and James Melville. 2018. 'UMAP: Uniform Manifold Approximation and Projection " + "for Dimension Reduction.' arXiv:1802.03426.](https://arxiv.org/abs/1802.03426)", + "[Understanding UMAP](https://pair-code.github.io/understanding-umap/)", + "[How to Use t-SNE Effectively](https://distill.pub/2016/misread-tsne/)", + ] + + @classmethod + def get_options(cls, parent_dataset=None, config=None) -> dict: + """ + Get processor options + + :param parent_dataset DataSet: An object representing the dataset that + the processor would be or was run on. + :param config ConfigManager|None config: Configuration reader (context-aware) + :return dict: Options for this processor + """ + return { + "algorithm": { + "type": UserInput.OPTION_CHOICE, + "help": "Reduction algorithm", + "options": { + "umap": "UMAP (recommended)", + "tsne": "t-SNE", + "pca": "PCA", + }, + "default": "umap", + "tooltip": "UMAP keeps both local and global structure and is usually the most readable. t-SNE " + "separates local clusters sharply but its global layout means little. PCA is fast and " + "fully deterministic, but flattens subtle structure.", + }, + "n_neighbors": { + "type": UserInput.OPTION_TEXT, + "help": "UMAP: neighbours", + "default": 15, + "min": 2, + "max": 200, + "coerce_type": int, + "tooltip": "How much of the dataset each point is compared against. Low values emphasise small local " + "clusters, high values emphasise overall shape.", + "requires": "algorithm==umap", + }, + "min_dist": { + "type": UserInput.OPTION_TEXT, + "help": "UMAP: minimum distance", + "default": 0.1, + "min": 0.0, + "max": 0.99, + "coerce_type": float, + "tooltip": "How tightly points may be packed together. Lower values give denser clumps.", + "requires": "algorithm==umap", + }, + "perplexity": { + "type": UserInput.OPTION_TEXT, + "help": "t-SNE: perplexity", + "default": 30, + "min": 2, + "max": 100, + "coerce_type": int, + "tooltip": "Roughly how many neighbours each point is balanced against. Lowered automatically if the " + "dataset is too small.", + "requires": "algorithm==tsne", + }, + "amount": { + "type": UserInput.OPTION_TEXT, + "help": "No. of items", + "default": 2000, + "min": 0, + "coerce_type": int, + "tooltip": "Use '0' for all items. Reduction slows down sharply on very large datasets.", + }, + "max_text_length": { + "type": UserInput.OPTION_TEXT, + "help": "Hover text length", + "default": 250, + "min": 20, + "max": 2000, + "coerce_type": int, + "tooltip": "Characters of each item's text to show on hover. Lower this if the result file gets too " + "large.", + }, + } + + def reduce(self, vectors, point_count): + """ + Reduce vectors to two dimensions. + + Imports are deliberately local: `ModuleCollector` imports every + processor at startup, and importing UMAP pulls in numba, whose JIT setup + makes that noticeably slower for a module most installs never run. + + :param vectors: 2D numpy array of embeddings, one row per item + :param int point_count: Number of items + :return: 2D numpy array of `(x, y)` coordinates + """ + algorithm = self.parameters.get("algorithm", "umap") + + if algorithm == "pca": + from sklearn.decomposition import PCA + return PCA(n_components=2, random_state=42).fit_transform(vectors) + + if algorithm == "tsne": + from sklearn.manifold import TSNE + # perplexity must stay below the item count or sklearn refuses to run + perplexity = min(self.parameters.get("perplexity", 30), max(2.0, (point_count - 1) / 3)) + self.dataset.log(f"Running t-SNE with perplexity {perplexity}") + return TSNE(n_components=2, perplexity=perplexity, metric="cosine", init="pca", + random_state=42).fit_transform(vectors) + + import umap + # n_neighbors cannot exceed the number of other points available + n_neighbors = min(self.parameters.get("n_neighbors", 15), point_count - 1) + min_dist = self.parameters.get("min_dist", 0.1) + self.dataset.log(f"Running UMAP with n_neighbors={n_neighbors}, min_dist={min_dist}") + + # cosine is the metric embeddings are meant to be compared under, and a + # fixed random_state keeps the map reproducible - 4CAT results should be + # retraceable, which a different layout on every run would undermine + return umap.UMAP(n_components=2, n_neighbors=n_neighbors, min_dist=min_dist, metric="cosine", + random_state=42).fit_transform(vectors) + + def process(self): + """ + Reduce the embeddings and write the interactive map. + """ + limit = self.parameters.get("amount", 2000) + max_processed = min(limit, self.source_dataset.num_rows) if limit else self.source_dataset.num_rows + max_text_length = self.parameters.get("max_text_length", 250) + + vectors = [] + labels = [] + dimensions = None + + self.dataset.update_status("Reading embeddings") + for item in self.source_dataset.iterate_items(self): + if self.interrupted: + raise ProcessorInterruptedException("Interrupted while reading embeddings") + + if len(vectors) >= max_processed: + break + + vector = item.get("embedding") + # map_item flattens the vector to a space-separated string; the raw + # NDJSON keeps it as a list. Accept either. + if isinstance(vector, str): + vector = [float(value) for value in vector.split() if value] + + if not vector: + continue + + if dimensions is None: + dimensions = len(vector) + elif len(vector) != dimensions: + # a ragged matrix cannot be reduced; this means the dataset mixes + # models, which should not happen but is worth saying out loud + self.dataset.finish_with_error("The parent dataset contains embeddings of different sizes, so they " + "cannot be mapped together.") + return + + vectors.append(vector) + text = str(item.get("text", "") or "") + labels.append(text[:max_text_length] + ("…" if len(text) > max_text_length else "")) + + point_count = len(vectors) + if point_count < 3: + self.dataset.finish_with_error(f"Not enough embeddings to build a map (found {point_count}, need at " + f"least 3).") + return + + self.dataset.update_status(f"Reducing {point_count:,} embeddings of {dimensions} dimensions to 2D; this may " + f"take a while") + coordinates = self.reduce(np.asarray(vectors, dtype=np.float32), point_count) + + self.dataset.update_status("Rendering map") + with self.dataset.get_results_path().open("w", encoding="utf-8") as outfile: + outfile.write(self.get_html(coordinates, labels)) + + self.dataset.update_status(f"Mapped {point_count:,} items", is_final=True) + self.dataset.finish(point_count) + + def get_html(self, coordinates, labels) -> str: + """ + Build the self-contained HTML page. + + The result is rendered into a 4CAT page with `{{ html|safe }}` as well as + opened directly when downloaded, so it carries no ``/`` + wrapper, scopes its CSS to one container, and keeps its JavaScript in an + IIFE rather than touching globals. + + It does carry a charset declaration: 4CAT's own page declares UTF-8, but + a downloaded file has nothing to inherit from, and item text is routinely + non-ASCII. Without it the browser guesses, and guesses latin-1. + + :param coordinates: 2D numpy array of `(x, y)` coordinates + :param list labels: Hover text per point, in the same order + :return str: HTML + """ + # normalise to 0-1 so the client only deals with its own viewport + mins = coordinates.min(axis=0) + spans = np.where(coordinates.max(axis=0) - mins == 0, 1, coordinates.max(axis=0) - mins) + normalised = (coordinates - mins) / spans + + points = [[round(float(x), 5), round(float(y), 5)] for x, y in normalised] + + # Item text is untrusted - it comes from whatever platform the dataset + # was collected from - and this page is injected into 4CAT's own DOM + # unescaped. Escaping the angle brackets and ampersand means no item can + # close the script tag or inject markup, whatever it contains. + payload = json.dumps({"points": points, "labels": labels}, ensure_ascii=False) + payload = payload.replace("<", "\\u003c").replace(">", "\\u003e").replace("&", "\\u0026") + + container = f"embedding-map-{self.dataset.key}" + subtitle = f"{len(points):,} items · {self.parameters.get('algorithm', 'umap').upper()}" + + return HTML_TEMPLATE % {"container": container, "payload": payload, "subtitle": subtitle} + + +HTML_TEMPLATE = """ +
+ +
%(subtitle)s · scroll to zoom, drag to pan
+ +
+
+ +""" diff --git a/setup.py b/setup.py index 8e19534f0..b7c094df0 100644 --- a/setup.py +++ b/setup.py @@ -68,6 +68,7 @@ "scenedetect[opencv]>=0.7", # video_scene_identifier "scikit-learn", # image_wall, histwords, tf_idf, topic_modeling, classification_evaluation, confusion_matrix "Telethon~=1.36.0", # search_telegram, download_telegram_videos, download-telegram-images + "umap-learn~=0.5.5", # embedding_map "unidecode~=1.3", # accent_fold "wordcloud~=1.8", # word-cloud # The https://github.com/akamhy/videohash is not being maintained anymore; these are two patches diff --git a/tests/test_llm_models.py b/tests/test_llm_models.py new file mode 100644 index 000000000..bd68beedf --- /dev/null +++ b/tests/test_llm_models.py @@ -0,0 +1,179 @@ +""" +Tests for LLM model task capabilities: + +- `common/lib/llm/models.py` — the task filter that keeps generative and + embedding models out of each other's processors. +- `common/lib/llm/clients/ollama_client.py` — deriving tasks and media types + from Ollama's reported capabilities. + +These import neither `LLMAdapter` nor any server, so they run on host Python +without the LangChain stack installed. +""" +import pytest + +from common.lib.exceptions import LLMServerException +from common.lib.llm.clients.ollama_client import OllamaClient +from common.lib.llm.models import get_model_library, supports_task + + +class DummyConfig: + """Minimal stand-in for the 4CAT config reader.""" + + def __init__(self, settings=None): + self.settings = settings or {} + + def get(self, key, default=None): + return self.settings.get(key, default) + + +def make_meta(capabilities): + """Build an Ollama entry shaped like the one `list_models()` produces.""" + return {"model": "test-model", "metadata": {"capabilities": capabilities}} + + +@pytest.fixture +def client(): + server = {"name": "Test Ollama", "type": "ollama", "url": "http://localhost:11434", + "auth_header": "", "auth_key": "", "_id": "ollama-test"} + return OllamaClient(DummyConfig(), server, log=None) + + +# --------------------------------------------------------------------------- # +# supports_task + +def test_supports_task_reads_declared_tasks(): + assert supports_task({"supported_tasks": ["embed"]}, "embed") + assert not supports_task({"supported_tasks": ["embed"]}, "generate") + assert supports_task({"supported_tasks": ["generate", "embed"]}, "generate") + + +def test_supports_task_defaults_legacy_entries_to_generative(): + """Entries built before `supported_tasks` existed all predate embedding.""" + assert supports_task({}, "generate") + assert not supports_task({}, "embed") + + +# --------------------------------------------------------------------------- # +# OllamaClient.parse_supported_tasks + +def test_parse_tasks_maps_completion_and_embedding(client): + assert client.parse_supported_tasks(make_meta(["completion"])) == ["generate"] + assert client.parse_supported_tasks(make_meta(["embedding"])) == ["embed"] + assert client.parse_supported_tasks(make_meta(["completion", "embedding"])) == ["generate", "embed"] + + +def test_parse_tasks_ignores_orthogonal_capabilities(client): + """ + `qwen3-embedding:latest` reports ["tools", "embedding"] and no "completion". + Treating any non-embedding capability as evidence of generation would offer + it in the prompter, where it can only fail. + """ + assert client.parse_supported_tasks(make_meta(["tools", "embedding"])) == ["embed"] + assert client.parse_supported_tasks(make_meta(["vision", "completion"])) == ["generate"] + + +def test_parse_tasks_falls_back_to_generative_when_unknown(client): + """An unreachable /api/show must not hide a model from every processor.""" + assert client.parse_supported_tasks(None) == ["generate"] + assert client.parse_supported_tasks({"model": "x"}) == ["generate"] + assert client.parse_supported_tasks(make_meta([])) == ["generate"] + + +def test_parse_tasks_reads_inline_capabilities(client): + """/api/tags reports capabilities inline, without a `metadata` key.""" + assert client.parse_supported_tasks({"model": "x", "capabilities": ["embedding"]}) == ["embed"] + + +# --------------------------------------------------------------------------- # +# OllamaClient.parse_supported_media_types + +def test_embedding_model_still_consumes_text(client): + """Embedding is a task, not a media type - an embedder still takes text in.""" + assert client.parse_supported_media_types(make_meta(["embedding"])) == ["text"] + assert "embedding" not in client.parse_supported_media_types(make_meta(["embedding"])) + + +def test_media_types_still_detect_vision(client): + assert client.parse_supported_media_types(make_meta(["completion", "vision"])) == ["text", "image"] + + +def test_media_types_unknown_stays_empty(client): + assert client.parse_supported_media_types(None) == [] + + +# --------------------------------------------------------------------------- # +# OllamaClient.embed + +def test_embed_refuses_media(client): + """ + Ollama's /api/embed has no image parameter and discards one silently, so a + media request must fail loudly here rather than come back as a text-only + vector that looks valid. + """ + with pytest.raises(LLMServerException, match="multimodal"): + client.embed("mxbai-embed-large:latest", ["some text"], media=[""]) + + +def test_embed_refuses_media_before_reaching_the_server(client): + """The guard must not depend on the server being up, or on there being text.""" + with pytest.raises(LLMServerException): + client.embed("mxbai-embed-large:latest", [], media=[""]) + + +def test_embed_short_circuits_on_empty_input(client): + """No inputs means no request; an empty list is not an error.""" + assert client.embed("mxbai-embed-large:latest", []) == [] + assert client.embed("mxbai-embed-large:latest", [], media=None) == [] + + +# --------------------------------------------------------------------------- # +# get_model_library + +@pytest.fixture +def config(): + servers = {"ollama-test": {"name": "Test Ollama", "type": "ollama", "url": "http://localhost:11434", + "_id": "ollama-test"}} + models = { + "gen": {"name": "Generative", "server": "ollama-test", "supported_tasks": ["generate"]}, + "emb": {"name": "Embedder", "server": "ollama-test", "supported_tasks": ["embed"]}, + "both": {"name": "Both", "server": "ollama-test", "supported_tasks": ["generate", "embed"]}, + "legacy": {"name": "Legacy", "server": "ollama-test"}, + "disabled": {"name": "Disabled", "server": "ollama-test", "supported_tasks": ["embed"]}, + } + return DummyConfig({ + "llm.available_models": models, + "llm.enabled_models": ["gen", "emb", "both", "legacy"], + "llm.servers": servers, + "llm.access": True, + }) + + +def test_library_separates_tasks(config): + assert set(get_model_library(config, task="generate")["Test Ollama"]) == {"gen", "both", "legacy"} + assert set(get_model_library(config, task="embed")["Test Ollama"]) == {"emb", "both"} + + +def test_library_defaults_to_generative(config): + assert get_model_library(config) == get_model_library(config, task="generate") + + +def test_library_respects_enabled_models(config): + for task in ("generate", "embed"): + assert "disabled" not in get_model_library(config, task=task).get("Test Ollama", {}) + + +def test_library_without_local_access_drops_local_models(config): + """Without `llm.access` only third-party models may be offered.""" + config.settings["llm.access"] = False + assert get_model_library(config, task="generate") == {} + assert get_model_library(config, task="embed") == {} + + +def test_library_empty_when_no_model_supports_task(): + config = DummyConfig({ + "llm.available_models": {"gen": {"name": "Generative", "server": "s", "supported_tasks": ["generate"]}}, + "llm.enabled_models": ["gen"], + "llm.servers": {"s": {"name": "Server", "_id": "s"}}, + "llm.access": True, + }) + assert get_model_library(config, task="embed") == {} From 3bdd525a12186aae81ab20204a3c64b171fa0274 Mon Sep 17 00:00:00 2001 From: sal-uva Date: Wed, 16 Sep 2026 19:31:12 -0400 Subject: [PATCH 02/24] feat: update llms.json --- common/assets/llms.json | 57 ++++++++++++++++++++++++----------------- 1 file changed, 33 insertions(+), 24 deletions(-) diff --git a/common/assets/llms.json b/common/assets/llms.json index 75fbf4d35..015c7078e 100644 --- a/common/assets/llms.json +++ b/common/assets/llms.json @@ -1,4 +1,25 @@ [ + { + "model": "gpt-5.6-sol", + "name": "[OpenAI] GPT-5.6 Sol", + "model_card": "https://developers.openai.com/api/docs/models/gpt-5.6-sol", + "provider": "openai" + },{ + "model": "gpt-5.6-terra", + "name": "[OpenAI] GPT-5.6 Terra", + "model_card": "https://platform.openai.com/docs/models/gpt-5.6-terra", + "provider": "openai" + },{ + "model": "gpt-5.6-luna", + "name": "[OpenAI] GPT-5.6 Luna", + "model_card": "https://platform.openai.com/docs/models/gpt-5.6-luna", + "provider": "openai" + },{ + "model": "gpt-5.5", + "name": "[OpenAI] GPT-5.5", + "model_card": "https://platform.openai.com/docs/models/gpt-5.5", + "provider": "openai" + }, { "model": "gpt-5.5", "name": "[OpenAI] GPT-5.5", @@ -48,44 +69,32 @@ "provider": "google" }, { - "model": "gemini-3.5-flash", - "name": "[Google] Gemini 3.5 Flash", - "model_card": "https://ai.google.dev/gemini-api/docs/models/gemini-3.5-flash", + "model": "gemini-3.8-flash", + "name": "[Google] Gemini 3.8 Flash", + "model_card": "https://ai.google.dev/gemini-api/docs/models/gemini-3.8-flash", "provider": "google" }, { - "model": "gemini-3-flash-preview", - "name": "[Google] Gemini 3 Flash", - "model_card": "https://cloud.google.com/vertex-ai/generative-ai/docs/models/gemini/3-flash", + "model": "gemini-3.5-flash-lite-preview", + "name": "[Google] Gemini 3.5 Flash Lite", + "model_card": "https://docs.cloud.google.com/vertex-ai/generative-ai/docs/models/gemini/3-5-flash-lite", "provider": "google" }, { - "model": "gemini-3.1-flash-lite-preview", - "name": "[Google] Gemini 3.1 Flash Lite", - "provider": "google", - "model_card": "https://docs.cloud.google.com/vertex-ai/generative-ai/docs/models/gemini/3-1-flash-lite" - }, - { - "model": "claude-fable-5", - "name": "[Anthropic] Claude Fable 5", + "model": "claude-fable-5-1", + "name": "[Anthropic] Claude Fable 5.1", "model_card": "https://www.anthropic.com/claude/fable", "provider": "anthropic" }, { - "model": "claude-opus-4-8", - "name": "[Anthropic] Claude Opus 4.8", - "model_card": "https://www.anthropic.com/claude/opus", - "provider": "anthropic" - }, - { - "model": "claude-opus-4-6", - "name": "[Anthropic] Claude Opus 4.6", + "model": "claude-opus-5", + "name": "[Anthropic] Claude Opus 5", "model_card": "https://www.anthropic.com/claude/opus", "provider": "anthropic" }, { - "model": "claude-sonnet-4-6", - "name": "[Anthropic] Claude Sonnet 4.6 (latest)", + "model": "claude-sonnet-5", + "name": "[Anthropic] Claude Sonnet 5 (latest)", "model_card": "https://www.anthropic.com/claude/sonnet", "provider": "anthropic" }, From 5eeef7b0c9c55c02b46b5cf35904cee3b3193fcf Mon Sep 17 00:00:00 2001 From: sal-uva Date: Wed, 16 Sep 2026 20:49:08 -0400 Subject: [PATCH 03/24] feat: - remove models.py, integrate embedding() into OpenAI and Ollama clients. Needs some unfortunate scaffolding for vLLM though... - abstract embed_media processor with embed_image and embed_video as child processors - improve annotation writing - misc changes --- common/lib/config_definition.py | 4 +- common/lib/llm/clients/ollama_client.py | 18 +- common/lib/llm/clients/openai_client.py | 169 ++++++- common/lib/llm/llm_client.py | 101 +++- common/lib/llm/models.py | 58 --- processors/machine_learning/embed_images.py | 135 +++++ processors/machine_learning/embed_media.py | 465 ++++++++++++++++++ ...erate_text_embeddings.py => embed_text.py} | 74 ++- processors/machine_learning/embed_videos.py | 152 ++++++ .../machine_learning/embedding_similarity.py | 129 +++-- processors/machine_learning/llm_prompter.py | 2 +- processors/machine_learning/prompt_compass.py | 2 +- processors/visualisation/embedding_map.py | 11 +- tests/test_embedding_annotations.py | 350 +++++++++++++ tests/test_embedding_processors.py | 236 +++++++++ tests/test_llm_models.py | 213 +++++++- 16 files changed, 1987 insertions(+), 132 deletions(-) delete mode 100644 common/lib/llm/models.py create mode 100644 processors/machine_learning/embed_images.py create mode 100644 processors/machine_learning/embed_media.py rename processors/machine_learning/{generate_text_embeddings.py => embed_text.py} (77%) create mode 100644 processors/machine_learning/embed_videos.py create mode 100644 tests/test_embedding_annotations.py create mode 100644 tests/test_embedding_processors.py diff --git a/common/lib/config_definition.py b/common/lib/config_definition.py index d913a2bdc..2ccf0ccbf 100644 --- a/common/lib/config_definition.py +++ b/common/lib/config_definition.py @@ -636,13 +636,13 @@ "type": UserInput.OPTION_TEXT, "help": "Authentication Header", "default": "", - "tooltip": "The HTTP header used to authenticate with the server (e.g. 'X-API-KEY', 'Authorization'). Passed with the Authentication Key as value.", + "tooltip": "The HTTP header used to authenticate with the server (e.g. 'X-API-KEY', 'Authorization'). Passed with the Authentication Key as value. For a bearer token you can write either 'Authorization' with 'Bearer ' as the value, or 'Authorization: Bearer' with just the key.", }, "auth_key": { "type": UserInput.OPTION_TEXT, "default": "", "help": "Authentication Key", - "tooltip": "The API key to access the LLM server, if required.", + "tooltip": "The API key to access the LLM server, if required. 'Bearer ' for bearer tokens.", }, } }, diff --git a/common/lib/llm/clients/ollama_client.py b/common/lib/llm/clients/ollama_client.py index 501634177..d6f6b03c5 100644 --- a/common/lib/llm/clients/ollama_client.py +++ b/common/lib/llm/clients/ollama_client.py @@ -134,17 +134,12 @@ def embed(self, model_id: str, inputs: list, media: list | None = None, timeout: The endpoint accepts a list and returns one vector per item, so a caller that batches its inputs gets one round-trip per batch rather than per - item. + item. NOTE: Ollama currently only supports text embeddings. - Text only. `/api/embed` takes `model`, `input`, `truncate`, `options`, - `keep_alive` and `dimensions` - there is no image parameter, and Ollama - *silently discards* an `images` field rather than rejecting it, handing - back a text-only vector that looks entirely valid (verified against - Ollama 0.34.1; see ollama/ollama#5304 and #7677, both still open). So - refuse media here explicitly: a quiet wrong answer is worse than a loud - failure, and this is the only place that can tell the difference. + todo: revise later, since Ollama notes that OpenAI-like embedding endpoints are coming soon, + https://ollama.com/blog/embedding-models - :param str model_id: Ollama model name, e.g. `"mxbai-embed-large:latest"`. + :param str model_id: Ollama model name. :param list inputs: Strings to embed. :param list media: Not supported; passing any raises. :param int timeout: Request timeout in seconds. @@ -155,9 +150,8 @@ def embed(self, model_id: str, inputs: list, media: list | None = None, timeout: """ if media: raise LLMServerException( - "Ollama cannot produce multimodal embeddings: its /api/embed endpoint has no image parameter and " - "ignores one without reporting an error, which would yield text-only vectors that look valid. Use a " - "server that supports multimodal embedding instead.") + "Ollama does not support non-text inputs yet. Use a " + "server and model that supports multimodal embeddings, like vLLM.") if not inputs: return [] diff --git a/common/lib/llm/clients/openai_client.py b/common/lib/llm/clients/openai_client.py index 7cdf92050..8cac3b4ab 100644 --- a/common/lib/llm/clients/openai_client.py +++ b/common/lib/llm/clients/openai_client.py @@ -5,6 +5,9 @@ some useful API endpoints exclusive to it that we can benefit from, so use the dedicated class for tht instead. """ +import requests + +from common.lib.exceptions import LLMServerException from common.lib.llm.llm_client import LLMServerClient @@ -15,6 +18,167 @@ class OpenAICompatibleClient(LLMServerClient): _models_info_key = "models" _model_id_key = "key" + # vLLM and any strict OpenAI-compatible server expose the standard + # /v1/models, returning {"object": "list", "data": [{"id": ...}]}. The + # /api/v1/models shape below is kept as a fallback for servers that only + # speak that dialect, so existing connections keep working. + _openai_models_path = "/v1/models" + + def get_status(self) -> bool | int: + """ + Check whether the server is reachable. + + Probes the same two model-listing dialects `list_models()` does. Without + this a vLLM server reads as down: the base class only knows + `_models_info_path`, which vLLM answers with a 404 even while healthy. + + :return: `200` if either path answers, else the last status code seen, + or `False` if the server could not be reached at all. + """ + last_status = False + for path in (self._openai_models_path, self._models_info_path): + try: + response = self._session.get( + f"{self.base_url}{path}", headers=self._headers, timeout=self.timeout) + if response.status_code == 200: + return 200 + last_status = response.status_code + except requests.RequestException as e: + if self.log: + self.log.warning(f"{self.__class__.__name__}: server not available at {self.base_url}{path}: {e}") + + return last_status + + def list_models(self) -> list[dict]: + """ + List available models, trying the standard OpenAI path first. + + Tries `/v1/models` and falls back to this class's `_models_info_path`. + vLLM often does not prepend 'api/'... + + :return list[dict]: List of models available, or `[]` on failure. + """ + try: + response = self._session.get( + f"{self.base_url}{self._openai_models_path}", + headers=self._headers, + timeout=self.timeout, + ) + if response.status_code == 200: + models = response.json().get("data", []) + if models: + return [{**model, self._model_id_key: model.get("id")} for model in models] + except (requests.RequestException, ValueError) as e: + if self.log: + self.log.debug( + f"{self.__class__.__name__}: {self._openai_models_path} not usable at {self.base_url} ({e}), " + f"falling back to {self._models_info_path}") + + return super().list_models() + + def embed(self, model_id: str, inputs: list, media: list | None = None, timeout: int = 300) -> list[list[float]]: + """ + Embed via the server's OpenAI-compatible `/v1/embeddings` endpoint. + + Two request shapes, because the OpenAI spec only covers the first: + + * text only -> `{"input": [...]}`, the standard batched form, one vector + per input. + * with media -> `{"messages": [...]}`, vLLM's format, where the + content parts carry the media. A conversation is a single input, so + this embeds exactly one item per request; `inputs` and `media` are + paired by index and each pair becomes its own call. + + A server that is not running a pooling/embedding model has no + `/v1/embeddings` route at all and answers 404. + + :param str model_id: Model ID within this server's context. + :param list inputs: Texts to embed. With media, `inputs[i]` is the text + accompanying `media[i]` and may be an empty string. + :param list media: Optional media descriptors, one per input - see + `LLMServerClient.embed`. + :param int timeout: Request timeout in seconds. + :returns list[list[float]]: One vector per input, in input order. + :raises LLMServerException: On transport failure, an error response, or + a vector count that does not match the input count. + """ + if not inputs and not media: + return [] + + if media and len(media) != len(inputs): + raise LLMServerException( + f"Got {len(media)} media items for {len(inputs)} inputs; each input needs exactly one media entry " + f"(use an empty string as the text if there is none)") + + if not media: + return self._post_embeddings({"model": model_id, "input": list(inputs)}, timeout, len(inputs)) + + vectors = [] + for text, item_media in zip(inputs, media): + content = [self._media_content_part(item_media)] + if text: + content.append({"type": "text", "text": text}) + + payload = {"model": model_id, "messages": [{"role": "user", "content": content}]} + vectors += self._post_embeddings(payload, timeout, 1) + + return vectors + + @staticmethod + def _media_content_part(item_media: dict) -> dict: + """ + Build one OpenAI-style content part for a media descriptor. + + vLLM keys the part by modality - `video_url` for video, `image_url` for + images - and accepts a `data:` URI in place of an http URL. + + :param dict item_media: Media descriptor (`type`, `mime`, `data`). + :return dict: Content part for the `messages` payload. + """ + media_type = item_media.get("type", "image") + mime = item_media.get("mime") or f"{media_type}/*" + url = item_media.get("url") or f"data:{mime};base64,{item_media.get('data', '')}" + + key = "video_url" if media_type == "video" else "image_url" + return {"type": key, key: {"url": url}} + + def _post_embeddings(self, payload: dict, timeout: int, expected: int) -> list[list[float]]: + """ + POST one embeddings request and pull the vectors out of the response. + + :param dict payload: Request body. + :param int timeout: Request timeout in seconds. + :param int expected: How many vectors the response must contain. + :return list[list[float]]: Vectors, ordered by the response `index` + field - the API does not promise to return them in input order. + """ + url = f"{self.base_url}/v1/embeddings" + try: + response = self._session.post(url, headers=self._headers, json=payload, timeout=timeout) + except requests.RequestException as e: + raise LLMServerException(f"Could not reach {url}: {e}") + + if response.status_code == 404: + raise LLMServerException( + f"{url} does not exist on this server.") + + if response.status_code != 200: + raise LLMServerException( + f"Server returned status {response.status_code} while embedding with model " + f"'{payload.get('model')}': {response.text[:500]}") + + try: + data = response.json().get("data", []) + except ValueError as e: + raise LLMServerException(f"Server returned invalid JSON while embedding: {e}") + + if len(data) != expected: + raise LLMServerException( + f"Server returned {len(data)} embeddings for {expected} input(s); cannot map vectors back to items") + + ordered = sorted(data, key=lambda row: row.get("index", 0)) + return [row["embedding"] for row in ordered] + def parse_supported_media_types(self, meta: dict) -> list[str]: """ Derive the media types a model supports from its LiteLLM metadata. @@ -54,6 +218,9 @@ def format_display_name(self, meta: dict) -> str: if meta.get("params_string"): extra_bits.append(meta["params_string"]) - model_name += f" ({', '.join(extra_bits)})" + # only append the parenthetical when there is something to put in it - + # a plain /v1/models entry has neither publisher nor size + if extra_bits: + model_name += f" ({', '.join(extra_bits)})" return model_name diff --git a/common/lib/llm/llm_client.py b/common/lib/llm/llm_client.py index 1508e6e0a..53d0c9ac3 100644 --- a/common/lib/llm/llm_client.py +++ b/common/lib/llm/llm_client.py @@ -12,6 +12,8 @@ import requests import re +from common.lib.exceptions import LLMServerException + class LLMServerClient: _headers = {} server_config = {} @@ -59,14 +61,29 @@ def __init__(self, config, server_config: dict, timeout: int = 10, log=None) -> self.base_url = server_config["url"].rstrip("/") if self.base_url.endswith("v1"): - # get rid of the 'v1' - we'll add this in the path - self.base_url = f"{self.base_url[:-2]}" + # get rid of the 'v1' - we'll add this in the path. + self.base_url = self.base_url[:-2].rstrip("/") self._session = requests.Session() self._headers = {"Content-Type": "application/json"} if self.auth_type: - self._headers[self.auth_type] = self.auth_key + # The setting asks for a header *name*, but "Authorization: Bearer" + # is the natural thing to type and is what people paste from curl + # examples. As a header name that is malformed - requests raises + # InvalidHeader, which surfaces as "server unavailable" and sends + # people hunting for a network problem. So accept both spellings: + # anything after a colon is treated as a prefix on the value. + header_name, _, value_prefix = self.auth_type.partition(":") + header_name = header_name.strip() + value_prefix = value_prefix.strip() + + auth_value = self.auth_key or "" + if value_prefix and not auth_value.lower().startswith(value_prefix.lower()): + auth_value = f"{value_prefix} {auth_value}".strip() + + if header_name: + self._headers[header_name] = auth_value self.log = log @@ -137,20 +154,32 @@ def parse_supported_tasks(self, meta: dict) -> list[str]: """ Derive the tasks a model can be used for from its metadata. - The default is `["generate"]`: every server 4CAT talks to serves chat - completions, while embedding endpoints are the exception. + Most servers do not report this. An OpenAI-compatible `/v1/models` lists + an id and nothing else, so there is no capability field to read. The + fallback is therefore the model's own name: an id containing "embed" + (Qwen3-VL-Embedding, mxbai-embed-large, nomic-embed-text) reads as an + embedder, anything else as generative. + + # todo: check if there's a more rigid way of separating later. :param dict meta: Model metadata, or `None` if unavailable. :returns list[str]: Supported tasks - `"generate"`, `"embed"`, or both. """ - return ["generate"] + try: + model_id = str(self.get_model_id(meta) or "") + except (KeyError, TypeError): + # nothing to go on; assume generative, which is how every server + # behaved before embedding support existed + return ["generate"] + + return ["embed"] if "embed" in model_id.lower() else ["generate"] def embed(self, model_id: str, inputs: list, media: list | None = None, timeout: int = 300) -> list[list[float]]: """ Embed one or more inputs, returning one vector per input. Deliberately *not* routed through `LLMAdapter`/LangChain. LangChain's - embedding interface is text-only (`embed_documents(list[str])`), so it + embedding interface is text-only, so it cannot handle multimodal embeddings. A client that cannot embed media **must raise** when `media` is given @@ -162,12 +191,21 @@ def embed(self, model_id: str, inputs: list, media: list | None = None, timeout: :param str model_id: Model ID *within this server's context* (i.e. a `local_id`, not a global model ID). :param list inputs: Inputs to embed. - :param list media: Media to embed alongside `inputs`, for servers that - support multimodal embedding. The payload shape is server-specific. + :param list media: Optional media to embed, one entry per input and + paired by index, for servers that support multimodal embedding. Each + entry is a descriptor dict:: + + {"type": "video"|"image", "mime": "video/mp4", "data": ""} + + or, in place of `data`, a `"url"` the server can fetch itself. Each + client translates this to whatever its own API expects. :param int timeout: Request timeout in seconds. :returns list[list[float]]: One vector per input, in input order. """ - raise NotImplementedError(f"{self.__class__.__name__} does not support embedding") + raise LLMServerException( + f"Embedding is not supported for {self.server_config.get('type', 'this')} connections, so the model " + f"'{model_id}' cannot be used to generate embeddings. Use a model on an Ollama or OpenAI-compatible " + f"connection instead.") def get_model_card_url(self, meta: dict) -> str: """ @@ -233,4 +271,45 @@ def get_global_model_id(self, meta: dict) -> str: """ domain = re.sub(r"^https?://", "", self.server_config["url"]) domain = domain.rstrip("/") - return f"{self.server_config['type']}://{domain}/{self.get_model_id(meta)}" \ No newline at end of file + return f"{self.server_config['type']}://{domain}/{self.get_model_id(meta)}" + + +def supports_task(model: dict, task: str = "generate") -> bool: + """ + Check whether a model entry can be used for a given task. + + :param dict model: A single `llm.available_models` entry. + :param str task: Task to check for - `"generate"` or `"embed"`. + :return bool: Whether the model supports the task. + """ + return task in model.get("supported_tasks", ["generate"]) + + +def get_model_library(config, task: str = "generate") -> dict: + """ + Get the LLM models available for a given task, grouped by server. + + :param config: 4CAT config reader (context-aware, so per-user `llm.access` + is respected) + :param str task: Task the model must support - `"generate"` or `"embed"`. + :return dict: `{server name: {model ID: model display name}}`, shaped for + a `UserInput.OPTION_CHOICE` option. + """ + available_models = config.get("llm.available_models", {}) + enabled_model_ids = config.get("llm.enabled_models", []) + servers = config.get("llm.servers", {}) + if not config.get("llm.access"): + enabled_model_ids = [_ for _ in enabled_model_ids if _.startswith("thirdparty")] + + models_option = {} + for key, value in {k: v for k, v in available_models.items() if k in enabled_model_ids}.items(): + if not supports_task(value, task): + continue + + server = servers[value["server"]] + if server["name"] not in models_option: + models_option[server["name"]] = {} + + models_option[server["name"]][key] = value["name"] + + return models_option diff --git a/common/lib/llm/models.py b/common/lib/llm/models.py deleted file mode 100644 index dd8692a66..000000000 --- a/common/lib/llm/models.py +++ /dev/null @@ -1,58 +0,0 @@ -""" -Present the configured LLM model inventory to processors. - -Kept separate from `common.lib.llm.adapter` on purpose: that module pulls in the -whole LangChain stack, while this one only reads 4CAT settings. A processor that -merely needs to list models or a test that needs to check the filtering can -import this without the heavy dependencies. -""" - - -def supports_task(model: dict, task: str = "generate") -> bool: - """ - Check whether a model entry can be used for a given task. - - Entries built before `supported_tasks` existed carry no task info. They all - predate embedding support, so they read as generative rather than vanishing - from every processor until the inventory is next refreshed. - - :param dict model: A single `llm.available_models` entry. - :param str task: Task to check for - `"generate"` or `"embed"`. - :return bool: Whether the model supports the task. - """ - return task in model.get("supported_tasks", ["generate"]) - - -def get_model_library(config, task: str = "generate") -> dict: - """ - Get the LLM models available for a given task, grouped by server. - - Only models that are both enabled and capable of `task` are returned, so a - prompting processor never offers an embedding model and an embedding - processor never offers a generative one. Without this filter either choice - produces a dataset that only fails once the request reaches the server. - - :param config: 4CAT config reader (context-aware, so per-user `llm.access` - is respected) - :param str task: Task the model must support - `"generate"` or `"embed"`. - :return dict: `{server name: {model ID: model display name}}`, shaped for - a `UserInput.OPTION_CHOICE` option. - """ - available_models = config.get("llm.available_models", {}) - enabled_model_ids = config.get("llm.enabled_models", []) - servers = config.get("llm.servers", {}) - if not config.get("llm.access"): - enabled_model_ids = [_ for _ in enabled_model_ids if _.startswith("thirdparty")] - - models_option = {} - for key, value in {k: v for k, v in available_models.items() if k in enabled_model_ids}.items(): - if not supports_task(value, task): - continue - - server = servers[value["server"]] - if server["name"] not in models_option: - models_option[server["name"]] = {} - - models_option[server["name"]][key] = value["name"] - - return models_option diff --git a/processors/machine_learning/embed_images.py b/processors/machine_learning/embed_images.py new file mode 100644 index 000000000..b85d6b246 --- /dev/null +++ b/processors/machine_learning/embed_images.py @@ -0,0 +1,135 @@ +""" +Generate an embedding per image via a multimodal embedding model. +""" +from PIL import Image, ImageOps, UnidentifiedImageError + +from common.lib.compatibility import Compatibility +from common.lib.user_input import UserInput +from processors.machine_learning.embed_media import EmbedMedia + +__author__ = "Sal Hagen" +__credits__ = ["Sal Hagen"] +__maintainer__ = "Sal Hagen" +__email__ = "4cat@oilab.eu" + + +class EmbedImages(EmbedMedia): + """ + Send each image to a multimodal embedding model and store the vector it + returns, so images can be compared, mapped and searched by content. + """ + type = "image-embeddings" # job type ID + category = "Visual" # category + title = "Generate image embeddings" # title displayed in UI + description = ("Generate a numerical representation (an 'embedding') of each image using a multimodal embedding " + "model. This allows images to be compared and clustered. Images are compressed before sending. " + "Requires an embedding model that accepts images, such as Qwen3-VL-Embedding.") + extension = "ndjson" # extension of result file, used internally and in UI + + media_label = "image" + media_label_plural = "images" + model_requirement_note = "not only text" + working_filename = "compressed.jpg" + annotation_label = "image embedding" + + compatibility = Compatibility( + media_types={"image"}, + type_prefixes={"image-downloader"}, + preferred_followups=["embedding-map", "embedding-similarity"], + ) + + @classmethod + def get_media_options(cls, config=None) -> dict: + """ + Options controlling how an image is made smaller before it is sent. + + :param ConfigManager|None config: Configuration reader (context-aware) + :return dict: Options for this processor + """ + return { + "compression_info": { + "type": UserInput.OPTION_INFO, + "help": "Compression speeds up requests. Reducing the size helps the most, and most models scale " + "images down to a few hundred pixels anyway. Very large images can also exceed the model's " + "context window and be rejected.", + }, + "max_size": { + "type": UserInput.OPTION_TEXT, + "help": "Max width/height in pixels", + "default": 448, + "min": 64, + "max": 1920, + "coerce_type": int, + "tooltip": "Images larger than this are scaled down, keeping their aspect ratio.", + }, + "quality": { + "type": UserInput.OPTION_TEXT, + "help": "Compression level", + "default": 85, + "min": 1, + "max": 95, + "coerce_type": int, + "tooltip": "Min: 1, max: 95. JPEG quality: lower means smaller files and lower quality. 85 is a " + "reasonable trade-off, below 50 makes the image grainy.", + }, + } + + def skip_reason(self, media_path) -> str | None: + """ + Leave out files Pillow cannot open as a bitmap. + + SVGs are vector files and are common in image archives, so they are + named explicitly rather than reported as a generic failure. + + :param Path media_path: File that would be embedded + :return str|None: Why the file is skipped, or `None` to embed it. + """ + if media_path.suffix.lower() == ".svg": + return "SVG files cannot be embedded" + + return None + + def prepare_media(self, media_path, output_path) -> tuple: + """ + Re-encode an image smaller and return it as a base64 media descriptor. + + Pixel size matters most: models scale images to a fixed resolution + anyway, so anything larger is bytes spent on nothing. Everything is + written out as JPEG - the model reads a bitmap, so a PNG's lossless + compression only costs transfer size. + + :param Path media_path: Image to prepare + :param Path output_path: Where to write the re-encoded image + :return tuple: `(media descriptor, original size, sent size)` + """ + output_path.unlink(missing_ok=True) + max_size = self.parameters.get("max_size", 448) + + try: + with Image.open(media_path) as image: + # lets the JPEG decoder skip straight to roughly the size we + # want instead of decoding at full resolution first + image.draft("RGB", (max_size, max_size)) + + # photos carry their rotation in EXIF rather than in the pixels; + # without this a sideways photo is embedded sideways + image = ImageOps.exif_transpose(image) + + # JPEG has no alpha channel, and paletted or 16-bit images need + # converting before they can be saved as one + if image.mode != "RGB": + image = image.convert("RGB") + + image.thumbnail((max_size, max_size), Image.LANCZOS) + image.save(output_path, "JPEG", quality=self.parameters.get("quality", 85), optimize=True) + except (UnidentifiedImageError, OSError, ValueError) as e: + raise RuntimeError(f"could not read image: {e}") + + original_size = media_path.stat().st_size + sent_size = output_path.stat().st_size + self.dataset.log(f"{media_path.name}: {original_size / 1024:.0f} KB -> {sent_size / 1024:.0f} KB") + + payload = {"type": "image", "mime": "image/jpeg", "data": self.encode(output_path)} + output_path.unlink(missing_ok=True) + + return payload, original_size, sent_size diff --git a/processors/machine_learning/embed_media.py b/processors/machine_learning/embed_media.py new file mode 100644 index 000000000..0c5a5071d --- /dev/null +++ b/processors/machine_learning/embed_media.py @@ -0,0 +1,465 @@ +""" +Shared behaviour for processors that embed media files via a multimodal model. + +`EmbedVideos` and `EmbedImages` differ only in how a file is made smaller before +it is sent; everything around that - resolving the model, walking the archive, +batching, writing results - is identical, and lives here. + +This class is abstract (`prepare_media` has no implementation), so +`ModuleCollector` skips it: `is_4cat_class()` ignores anything +`inspect.isabstract()` reports, and also ignores classes whose `__module__` is +not the file being scanned, which covers the subclass imports. +""" +import abc +import base64 +import json +import time +import zipfile + +from datetime import datetime + +from backend.lib.processor import BasicProcessor +from common.lib.exceptions import LLMServerException, ProcessorInterruptedException, QueryParametersException +from common.lib.item_mapping import MappedItem +from common.lib.llm.llm_client import LLMServerClient, get_model_library +from common.lib.user_input import UserInput + +__author__ = "Sal Hagen" +__credits__ = ["Sal Hagen"] +__maintainer__ = "Sal Hagen" +__email__ = "4cat@oilab.eu" + + +def read_post_id_map(archive_path, log=None) -> dict: + """ + Map each media filename in an archive to the posts it was downloaded for. + + A media archive's `.metadata.json` is keyed by source URL, and each entry + carries the `post_ids` it came from plus the `files` written for it. One + media file can belong to several posts, and its own filename is a hash + rather than a post ID, so anything annotating posts has to go through this. + + Module-level rather than a method, because processors that merely *read* a + media embedding - and so never subclass `EmbedMedia` - need it too. + + :param Path archive_path: Archive to read the metadata from + :param callable log: Optional logger for explaining a failed read + :return dict: `{filename without extension: [post ID, ...]}` + """ + post_ids = {} + if not archive_path or not zipfile.is_zipfile(archive_path): + return post_ids + + try: + with zipfile.ZipFile(archive_path) as archive: + if ".metadata.json" not in archive.namelist(): + return post_ids + + metadata = json.loads(archive.read(".metadata.json")) + except (zipfile.BadZipFile, json.JSONDecodeError, KeyError, OSError) as e: + if log: + log(f"Could not read .metadata.json, post IDs are unavailable: {e}") + return post_ids + + for entry in metadata.values(): + if not isinstance(entry, dict): + continue + for file in entry.get("files", []): + filename = file.get("filename") if isinstance(file, dict) else None + if filename: + post_ids[".".join(filename.split(".")[:-1])] = entry.get("post_ids", []) + + return post_ids + + +class EmbedMedia(BasicProcessor): + """ + Send each media file to a multimodal embedding model and store the vector it + returns, so the files can be compared, mapped and searched by content. + """ + + media_label = "media file" + media_label_plural = "media files" + model_requirement_note = "not only text" + working_filename = "compressed" + + annotation_label = "media_embedding" + annotation_batch_size = 100 + + @classmethod + def get_queue_id(cls, remote_id, details, dataset) -> str: + """ + Shared queue for locally hosted models + + :param str remote_id: Job item ID + :param dict details: Job details + :param DataSet dataset: Dataset to run job for + :return str: Queue ID + """ + # Unique queue for locally hosted models; used by other local model processors as well + return "local_models" + + @classmethod + def is_compatible_with(cls, module=None, config=None): + """ + Media datasets, but only where an embedding model is available. + + :param module: Dataset or processor to check against + :param ConfigManager|None config: Context-aware configuration reader + :return bool: + """ + if config is None or not cls.compatibility.is_compatible_with(module, config=config): + return False + + return bool(get_model_library(config, task="embed")) + + @classmethod + def get_media_options(cls, config=None) -> dict: + """ + Options controlling how a file is made smaller before it is sent. + + Subclasses override this; the surrounding model and amount options are + the same everywhere and are added by `get_options()`. + + :param ConfigManager|None config: Configuration reader (context-aware) + :return dict: Options for this processor + """ + return {} + + @classmethod + def get_options(cls, parent_dataset=None, config=None) -> dict: + """ + Get processor options + + :param parent_dataset DataSet: An object representing the dataset that + the processor would be or was run on. + :param config ConfigManager|None config: Configuration reader (context-aware) + :return dict: Options for this processor + """ + options = { + "model_info": { + "type": UserInput.OPTION_INFO, + "help": f"The model must accept **{cls.media_label}**, {cls.model_requirement_note}. 4CAT cannot tell " + f"which models do, so every embedding model is listed here; one that cannot read " + f"{cls.media_label} will return an error.", + }, + "model": { + "type": UserInput.OPTION_CHOICE, + "help": "Embedding model", + "options": get_model_library(config, task="embed") if config else {}, + "default": "", + }, + "instruction": { + "type": UserInput.OPTION_TEXT, + "help": "[optional] Prompt", + "default": f"Represent the {cls.media_label}.", + "tooltip": f"A small instruction sent alongside each {cls.media_label}. Models like " + f"Qwen3-VL-Embedding use this to steer what the embedding captures; tailoring it to your " + f"research question can improve results.", + }, + } + + options.update(cls.get_media_options(config)) + + options["amount"] = { + "type": UserInput.OPTION_TEXT, + "help": f"No. of {cls.media_label_plural}", + "default": 100, + "min": 0, + "coerce_type": int, + "tooltip": f"Use '0' for all {cls.media_label_plural}. Each {cls.media_label} is a separate request, so " + f"this is slow for large datasets.", + } + + options["save_annotations"] = { + "type": UserInput.OPTION_ANNOTATION, + "label": cls.annotation_label, + "tooltip": "Add embeddings scores as annotations to the parent dataset.", + "default": False, + "hide_in_explorer": True + } + + return options + + @classmethod + def validate_query(cls, query, request, config): + """ + Validate input + + :param dict query: Query parameters, as submitted by the user via the web interface + :param request: Flask request + :param ConfigManager config: Configuration reader + :return dict: Validated query parameters + """ + allowed_models = { + model_id + for server_models in get_model_library(config, task="embed").values() + for model_id in server_models + } + + if query.get("model") not in allowed_models: + raise QueryParametersException(f"The '{query.get('model')}' embedding model is not currently available.") + + return query + + def check_requirements(self) -> str | None: + """ + Check anything this processor needs beyond a model, before starting. + + Runs once, before any file is read, so a missing dependency is reported + immediately rather than after the first item fails. Subclasses that need + an external tool override this and can store what they resolved on + `self`. + + :return str|None: An error message, or `None` when everything is present. + """ + return None + + def skip_reason(self, media_path) -> str | None: + """ + Decide whether to leave a file out, before any work is done on it. + + :param Path media_path: File that would be embedded + :return str|None: Why the file is skipped, or `None` to embed it. + """ + return None + + @abc.abstractmethod + def prepare_media(self, media_path, output_path) -> tuple: + """ + Make one file smaller and return it as a base64 media descriptor. + + The file travels inside the request body, so the point is to send as few + bytes as possible without losing what the model looks at. + + :param Path media_path: File to prepare + :param Path output_path: Where to write the smaller version + :return tuple: `(media descriptor, original size, sent size)` + """ + pass + + def resolve_model(self): + """ + Resolve the chosen model to a server and a client for it. + + :return tuple|None: `(model metadata, client)`, or `None` when + something is missing - in which case the dataset has been finished + with an error explaining what. + """ + model_id = self.parameters.get("model") + available_models = { + k: v for k, v in self.config.get("llm.available_models", {}).items() + if k in self.config.get("llm.enabled_models", []) + } + + if model_id not in available_models: + self.dataset.finish_with_error(f"Embedding model '{model_id}' is no longer available. Ask an admin to " + f"enable it, or pick another model.") + return None + + model = available_models[model_id] + server = self.config.get("llm.servers", {}).get(model["server"]) + if not server: + self.dataset.finish_with_error(f"The LLM server for model '{model_id}' is no longer configured.") + return None + + try: + client = LLMServerClient.get_client(self.config, server, self.log) + except ValueError as e: + self.dataset.finish_with_error(str(e)) + return None + + return model, client + + @staticmethod + def encode(path) -> str: + """ + Read a file and base64-encode it for transport in a request body. + + :param Path path: File to encode + :return str: Base64-encoded contents + """ + with path.open("rb") as infile: + return base64.b64encode(infile.read()).decode("utf-8") + + def load_post_id_map(self) -> dict: + """ + Map each media filename to the posts it was downloaded for. + + Read straight from the source archive rather than the staging area: the + staging area is filled lazily as items are iterated, so the metadata is + not reliably there before the first file is handled - and annotations + are written as we go. + + :return dict: `{filename without extension: [post ID, ...]}` + """ + post_ids = read_post_id_map(self.source_file, log=self.dataset.log) + self.dataset.log(f"Found post IDs for {len(post_ids):,} {self.media_label_plural} in .metadata.json") + return post_ids + + def process(self): + """ + Compress each file, embed it, and write one vector per file. + """ + resolved = self.resolve_model() + if not resolved: + return + + model, client = resolved + + requirement_error = self.check_requirements() + if requirement_error: + self.dataset.finish_with_error(requirement_error) + return + + instruction = self.parameters.get("instruction", "").strip() + limit = self.parameters.get("amount", 100) + max_processed = min(limit, self.source_dataset.num_rows) if limit else self.source_dataset.num_rows + + staging_area = self.dataset.get_staging_area() + output_path = staging_area.joinpath(self.working_filename) + + save_annotations = self.parameters.get("save_annotations", False) + post_id_map = self.load_post_id_map() if save_annotations else {} + annotations = [] + + embedded = 0 + skipped = 0 + processed = 0 + total_sent = 0 + + self.dataset.update_status(f"Embedding {self.media_label_plural} with {model['local_id']}") + + with self.dataset.get_results_path().open("w", encoding="utf-8") as outfile: + for item in self.source_dataset.iterate_items(self, staging_area=staging_area, + immediately_delete=False): + if self.interrupted: + raise ProcessorInterruptedException(f"Interrupted while embedding {self.media_label_plural}") + + if processed >= max_processed: + break + + media_path = item.file if hasattr(item, "file") else None + if not media_path or media_path.name == ".metadata.json": + # the archive carries its metadata alongside the media files + continue + + reason = self.skip_reason(media_path) + if reason: + self.dataset.log(f"Skipping {media_path.name}: {reason}") + skipped += 1 + continue + + processed += 1 + self.dataset.update_status(f"Embedding {self.media_label} {processed:,}/{max_processed:,} " + f"({media_path.name})") + + try: + payload, original_size, sent_size = self.prepare_media(media_path, output_path) + except ProcessorInterruptedException: + raise + except Exception as e: + self.dataset.log(f"Skipping {media_path.name}: could not compress ({e})") + skipped += 1 + continue + + if not payload: + self.dataset.log(f"Skipping {media_path.name}: compression produced no output") + skipped += 1 + continue + + try: + vector = client.embed(model["local_id"], [instruction], media=[payload])[0] + except LLMServerException as e: + if embedded: + self.dataset.finish_with_warning( + embedded, f"Not all {self.media_label_plural} were embedded: {e}") + else: + self.dataset.finish_with_error(str(e)) + return + + # the filename is a hash, so carry the posts it came from into + # the result: processors downstream cannot reach .metadata.json + # themselves and would otherwise annotate the hash + post_ids = post_id_map.get(media_path.stem, []) + + time_created = int(time.time()) + outfile.write(json.dumps({ + "id": media_path.stem, + "text": media_path.name, + "filename": media_path.name, + "post_ids": post_ids, + "embedding": vector, + "dimensions": len(vector), + "model": model["local_id"], + "original_bytes": original_size, + "sent_bytes": sent_size, + "time_created": datetime.fromtimestamp(time_created).strftime("%Y-%m-%d %H:%M:%S"), + "time_created_utc": time_created, + }) + "\n") + + if save_annotations: + # one vector can belong to several posts, and each gets its + # own annotation + for post_id in post_ids: + annotations.append({ + "item_id": post_id, + "label": self.annotation_label, + "value": " ".join([str(value) for value in vector]), + "type": "text", + }) + + embedded += 1 + total_sent += sent_size + + if embedded % self.annotation_batch_size == 0: + outfile.flush() + if annotations: + self.save_annotations(annotations, hide_in_explorer=True) + annotations = [] + + self.dataset.update_progress(processed / max_processed) + + if annotations: + self.save_annotations(annotations, hide_in_explorer=True) + + if not embedded: + self.dataset.finish_with_error(f"No {self.media_label_plural} could be embedded.") + return + + if save_annotations and not post_id_map: + self.dataset.log("No annotations were written: the archive has no .metadata.json linking files to posts.") + + status = f"Embedded {embedded:,} {self.media_label_plural} ({total_sent / 1024 / 1024:.1f} MB sent in total)" + if skipped: + status += f", skipped {skipped:,}" + self.dataset.update_status(status, is_final=True) + self.dataset.finish(embedded) + + @staticmethod + def map_item(item): + """ + Map a media embedding to a flat row. + + Deliberately the same shape `text-embeddings` produces, so that the + processors chaining off embeddings work on these unchanged. + + :param item: Item to map + :return MappedItem: Mapped item + """ + embedding = item.get("embedding", []) + + return MappedItem({ + "id": item.get("id"), + "text": item.get("text"), + "filename": item.get("filename"), + # without this the field is dropped by the mapper, and a processor + # reading these embeddings falls back to the filename hash + "post_ids": ", ".join([str(post_id) for post_id in item.get("post_ids", [])]), + "model": item.get("model"), + "dimensions": item.get("dimensions"), + "embedding": " ".join([str(value) for value in embedding]), + "original_bytes": item.get("original_bytes"), + "sent_bytes": item.get("sent_bytes"), + "timestamp": item.get("time_created"), + }) diff --git a/processors/machine_learning/generate_text_embeddings.py b/processors/machine_learning/embed_text.py similarity index 77% rename from processors/machine_learning/generate_text_embeddings.py rename to processors/machine_learning/embed_text.py index 4b8f20307..1ded53775 100644 --- a/processors/machine_learning/generate_text_embeddings.py +++ b/processors/machine_learning/embed_text.py @@ -10,8 +10,7 @@ from common.lib.compatibility import Compatibility from common.lib.exceptions import LLMServerException, ProcessorInterruptedException, QueryParametersException from common.lib.item_mapping import MappedItem -from common.lib.llm.llm_client import LLMServerClient -from common.lib.llm.models import get_model_library +from common.lib.llm.llm_client import LLMServerClient, get_model_library from common.lib.user_input import UserInput __author__ = "Sal Hagen" @@ -27,14 +26,21 @@ class GenerateTextEmbeddings(BasicProcessor): """ type = "text-embeddings" # job type ID category = "Text analysis" # category - title = "Generate text embeddings via LLM" # title displayed in UI + title = "Generate text embeddings" # title displayed in UI description = ("Generate a numerical representation (an 'embedding') of the text of each item using an LLM embedding " - "model. Embeddings can be used to e.g. retrieve similar kinds of texts. Requires an embedding model " - "be installed on this server.") + "model. Embeddings can be used to e.g. map, cluster, and retrieve similar kinds of texts. Requires " + "an embedding model to be installed on this server.") extension = "ndjson" # extension of result file, used internally and in UI compatibility = Compatibility(extensions={"csv", "ndjson"}) + #: Hard-coded label the embeddings are written back under, when saved as + #: annotations. Not user-editable: a vector is only meaningful next to the + #: model that produced it, so a free-form label invites mismatched columns. + annotation_label = "text embedding" + #: How many annotations to buffer before writing them to the database + annotation_batch_size = 100 + references = [ "[Ollama embedding models](https://ollama.com/blog/embedding-models)", ] @@ -90,7 +96,8 @@ def get_options(cls, parent_dataset=None, config=None) -> dict: "type": UserInput.OPTION_TEXT, "help": "Column(s) to embed", "default": "body", - "tooltip": "Values of multiple columns are joined with a space before being embedded.", + "inline": True, + "tooltip": "Values of multiple columns are joined with a newline before being embedded.", }, "amount": { "type": UserInput.OPTION_TEXT, @@ -111,6 +118,16 @@ def get_options(cls, parent_dataset=None, config=None) -> dict: "tooltip": "How many items to send to the server at once. Higher is faster but uses more memory on the " "server; lower this if requests time out.", }, + "save_annotations": { + "type": UserInput.OPTION_ANNOTATION, + # must match the label the annotations are written under, below: + # the form renders this as "Add