Delete retrieval

retrieval/candid_blog.py

@@ -1,70 +0,0 @@
-from typing import Dict, Tuple, Any
-
-
-def build_wp_candid_blog_knn_model_query(
-    query: str,
-    fields: Tuple[str] = (
-        "content",
-        "title"
-    ),
-    k: int = 10,
-    model_id: str = "sentence-transformers__all-mpnet-base-v2"
-):
-    output = []
-
-    for f in fields:
-        output.append({
-            "field": f"embeddings.{f}.chunks.vector.predicted_value",
-            "k": k,
-            "num_candidates": 100,
-            "query_vector_builder": {
-                "text_embedding": {
-                    "model_id": model_id,
-                    "model_text": query
-                }
-            },
-            "boost": 1 / len(fields)
-        })
-    return {"knn": output}
-
-
-def build_card_html(doc: Dict[str, Any], height_px: int = 200, show_chunks=False) -> str:
-    url = f"{doc['link']}"
-    fields = ["title", "excerpt"]
-
-    fields_dict = {}
-    fields_len = 0
-    for field in fields:
-        if doc.get(field, None) is not None:
-            fields_dict[field] = doc[field]
-            fields_dict[field + "_txt"] = f"<div>{doc[field]}</div>"
-
-            if (fields_len + len(doc[field])) > 999:
-                rest_text_len = 999 - fields_len
-                if rest_text_len > 0:
-                    fields_dict[field + "_txt"] = f"<div>{doc[field][:rest_text_len] + '[...]'}</div>"
-                else: fields_dict[field + "_txt"] = f"<span>{'[...]'}</span>"
-            fields_len = fields_len + len(doc[field])
-        else:
-            fields_dict[field] = ""
-            fields_dict[field + "_txt"] = ""
-    html = f"""
-    <div style='height: {height_px}px; padding: 5px;'>
-        <div style='height: {height_px}px; border: 1px solid #febe10;'>
-            <span style='padding-left: 10px; display: inline-block; width: 100%;'>
-                <div>
-                    <span>
-                        <b>Candid blog post:</b>
-                        <a href='{url}' target='_blank' style='text-decoration: none;'>
-                            {doc['title']} 
-                        </a>
-                    </span>
-                    <br>
-                    <br>
-                    {fields_dict["excerpt_txt"]}
-                </div>
-            </span>
-        </div>
-    </div>
-    """
-    return html

retrieval/candid_help.py

@@ -1,41 +0,0 @@
-from typing import Dict, Any
-
-
-def build_card_html(doc: Dict[str, Any], height_px: int = 200, show_chunks=False) -> str:
-    url = f"{doc['link']}"
-    fields = ["title", "summary"]
-
-    fields_dict = {}
-    fields_len = 0
-    for field in fields:
-        if doc.get(field, None) is not None:
-            fields_dict[field] = doc[field]
-            fields_dict[field + "_txt"] = f"<div>{doc[field]}</div>"
-
-            if (fields_len + len(doc[field])) > 999:
-                rest_text_len = 999 - fields_len
-                if rest_text_len > 0:
-                    fields_dict[field + "_txt"] = f"<div>{doc[field][:rest_text_len] + '[...]'}</div>"
-                else: fields_dict[field + "_txt"] = f"<span>{'[...]'}</span>"
-            fields_len = fields_len + len(doc[field])
-        else:
-            fields_dict[field] = ""
-            fields_dict[field + "_txt"] = ""
-    html = f"""
-    <div style='height: {height_px}px; padding: 5px;'>
-        <div style='height: {height_px}px; border: 1px solid #febe10;'>
-            <span style='padding-left: 10px; display: inline-block; width: 100%;'>
-                <div>
-                    <span>
-                        <b>Candid help article:</b>
-                        <a href='{url}' target='_blank' style='text-decoration: none;'>
-                            {doc['title']} 
-                        </a>
-                    </span>
-                    <br>
-                </div>
-            </span>
-        </div>
-    </div>
-    """
-    return html

retrieval/candid_learning.py

@@ -1,41 +0,0 @@
-from typing import Dict, Any
-
-
-def build_card_html(doc: Dict[str, Any], height_px: int = 200, show_chunks=False) -> str:
-    url = f"{doc['url']}"
-    fields = ["title", "excerpt"]
-
-    fields_dict = {}
-    fields_len = 0
-    for field in fields:
-        if doc.get(field, None) is not None:
-            fields_dict[field] = doc[field]
-            fields_dict[field + "_txt"] = f"<div>{doc[field]}</div>"
-
-            if (fields_len + len(doc[field])) > 999:
-                rest_text_len = 999 - fields_len
-                if rest_text_len > 0:
-                    fields_dict[field + "_txt"] = f"<div>{doc[field][:rest_text_len] + '[...]'}</div>"
-                else: fields_dict[field + "_txt"] = f"<span>{'[...]'}</span>"
-            fields_len = fields_len + len(doc[field])
-        else:
-            fields_dict[field] = ""
-            fields_dict[field + "_txt"] = ""
-    html = f"""
-    <div style='height: {height_px}px; padding: 5px;'>
-        <div style='height: {height_px}px; border: 1px solid #febe10;'>
-            <span style='padding-left: 10px; display: inline-block; width: 100%;'>
-                <div>
-                    <span>
-                        <b>Candid Learning resource:</b>
-                        <a href='{url}' target='_blank' style='text-decoration: none;'>
-                            {doc['title']} 
-                        </a>
-                    </span>
-                    <br>
-                </div>
-            </span>
-        </div>
-    </div>
-    """
-    return html

retrieval/config.py

@@ -1,26 +0,0 @@
-import os
-
-ONECANDID_QA = {
-    'ES_URL': os.getenv('ONECANDID_URL_QA'),
-    "ES_CLOUD_ID": os.getenv("ONECANDID_CLOUD_ID"),
-    "ES_API_KEY": os.getenv("ONECANDID_API_KEY")
-}
-
-class Indices:
-    ISSUELAB_INDEX = "search-semantic-issuelab_v1"
-    ISSUELAB_INDEX_ELSER = "search-semantic-issuelab-elser_ve2"
-    YOUTUBE_INDEX = "search-semantic-youtube_v1"
-    YOUTUBE_INDEX_ELSER = "search-semantic-youtube-elser_ve1"
-    CANDID_BLOG_INDEX = "search-semantic-candid-blog_v1"
-    CANDID_BLOG_INDEX_ELSER = "search-semantic-candid-blog-elser_ve2"
-    CANDID_LEARNING_INDEX_ELSER = "search-semantic-candid-learning_ve1"
-    CANDID_HELP_INDEX_ELSER = "search-semantic-candid-help-elser_ve1"
-
-
-ALL_INDICES = (
-    "issuelab",
-    "youtube",
-    "candid_blog",
-    "candid_learning",
-    "candid_help"
-)

retrieval/elastic.py

@@ -1,320 +0,0 @@
-from typing import List, Tuple, Dict, Iterable, Iterator, Optional, Any
-from dataclasses import dataclass
-from functools import partial
-from itertools import groupby
-
-from pydantic import BaseModel, Field
-from langchain_core.documents import Document
-from langchain_core.tools import Tool
-
-from elasticsearch import Elasticsearch
-
-try:
-    from config import ONECANDID_QA, ALL_INDICES, Indices
-except ImportError:
-    from .config import ONECANDID_QA, ALL_INDICES, Indices
-
-
-@dataclass
-class ElasticHitsResult:
-    """Dataclass for Elasticsearch hits results
-    """
-    index: str
-    id: Any
-    score: float
-    source: Dict[str, Any]
-    inner_hits: Dict[str, Any]
-
-
-class RetrieverInput(BaseModel):
-    """Input to the Elasticsearch retriever."""
-    user_input: str = Field(description="query to look up in retriever")
-
-
-def build_text_expansion_query(
-    query: str,
-    fields: Tuple[str],
-    model_id: str = ".elser_model_2_linux-x86_64"
-) -> Dict[str, Any]:
-
-    output = []
-
-    for f in fields:
-        output.append({
-            "nested": {
-                "path": f"embeddings.{f}.chunks",
-                "query": {
-                    "text_expansion": {
-                        f"embeddings.{f}.chunks.vector": {
-                            "model_id": model_id,
-                            "model_text": query,
-                            "boost": 1 / len(fields)
-                        }
-                    }
-                },
-                 "inner_hits": {
-                    "_source": False,
-                    "size": 2,
-                    "fields": [f"embeddings.{f}.chunks.chunk"]
-                }
-            }
-        })
-    return {"query": {"bool": {"should": output}}}
-
-
-def query_builder(query: str, indices: List[str], **kwargs):
-    queries = []
-    if indices is None:
-        indices = list(ALL_INDICES)
-
-    for index in indices:
-        if index == "issuelab":
-            q = build_text_expansion_query(
-                query=query,
-                fields=("description", "content", "combined_issuelab_findings", "combined_item_description")
-            )
-            q["_source"] = {"excludes": ["embeddings"]}
-            q["size"] = 1
-            queries.extend([{"index": Indices.ISSUELAB_INDEX_ELSER}, q])
-        elif index == "youtube":
-            q = build_text_expansion_query(
-                query=query,
-                fields=("captions_cleaned", "description_cleaned", "title")
-            )
-            # text_cleaned duplicates captions_cleaned
-            q["_source"] = {"excludes": ["embeddings", "captions", "description", "text_cleaned"]}
-            q["size"] = 2
-            queries.extend([{"index": Indices.YOUTUBE_INDEX_ELSER}, q])
-        elif index == "candid_blog":
-            q = build_text_expansion_query(
-                query=query,
-                fields=("content", "title")
-            )
-            q["_source"] = {"excludes": ["embeddings"]}
-            q["size"] = 2
-            queries.extend([{"index": Indices.CANDID_BLOG_INDEX_ELSER}, q])
-        elif index == "candid_learning":
-            q = build_text_expansion_query(
-                query=query,
-                fields=("content", "title", "training_topics", "staff_recommendations")
-            )
-            q["_source"] = {"excludes": ["embeddings"]}
-            q["size"] = 2
-            queries.extend([{"index": Indices.CANDID_LEARNING_INDEX_ELSER}, q])
-        elif index == "candid_help":
-            q = build_text_expansion_query(
-                query=query,
-                fields=("content", "combined_article_description")
-            )
-            q["_source"] = {"excludes": ["embeddings"]}
-            q["size"] = 2
-            queries.extend([{"index": Indices.CANDID_HELP_INDEX_ELSER}, q])
-
-    return queries
-
-
-def multi_search(queries: List[ElasticHitsResult]):
-    results = []
-    with Elasticsearch(
-        cloud_id=ONECANDID_QA["ES_CLOUD_ID"],
-        api_key=ONECANDID_QA["ES_API_KEY"],
-        verify_certs=False,
-        request_timeout=60 * 3
-    ) as es:
-        for query_group in es.msearch(body=queries).get("responses", []):
-            for hit in query_group.get("hits", {}).get("hits", []):
-                hit = ElasticHitsResult(
-                    index=hit["_index"],
-                    id=hit["_id"],
-                    score=hit["_score"],
-                    source=hit["_source"],
-                    inner_hits=hit.get("inner_hits", {})
-                )
-                results.append(hit)
-    return results
-
-
-def get_query_results(search_text: str, indices: Optional[List[str]] = None):
-    queries = query_builder(query=search_text, indices=indices)
-    return multi_search(queries)
-
-
-def reranker(query_results: Iterable[ElasticHitsResult]) -> Iterator[ElasticHitsResult]:
-    """Reranks Elasticsearch hits coming from multiple indicies/queries which may have scores on different scales.
-    This will shuffle results
-
-    Parameters
-    ----------
-    query_results : Iterable[ElasticHitsResult]
-
-    Yields
-    ------
-    Iterator[ElasticHitsResult]
-    """
-
-    results: List[ElasticHitsResult] = []
-    for _, data in groupby(query_results, key=lambda x: x.index):
-        data = list(data)
-        max_score = max(data, key=lambda x: x.score).score
-        min_score = min(data, key=lambda x: x.score).score
-
-        for d in data:
-            d.score = (d.score - min_score) / (max_score - min_score + 1e-9)
-            results.append(d)
-
-    yield from sorted(results, key=lambda x: x.score, reverse=True)
-
-
-def get_results(user_input: str, indices: List[str]) -> List[ElasticHitsResult]:
-    output = ["Search didn't return any Candid sources"]
-    page_content=[]
-    content = "Search didn't return any Candid sources"
-    results = get_query_results(search_text=user_input, indices=indices)
-    if results:
-        output = get_reranked_results(results)
-        for doc in output:
-            page_content.append(doc.page_content)
-        content = "/n/n".join(page_content)
-    # for the tool we need to return a tuple for content_and_artifact type
-    return content, output
-
-
-def get_context(field_name: str, hit: ElasticHitsResult, context_length: int = 1024) -> str:
-    """Pads the relevant chunk of text with context before and after
-
-    Parameters
-    ----------
-    field_name : str
-        a field with the long text that was chunked into pieces
-    hit : ElasticHitsResult
-    context_length : int, optional
-        length of text to add before and after the chunk, by default 1024
-
-    Returns
-    -------
-    str
-        longer chunks stuffed together
-    """
-
-    chunks_with_context = []
-    long_text = hit.source.get(f"{field_name}", "")
-    inner_hits_field = f"embeddings.{field_name}.chunks"
-    inner_hits = hit.inner_hits
-    found_chunks = inner_hits.get(inner_hits_field, {})
-    if found_chunks:
-        hits = found_chunks.get("hits", {}).get("hits", [])
-        for h in hits:
-            chunk = h.get("fields", {})[inner_hits_field][0]["chunk"][0]
-            chunk = chunk[3:-3] # cutting the middle because we may have tokenizing artefacts there
-            # Find the start and end indices of the chunk in the large text
-            start_index = long_text.find(chunk)
-            if start_index != -1: # Chunk is found
-                end_index = start_index + len(chunk)
-                pre_start_index = max(0, start_index - context_length)
-                post_end_index = min(len(long_text), end_index + context_length)
-                context = long_text[pre_start_index:post_end_index]
-                chunks_with_context.append(context)
-    chunks_with_context_txt = '\n\n'.join(chunks_with_context)
-
-    return chunks_with_context_txt
-
-
-def process_hit(hit: ElasticHitsResult) -> Document | None:
-    if "issuelab-elser" in hit.index:
-        combined_item_description = hit.source.get("combined_item_description", "") # title inside
-        description = hit.source.get("description", "")
-        combined_issuelab_findings = hit.source.get("combined_issuelab_findings", "")
-        # we only need to process long texts
-        chunks_with_context_txt = get_context("content", hit, context_length=12)
-        doc = Document(
-            page_content='\n\n'.join([
-                combined_item_description,
-                combined_issuelab_findings,
-                description,
-                chunks_with_context_txt
-            ]),
-            metadata={
-                "source": "IssueLab",
-                "source_id": hit.source["resource_id"],
-                "url": hit.source.get("permalink", "")
-            }
-        )
-    elif "youtube" in hit.index:
-        title = hit.source.get("title", "")
-        # we only need to process long texts
-        description_cleaned_with_context_txt = get_context("description_cleaned", hit, context_length=12)
-        captions_cleaned_with_context_txt = get_context("captions_cleaned", hit, context_length=12)
-        doc = Document(
-            page_content='\n\n'.join([title, description_cleaned_with_context_txt, captions_cleaned_with_context_txt]),
-            metadata={
-                "source": "Candid Youtube",
-                "source_id": hit.source['video_id'],
-                "url": f"https://www.youtube.com/watch?v={hit.source['video_id']}"
-            }
-        )
-    elif "candid-blog" in hit.index:
-        excerpt = hit.source.get("excerpt", "")
-        title = hit.source.get("title", "")
-        # we only need to process long texts
-        content_with_context_txt = get_context("content", hit, context_length=12)
-        doc = Document(
-            page_content='\n\n'.join([title, excerpt, content_with_context_txt]),
-            metadata={
-                "source": "Candid Blog",
-                "source_id": hit.source["id"],
-                "url": hit.source["link"]
-            }
-        )
-    elif "candid-learning" in hit.index:
-        title = hit.source.get("title", "")
-        content_with_context_txt = get_context("content", hit, context_length=12)
-        training_topics = hit.source.get("training_topics", "")
-        staff_recommendations = hit.source.get("staff_recommendations", "")
-
-        doc = Document(
-            page_content='\n\n'.join([title, staff_recommendations, training_topics, content_with_context_txt]),
-            metadata={
-                "source": "Candid Learning",
-                "source_id": hit.source["post_id"],
-                "url": hit.source.get("url", "")
-            }
-        )
-    elif "candid-help" in hit.index:
-        title = hit.source.get("title", "")
-        content_with_context_txt = get_context("content", hit, context_length=12)
-        combined_article_description = hit.source.get("combined_article_description", "")
-
-        doc = Document(
-            page_content='\n\n'.join([combined_article_description, content_with_context_txt]),
-            metadata={
-                "source": "Candid Help",
-                "source_id": hit.source["id"],
-                "url": hit.source.get("link", "")
-            }
-        )
-    else:
-        doc = None
-    return doc
-
-
-def get_reranked_results(results: List[ElasticHitsResult]) -> List[Document]:
-    output = []
-    for r in reranker(results):
-        hit = process_hit(r)
-        output.append(hit)
-    return output
-
-
-def retriever_tool(indices: List[str]) -> Tool:
-    # cannot use create_retriever_tool because it only provides content losing all metadata on the way
-    # https://python.langchain.com/docs/how_to/custom_tools/#returning-artifacts-of-tool-execution
-    return Tool(
-        name="retrieve_social_sector_information",
-        func=partial(get_results, indices=indices),
-        description=(
-            "Return additional information about social and philanthropic sector, "
-            "including nonprofits (NGO), grants, foundations, funding, RFP, LOI, Candid."
-        ),
-        args_schema=RetrieverInput,
-        response_format="content_and_artifact"
-    )

retrieval/elastic_qa.py

@@ -1,194 +0,0 @@
-from typing import List, Dict, Any, Optional
-import re
-import os
-
-import numpy as np
-
-from langchain_huggingface.embeddings import HuggingFaceEndpointEmbeddings
-from langchain_core.documents import Document
-
-from ...socialspark.retrieval_qa.elastic import ElasticHitsResult, ElasticsearchQABase
-
-from .elastic import query_builder
-from .config import ALL_INDICES
-
-timestamp_pattern = re.compile(r"^(\d:\d{2}:\d{2}\.\d{3}),(\d:\d{2}:\d{2}\.\d{3})\n(.*)$")
-
-
-def parse_youtube_captions(caption_text: str) -> List[str]:
-    timestamp_blocks = re.split(r'\n{2}', caption_text)
-    parsed_blocks = [m.groups() for block in timestamp_blocks if (m:=timestamp_pattern.match(block)) if not None]
-    video_length_mins = int(parsed_blocks[-1][0].split(':')[1])
-
-    # what is the block timing in minutes which gives 10 chunks? keep minimum 1 minute breaks to maintain context
-    minute_difference = min(video_length_mins // 10, 1)
-
-    prev_minute_start = 0
-    texts = []
-    text_block = []
-    for start, _, text in parsed_blocks:
-        text_block.append(text)
-
-        current_minute = int(start.split(':')[1])
-        if (current_minute - prev_minute_start) >= minute_difference:
-            texts.append(' '.join(text_block))
-            text_block.clear()
-            prev_minute_start = current_minute
-
-    return texts
-
-
-def parse_candid_learning(text: str) -> List[str]:
-    texts = []
-    for block in map(str.strip, re.split(r'\n{1,}', text, flags=re.I | re.M)):
-        if (
-            'back to top' in block.lower()
-            or 'table of contents' in block.lower()
-        ):
-            continue
-
-        texts.append(block)
-    return texts
-
-
-class ElasticsearchQA(ElasticsearchQABase):
-
-    indices: Optional[List[str]] = ALL_INDICES
-    embedding: Any = HuggingFaceEndpointEmbeddings(
-        huggingfacehub_api_token=os.getenv("HF_API_KEY"),
-        # model="sentence-transformers/all-mpnet-base-v2"
-        model="mixedbread-ai/mxbai-embed-large-v1"
-    )
-
-    def build_query(self, query: str, **kwargs) -> List[Dict[str, Any]]:
-        queries = query_builder(query=query, indices=self.indices)
-        return queries
-
-    def sub_section_alignment(self, query: str, document: List[str]) -> str:
-        question_vector = np.array(self.embedding.embed_query(query), dtype='float32')
-        vectors = np.array(self.embedding.embed_documents(document), dtype='float32')
-        vectors = np.array(vectors, dtype='float32')
-
-        size = max(5, int(0.1 * len(document)))
-
-        doc_vector = vectors.sum(axis=0, keepdims=True)
-        vectors /= np.linalg.norm(vectors, ord=2.0, axis=-1, keepdims=True)
-        doc_vector /= np.linalg.norm(doc_vector, ord=2.0, axis=-1, keepdims=True)
-        question_vector /= np.linalg.norm(question_vector, ord=2.0, axis=-1, keepdims=True)
-        # similarity = (doc_vector * vectors).sum(axis=-1)
-        similarity = (question_vector * vectors).sum(axis=-1)
-
-        return '\n'.join(text for text, _ in sorted(zip(document, similarity), key=lambda x: 1 - x[-1])[:size])
-
-    def process_hit(self, hit: ElasticHitsResult, q: str) -> Document | None:
-        if "news" in hit.index:
-            doc = Document(
-                page_content='\n\n'.join(v for k, v in hit.source["texts"].items() if v),
-                metadata={
-                    "source": "news",
-                    "source_id": hit.source['metadata']['link']
-                }
-            )
-        elif "transactions" in hit.index:
-            doc = Document(
-                page_content='\n\n'.join(v for k, v in hit.source["semantic_texts"].items() if v),
-                metadata={
-                    "source": "cds-transactions",
-                    "source_id": hit.source["id"]
-                }
-            )
-        elif "organizations" in hit.index:
-            source = hit.source
-            org_gen = source.get("combined_organization_description_general", "")
-            org_fin = source.get("combined_organization_description_financial", "")
-            org_contact = source.get("combined_organization_description_contacts", "")
-            mission = ""
-            if source.get("mission_statement", None) is not None:
-                mission = source.get("mission_statement", "")
-            keyword = ""
-            if source.get("keyword", None) is not None:
-                keyword = source.get("keyword", "")
-            programs = source.get("programs", "")
-            doc = Document(
-                page_content='\n\n'.join([org_gen, mission, keyword, org_fin, org_contact, programs]),
-                metadata={
-                    "source": "UP-organizations-QA",
-                    "source_id": hit.source["candid_entity_id"]
-                }
-            )
-        # elif "issuelab" in hit.index:
-        #     doc = Document(
-        #         page_content='\n\n'.join(v for k, v in hit.source["semantic_texts"].items() if v),
-        #         metadata={
-        #             "source": "IssueLab",
-        #             "source_id": hit.source["resource_id"]
-        #         }
-        #     )
-        elif "issuelab-elser" in hit.index:
-            title = hit.source.get("title", "")
-            description = hit.source.get("description", "")
-            doc = Document(
-                page_content='\n\n'.join([title, description]),
-                metadata={
-                    "source": "IssueLab",
-                    "source_id": hit.source["resource_id"],
-                    "url": hit.source.get("permalink", "")
-                }
-            )
-        elif "youtube" in hit.index:
-            title = hit.source.get("title", "")
-            summary = self.sub_section_alignment(
-                query=q,
-                document=parse_youtube_captions(hit.source.get("text"))
-            )
-            doc = Document(
-                # page_content='\n\n'.join([title]),
-                page_content=summary,
-                metadata={
-                    "source": "Candid's Youtube channel",
-                    "source_id": hit.source['video_id'],
-                    "url": f"https://www.youtube.com/watch?v={hit.source['video_id']}"
-                }
-            )
-        elif "candid-blog" in hit.index:
-            excerpt = hit.source.get("excerpt", "")
-            title = hit.source.get("title", "")
-            doc = Document(
-                page_content='\n\n'.join([title, excerpt]),
-                metadata={
-                    "source": "Candid Blog",
-                    "source_id": hit.source["id"],
-                    "url": hit.source["link"]
-                }
-            )
-        elif "candid-learning" in hit.index:
-            # content = hit.source.get("content", "")
-            title = hit.source.get("title", "")
-            summary = self.sub_section_alignment(
-                query=q,
-                document=parse_candid_learning(hit.source.get("content", ""))
-            )
-            doc = Document(
-                # page_content='\n\n'.join([title]),
-                page_content=summary,
-                metadata={
-                    "source": "Candid Learning",
-                    "source_id": hit.source["post_id"],
-                    "url": hit.source.get("url", "")
-                }
-            )
-        elif "candid-help" in hit.index:
-            title = hit.source.get("title", "")
-            content = hit.source.get("content", "")
-
-            doc = Document(
-                page_content='\n\n'.join([title, content]),
-                metadata={
-                    "source": "Candid Help",
-                    "source_id": hit.source["id"],
-                    "url": hit.source.get("link", "")
-                }
-            )
-        else:
-            doc = None
-        return doc

retrieval/issuelab.py

@@ -1,50 +0,0 @@
-from typing import Dict, Any
-
-
-def issuelab_card_html(doc: Dict[str, Any], height_px: int = 200, show_chunks=False) -> str:
-    chunks_html = ""
-    if show_chunks:
-        cleaned_text = []
-        for k, v in doc["inner_hits"].items():
-            hits = v["hits"]["hits"]
-            for h in hits:
-                for k1, v1 in h["fields"].items():
-                    # we don't want other chunks
-                    if "content" in k1:
-                        cleaned_text.append(f"<div><p>{v1[0]['chunk'][0]}</p></div>")
-
-        chunks_html ="<span><b>Relevant parts of the content:</b></span>" + "<br>".join(cleaned_text)
-
-    html = f"""
-    <div style='height: auto; padding: 5px;'>
-        <div style='border: 1px solid #febe10;'>
-            <span style='display: inline-block; height: {height_px - 10}px; padding: 5px; vertical-align: top;'>
-                <img
-                    src='{doc['cover_graphic_small']}'
-                    style='max-height: 100%; overflow: hidden; border-radius: 3%;'
-                >
-            </span>
-
-            <span style='padding: 10px; display: inline-block; width: 70%;'>
-                <div>
-                    <span><b>Issuelab ID:</b> {doc['resource_id']}</span>
-                    <br>
-                    <span>
-                        <a href='{doc['issuelab_url']}' target='_blank' style='text-decoration: none;'>
-                            {doc['title']}
-                        </a>
-                    </span>
-                    <br>
-
-                    <span><b>Description:</b> {doc['description']}</span>
-                    <br>
-                    <div>{doc['combined_item_description']}</div>
-                    <br>
-                    <div>{chunks_html}</div>
-                    
-                </div>
-            </span>
-        </div>
-    </div>
-    """
-    return html

retrieval/youtube.py

@@ -1,82 +0,0 @@
-from typing import Dict, Tuple, Any
-
-
-def build_knn_model_query(
-    query: str,
-    fields: Tuple[str] = (
-        "captions_cleaned",
-        "description_cleaned",
-        "title"
-    ),
-    k: int = 10,
-    model_id: str = "sentence-transformers__all-mpnet-base-v2"
-):
-    output = []
-
-    for f in fields:
-        output.append({
-            "field": f"embeddings.{f}.chunks.vector.predicted_value",
-            "k": k,
-            "num_candidates": 100,
-            "query_vector_builder": {
-                "text_embedding": {
-                    "model_id": model_id,
-                    "model_text": query
-                }
-            },
-            "boost": 1 / len(fields)
-        })
-    return {"knn": output}
-
-
-def build_card_html(doc: Dict[str, Any], height_px: int = 200, show_chunks=False) -> str:
-    url = f"https://www.youtube.com/watch?v={doc['video_id']}"
-    fields = ["title", "description_cleaned"]
-
-    fields_dict = {}
-    fields_len = 0
-    for field in fields:
-        if doc.get(field, None) is not None:
-            fields_dict[field] = doc[field]
-            fields_dict[field + "_txt"] = f"<div>{doc[field]}</div>"
-
-            if (fields_len + len(doc[field])) > 999:
-                rest_text_len = 999 - fields_len
-                if rest_text_len > 0:
-                    fields_dict[field + "_txt"] = f"<div>{doc[field][:rest_text_len] + '[...]'}</div>"
-                else: fields_dict[field + "_txt"] = f"<span>{'[...]'}</span>"
-            fields_len = fields_len + len(doc[field])
-        else:
-            fields_dict[field] = ""
-            fields_dict[field + "_txt"] = ""
-    html = f"""
-    <div style='height: {height_px}px; padding: 5px;'>
-        <div style='height: {height_px}px; border: 1px solid #febe10;'>
-            <span style='padding-left: 10px; display: inline-block; width: 100%;'>
-                <div>
-                    <span>
-                        <b>Candid Youtube video:</b>
-                        <a href='{url}' target='_blank' style='text-decoration: none;'>
-                            {doc['title']} 
-                        </a>
-                    </span>
-                    <iframe 
-                        width="426"
-                        height="240"
-                        src="https://www.youtube.com/embed/{doc['video_id']}?si=0-y6eRrOzXTUSBDY"
-                        title="YouTube video player"
-                        frameborder="0"
-                        allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture; web-share"
-                        referrerpolicy="strict-origin-when-cross-origin"
-                        allowfullscreen
-                        style="display: inline-block; float: left;padding-right: 10px;padding-top: 5px;">
-                    </iframe>
-                    <br>
-                    <br>
-                    {fields_dict["description_cleaned_txt"]}
-                </div>
-            </span>
-        </div>
-    </div>
-    """
-    return html