From 054fb5dd2045a7b1c2b6cdd43bff491a38855c08 Mon Sep 17 00:00:00 2001 From: Thomas Wood Date: Mon, 12 Oct 2026 14:21:05 +0100 Subject: [PATCH] Clean up obsolete catalogue instrument functions --- src/harmony/matching/matcher.py | 391 -------------------- src/harmony/schemas/catalogue_instrument.py | 9 - src/harmony/schemas/catalogue_question.py | 10 - src/harmony/schemas/requests/text.py | 13 - src/harmony/schemas/responses/text.py | 6 - 5 files changed, 429 deletions(-) delete mode 100644 src/harmony/schemas/catalogue_instrument.py delete mode 100644 src/harmony/schemas/catalogue_question.py diff --git a/src/harmony/matching/matcher.py b/src/harmony/matching/matcher.py index ac39444a..956dfc2f 100644 --- a/src/harmony/matching/matcher.py +++ b/src/harmony/matching/matcher.py @@ -40,8 +40,6 @@ from harmony.matching.hdbscan_clustering import cluster_questions_hdbscan_from_embeddings from harmony.matching.instrument_to_instrument_similarity import get_instrument_similarity from harmony.matching.negator import negate -from harmony.schemas.catalogue_instrument import CatalogueInstrument -from harmony.schemas.catalogue_question import CatalogueQuestion from harmony.schemas.requests.text import ( Instrument, Question, @@ -223,396 +221,7 @@ def create_full_text_vectors( return text_vectors, new_vectors_dict -def match_instruments_with_catalogue_instruments( - instruments: List[Instrument], - catalogue_data: dict, - vectorisation_function: Callable, - texts_cached_vectors: dict[str, List[float]], - is_negate: bool = True -) -> tuple[List[Instrument], List[CatalogueInstrument]]: - """ - Match instruments with catalogue instruments. - - :param instruments: The instruments. - :param catalogue_data: The catalogue data. - :param vectorisation_function: A function to vectorize a text. - :param texts_cached_vectors: A dictionary of already cached vectors from texts (key is the text and value is the vector). - :return: Index 0 in the tuple contains the list of instruments that now each contain the best instrument matches from the catalog. - Index 1 in the tuple contains a list of closest instrument matches from the catalog for all the instruments. - """ - - # Gather all questions - all_questions: List[str] = [] - for instrument in instruments: - all_questions.extend([q.question_text for q in instrument.questions]) - all_questions = list(set(all_questions)) - - # Create text vectors for all questions in all the uploaded instruments - all_instruments_text_vectors, _ = create_full_text_vectors( - all_questions=all_questions, - query=None, - vectorisation_function=vectorisation_function, - texts_cached_vectors=texts_cached_vectors, - is_negate=is_negate, - question_languages=get_question_languages(instruments) - ) - - # For each instrument, find the best instrument matches for it in the catalogue - for instrument in instruments: - instrument.closest_catalogue_instrument_matches = ( - match_questions_with_catalogue_instruments( - questions=instrument.questions, - catalogue_data=catalogue_data, - all_instruments_text_vectors=all_instruments_text_vectors, - questions_are_from_one_instrument=True, - ) - ) - - # Gather all questions from all instruments and find the best instrument matches in the catalogue - all_instrument_questions: List[Question] = [] - for instrument in instruments: - all_instrument_questions.extend(instrument.questions) - closest_catalogue_instrument_matches = match_questions_with_catalogue_instruments( - questions=all_instrument_questions, - catalogue_data=catalogue_data, - all_instruments_text_vectors=all_instruments_text_vectors, - questions_are_from_one_instrument=False, - ) - - return instruments, closest_catalogue_instrument_matches - - -def match_questions_with_catalogue_instruments( - questions: List[Question], - catalogue_data: dict, - all_instruments_text_vectors: List[TextVector], - questions_are_from_one_instrument: bool, -) -> List[CatalogueInstrument]: - """ - Match questions with catalogue instruments. - Each question from the list will receive the closest instrument match for it. - The closest instrument match for all questions is returned as a result of this function. - - :param questions: The questions. - :param catalogue_data: The catalogue data. - :param all_instruments_text_vectors: A list of text vectors of all questions found in all the instruments uploaded. - :param questions_are_from_one_instrument: If the questions provided are coming from one instrument only. - - :return: A list of closest instrument matches for the questions provided. - """ - - # Catalogue data - catalogue_instrument_idx_to_catalogue_questions_idx: List[List[int]] = catalogue_data[ - "instrument_idx_to_question_idx" - ] - all_catalogue_questions_embeddings_concatenated: np.ndarray = catalogue_data[ - "all_embeddings_concatenated" - ] - all_catalogue_instruments: List[dict] = catalogue_data["all_instruments"] - all_catalogue_questions: List[str] = catalogue_data["all_questions"] - - # No embeddings = nothing to find - if len(all_catalogue_questions_embeddings_concatenated) == 0: - return [] - - # All instruments text vectors to dict - all_instruments_text_vectors_dict = { - text_vector.text: text_vector.vector for text_vector in all_instruments_text_vectors - } - - # The total number of questions we received as input. - num_input_questions = len(questions) - - # Get an array of dimensions. - # (number of input questions) x (number of dimensions of LLM - typically 768, 384, 500, 512, etc.) - vectors = np.array( - [all_instruments_text_vectors_dict[question.question_text] for question in questions] - ) - - # Get a 2D array of (number of input questions) x (number of questions in catalogue). - # E.g. index 0 (matches for the first input question) will contain a list of matches for each question in the - # catalogue. So the best match for the first input question is the highest similarity found in index 0. - catalogue_similarities = cosine_similarity( - vectors, all_catalogue_questions_embeddings_concatenated - ) - - # Get a 1D array of length (number of input questions). - # For each input question, this is the index of the single closest matching question text in our catalogues. - # Note that each question text in the catalogue (vector index) is unique, and we must later do a further mapping to - # find out which instrument(s) it occurs in. - idxs_of_top_questions_matched_in_catalogue = np.argmax(catalogue_similarities, axis=1) - - # Get a set of all the top matching question text indices in our catalogue. - # idxs_of_top_questions_matched_in_catalogue_set = set(idxs_of_top_questions_matched_in_catalogue) - - # This keeps track of each instrument matches how many question items in the query - # e.g. if the first instrument in our catalogue (instrument 0) matches 4 items, then this dictionary will - # contain {0: 4}. - # instrument_idx_to_num_matching_items_with_query = {} - - # This dictionary will contain the index of the instrument and the cosine similarities to the top matched questions - # in that instrument e.g. {50: [ ... ]} - instrument_idx_to_cosine_similarities_top_match: dict[int, list[float]] = {} - - # This keeps track of how many question items in total are contained in each instrument, irrespective of the - # number of matches. - # This is needed for stats such as precision and recall. - instrument_idx_to_total_num_question_items_present = {} - - # Find any instruments matching - input_question_idx_to_matching_instruments: List[List[dict]] = [] - for input_question_idx in range(len(questions)): - input_question_idx_to_matching_instruments.append([]) - for input_question_idx in range(len(questions)): - top_match_catalogue_question_idx = idxs_of_top_questions_matched_in_catalogue[ - input_question_idx - ] - for instrument_idx, question_idxs_in_this_instrument in enumerate( - catalogue_instrument_idx_to_catalogue_questions_idx - ): - if top_match_catalogue_question_idx in question_idxs_in_this_instrument: - instrument_from_catalogue = all_catalogue_instruments[instrument_idx] - if not any( - x["instrument_name"] == instrument_from_catalogue["instrument_name"] - for x in input_question_idx_to_matching_instruments[input_question_idx] - ): - input_question_idx_to_matching_instruments[ - input_question_idx - ].append(instrument_from_catalogue) - - # For each catalogue instrument get the total number of question matches in the query - # For each catalogue instrument get the total number of questions - for instrument_idx, question_idxs_in_this_instrument in enumerate( - catalogue_instrument_idx_to_catalogue_questions_idx - ): - catalogue_question_idxs_in_this_instrument_set = set( - question_idxs_in_this_instrument - ) - # instrument_idx_to_num_matching_items_with_query[instrument_idx] = len( - # catalogue_question_idxs_in_this_instrument_set.intersection( - # idxs_of_top_questions_matched_in_catalogue_set - # ) - # ) - instrument_idx_to_total_num_question_items_present[instrument_idx] = len( - catalogue_question_idxs_in_this_instrument_set - ) - - # Question similarity with catalogue questions - for idx, question in enumerate(questions): - seen_in_instruments: List[CatalogueInstrument] = [] - for instrument in input_question_idx_to_matching_instruments[idx]: - instrument_name = instrument["instrument_name"] - instrument_url = instrument["metadata"].get("url", "") - source = instrument["metadata"]["source"].upper() - sweep = instrument["metadata"].get("sweep_id", "") - seen_in_instruments.append( - CatalogueInstrument( - instrument_name=instrument_name, - instrument_url=instrument_url, - source=source, - sweep=sweep, - ) - ) - - question.closest_catalogue_question_match = CatalogueQuestion( - question=all_catalogue_questions[idxs_of_top_questions_matched_in_catalogue[idx]], - seen_in_instruments=seen_in_instruments, - ) - - # Instrument index to list of cosine similarities top question match - for input_question_idx, idx_top_input_question_match_in_catalogue in enumerate( - idxs_of_top_questions_matched_in_catalogue - ): - for ( - catalogue_instrument_idx, - catalogue_question_idxs_in_this_instrument, - ) in enumerate(catalogue_instrument_idx_to_catalogue_questions_idx): - catalogue_question_idxs_set = set( - catalogue_question_idxs_in_this_instrument - ) - if idx_top_input_question_match_in_catalogue in catalogue_question_idxs_set: - # Create the list if it doesn't exist yet - if not instrument_idx_to_cosine_similarities_top_match.get( - catalogue_instrument_idx - ): - instrument_idx_to_cosine_similarities_top_match[ - catalogue_instrument_idx - ] = [] - - # Add the cosine similarity - instrument_idx_to_cosine_similarities_top_match[ - catalogue_instrument_idx - ].append( - catalogue_similarities[input_question_idx][ - idx_top_input_question_match_in_catalogue - ] - ) - - # Keep track of the instrument id and the count of top question matches that belong to it - instrument_idx_to_top_matches_ct = { - k: len(v) for k, v in instrument_idx_to_cosine_similarities_top_match.items() - } - - # Calculate the average for each list of cosine similarities from instruments - instrument_idx_to_cosine_similarities_average: dict[int, float] = {} - for ( - instrument_idx, - cosine_similarities, - ) in instrument_idx_to_cosine_similarities_top_match.items(): - instrument_idx_to_cosine_similarities_average[instrument_idx] = ( - statistics.mean(cosine_similarities) - ) - - instrument_idx_to_score = {} - for instrument_idx, average_sim in instrument_idx_to_cosine_similarities_average.items(): - score = average_sim * (0.1 + instrument_idx_to_top_matches_ct.get(instrument_idx, 0)) - instrument_idx_to_score[instrument_idx] = score - - # Find the top 10 best instrument idx matches, index 0 containing the best match etc. - top_n_catalogue_instrument_idxs = sorted( - instrument_idx_to_score, - key=instrument_idx_to_score.get, - reverse=True - )[:200] - - # Create a list of CatalogueInstrument for each top instrument - top_instruments: List[CatalogueInstrument] = [] - for top_catalogue_instrument_idx in top_n_catalogue_instrument_idxs: - top_catalogue_instrument = all_catalogue_instruments[top_catalogue_instrument_idx] - num_questions_in_ref_instrument = ( - instrument_idx_to_total_num_question_items_present[ - top_catalogue_instrument_idx - ] - ) - num_top_match_questions = instrument_idx_to_top_matches_ct[ - top_catalogue_instrument_idx - ] - - instrument_name = top_catalogue_instrument["instrument_name"] - instrument_url = top_catalogue_instrument["metadata"].get("url", "") - source = top_catalogue_instrument["metadata"]["source"].upper() - sweep = top_catalogue_instrument["metadata"].get("sweep_id", "") - - if questions_are_from_one_instrument: - info = ( - f"{instrument_name} Sweep {sweep if sweep else 'UNKNOWN'} matched {num_top_match_questions} " - f"question(s) in your instrument, your instrument contains {num_input_questions} question(s). " - f"The reference instrument contains {num_questions_in_ref_instrument} question(s)." - ) - else: - info = ( - f"{instrument_name} Sweep {sweep if sweep else 'UNKNOWN'} matched {num_top_match_questions} " - f"question(s) in all of your instruments, your instruments contains {num_input_questions} " - f"question(s). The reference instrument contains {num_questions_in_ref_instrument} question(s)." - ) - - top_instruments.append(CatalogueInstrument( - instrument_name=instrument_name, - instrument_url=instrument_url, - source=source, - sweep=sweep, - metadata={ - "info": info, - "num_matched_questions": num_top_match_questions, - "num_ref_instrument_questions": num_questions_in_ref_instrument, - "mean_cosine_similarity": instrument_idx_to_cosine_similarities_average.get( - top_catalogue_instrument_idx) - }, - )) - - return top_instruments - - -def match_query_with_catalogue_instruments( - query: str, - catalogue_data: dict, - vectorisation_function: Callable, - texts_cached_vectors: dict[str, List[float]], - max_results: int = 100, - is_negate: bool = True -) -> dict[str, Union[list, dict]]: - """ - Match query with catalogue instruments. - - :param query: The query. - :param catalogue_data: The catalogue data. - :param vectorisation_function: A function to vectorize a text. - :param texts_cached_vectors: A dictionary of already cached text vectors (text to vector). - :param max_results: The max amount of instruments to return. - :return: A dict containing the list of instruments (up to 100) and the new text vectors. - E.g. {"instruments": [...], "new_text_vectors": {...}}. - """ - - response = {"instruments": [], "new_text_vectors": {}} - - # Catalogue data - catalogue_instrument_idx_to_catalogue_questions_idx: List[List[int]] = ( - catalogue_data["instrument_idx_to_question_idx"] - ) - all_catalogue_questions_embeddings_concatenated: np.ndarray = catalogue_data[ - "all_embeddings_concatenated" - ] - all_catalogue_instruments: List[dict] = catalogue_data["all_instruments"] - - # No embeddings = nothing to find - if len(all_catalogue_questions_embeddings_concatenated) == 0: - return response - - # Text vectors - text_vectors, new_text_vectors = create_full_text_vectors( - all_questions=[], - query=query, - vectorisation_function=vectorisation_function, - texts_cached_vectors=texts_cached_vectors, - is_negate=is_negate - ) - - # Get an array of dimensions - vectors = np.array([text_vectors[0].vector]) - - # Get a 2D array of 1 x (number of questions in catalogue) - catalogue_similarities = cosine_similarity( - vectors, all_catalogue_questions_embeddings_concatenated - ) - # Get the catalogue questions similarities for the query - catalogue_questions_similarities_for_query = catalogue_similarities[0].tolist() - - # Get indexes of top matching questions in the catalogue - # The first index contains the best match - top_catalogue_questions_matches_idxs = [ - catalogue_questions_similarities_for_query.index(i) - for i in heapq.nlargest(max_results, catalogue_questions_similarities_for_query) - ] - - # A dict of matching instruments - # The key is the name of the instrument and the value is the instrument - instrument_matches: OrderedDict[str, Instrument] = OrderedDict() - - # Find the matching instruments by looking for the instrument of the top catalogue questions matches indexes - # Loop through indexes of top matched catalogue question - for top_catalogue_question_match_idx in top_catalogue_questions_matches_idxs: - # Loop through instrument index with its question indexes - for catalogue_instrument_idx, catalogue_instrument_questions_idxs in enumerate( - catalogue_instrument_idx_to_catalogue_questions_idx - ): - # Check if the index of the top matched catalogue question is in the catalogue instrument's question indexes - if top_catalogue_question_match_idx in catalogue_instrument_questions_idxs: - catalogue_instrument = all_catalogue_instruments[ - catalogue_instrument_idx - ] - - # Add the instrument to the dict if it wasn't already added - instrument_name = catalogue_instrument["instrument_name"] - if instrument_name not in instrument_matches: - instrument_matches[instrument_name] = Instrument.model_validate( - catalogue_instrument - ) - - response["instruments"] = [x for x in instrument_matches.values()] - response["new_text_vectors"] = new_text_vectors - - return response # diff --git a/src/harmony/schemas/catalogue_instrument.py b/src/harmony/schemas/catalogue_instrument.py deleted file mode 100644 index 505599cd..00000000 --- a/src/harmony/schemas/catalogue_instrument.py +++ /dev/null @@ -1,9 +0,0 @@ -from pydantic import BaseModel, Field - - -class CatalogueInstrument(BaseModel): - instrument_name: str = Field(description="Instrument name") - instrument_url: str = Field(description="Instrument URL") - source: str = Field(description="Source") - sweep: str = Field(description="Sweep") - metadata: dict = Field(default=None, description="Metadata") diff --git a/src/harmony/schemas/catalogue_question.py b/src/harmony/schemas/catalogue_question.py deleted file mode 100644 index a18721a2..00000000 --- a/src/harmony/schemas/catalogue_question.py +++ /dev/null @@ -1,10 +0,0 @@ -from pydantic import BaseModel, Field - -from harmony.schemas.catalogue_instrument import CatalogueInstrument - - -class CatalogueQuestion(BaseModel): - question: str = Field(description="The catalogue question") - seen_in_instruments: list[CatalogueInstrument] = Field( - description="The instruments from the catalogue were the question was seen in" - ) diff --git a/src/harmony/schemas/requests/text.py b/src/harmony/schemas/requests/text.py index 5106a540..eabe3f20 100644 --- a/src/harmony/schemas/requests/text.py +++ b/src/harmony/schemas/requests/text.py @@ -28,8 +28,6 @@ import uuid from typing import Any, Dict, List, Optional from pydantic import ConfigDict, BaseModel, Field -from harmony.schemas.catalogue_instrument import CatalogueInstrument -from harmony.schemas.catalogue_question import CatalogueQuestion from harmony.schemas.enums.file_types import FileType from harmony.schemas.enums.languages import Language @@ -69,12 +67,6 @@ class Question(BaseModel): topics_strengths: Optional[dict] = Field(None, description="Automated list of topics identified by model with strength of topic") nearest_match_from_mhc_auto: Optional[dict] = Field(None, description="Automatically identified nearest MHC match") - closest_catalogue_question_match: Optional[CatalogueQuestion] = Field( - None, description="The closest question match in the catalogue for the question" - ) - seen_in_catalogue_instruments: list[CatalogueInstrument] = Field( - default=None, description="The instruments from the catalogue were the question was seen in" - ) model_config = ConfigDict( json_schema_extra={ "example": { @@ -101,11 +93,6 @@ class Instrument(BaseModel): language: Language = Field(Language.English, description="The ISO 639-2 (alpha-2) encoding of the instrument language") questions: List[Question] = Field(description="The items inside the instrument") - closest_catalogue_instrument_matches: List[CatalogueInstrument] = Field( - None, - description="The closest instrument matches in the catalogue for the instrument, the first index " - "contains the best match etc" - ) model_config = ConfigDict( json_schema_extra={ "example": { diff --git a/src/harmony/schemas/responses/text.py b/src/harmony/schemas/responses/text.py index a63d7a51..75a1f8e2 100644 --- a/src/harmony/schemas/responses/text.py +++ b/src/harmony/schemas/responses/text.py @@ -29,7 +29,6 @@ from pydantic import BaseModel, Field, RootModel -from harmony.schemas.catalogue_instrument import CatalogueInstrument from harmony.schemas.requests.text import Instrument from harmony.schemas.requests.text import Question @@ -90,11 +89,6 @@ class MatchResponse(BaseModel): query_similarity: List = Field( None, description="Similarity metric between query string and items" ) - closest_catalogue_instrument_matches: List[CatalogueInstrument] = Field( - default=[], - description="The closest catalogue instrument matches in the catalogue for all the instruments, " - "the first index contains the best match etc." - ) instrument_to_instrument_similarities: List[InstrumentToInstrumentSimilarity] = Field( None, description="A list of similarity values (precision, recall, F1) between instruments" )