Upload folder using huggingface_hub

app/concepts.py

@@ -0,0 +1,139 @@
+"""Concept-vector search: nearest papers to added concepts minus removed ones.
+
+The document-level analogue of ``king − man + woman``. Each concept is embedded
+once, the vectors are combined into one direction, and the index is searched by
+nearest neighbour to it. Every match carries its cosine to each supplied concept
+so the answer stays legible: what pulled a paper in, and what should have pushed
+it away. Nothing here scores novelty or likelihood — only cosines and counts.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+import numpy as np
+
+from app.config import Settings, settings
+from app.embeddings import QueryEmbedder
+from app.gapmap import _bucket, combine
+from app.models import ConceptSearchRequest
+
+VERSION = "concepts-v1"
+
+
+class ConceptError(Exception):
+    """A concept search that cannot honestly return matches."""
+
+    status = 503
+
+
+class EmbeddingsUnavailable(ConceptError):
+    status = 503
+
+
+class ConceptsCancelOut(ConceptError):
+    status = 422
+
+
+def _unit(values) -> np.ndarray | None:
+    vector = np.asarray(values, dtype=np.float32)
+    norm = float(np.linalg.norm(vector))
+    return None if norm < 1e-6 else vector / norm
+
+
+def _source(document: dict) -> str:
+    source = document.get("source") or "openalex"
+    return "clinicaltrials" if source == "ctgov" else str(source)
+
+
+class ConceptVectorService:
+    def __init__(self, repository, config: Settings = settings):
+        self.repo = repository
+        self.config = config
+        self.embedder = QueryEmbedder(config)
+
+    async def embed(self, text: str) -> list[float] | None:
+        return await self.embedder.embed(text)
+
+    async def resolve(
+        self, positive: list[str], negative: list[str]
+    ) -> tuple[np.ndarray, dict[str, np.ndarray]]:
+        """Embed each distinct concept once and combine them into one direction.
+
+        The per-term vectors come back with the combination so a caller can
+        explain a result against each concept, or reuse the direction for a
+        query of its own without paying for the embeddings twice.
+        """
+        terms = list(dict.fromkeys([*positive, *negative]))
+        try:
+            vectors = await asyncio.gather(*(self.embed(term) for term in terms))
+        except Exception as exc:
+            raise EmbeddingsUnavailable(
+                f"The embedding model could not be loaded ({type(exc).__name__}), "
+                "so the concepts could not be turned into vectors."
+            ) from exc
+        if any(vector is None for vector in vectors):
+            raise EmbeddingsUnavailable(
+                "Embeddings are disabled, so the concepts could not be turned into vectors."
+            )
+        by_term = {
+            term: np.asarray(vector, dtype=np.float32) for term, vector in zip(terms, vectors)
+        }
+        combined = combine(
+            [by_term[term] for term in positive], [by_term[term] for term in negative]
+        )
+        if combined is None:
+            raise ConceptsCancelOut(
+                "The concepts cancel each other out, so the search has no direction."
+            )
+        return combined, by_term
+
+    async def search(self, request: ConceptSearchRequest) -> dict:
+        combined, by_term = await self.resolve(request.positive, request.negative)
+        filters = (
+            request.filters.model_dump()
+            if request.filters is not None and request.filters.active
+            else None
+        )
+        hits = await self.repo.knn_studies(
+            [float(value) for value in combined], limit=request.limit, filters=filters
+        )
+        signed = [(term, "positive") for term in request.positive]
+        signed += [(term, "negative") for term in request.negative]
+        warnings: list[str] = []
+        matches = []
+        for hit in hits:
+            vector = _unit(hit.get("embedding") or [])
+            if vector is None and not warnings:
+                warnings.append(
+                    "Some matches were stored without a vector, so their per-concept "
+                    "cosines are missing."
+                )
+            matches.append(
+                {
+                    "id": str(hit.get("id", "")),
+                    "title": hit.get("title", ""),
+                    "year": hit.get("year") if isinstance(hit.get("year"), int) else None,
+                    "url": hit.get("url") or "",
+                    "source": _source(hit),
+                    "verdict": _bucket(hit),
+                    "citations": hit.get("cited_by_count") or 0,
+                    "cosine": round(float(hit.get("cosine", 0.0)), 4),
+                    "concepts": [
+                        {
+                            "text": term,
+                            "sign": sign,
+                            "cosine": round(float(vector @ by_term[term]), 4),
+                        }
+                        for term, sign in signed
+                    ]
+                    if vector is not None
+                    else [],
+                }
+            )
+        return {
+            "version": VERSION,
+            "concepts": [{"text": term, "sign": sign} for term, sign in signed],
+            "matches": matches,
+            "warnings": warnings,
+        }

app/embeddings.py

@@ -4,6 +4,7 @@ No hosted embedding endpoint is used. Loading/downloading a model is deferred
 until embeddings are requested, so BM25 and metadata-only operation stay light.
 """
 
+import asyncio
 from functools import lru_cache
 
 DEFAULT_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
@@ -46,3 +47,22 @@ class Embedder:
 @lru_cache(maxsize=4)
 def get_embedder(model_name: str = DEFAULT_MODEL, device: str = "cpu") -> Embedder:
     return Embedder(model_name=model_name, device=device)
+
+
+class QueryEmbedder:
+    """One query embedding per call, loaded on first use and encoded off the loop.
+
+    ``None`` rather than a zero vector when embeddings are disabled: a caller
+    must say it could not embed, not search with a direction it invented.
+    """
+
+    def __init__(self, config):
+        self.config = config
+        self.embedder: Embedder | None = None
+
+    async def embed(self, text: str) -> list[float] | None:
+        if not self.config.embeddings_enabled:
+            return None
+        if self.embedder is None:
+            self.embedder = get_embedder(self.config.embedding_model, self.config.embedding_device)
+        return await asyncio.to_thread(self.embedder.embed_query, text)

app/gapmap_service.py

@@ -14,6 +14,7 @@ import asyncio
 import numpy as np
 
 from app.config import Settings, settings
+from app.embeddings import QueryEmbedder
 from app.gapmap import (
     _bucket,
     _corpus,
@@ -71,18 +72,12 @@ class GapMapService:
     def __init__(self, repository, config: Settings = settings):
         self.repo = repository
         self.config = config
-        self.embedder = None
+        self.embedder = QueryEmbedder(config)
         self._built: dict | None = None
         self._lock = asyncio.Lock()
 
     async def embed(self, text: str) -> list[float] | None:
-        if not self.config.embeddings_enabled:
-            return None
-        if self.embedder is None:
-            from app.embeddings import get_embedder
-
-            self.embedder = get_embedder(self.config.embedding_model, self.config.embedding_device)
-        return await asyncio.to_thread(self.embedder.embed_query, text)
+        return await self.embedder.embed(text)
 
     async def build(self, refresh: bool = False) -> dict:
         async with self._lock:

app/main.py

@@ -11,10 +11,11 @@ from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse, StreamingResponse
 from fastapi.staticfiles import StaticFiles
 
+from app.concepts import ConceptError, ConceptVectorService
 from app.config import settings
 from app.gapmap_service import GapMapService
 from app.llm import LLMService
-from app.models import MapRequest, NoveltyRequest, SearchRequest
+from app.models import ConceptSearchRequest, MapRequest, NoveltyRequest, SearchRequest
 from app.novelty import NoveltyEngine
 from app.pipeline import SearchPipeline
 from app.repository import ElasticRepository
@@ -30,6 +31,7 @@ async def lifespan(app: FastAPI):
     app.state.pipeline = SearchPipeline(repo, llm)
     app.state.novelty = NoveltyEngine(llm, repo=repo)
     app.state.gapmap = GapMapService(repo)
+    app.state.concepts = ConceptVectorService(repo)
     yield
     await app.state.novelty.fulltext.close()
     await llm.close()
@@ -170,6 +172,21 @@ async def gap_map(body: MapRequest, request: Request):
         ) from None
 
 
+@app.post("/concepts/search")
+@app.post("/api/concepts/search", include_in_schema=False)
+async def concept_search(body: ConceptSearchRequest, request: Request):
+    try:
+        return await request.app.state.concepts.search(body)
+    except ConceptError as exc:
+        raise HTTPException(status_code=exc.status, detail=str(exc)) from None
+    except Exception as exc:
+        logger.error("Concept search failed: %s", type(exc).__name__)
+        raise HTTPException(
+            status_code=503,
+            detail="Elasticsearch could not complete the concept search. Check the backend connection and index setup.",
+        ) from None
+
+
 @app.get("/studies/{study_id:path}")
 @app.get("/api/studies/{study_id:path}", include_in_schema=False)
 async def study_detail(study_id: str, request: Request):

app/models.py

@@ -240,6 +240,23 @@ class MapRequest(BaseModel):
     refresh: bool = False
 
 
+ConceptTerm = Annotated[str, Field(min_length=2, max_length=200)]
+
+
+class ConceptSearchRequest(BaseModel):
+    """Concepts to add and to subtract before the index is searched by nearest neighbour.
+
+    The negatives are part of the query direction, not a post-filter: a paper is
+    pushed away by a subtracted concept rather than removed for mentioning it.
+    """
+
+    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True, allow_inf_nan=False)
+    positive: list[ConceptTerm] = Field(min_length=1, max_length=8)
+    negative: list[ConceptTerm] = Field(default_factory=list, max_length=8)
+    limit: int = Field(default=20, ge=1, le=50)
+    filters: SearchFilters | None = None
+
+
 class Claim(BaseModel):
     """The testable content of a hypothesis, as retrieval terms and comparable facets."""
 

app/pipeline.py

@@ -14,6 +14,7 @@ from app.llm import LLMService, Usage
 from app.models import Pico, SearchRequest
 from app.repository import (
     BUCKETS,
+    EFFECT_DIRECTIONS,
     ElasticRepository,
     filter_clauses,
     lexical_query,
@@ -116,12 +117,21 @@ def recount(studies: list[dict], sesoi: float, effect_type: str) -> dict:
     """Bucket counts over screened studies, replacing the keyword-match aggregation."""
     counts = dict.fromkeys(BUCKETS, 0)
     reasons = dict.fromkeys(INCONCLUSIVE_REASONS, 0)
+    directions = dict.fromkeys(EFFECT_DIRECTIONS.values(), 0)
     for study in studies:
         verdict = assign_bucket(study, sesoi, effect_type)
         counts[verdict["bucket"]] += 1
         if verdict["inconclusive_reason"]:
             reasons[verdict["inconclusive_reason"]] += 1
-    return {"total": len(studies), "bucketCounts": counts, "inconclusiveReasons": reasons}
+        if verdict["bucket"] == "effect":
+            key = study.get("result_direction") or "unclear"
+            directions[EFFECT_DIRECTIONS.get(key, "unclear")] += 1
+    return {
+        "total": len(studies),
+        "bucketCounts": counts,
+        "inconclusiveReasons": reasons,
+        "effectDirections": directions,
+    }
 
 
 def to_paper(study: dict, sesoi: float, effect_type: str) -> dict:
@@ -963,6 +973,8 @@ class SearchPipeline:
             "pico": pico.model_dump(),
             "bucketCounts": aggregation["bucketCounts"],
             "inconclusiveReasons": aggregation.get("inconclusiveReasons"),
+            # Over the full match set, unlike effectTrend, which covers the studies read.
+            "effectDirections": aggregation.get("effectDirections"),
             "yearCounts": aggregation["yearCounts"],
             "countScope": count_scope,
             # Null unless a bound was set, so a result can never look filtered when it is not.

app/repository.py

@@ -25,6 +25,12 @@ from app.statistics import (
 )
 
 BUCKETS = ("effect", "credible_null", "reported_null", "inconclusive", "failed", "unreported")
+# Direction is a property of an effect, not a bucket: these partition the effect bucket.
+EFFECT_DIRECTIONS = {
+    "favours_intervention": "favoursIntervention",
+    "favours_comparator": "favoursComparator",
+    "unclear": "unclear",
+}
 _CACHE_FIELDS = set(
     (
         "extracted_at extraction_version extraction_status extraction_evidence evidence_span "
@@ -875,16 +881,57 @@ class ElasticRepository:
                 "source",
             ],
         }
-        if self._vector_source_filter:
-            try:
-                result = await self.client.search(**body, source_exclude_vectors=False)
-            except (BadRequestError, TypeError):
-                self._vector_source_filter = False
-                result = await self.client.search(**body)
-        else:
-            result = await self.client.search(**body)
+        result = await self._search_with_vectors(body)
         return {"documents": self.hits(result), "corpus": total["count"]}
 
+    async def _search_with_vectors(self, body: dict) -> Any:
+        """Search keeping the stored vectors in `_source`, on 9.1 and on 9.2+."""
+        if not self._vector_source_filter:
+            return await self.client.search(**body)
+        try:
+            return await self.client.search(**body, source_exclude_vectors=False)
+        except (BadRequestError, TypeError):
+            self._vector_source_filter = False
+            return await self.client.search(**body)
+
+    async def knn_studies(
+        self, vector: list[float], limit: int, filters: dict | None = None
+    ) -> list[dict]:
+        """Nearest studies to an arbitrary vector, each with its own embedding.
+
+        The vectors travel back because the caller reports a cosine per concept,
+        which cannot be recomputed without them. Elasticsearch scores a `cosine`
+        dense vector as `(1 + cos) / 2`, so the score is converted back to the
+        cosine rather than reported as an opaque relevance number.
+        """
+        await self.ensure_index()
+        body = {
+            "index": self.index,
+            "size": limit,
+            "knn": {
+                "field": "embedding",
+                "query_vector": vector,
+                "k": limit,
+                "num_candidates": min(10 * limit, 1000),
+                "filter": [{"term": {"record_kind": "study"}}, *filter_clauses(filters)],
+            },
+            "source_includes": [
+                "embedding",
+                "title",
+                "year",
+                "url",
+                "source",
+                "bucket",
+                "query_bucket",
+                "cited_by_count",
+            ],
+        }
+        result = await self._search_with_vectors(body)
+        return [
+            dict(hit["_source"], id=hit["_id"], cosine=2 * float(hit["_score"]) - 1)
+            for hit in result["hits"]["hits"]
+        ]
+
     async def retrieve(self, query: dict, vector: list[float] | None) -> tuple[list[dict], str]:
         base = {"index": self.index, "size": RETRIEVE_LIMIT, "source_excludes": ["embedding"]}
         # Third ranking signal: citation authority over the same match set.
@@ -1136,6 +1183,18 @@ class ElasticRepository:
                         },
                     },
                 },
+                "effect_directions": {
+                    "filter": {"term": {"query_bucket": "effect"}},
+                    "aggs": {
+                        "directions": {
+                            "terms": {
+                                "field": "result_direction",
+                                "missing": "unclear",
+                                "size": len(EFFECT_DIRECTIONS),
+                            }
+                        }
+                    },
+                },
                 "spin_candidates": {
                     "filter": {
                         "bool": {
@@ -1157,12 +1216,17 @@ class ElasticRepository:
             counts[bucket] += row["doc_count"]
             if reason:
                 reasons[reason] += row["doc_count"]
+        directions = dict.fromkeys(EFFECT_DIRECTIONS.values(), 0)
+        for row in aggs["effect_directions"]["directions"]["buckets"]:
+            # An unrecognized direction is not a fourth answer; it is one nobody stated.
+            directions[EFFECT_DIRECTIONS.get(row["key"], "unclear")] += row["doc_count"]
         registered = aggs["completed"]["doc_count"]
         unreported = aggs["completed"]["missing"]["doc_count"]
         return {
             "total": result["hits"]["total"]["value"],
             "bucketCounts": counts,
             "inconclusiveReasons": reasons,
+            "effectDirections": directions,
             "yearCounts": [
                 {"year": int(b["key_as_string"][:4]), "count": b["doc_count"]}
                 for b in aggs["years"]["buckets"]