135 lines
4.9 KiB
Markdown
135 lines
4.9 KiB
Markdown
# M8.5 — Hybrid query worker: parallel retrieval + fusion
|
||
|
||
| Field | Value |
|
||
|---|---|
|
||
| Phase | M8 — Hybrid Search |
|
||
| Size | L — 2–3 days |
|
||
| Status | ✅ COMPLETE |
|
||
| Flags | — |
|
||
| Spec | inlined below |
|
||
| Blocks | M8.6, M8.7 |
|
||
| Depends | M8.1 (OpenSearch running), M8.2 (dual-write), M8.3 (query optimizer), M8.4 (RRF) |
|
||
|
||
## Goal
|
||
|
||
`HybridQueryWorker` orchestrates parallel retrieval from pgvector and OpenSearch, applies RRF fusion, and returns ranked results with score breakdown and latency metrics. It replaces `QueryWorker` as the primary query engine behind `GET /memory/query`.
|
||
|
||
## Design
|
||
|
||
**Execution flow:**
|
||
1. Call `QueryOptimizer::optimize_query()` → get `QueryContext` with `SearchStrategy`.
|
||
2. Generate embedding via `EmbeddingsClient::embed()`.
|
||
3. Execute strategy:
|
||
- **Hybrid:** `tokio::try_join!` pgvector top-50 + OpenSearch top-50. Fuse with RRF.
|
||
- **LexicalFirst:** OpenSearch top-200 → extract IDs → pgvector `WHERE id IN (...)` top-10.
|
||
- **SemanticOnly:** pgvector top-50 (fallback if OpenSearch unavailable).
|
||
- **LexicalOnly:** OpenSearch top-50 (fallback if embedding model unavailable).
|
||
4. Build response with score breakdown.
|
||
|
||
**Critical: use actual VectorStore API.**
|
||
|
||
The existing `VectorStore` exposes `search_l1(project, &embedding, limit)` and `search_l2(project, &embedding)`. The worker must call these — not invented methods.
|
||
|
||
For the cascading strategy (`LexicalFirst`), a new `search_l1_by_ids(project, &embedding, limit, &[chunk_id])` method is needed on `VectorStore`. This is a filtered pgvector query:
|
||
```sql
|
||
SELECT id, 1 - (embedding <=> $1) as score, text, source
|
||
FROM chunks
|
||
WHERE project = $2 AND id = ANY($3)
|
||
ORDER BY embedding <=> $1
|
||
LIMIT $4
|
||
```
|
||
|
||
**OpenSearch query:**
|
||
```json
|
||
{
|
||
"size": 50,
|
||
"query": {
|
||
"bool": {
|
||
"must": [{
|
||
"multi_match": {
|
||
"query": "...",
|
||
"fields": ["content^2", "section_title^1.5", "breadcrumb", "source"],
|
||
"type": "best_fields",
|
||
"fuzziness": "AUTO"
|
||
}
|
||
}],
|
||
"filter": [
|
||
{"term": {"project_id": "..."}},
|
||
{"terms": {"level": ["L0", "L1", "L2"]}}
|
||
]
|
||
}
|
||
}
|
||
}
|
||
```
|
||
|
||
**Response struct:**
|
||
```rust
|
||
pub struct HybridQueryResponse {
|
||
pub query: String,
|
||
pub project: String,
|
||
pub search_strategy: String,
|
||
pub results: Vec<HybridQueryResult>,
|
||
pub metrics: QueryMetrics,
|
||
}
|
||
|
||
pub struct HybridQueryResult {
|
||
pub id: String,
|
||
pub text: String,
|
||
pub source: String,
|
||
pub level: String,
|
||
pub breadcrumb: Vec<String>,
|
||
pub final_score: f32,
|
||
pub semantic_rank: Option<usize>,
|
||
pub lexical_rank: Option<usize>,
|
||
pub fusion_method: String,
|
||
}
|
||
|
||
pub struct QueryMetrics {
|
||
pub total_time_ms: u128,
|
||
pub semantic_time_ms: Option<u128>,
|
||
pub lexical_time_ms: Option<u128>,
|
||
pub fusion_time_ms: u128,
|
||
pub semantic_candidates: Option<usize>,
|
||
pub lexical_candidates: Option<usize>,
|
||
pub final_count: usize,
|
||
}
|
||
```
|
||
|
||
## Steps
|
||
|
||
1. Add `search_l1_by_ids()` to `VectorStore` (new SQL query with `id = ANY($3)`).
|
||
2. Make `OpenSearchClient::lexical_search()` public.
|
||
3. Implement `HybridQueryWorker::new()` taking `VectorStore`, `EmbeddingsClient`, `Option<OpenSearchClient>`.
|
||
4. Implement `query()` method with strategy dispatch.
|
||
5. Implement `retrieve_hybrid()` using `tokio::try_join!`.
|
||
6. Implement `retrieve_cascading()` (2-stage).
|
||
7. Implement fallback methods (`retrieve_semantic()`, `retrieve_lexical()`).
|
||
8. Wire RRF fusion into the result pipeline.
|
||
9. Build response with per-result rank tracking.
|
||
10. Write integration tests with mock VectorStore + mock OpenSearch.
|
||
|
||
## Acceptance
|
||
|
||
1. Hybrid query returns results from **both** engines — check `semantic_rank` and `lexical_rank` are both `Some` for documents appearing in both lists.
|
||
2. Cascading query's pgvector call receives only IDs from the OpenSearch narrowing step — verify with query log or mock.
|
||
3. If `OpenSearchClient` is `None`, strategy automatically falls back to `SemanticOnly`.
|
||
4. If embedding generation fails, strategy falls back to `LexicalOnly` (if OpenSearch available) or returns error.
|
||
5. `metrics.total_time_ms` is populated and < 500ms for test fixtures.
|
||
6. `metrics.semantic_time_ms` and `lexical_time_ms` are roughly equal (parallel execution, not serial).
|
||
7. Results are sorted by `final_score` descending.
|
||
8. `final_count` ≤ `RRFConfig.final_k`.
|
||
|
||
## Verify
|
||
|
||
```bash
|
||
cargo test -p mem-cli hybrid_query -- --nocapture
|
||
```
|
||
|
||
**False pass:** Worker always returns semantic-only results even when strategy is `Hybrid`. Check by asserting `lexical_rank.is_some()` on at least one result when OpenSearch is configured. Another false pass: serial execution disguised as parallel — assert `max(semantic_time_ms, lexical_time_ms) ≈ total - fusion_time_ms`, not `sum`.
|
||
|
||
## Artifacts
|
||
|
||
- `crates/mem-cli/src/hybrid_query_worker.rs` (rewrite from current stub)
|
||
- Modified `crates/mem-store/src/lib.rs` (add `search_l1_by_ids`)
|
||
- Modified `crates/mem-cli/src/opensearch_client.rs` (make `lexical_search` pub)
|