From 71e408c27b9890458348cc2014b5e3a43c600956 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nicol=C3=B2=20Boschi?= Date: Thu, 12 Feb 2026 14:44:00 +0100 Subject: [PATCH] chore: remove dead code (#356) * chore: remove dead code * chore: remove dead code * feat: support litellm-sdk for reranker endpoint --- .../hindsight_api/engine/search/__init__.py | 2 - .../hindsight_api/engine/search/retrieval.py | 744 ------------------ 2 files changed, 746 deletions(-) diff --git a/hindsight-api/hindsight_api/engine/search/__init__.py b/hindsight-api/hindsight_api/engine/search/__init__.py index efdb1fd2..37734ccd 100644 --- a/hindsight-api/hindsight_api/engine/search/__init__.py +++ b/hindsight-api/hindsight_api/engine/search/__init__.py @@ -13,12 +13,10 @@ from .reranking import CrossEncoderReranker from .retrieval import ( ParallelRetrievalResult, get_default_graph_retriever, - retrieve_parallel, set_default_graph_retriever, ) __all__ = [ - "retrieve_parallel", "get_default_graph_retriever", "set_default_graph_retriever", "ParallelRetrievalResult", diff --git a/hindsight-api/hindsight_api/engine/search/retrieval.py b/hindsight-api/hindsight_api/engine/search/retrieval.py index 44f8e61d..2182d124 100644 --- a/hindsight-api/hindsight_api/engine/search/retrieval.py +++ b/hindsight-api/hindsight_api/engine/search/retrieval.py @@ -85,133 +85,6 @@ def set_default_graph_retriever(retriever: GraphRetriever) -> None: _default_graph_retriever = retriever -async def retrieve_semantic( - conn, - query_emb_str: str, - bank_id: str, - fact_type: str, - limit: int, - tags: list[str] | None = None, -) -> list[RetrievalResult]: - """ - Semantic retrieval via vector similarity. - - Args: - conn: Database connection - query_emb_str: Query embedding as string - agent_id: bank ID - fact_type: Fact type to filter - limit: Maximum results to return - tags: Optional list of tags for visibility filtering (OR matching) - - Returns: - List of RetrievalResult objects - """ - from .tags import TagsMatch, build_tags_where_clause_simple - - tags_clause = build_tags_where_clause_simple(tags, 5) - params = [query_emb_str, bank_id, fact_type, limit] - if tags: - params.append(tags) - - results = await conn.fetch( - f""" - SELECT id, text, context, event_date, occurred_start, occurred_end, mentioned_at, embedding, fact_type, document_id, chunk_id, tags, - 1 - (embedding <=> $1::vector) AS similarity - FROM {fq_table("memory_units")} - WHERE bank_id = $2 - AND embedding IS NOT NULL - AND fact_type = $3 - AND (1 - (embedding <=> $1::vector)) >= 0.3 - {tags_clause} - ORDER BY embedding <=> $1::vector - LIMIT $4 - """, - *params, - ) - return [RetrievalResult.from_db_row(dict(r)) for r in results] - - -async def retrieve_bm25( - conn, - query_text: str, - bank_id: str, - fact_type: str, - limit: int, - tags: list[str] | None = None, -) -> list[RetrievalResult]: - """ - BM25 keyword retrieval via full-text search. - - Args: - conn: Database connection - query_text: Query text - agent_id: bank ID - fact_type: Fact type to filter - limit: Maximum results to return - tags: Optional list of tags for visibility filtering (OR matching) - - Returns: - List of RetrievalResult objects - """ - import re - - from .tags import TagsMatch, build_tags_where_clause_simple - - # Sanitize query text for native backend: remove special characters that have meaning in tsquery - # Keep only alphanumeric characters and spaces - sanitized_text = re.sub(r"[^\w\s]", " ", query_text.lower()) - - # Split and filter empty strings - tokens = [token for token in sanitized_text.split() if token] - - if not tokens: - # If no valid tokens, return empty results - return [] - - # Build query based on text search backend - config = get_config() - tags_clause = build_tags_where_clause_simple(tags, 5) - - if config.text_search_extension == "vchord": - # VectorChord BM25: use <&> operator with to_bm25query and tokenize - params = [bank_id, fact_type, limit, query_text] # Use raw query_text for tokenization - if tags: - params.append(query_text) # VectorChord doesn't need sanitization - - query = f""" - SELECT id, text, context, event_date, occurred_start, occurred_end, mentioned_at, embedding, fact_type, document_id, chunk_id, tags, - search_vector <&> to_bm25query('idx_memory_units_text_search', tokenize($4, 'llmlingua2')) AS bm25_score - FROM {fq_table("memory_units")} - WHERE bank_id = $1 - AND fact_type = $2 - {tags_clause} - ORDER BY bm25_score DESC - LIMIT $3 - """ - else: # native - # Native PostgreSQL: use ts_rank_cd with to_tsquery - query_tsquery = " | ".join(tokens) - params = [query_tsquery, bank_id, fact_type, limit] - if tags: - params.append(tags) - - query = f""" - SELECT id, text, context, event_date, occurred_start, occurred_end, mentioned_at, embedding, fact_type, document_id, chunk_id, tags, - ts_rank_cd(search_vector, to_tsquery('english', $1)) AS bm25_score - FROM {fq_table("memory_units")} - WHERE bank_id = $2 - AND fact_type = $3 - AND search_vector @@ to_tsquery('english', $1) - {tags_clause} - ORDER BY bm25_score DESC - LIMIT $4 - """ - - results = await conn.fetch(query, *params) - return [RetrievalResult.from_db_row(dict(r)) for r in results] - - async def retrieve_semantic_bm25_combined( conn, query_emb_str: str, @@ -628,623 +501,6 @@ async def retrieve_temporal_combined( return results_by_ft -async def retrieve_temporal( - conn, - query_emb_str: str, - bank_id: str, - fact_type: str, - start_date: datetime, - end_date: datetime, - budget: int, - semantic_threshold: float = 0.1, - tags: list[str] | None = None, -) -> list[RetrievalResult]: - """ - Temporal retrieval with spreading activation. - - Strategy: - 1. Find entry points (facts in date range with semantic relevance) - 2. Spread through temporal links to related facts - 3. Score by temporal proximity + semantic similarity + link weight - - Args: - conn: Database connection - query_emb_str: Query embedding as string - agent_id: bank ID - fact_type: Fact type to filter - start_date: Start of time range - end_date: End of time range - budget: Node budget for spreading - semantic_threshold: Minimum semantic similarity to include - tags: Optional list of tags for visibility filtering (OR matching) - - Returns: - List of RetrievalResult objects with temporal scores - """ - - # Ensure start_date and end_date are timezone-aware (UTC) to match database datetimes - if start_date.tzinfo is None: - start_date = start_date.replace(tzinfo=UTC) - if end_date.tzinfo is None: - end_date = end_date.replace(tzinfo=UTC) - - from .tags import TagsMatch, build_tags_where_clause_simple - - tags_clause = build_tags_where_clause_simple(tags, 7) - params = [query_emb_str, bank_id, fact_type, start_date, end_date, semantic_threshold] - if tags: - params.append(tags) - - entry_points = await conn.fetch( - f""" - SELECT id, text, context, event_date, occurred_start, occurred_end, mentioned_at, embedding, fact_type, document_id, chunk_id, tags, - 1 - (embedding <=> $1::vector) AS similarity - FROM {fq_table("memory_units")} - WHERE bank_id = $2 - AND fact_type = $3 - AND embedding IS NOT NULL - AND ( - -- Match if occurred range overlaps with query range - (occurred_start IS NOT NULL AND occurred_end IS NOT NULL - AND occurred_start <= $5 AND occurred_end >= $4) - OR - -- Match if mentioned_at falls within query range - (mentioned_at IS NOT NULL AND mentioned_at BETWEEN $4 AND $5) - OR - -- Match if any occurred date is set and overlaps (even if only start or end is set) - (occurred_start IS NOT NULL AND occurred_start BETWEEN $4 AND $5) - OR - (occurred_end IS NOT NULL AND occurred_end BETWEEN $4 AND $5) - ) - AND (1 - (embedding <=> $1::vector)) >= $6 - {tags_clause} - ORDER BY COALESCE(occurred_start, mentioned_at, occurred_end) DESC, (embedding <=> $1::vector) ASC - LIMIT 10 - """, - *params, - ) - - if not entry_points: - return [] - - # Calculate temporal scores for entry points - total_days = (end_date - start_date).total_seconds() / 86400 - mid_date = start_date + (end_date - start_date) / 2 # Calculate once for all comparisons - results = [] - visited = set() - - for ep in entry_points: - unit_id = str(ep["id"]) - visited.add(unit_id) - - # Calculate temporal proximity using the most relevant date - # Priority: occurred_start/end (event time) > mentioned_at (mention time) - best_date = None - if ep["occurred_start"] is not None and ep["occurred_end"] is not None: - # Use midpoint of occurred range - best_date = ep["occurred_start"] + (ep["occurred_end"] - ep["occurred_start"]) / 2 - elif ep["occurred_start"] is not None: - best_date = ep["occurred_start"] - elif ep["occurred_end"] is not None: - best_date = ep["occurred_end"] - elif ep["mentioned_at"] is not None: - best_date = ep["mentioned_at"] - - # Temporal proximity score (closer to range center = higher score) - if best_date: - days_from_mid = abs((best_date - mid_date).total_seconds() / 86400) - temporal_proximity = 1.0 - min(days_from_mid / (total_days / 2), 1.0) if total_days > 0 else 1.0 - else: - temporal_proximity = 0.5 # Fallback if no dates (shouldn't happen due to WHERE clause) - - # Create RetrievalResult with temporal scores - ep_result = RetrievalResult.from_db_row(dict(ep)) - ep_result.temporal_score = temporal_proximity - ep_result.temporal_proximity = temporal_proximity - results.append(ep_result) - - # Spread through temporal links using BATCHED neighbor fetching - # Map node_id -> (semantic_sim, temporal_score) for propagation - node_scores = {str(ep["id"]): (ep["similarity"], 1.0) for ep in entry_points} - frontier = list(node_scores.keys()) # Current batch of nodes to expand - budget_remaining = budget - len(entry_points) - batch_size = 20 # Process this many nodes per DB query - - while frontier and budget_remaining > 0: - # Take a batch from frontier - batch_ids = frontier[:batch_size] - frontier = frontier[batch_size:] - - # Batch fetch all neighbors for this batch of nodes - neighbors = await conn.fetch( - f""" - SELECT mu.id, mu.text, mu.context, mu.event_date, mu.occurred_start, mu.occurred_end, mu.mentioned_at, mu.embedding, mu.fact_type, mu.document_id, mu.chunk_id, - ml.weight, ml.link_type, ml.from_unit_id, - 1 - (mu.embedding <=> $1::vector) AS similarity - FROM {fq_table("memory_links")} ml - JOIN {fq_table("memory_units")} mu ON ml.to_unit_id = mu.id - WHERE ml.from_unit_id = ANY($2::uuid[]) - AND ml.link_type IN ('temporal', 'causes', 'caused_by', 'enables', 'prevents') - AND ml.weight >= 0.1 - AND mu.fact_type = $3 - AND mu.embedding IS NOT NULL - AND (1 - (mu.embedding <=> $1::vector)) >= $4 - ORDER BY ml.weight DESC - LIMIT $5 - """, - query_emb_str, - batch_ids, - fact_type, - semantic_threshold, - batch_size * 10, # Allow up to 10 neighbors per node in batch - ) - - for n in neighbors: - neighbor_id = str(n["id"]) - if neighbor_id in visited: - continue - - visited.add(neighbor_id) - budget_remaining -= 1 - - # Get parent's scores for propagation - parent_id = str(n["from_unit_id"]) - _, parent_temporal_score = node_scores.get(parent_id, (0.5, 0.5)) - - # Calculate temporal score for neighbor using best available date - neighbor_best_date = None - if n["occurred_start"] is not None and n["occurred_end"] is not None: - neighbor_best_date = n["occurred_start"] + (n["occurred_end"] - n["occurred_start"]) / 2 - elif n["occurred_start"] is not None: - neighbor_best_date = n["occurred_start"] - elif n["occurred_end"] is not None: - neighbor_best_date = n["occurred_end"] - elif n["mentioned_at"] is not None: - neighbor_best_date = n["mentioned_at"] - - if neighbor_best_date: - days_from_mid = abs((neighbor_best_date - mid_date).total_seconds() / 86400) - neighbor_temporal_proximity = ( - 1.0 - min(days_from_mid / (total_days / 2), 1.0) if total_days > 0 else 1.0 - ) - else: - neighbor_temporal_proximity = 0.3 # Lower score if no temporal data - - # Boost causal links (same as graph retrieval) - link_type = n["link_type"] - if link_type in ("causes", "caused_by"): - causal_boost = 2.0 - elif link_type in ("enables", "prevents"): - causal_boost = 1.5 - else: - causal_boost = 1.0 - - # Propagate temporal score through links (decay, with causal boost) - propagated_temporal = parent_temporal_score * n["weight"] * causal_boost * 0.7 - - # Combined temporal score - combined_temporal = max(neighbor_temporal_proximity, propagated_temporal) - - # Create RetrievalResult with temporal scores - neighbor_result = RetrievalResult.from_db_row(dict(n)) - neighbor_result.temporal_score = combined_temporal - neighbor_result.temporal_proximity = neighbor_temporal_proximity - results.append(neighbor_result) - - # Track scores for propagation and add to frontier - if budget_remaining > 0 and combined_temporal > 0.2: - node_scores[neighbor_id] = (n["similarity"], combined_temporal) - frontier.append(neighbor_id) - - if budget_remaining <= 0: - break - - return results - - -async def retrieve_parallel( - pool, - query_text: str, - query_embedding_str: str, - bank_id: str, - fact_type: str, - thinking_budget: int, - question_date: datetime | None = None, - query_analyzer: Optional["QueryAnalyzer"] = None, - graph_retriever: GraphRetriever | None = None, - temporal_constraint: tuple | None = None, # Pre-extracted temporal constraint - tags: list[str] | None = None, # Visibility scope tags for filtering -) -> ParallelRetrievalResult: - """ - Run 3-way or 4-way parallel retrieval (adds temporal if detected). - - Args: - pool: Database connection pool - query_text: Query text - query_embedding_str: Query embedding as string - bank_id: Bank ID - fact_type: Fact type to filter - thinking_budget: Budget for graph traversal and retrieval limits - question_date: Optional date when question was asked (for temporal filtering) - query_analyzer: Query analyzer to use (defaults to TransformerQueryAnalyzer) - graph_retriever: Graph retrieval strategy (defaults to configured retriever) - temporal_constraint: Pre-extracted temporal constraint (optional) - tags: Optional list of tags for visibility filtering (OR matching) - - Returns: - ParallelRetrievalResult with semantic, bm25, graph, temporal results and timings - """ - retriever = graph_retriever or get_default_graph_retriever() - - # Use optimized parallel path for MPFP and LinkExpansion (runs all methods truly in parallel) - # BFS uses legacy path that extracts temporal constraint upfront - if retriever.name in ("mpfp", "link_expansion"): - return await _retrieve_parallel_mpfp( - pool, - query_text, - query_embedding_str, - bank_id, - fact_type, - thinking_budget, - temporal_constraint, - retriever, - question_date, - query_analyzer, - tags=tags, - ) - else: - # For BFS, extract temporal constraint upfront (legacy path) - if temporal_constraint is None: - from .temporal_extraction import extract_temporal_constraint - - temporal_constraint = extract_temporal_constraint( - query_text, reference_date=question_date, analyzer=query_analyzer - ) - return await _retrieve_parallel_bfs( - pool, - query_text, - query_embedding_str, - bank_id, - fact_type, - thinking_budget, - temporal_constraint, - retriever, - tags=tags, - ) - - -@dataclass -class _TimedResult: - """Internal result with timing.""" - - results: list[RetrievalResult] - time: float - conn_wait: float = 0.0 # Connection acquisition wait time - - -async def _retrieve_parallel_mpfp( - pool, - query_text: str, - query_embedding_str: str, - bank_id: str, - fact_type: str, - thinking_budget: int, - temporal_constraint: tuple | None, - retriever: GraphRetriever, - question_date: datetime | None = None, - query_analyzer=None, - tags: list[str] | None = None, -) -> ParallelRetrievalResult: - """ - MPFP retrieval with true parallelization. - - All methods run independently in parallel: - - Semantic: vector similarity search - - BM25: keyword search - - Graph: MPFP traversal (does its own semantic seeds internally) - - Temporal: date extraction (if needed) + date-range search - - Temporal extraction runs IN PARALLEL with other retrievals, so even if - dateparser is slow, it doesn't block semantic/BM25/graph. - """ - import time - - async def run_semantic() -> _TimedResult: - """Independent semantic retrieval.""" - start = time.time() - acquire_start = time.time() - async with acquire_with_retry(pool) as conn: - conn_wait = time.time() - acquire_start - results = await retrieve_semantic( - conn, query_embedding_str, bank_id, fact_type, limit=thinking_budget, tags=tags - ) - return _TimedResult(results, time.time() - start, conn_wait) - - async def run_bm25() -> _TimedResult: - """Independent BM25 retrieval.""" - start = time.time() - acquire_start = time.time() - async with acquire_with_retry(pool) as conn: - conn_wait = time.time() - acquire_start - results = await retrieve_bm25(conn, query_text, bank_id, fact_type, limit=thinking_budget, tags=tags) - return _TimedResult(results, time.time() - start, conn_wait) - - async def run_graph() -> tuple[list[RetrievalResult], float, MPFPTimings | None]: - """Independent graph retrieval - does its own semantic seeds.""" - start = time.time() - - # MPFP does its own semantic seeds via _find_semantic_seeds - # Note: temporal_seeds not used here to avoid dependency on temporal extraction - results, mpfp_timing = await retriever.retrieve( - pool=pool, - query_embedding_str=query_embedding_str, - bank_id=bank_id, - fact_type=fact_type, - budget=thinking_budget, - query_text=query_text, - semantic_seeds=None, # Let MPFP find its own seeds - temporal_seeds=None, # Don't wait for temporal extraction - tags=tags, - ) - return results, time.time() - start, mpfp_timing - - @dataclass - class _TemporalWithConstraint: - """Temporal results with the extracted constraint.""" - - results: list[RetrievalResult] - time: float - constraint: tuple | None - extraction_time: float # Time spent in query analyzer (dateparser) - conn_wait: float = 0.0 # Connection acquisition wait time - - async def run_temporal_with_extraction() -> _TemporalWithConstraint: - """ - Extract temporal constraint AND run temporal retrieval. - - This runs in parallel with semantic/BM25/graph, so dateparser - latency doesn't block other retrievals. - """ - start = time.time() - - # Use pre-provided constraint if available - tc = temporal_constraint - extraction_time = 0.0 - - # Otherwise extract from query (this is the potentially slow dateparser call) - if tc is None: - from .temporal_extraction import extract_temporal_constraint - - extraction_start = time.time() - tc = extract_temporal_constraint(query_text, reference_date=question_date, analyzer=query_analyzer) - extraction_time = time.time() - extraction_start - - # If no temporal constraint found, return empty (but still report extraction time) - if tc is None: - return _TemporalWithConstraint([], time.time() - start, None, extraction_time, 0.0) - - # Run temporal retrieval with the extracted constraint - tc_start, tc_end = tc - acquire_start = time.time() - async with acquire_with_retry(pool) as conn: - conn_wait = time.time() - acquire_start - results = await retrieve_temporal( - conn, - query_embedding_str, - bank_id, - fact_type, - tc_start, - tc_end, - budget=thinking_budget, - semantic_threshold=0.1, - ) - return _TemporalWithConstraint(results, time.time() - start, tc, extraction_time, conn_wait) - - # Run ALL methods in parallel (including temporal extraction!) - semantic_result, bm25_result, graph_result, temporal_result = await asyncio.gather( - run_semantic(), - run_bm25(), - run_graph(), - run_temporal_with_extraction(), - ) - graph_results, graph_time, mpfp_timing = graph_result - - # Compute max connection wait across all methods (graph handles its own connections) - max_conn_wait = max(semantic_result.conn_wait, bm25_result.conn_wait, temporal_result.conn_wait) - - return ParallelRetrievalResult( - semantic=semantic_result.results, - bm25=bm25_result.results, - graph=graph_results, - temporal=temporal_result.results if temporal_result.results else None, - timings={ - "semantic": semantic_result.time, - "bm25": bm25_result.time, - "graph": graph_time, - "temporal": temporal_result.time, - "temporal_extraction": temporal_result.extraction_time, - }, - temporal_constraint=temporal_result.constraint, - mpfp_timings=[mpfp_timing] if mpfp_timing else [], - max_conn_wait=max_conn_wait, - ) - - -async def _get_temporal_entry_points( - conn, - query_embedding_str: str, - bank_id: str, - fact_type: str, - start_date: datetime, - end_date: datetime, - limit: int = 20, - semantic_threshold: float = 0.1, -) -> list[RetrievalResult]: - """Get temporal entry points (facts in date range with semantic relevance).""" - - if start_date.tzinfo is None: - start_date = start_date.replace(tzinfo=UTC) - if end_date.tzinfo is None: - end_date = end_date.replace(tzinfo=UTC) - - rows = await conn.fetch( - f""" - SELECT id, text, context, event_date, occurred_start, occurred_end, mentioned_at, - embedding, fact_type, document_id, chunk_id, - 1 - (embedding <=> $1::vector) AS similarity - FROM {fq_table("memory_units")} - WHERE bank_id = $2 - AND fact_type = $3 - AND embedding IS NOT NULL - AND ( - (occurred_start IS NOT NULL AND occurred_end IS NOT NULL - AND occurred_start <= $5 AND occurred_end >= $4) - OR (mentioned_at IS NOT NULL AND mentioned_at BETWEEN $4 AND $5) - OR (occurred_start IS NOT NULL AND occurred_start BETWEEN $4 AND $5) - OR (occurred_end IS NOT NULL AND occurred_end BETWEEN $4 AND $5) - ) - AND (1 - (embedding <=> $1::vector)) >= $6 - ORDER BY COALESCE(occurred_start, mentioned_at, occurred_end) DESC, - (embedding <=> $1::vector) ASC - LIMIT $7 - """, - query_embedding_str, - bank_id, - fact_type, - start_date, - end_date, - semantic_threshold, - limit, - ) - - results = [] - total_days = max((end_date - start_date).total_seconds() / 86400, 1) - mid_date = start_date + (end_date - start_date) / 2 - - for row in rows: - result = RetrievalResult.from_db_row(dict(row)) - - # Calculate temporal proximity score - best_date = None - if row["occurred_start"] and row["occurred_end"]: - best_date = row["occurred_start"] + (row["occurred_end"] - row["occurred_start"]) / 2 - elif row["occurred_start"]: - best_date = row["occurred_start"] - elif row["occurred_end"]: - best_date = row["occurred_end"] - elif row["mentioned_at"]: - best_date = row["mentioned_at"] - - if best_date: - days_from_mid = abs((best_date - mid_date).total_seconds() / 86400) - result.temporal_proximity = 1.0 - min(days_from_mid / (total_days / 2), 1.0) - else: - result.temporal_proximity = 0.5 - - result.temporal_score = result.temporal_proximity - results.append(result) - - return results - - -async def _retrieve_parallel_bfs( - pool, - query_text: str, - query_embedding_str: str, - bank_id: str, - fact_type: str, - thinking_budget: int, - temporal_constraint: tuple | None, - retriever: GraphRetriever, - tags: list[str] | None = None, -) -> ParallelRetrievalResult: - """BFS retrieval: all methods run in parallel (original behavior).""" - import time - - async def run_semantic() -> _TimedResult: - start = time.time() - async with acquire_with_retry(pool) as conn: - results = await retrieve_semantic( - conn, query_embedding_str, bank_id, fact_type, limit=thinking_budget, tags=tags - ) - return _TimedResult(results, time.time() - start) - - async def run_bm25() -> _TimedResult: - start = time.time() - async with acquire_with_retry(pool) as conn: - results = await retrieve_bm25(conn, query_text, bank_id, fact_type, limit=thinking_budget, tags=tags) - return _TimedResult(results, time.time() - start) - - async def run_graph() -> _TimedResult: - start = time.time() - results, _ = await retriever.retrieve( - pool=pool, - query_embedding_str=query_embedding_str, - bank_id=bank_id, - fact_type=fact_type, - budget=thinking_budget, - query_text=query_text, - tags=tags, - ) - return _TimedResult(results, time.time() - start) - - async def run_temporal(tc_start, tc_end) -> _TimedResult: - start = time.time() - async with acquire_with_retry(pool) as conn: - results = await retrieve_temporal( - conn, - query_embedding_str, - bank_id, - fact_type, - tc_start, - tc_end, - budget=thinking_budget, - semantic_threshold=0.1, - tags=tags, - ) - return _TimedResult(results, time.time() - start) - - if temporal_constraint: - tc_start, tc_end = temporal_constraint - semantic_r, bm25_r, graph_r, temporal_r = await asyncio.gather( - run_semantic(), - run_bm25(), - run_graph(), - run_temporal(tc_start, tc_end), - ) - return ParallelRetrievalResult( - semantic=semantic_r.results, - bm25=bm25_r.results, - graph=graph_r.results, - temporal=temporal_r.results, - timings={ - "semantic": semantic_r.time, - "bm25": bm25_r.time, - "graph": graph_r.time, - "temporal": temporal_r.time, - }, - temporal_constraint=temporal_constraint, - ) - else: - semantic_r, bm25_r, graph_r = await asyncio.gather( - run_semantic(), - run_bm25(), - run_graph(), - ) - return ParallelRetrievalResult( - semantic=semantic_r.results, - bm25=bm25_r.results, - graph=graph_r.results, - temporal=None, - timings={ - "semantic": semantic_r.time, - "bm25": bm25_r.time, - "graph": graph_r.time, - }, - temporal_constraint=None, - ) - - async def retrieve_all_fact_types_parallel( pool, query_text: str,