From 97c2e1710c157e39cae2861a6d2ce670f4dfbe69 Mon Sep 17 00:00:00 2001 From: Hitoshi-Hub Date: Sat, 18 Apr 2026 12:09:21 +0300 Subject: [PATCH 1/2] =?UTF-8?q?=D1=81=D0=B4=D0=B5=D0=BB=D0=B0=D0=BB=20?= =?UTF-8?q?=D0=BF=D0=B5=D1=80=D0=B2=D1=8B=D0=B5=203=20=D0=B7=D0=B0=D0=B4?= =?UTF-8?q?=D0=B0=D1=87=D0=B8=20=D0=B8=D0=B7=20todo=5Fpeople.md?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- search/main.py | 132 ++++++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 125 insertions(+), 7 deletions(-) diff --git a/search/main.py b/search/main.py index c5063aa..6a0df49 100644 --- a/search/main.py +++ b/search/main.py @@ -270,6 +270,111 @@ async def qdrant_search( return response.points +async def qdrant_search_dense_only( + client: AsyncQdrantClient, + dense_vector: list[float], + question_data: Question, +) -> Any | None: + must_conditions: [] + + if question_data.date_range: + must_conditions.append( + models.FieldCondition( + key="metadata.start", + range=models.Range( + gte=question_data.date_range.from_, + lte=question_data.date_range.to_, + ), + ) + ) + + if question_data.asker: + must_conditions.append( + models.FieldCondition( + key="metadata.participants", + match=models.MatchValue(value=question_data.asker), + ) + ) + + search_filter = models.Filter(must=must_conditions) if must_conditions else None + + response = await client.query_points( + collection_name=QDRANT_COLLECTION_NAME, + prefetch=[ + models.Prefetch( + query=dense_vector, + using=QDRANT_DENSE_VECTOR_NAME, + limit=DENSE_PREFETCH_K, + filter=search_filter, + ), + ], + query=models.FusionQuery(fusion=models.Fusion.RRF), + limit=RETRIEVE_K, + with_payload=True, + ) + + if not response.points: + return None + + return response.points + + +def collect_query_variants(question: Question) -> list[str]: + variants: list[str] = [] + seen: set[str] = set() + + def add_query(text: str | None) -> None: + if text is None: + return + normalized = text.strip() + if not normalized: + return + if normalized in seen: + return + seen.add(normalized) + variants.append(normalized) + + add_query(question.search_text) + add_query(question.text) + + for variant in question.variants or []: + add_query(variant) + + return variants + + +def collect_hyde_queries(question: Question, base_queries: list[str]) -> list[str]: + hyde_queries: list[str] = [] + seen: set[str] = set(base_queries) + + for hyde_query in question.hyde or []: + normalized = hyde_query.strip() + if not normalized: + continue + if normalized in seen: + continue + seen.add(normalized) + hyde_queries.append(normalized) + + return hyde_queries + + +def deduplicate_points(points: list[Any]) -> list[Any]: + unique_points: list[Any] = [] + seen_ids: set[str] = set() + + for point in points: + point_id = str(getattr(point, "id", "")) + if not point_id: + continue + if point_id in seen_ids: + continue + seen_ids.add(point_id) + unique_points.append(point) + + return unique_points + + def extract_message_ids(point: Any) -> list[str]: payload = point.payload or {} metadata = payload.get("metadata") or {} @@ -334,18 +439,31 @@ async def health() -> dict[str, str]: @app.post("/search", response_model=SearchAPIResponse) async def search(payload: SearchAPIRequest) -> SearchAPIResponse: - query = payload.question.text.strip() - if not query: - raise HTTPException(status_code=400, detail="question.text is required") + queries = collect_query_variants(payload.question) + if not queries: + raise HTTPException(status_code=400, detail="question.search_text or question.text is required") + hyde_queries = collect_hyde_queries(payload.question, queries) + query = queries[0] client: httpx.AsyncClient = app.state.http qdrant: AsyncQdrantClient = app.state.qdrant - dense_vector = await embed_dense(client, query) - sparse_vector = await embed_sparse(query) - best_points = await qdrant_search(qdrant, dense_vector, sparse_vector, payload.question) + all_points: list[Any] = [] + for query_variant in queries: + dense_vector = await embed_dense(client, query_variant) + sparse_vector = await embed_sparse(query_variant) + points = await qdrant_search(qdrant, dense_vector, sparse_vector, payload.question) + if points: + all_points.extend(list(points)) - if best_points is None: + for hyde_query in hyde_queries: + hyde_dense_vector = await embed_dense(client, hyde_query) + hyde_points = await qdrant_search_dense_only(qdrant, hyde_dense_vector, payload.question) + if hyde_points: + all_points.extend(list(hyde_points)) + + best_points = deduplicate_points(all_points) + if not best_points: return SearchAPIResponse(results=[]) best_points = await rerank_points(client, query, list(best_points)) From bb8f4f79e4451f6ab0ecc894602956147ebf6924 Mon Sep 17 00:00:00 2001 From: Hitoshi-Hub Date: Sat, 18 Apr 2026 12:16:57 +0300 Subject: [PATCH 2/2] =?UTF-8?q?=D0=B4=D0=BE=D0=B4=D0=B5=D0=BB=D0=B0=D0=BB?= =?UTF-8?q?=20=D0=B2=D1=81=D0=B5=20=D0=B7=D0=B0=D0=B4=D0=B0=D1=87=D0=B8=20?= =?UTF-8?q?=D0=BF=D0=BE=20P0=20=D0=B8=20=D0=B4=D0=BE=D0=B1=D0=B0=D0=B2?= =?UTF-8?q?=D0=B8=D0=BB=20=D0=B2=D1=81=D0=B5=20=D0=B2=20changes.md=20(?= =?UTF-8?q?=D0=BF=D0=BE=20=D0=BF=D1=80=D0=BE=D0=BC=D0=BF=D1=82=D1=83)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .ai_update/changes.md | 27 +++++++++++++++ search/main.py | 76 ++++++++++++++++++++++++++++++++++++------- 2 files changed, 92 insertions(+), 11 deletions(-) create mode 100644 .ai_update/changes.md diff --git a/.ai_update/changes.md b/.ai_update/changes.md new file mode 100644 index 0000000..928a88a --- /dev/null +++ b/.ai_update/changes.md @@ -0,0 +1,27 @@ +# AI Update Log + +## Scope +- File: `search/main.py` +- Purpose: fixed and extended retrieval/rerank pipeline according to TODO items. + +## Done Changes +- Switched base query selection to `question.search_text` with fallback to `question.text`. +- Added support for `question.variants` as additional query variants in retrieval. +- Added support for `question.hyde` as additional dense-only queries. +- Added support for `question.keywords` as the primary source for sparse query text with fallback to current query variant. +- Stopped losing retrieval candidates after rerank: rerank is applied to head (`RERANK_LIMIT`), tail candidates are preserved. +- Added deduplication of retrieval points by Qdrant point id before rerank. +- Implemented score aggregation by `message_id` (sum of chunk scores mapped to same message). +- Limited final response to `top-50` message ids via `FINAL_TOP_K = 50`. +- Final output message ids are now selected from aggregated scores (sorted by score desc, tie-break by message_id). + +## Notes +- Earlier step introduced direct `message_ids` deduplication before response. +- Current logic supersedes this by ranking and selecting unique `message_id` values from aggregated scores. + +## Verification +- Syntax check passed after each main change: `python -m py_compile search/main.py`. + +## How To Use This Log +- Treat this file as the source of truth for already completed `search/main.py` tasks. +- On next tasks, read this file first to avoid duplicate edits. diff --git a/search/main.py b/search/main.py index 6a0df49..d7f1c21 100644 --- a/search/main.py +++ b/search/main.py @@ -175,6 +175,7 @@ DENSE_PREFETCH_K = 10 SPRASE_PREFETCH_K = 30 RETRIEVE_K = 20 RERANK_LIMIT = 10 +FINAL_TOP_K = 50 async def embed_dense(client: httpx.AsyncClient, text: str) -> list[float]: # Dense endpoint ожидает OpenAI-compatible body с input как списком строк. @@ -359,6 +360,25 @@ def collect_hyde_queries(question: Question, base_queries: list[str]) -> list[st return hyde_queries +def build_sparse_query_text(question: Question, fallback_query: str) -> str: + keywords: list[str] = [] + seen: set[str] = set() + + for keyword in question.keywords or []: + normalized = keyword.strip() + if not normalized: + continue + if normalized in seen: + continue + seen.add(normalized) + keywords.append(normalized) + + if keywords: + return " ".join(keywords) + + return fallback_query + + def deduplicate_points(points: list[Any]) -> list[Any]: unique_points: list[Any] = [] seen_ids: set[str] = set() @@ -375,6 +395,13 @@ def deduplicate_points(points: list[Any]) -> list[Any]: return unique_points +def extract_point_score(point: Any) -> float: + score = getattr(point, "score", 0.0) + if score is None: + return 0.0 + return float(score) + + def extract_message_ids(point: Any) -> list[str]: payload = point.payload or {} metadata = payload.get("metadata") or {} @@ -414,21 +441,49 @@ async def rerank_points( client: httpx.AsyncClient, query: str, points: list[Any], -) -> list[Any]: - rerank_candidates = points[:10] +) -> list[tuple[Any, float]]: + rerank_candidates = points[:RERANK_LIMIT] + tail_candidates = points[RERANK_LIMIT:] rerank_targets = [point.payload.get("page_content") for point in rerank_candidates] scores = await get_rerank_scores(client, query, rerank_targets) reranked_candidates = [ - point - for _, point in sorted( + (point, float(score)) + for score, point in sorted( zip(scores, rerank_candidates, strict=True), key=lambda item: item[0], reverse=True, ) ] + tail_with_scores = [ + (point, extract_point_score(point)) + for _, point in sorted( + [(extract_point_score(point), point) for point in tail_candidates], + key=lambda item: item[0], + reverse=True, + ) + ] - return reranked_candidates + return reranked_candidates + tail_with_scores + + +def aggregate_message_scores(scored_points: list[tuple[Any, float]]) -> dict[str, float]: + aggregated_scores: dict[str, float] = {} + + for point, point_score in scored_points: + point_message_ids = set(extract_message_ids(point)) + for message_id in point_message_ids: + aggregated_scores[message_id] = aggregated_scores.get(message_id, 0.0) + point_score + + return aggregated_scores + + +def select_top_message_ids(aggregated_scores: dict[str, float], limit: int) -> list[str]: + sorted_items = sorted( + aggregated_scores.items(), + key=lambda item: (-item[1], item[0]), + ) + return [message_id for message_id, _ in sorted_items[:limit]] # Ваш сервис должен имплементировать оба этих метода @@ -451,7 +506,8 @@ async def search(payload: SearchAPIRequest) -> SearchAPIResponse: all_points: list[Any] = [] for query_variant in queries: dense_vector = await embed_dense(client, query_variant) - sparse_vector = await embed_sparse(query_variant) + sparse_query_text = build_sparse_query_text(payload.question, query_variant) + sparse_vector = await embed_sparse(sparse_query_text) points = await qdrant_search(qdrant, dense_vector, sparse_vector, payload.question) if points: all_points.extend(list(points)) @@ -466,11 +522,9 @@ async def search(payload: SearchAPIRequest) -> SearchAPIResponse: if not best_points: return SearchAPIResponse(results=[]) - best_points = await rerank_points(client, query, list(best_points)) - - message_ids: list[str] = [] - for point in best_points: - message_ids += extract_message_ids(point) + scored_points = await rerank_points(client, query, list(best_points)) + aggregated_scores = aggregate_message_scores(scored_points) + message_ids = select_top_message_ids(aggregated_scores, FINAL_TOP_K) return SearchAPIResponse( results=[SearchAPIItem(message_ids=message_ids)]