airweave / rules

---
globs: **/search/**,**/search.py,**/domains/search/**
alwaysApply: false
---
# Airweave Search Rules

> **WARNING — PARTIALLY OUTDATED (March 2026)**
>
> This file documents the **legacy V1 search module** (`airweave/search/`), which
> uses an operation-based pipeline with Qdrant, Cerebras/OpenAI/Groq providers,
> and structured-output LLM calls.
>
> The **new V2 search** lives in `airweave/domains/search/` and implements a
> three-tier architecture:
>
> - **Instant** — direct vector search, no LLM
> - **Classic** — vector search + reranking + answer generation (replaces V1)
> - **Agentic** — multi-turn tool-calling agent loop with search/count/navigate/read/collect/finish tools
>
> Key differences from V1:
> - **Vector DB**: Vespa (not Qdrant)
> - **LLM adapters**: `airweave/adapters/llm/` with Together AI + Anthropic fallback (not Cerebras/Groq)
> - **Reranker**: `airweave/adapters/reranker/` with Cohere
> - **Filters**: `domains/search/types/filters.py` — FilterableField/FilterOperator/FilterGroup (not Qdrant native filters)
> - **Streaming**: SSE events defined in `domains/search/agentic/events.py` (tool_call, thinking, search_results, etc.)
> - **API**: `POST /collections/{id}/search/{tier}` where tier is `instant`, `classic`, or `agentic`
> - **Config**: `domains/search/config.py` — model specs, token budgets, tier defaults
>
> **When working on `domains/search/`**, prefer reading the actual code over this
> file. The sections below remain accurate for the legacy `airweave/search/` module only.

## Overview (Legacy V1)

The search module (`@search/`) implements a **modular, pipeline-based architecture** with composable operations.  It aims to maintain search quality and flexibility.

## Core Architecture

### Operation-Based Pipeline

```python
SearchRequest → SearchFactory → SearchContext → SearchOrchestrator → SearchResponse
                                    ↓
                            [Operations Pipeline]
```

Each operation:
- Implements `SearchOperation` abstract base class
- Declares dependencies explicitly
- Reads/writes to shared state dictionary
- Can be optional (graceful failure)
- Executes asynchronously

### Request Flow
1. **Endpoint** (`api/v1/endpoints/search.py`) → Creates/receives `SearchRequest`
2. **SearchService.search()** → Main entry point
3. **SearchFactory.build()** → Creates `SearchContext` with enabled operations
4. **SearchOrchestrator.run()** → Executes operations in dependency order
5. **Operations** → Execute in topologically sorted order
6. **Qdrant destination** → Vector search execution
7. **Data Persistence** → Save search query to `search_queries` table via `CRUDSearchQuery`

### API Endpoints

Search endpoints are defined in `api/v1/endpoints/search.py` and mounted under `/collections` prefix in `api/v1/api.py`:

```python
# In api/v1/api.py
from airweave.api.v1.endpoints import search
api_router.include_router(search.router, prefix="/collections", tags=["collections"])
```

**Available Endpoints:**

1. **GET `/collections/{readable_id}/search`** - Legacy endpoint (DEPRECATED)
   - Maintained for backwards compatibility
   - Query parameters: `query`, `response_type`, `limit`, `offset`, `recency_bias`
   - Returns `LegacySearchResponse` with `status` field
   - Automatically converts to new format internally
   - Adds deprecation headers

2. **POST `/collections/{readable_id}/search`** - Main search endpoint (RECOMMENDED)
   - Accepts both `SearchRequest` (new) and `LegacySearchRequest` (old) schemas
   - Supports Qdrant native filters via `filter` field
   - Full control over all search features
   - Returns `SearchResponse` (new) or `LegacySearchResponse` (legacy) based on input schema
   - Adds deprecation headers for legacy requests

3. **POST `/collections/{readable_id}/search/stream`** - Streaming search with SSE
   - Accepts both `SearchRequest` and `LegacySearchRequest`
   - Returns Server-Sent Events (SSE) stream
   - Real-time progress updates via Redis pubsub
   - Automatically converts legacy requests

4. **GET `/collections/internal/filter-schema`** - Filter schema endpoint
   - Returns Qdrant Filter JSON schema for frontend validation
   - Public endpoint for building UI filter builders

All endpoints use the same underlying `SearchService.search()`, ensuring consistent behavior and quality.

### Input/Output Schemas

**SearchRequest** (new schema - `schemas/search.py`):
```python
query: str                                   # Search text (required)
retrieval_strategy: Optional[RetrievalStrategy]  # "hybrid", "neural", or "keyword"
filter: Optional[QdrantFilter]               # Qdrant native filter object
offset: Optional[int]                        # Pagination offset
limit: Optional[int]                         # Results per page
temporal_relevance: Optional[float]          # Recency weight (0-1, default: 0.3)
expand_query: Optional[bool]                 # Generate query variations
interpret_filters: Optional[bool]            # Extract filters from natural language
rerank: Optional[bool]                       # LLM-based reranking
generate_answer: Optional[bool]              # AI-generated completion
```

**SearchResponse** (new schema - `schemas/search.py`):
```python
results: List[Dict]                          # Search results
completion: Optional[str]                    # AI-generated answer (if generate_answer=True)
```

**LegacySearchRequest** (old schema - `schemas/search_legacy.py`):
```python
query: str
response_type: ResponseType                  # "raw" or "completion"
search_method: Optional[str]                 # "hybrid", "neural", "keyword"
expansion_strategy: Optional[QueryExpansionStrategy]  # "auto", "llm", "no_expansion"
recency_bias: Optional[float]                # Recency weight (0-1)
enable_reranking: Optional[bool]             # LLM reranking
enable_query_interpretation: Optional[bool]  # Filter extraction
# ... other legacy fields
```

**LegacySearchResponse** (old schema):
```python
results: List[Dict]
response_type: ResponseType
completion: Optional[str]
status: SearchStatus                         # "success", "no_results", etc.
```

Result payload includes: `entity_id`, `source_name`, `md_content`, `metadata`, `score`, `breadcrumbs`, `url`

## Key Components

### Search Operations (`search/operations/`)

1. **QueryExpansion** (`query_expansion.py`): Generates query variants using LLM
2. **QueryInterpretation** (`query_interpretation.py`): LLM-based filter extraction from natural language
3. **EmbedQuery** (`embed_query.py`): Generates dense (neural) and sparse (BM25) embeddings (optional for federated-only collections)
4. **UserFilter** (`user_filter.py`): Applies and merges user-provided Qdrant filters
5. **TemporalRelevance** (`temporal_relevance.py`): Dynamic time-based decay configuration
6. **Retrieval** (`retrieval.py`): Executes vector search in Qdrant (hybrid/neural/keyword) (optional for federated-only collections)
7. **FederatedSearch** (`federated_search.py`): Searches federated sources (e.g., Slack) at query time and merges with vector results using RRF
8. **Reranking** (`reranking.py`): Post-retrieval reranking using LLM providers
9. **GenerateAnswer** (`generate_answer.py`): AI-generated completions from search results

### Hybrid Search

True hybrid search combining:
- **Neural embeddings**: Semantic similarity
- **BM25 sparse embeddings**: Keyword matching
- **Fusion**: Reciprocal Rank Fusion (RRF)
- Default: `"hybrid"` for optimal quality

### Federated Search

For sources with strict rate limits or massive data volumes (e.g., Slack), **federated search** queries the source's API at search time instead of syncing all data:

**How It Works:**
1. Collection contains both vector-synced sources and federated sources
2. User submits search query
3. LLM extracts exactly 5 keywords from query and expansions
4. Federated sources are searched in parallel with vector database
5. Each federated source searches with all keywords concurrently
6. Results are deduplicated by entity_id across keywords
7. Federated and vector results are merged using Reciprocal Rank Fusion (RRF)
8. Final merged results returned to user

**Key Features:**
- Parallel execution across all federated sources
- Concurrent keyword searches per source
- Automatic deduplication (1.5x padding for ~33% duplication rate)
- RRF scoring for fair ranking across source types
- Graceful degradation (continues if one source fails)
- Real-time progress events via SSE

**Configuration:**
- Factory automatically detects federated sources in collection
- Conditionally omits EmbedQuery/Retrieval for federated-only collections
- Adds FederatedSearch operation when federated sources present

### Dynamic Temporal Relevance

```python
final_score = similarity × (1 - weight + weight × decay)
```
- Computes decay dynamically from actual collection time range
- Respects filters when determining time ranges
- Linear decay with configurable weight (default: 0.3)
- Applied at Qdrant level for efficiency via `DecayConfig`

### Configuration Defaults

Defaults are defined in `search/defaults.yml` and loaded via `SearchFactory`:
- `retrieval_strategy: hybrid` - Combines neural + keyword search
- `temporal_relevance: 0.3` - Moderate recency weighting
- `expand_query: true` - Generate query variations
- `interpret_filters: false` - Manual filter control by default
- `rerank: true` - LLM-based reranking enabled
- `generate_answer: true` - AI completion generation enabled
- `offset: 0`, `limit: 1000` - Pagination defaults

Provider and model configurations are also in `defaults.yml`:
- `provider_models`: Model specs per provider (Cerebras, OpenAI, Groq, Cohere)
- `operation_preferences`: Provider selection order for each operation with fallback support

**Provider Capabilities:**
- **Cerebras**: LLM only (llm: gpt-oss-120b, 131K context window)
- **OpenAI**: Full stack (LLM: gpt-5/gpt-5-nano, embeddings: text-embedding-3-small, rerank: gpt-5-nano)
- **Groq**: LLM and rerank (llm: openai/gpt-oss-120b/20b, rerank: openai/gpt-oss-120b)
- **Cohere**: Rerank only (rerank: rerank-v3.5)

**Default Provider Preferences** (from `defaults.yml`):
- Query expansion: Cerebras → Groq → OpenAI
- Query interpretation: Cerebras → Groq → OpenAI
- Embeddings: OpenAI only (no fallback - embeddings must be consistent)
- Reranking: Cohere → Groq → OpenAI
- Answer generation: Cerebras → Groq → OpenAI
- Federated search: Cerebras → Groq → OpenAI

## Entity Architecture

### AirweaveField System

Field annotation system extending Pydantic fields:

```python
class MyEntity(ChunkEntity):
    name: str = AirweaveField(..., embeddable=True)
    created_at: datetime = AirweaveField(..., is_created_at=True)
    content: str = AirweaveField(..., embeddable=True)
```

- **Type-safe metadata** alongside field definitions
- **Embeddable marking** for search embeddings
- **Timestamp harmonization** for recency handling

### Entity Hierarchy

```
BaseEntity
├── ChunkEntity (searchable, embeddable)
│   ├── Domain entities (Jira, Linear, Notion, etc.)
│   └── Generated chunks from FileEntity
├── FileEntity (file-based content)
└── CodeFileEntity (specialized for code)
```

### Embeddable Text Generation

ChunkEntity builds structured markdown representation:
- Includes source, type, breadcrumbs, content
- Respects `embeddable=True` markers
- Caps at 12,000 characters

## Advanced Features

### Query Expansion
- Boolean control: `expand_query: true/false`
- When enabled, generates up to 4 query variations using LLM
- Provides diverse paraphrases to improve recall
- Includes keyword-forward and normalized variants
- Expands abbreviations and synonyms

### Provider System (`search/providers/`)
- **Cerebras** (`cerebras.py`): Ultra-fast LLM inference for structured output and text generation (no embeddings or reranking)
- **OpenAI** (`openai.py`): Complete provider - LLM, embeddings, reranking
- **Groq** (`groq.py`): Fast LLM inference and reranking (no embeddings)
- **Cohere** (`cohere.py`): Specialized reranking API
- Automatic provider selection based on available API keys
- Provider fallback on rate limits (429) and server errors (5xx)
- Each operation tries providers in preference order from `defaults.yml`
- Graceful degradation: if first provider fails with retryable error, automatically falls back to next provider

### Hybrid Search Prefetch
- Large prefetch limits for broad candidate sets
- Ensures both neural and BM25 contribute meaningfully
- Enables effective temporal relevance across full collection
- Rerank multiplier (2x) fetches extra candidates for better reranking

## System Metadata

### AirweaveSystemMetadata

Centralized tracking:
- **Vectors**: Dense and sparse embeddings
- **Timestamps**: Harmonized `airweave_created_at/updated_at`
- **Sync tracking**: `sync_id`, `sync_job_id`
- **Content hash**: Change detection
- **Skip flag**: Processing control

### Vector Storage

```python
vectors: Optional[List[List[float] | SparseEmbedding | None]]
# Index 0: Dense neural embedding
# Index 1: Sparse BM25 embedding
```

## Performance & Quality

### Optimizations
- Async-first design throughout
- Parallel operation execution via orchestrator
- Provider-based architecture for flexibility
- Bulk search APIs for multiple queries
- Smart prefetching for reranking
- Token budget management for LLM operations

### Intelligent Defaults (from `defaults.yml`)
- Query expansion: `true` (enabled)
- LLM reranking: `true` (enabled)
- Retrieval strategy: `hybrid` (neural + keyword)
- Temporal relevance: `0.3` (moderate recency)
- Generate answer: `true` (AI completion)
- Interpret filters: `false` (manual control)

### Provider Fallback System

**Automatic Failover**: Operations automatically try providers in preference order when encountering retryable errors.

**Retryable Errors** (trigger fallback):
- Rate limits (HTTP 429)
- Server errors (HTTP 500, 502, 503, 504)
- Queue exceeded errors
- Too many requests errors

**Non-Retryable Errors** (fail immediately):
- Authentication errors (HTTP 401, 403)
- Validation errors (HTTP 400, 422)
- Not found errors (HTTP 404)

**Fallback Behavior**:
1. Operation tries first provider in preference list
2. If retryable error occurs, logs warning and tries next provider
3. If non-retryable error occurs, fails immediately
4. If all providers fail with retryable errors, fails entire search
5. Successful provider logged at INFO level for observability

**Example Flow**:
```
[QueryExpansion] Attempting with provider CerebrasProvider (1/3)
[QueryExpansion] Provider CerebrasProvider failed with retryable error: 429. Trying next provider...
[QueryExpansion] Attempting with provider GroqProvider (2/3)
[QueryExpansion] ✓ Succeeded with fallback provider GroqProvider
```

**Implementation**: `SearchOperation._execute_with_provider_fallback()` in `operations/_base.py`

### Graceful Degradation
- Provider fallback: Automatically tries alternative providers on rate limits/errors
- Query interpretation failure → continues without extracted filters
- Reranking failure → returns unranked results
- Query expansion failure → uses original query only

## Implementation Details

### Filter Conversion
Qdrant requires dict format, not Filter objects:
```python
# Convert Filter to dict before passing to Qdrant
filter_dict = filter.model_dump(exclude_none=True) if filter else None
await destination.search(filter=filter_dict)
```

### Result Cleaning
Automatically removes sensitive/internal fields:
- `id`, `vector` from payload
- `download_url`, `local_path`, `file_uuid`, `checksum`
- Parses JSON strings in `metadata`, `sync_metadata`

### Error Handling
HTTP status mapping in endpoints:
- Connection errors → 503 (service unavailable)
- Not found → 404
- Invalid filter → 422 (unprocessable entity)
- Others → 500

### Critical Gotchas
1. **Filter format**: Must convert to dict for Qdrant (done automatically in operations)
2. **Offset with expansion**: Pagination applied after deduplication in bulk search
3. **Provider dependencies**: Requires at least OpenAI API key for embeddings; other providers optional but recommended for fallback
4. **Source names**: Case-sensitive in filters (e.g., "GitHub" not "github")
5. **Embedding model**: Auto-selected from provider_models configuration; no fallback (must be consistent)
6. **Legacy compatibility**: Both old and new schemas supported via `legacy_adapter.py`
7. **Operation dependencies**: Orchestrator resolves execution order via topological sort
8. **Provider validation**: Token budgeting and validation use the **actual provider** being called, not `providers[0]`
9. **Federated keywords**: Exactly 5 keywords extracted (fixed-size tuple for Cerebras compatibility)
10. **Provider fallback**: Operations automatically try providers in order; log which provider succeeded at INFO level

## Search Persistence & Analytics

### Data Persistence

Every search operation is automatically persisted to the `search_queries` table for:
- **Audit trails** and compliance
- **Performance analytics** and optimization
- **User behavior analysis** and insights
- **Search evolution tracking** over time

**SearchQuery Model** (`airweave.models.search_query`):
```python
class SearchQuery(OrganizationBase, UserMixin):
    # Core search data
    query_text: str                    # Full search query
    query_length: int                  # Character count
    is_streaming: bool                 # Whether this was a streaming search
    retrieval_strategy: str            # "hybrid", "neural", or "keyword"

    # Search parameters
    limit: int                         # Results limit
    offset: int                        # Pagination offset
    temporal_relevance: float          # Recency weight (0-1)
    filter: Optional[Dict]             # Applied Qdrant filter

    # Performance metrics
    duration_ms: int                   # Execution time
    results_count: int                 # Results returned

    # Feature usage tracking
    expand_query: bool                 # Query expansion enabled
    interpret_filters: bool            # Query interpretation enabled
    rerank: bool                       # LLM reranking enabled
    generate_answer: bool              # AI completion enabled

    # Relationships
    collection_id: UUID                # Collection searched
    user_id: Optional[UUID]            # User who searched (null for API)
    api_key_id: Optional[UUID]         # API key used (null for users)
```

**CRUDSearchQuery** (`airweave.crud.crud_search_query`):
- Inherits from `CRUDBaseOrganization` for organization scoping
- Provides `get_user_search_history()` for user experience features
- Supports analytics queries for collection performance

### Analytics Integration

**Database-First Analytics**: All search data is persisted to the `search_queries` table, providing a comprehensive analytics foundation:

- **Search performance metrics** (duration, results count)
- **Feature adoption tracking** (expansion, reranking, interpretation)
- **User behavior analysis** (query patterns, search evolution)
- **Collection analytics** (usage patterns, performance trends)

**PostHog Integration**: For real-time dashboards, search data can be exported from the database to PostHog via batch jobs or streaming pipelines.

### Error Handling

Search persistence is **non-blocking**:
- If persistence fails → Search continues, error logged
- Core search functionality is never compromised

## Best Practices

### When Working with Search

1. **Always use the new schemas** (`SearchRequest`/`SearchResponse`) for new code
2. **Let defaults do the work** - Most parameters are optional; factory applies intelligent defaults
3. **Use the service singleton** - Import from `airweave.search.service import service`
4. **Handle both schemas in endpoints** - Accept `Union[SearchRequest, LegacySearchRequest]` for compatibility
5. **Log with context** - Operations receive `ctx: ApiContext` with pre-configured logger
6. **Emit events for streaming** - Use `EventEmitter` for progress updates
7. **Validate providers early** - Factory raises clear errors for missing API keys

### Adding New Operations

To add a new search operation:

1. **Create operation class** in `search/operations/your_operation.py`:
   ```python
   from ._base import SearchOperation

   class YourOperation(SearchOperation):
       def depends_on(self) -> List[str]:
           return ["EmbedQuery"]  # Declare dependencies

       async def execute(self, context, state, ctx, emitter):
           # Read from state
           embeddings = state.get("dense_embeddings")

           # Do work
           result = await self._process(embeddings)

           # Write to state
           state["your_result"] = result

           # Emit events
           await emitter.emit("your_event", {"data": result}, op_name=self.__class__.__name__)
   ```

2. **Add to SearchContext** in `search/context.py`:
   ```python
   your_operation: Optional[YourOperation] = Field(default=None)
   ```

3. **Add to factory** in `search/factory.py`:
   ```python
   search_context = SearchContext(
       # ... existing fields
       your_operation=YourOperation() if some_condition else None,
   )
   ```

4. **Update orchestrator** - No changes needed! It automatically discovers operations from context

### Modifying Prompts

System prompts are in `search/prompts/`:
- Each prompt is a module constant (e.g., `QUERY_EXPANSION_SYSTEM_PROMPT`)
- Use `.format()` for dynamic values (e.g., available fields)
- Test with different LLM providers (prompt quality varies)
- Keep prompts focused and concise for better results

### Working with Providers

Providers implement capabilities (LLM, embeddings, reranking):
- **BaseProvider** defines the interface with `is_retryable_error()` static method for error classification
- Each provider declares its supported capabilities via `ProviderModelSpec`
- Factory initializes **all available providers** for each operation in preference order
- Operations receive provider lists and automatically fall back on errors
- Providers handle tokenization, budgeting, and API calls
- Token validation and budgeting happen **per-provider** (not using first provider)
- Add new providers by implementing `BaseProvider` and adding to `defaults.yml`

**Provider Fallback Pattern**:
```python
# Operations use _execute_with_provider_fallback from base class
async def call_provider(provider: BaseProvider) -> ResultType:
    # Validate for THIS specific provider
    self._validate_for_provider(input, provider, ctx)
    # Call provider method
    return await provider.method(args)

result = await self._execute_with_provider_fallback(
    providers=self.providers,
    operation_call=call_provider,
    operation_name="OperationName",
    ctx=ctx,
)
```

**Exception**: `EmbedQuery` uses a single provider (no fallback) because embedding models must remain consistent within a collection for vector similarity to work.

## Legacy Adapter & Migration

### Legacy Support (`search/legacy_adapter.py`)

The system maintains backwards compatibility with the old search API:

**Converting Legacy Requests:**
- `response_type` → `generate_answer` (enum to boolean)
- `search_method` → `retrieval_strategy` (renamed)
- `expansion_strategy` → `expand_query` (enum to boolean)
- `recency_bias` → `temporal_relevance` (renamed)
- `enable_reranking` → `rerank` (renamed)
- `enable_query_interpretation` → `interpret_filters` (renamed)
- `score_threshold` → Removed (no longer supported)

**Converting Legacy Responses:**
- New `SearchResponse` → `LegacySearchResponse` with `status` and `response_type` fields
- Status inferred from result count and quality
- Response type preserved from request

**Deprecation Headers:**
When using legacy schemas, endpoints add HTTP headers:
- `X-API-Deprecation: true`
- `X-API-Deprecation-Message: <migration guidance>`

### Migration Path

**For API consumers:**
1. Update request schema to use new field names
2. Update response handling (remove status checks)
3. Test with new schema before removing old code
4. Monitor deprecation headers

**For internal code:**
- Always use new `SearchRequest` and `SearchResponse` schemas
- Legacy adapter handles conversion automatically
- Service layer only processes new schemas

## Design Principles

1. **Modularity**: Self-contained operations in `search/operations/`
2. **Configurability**: Centralized via `SearchContext` and `defaults.yml`
3. **Extensibility**: Plug-in new operations via `SearchOperation` interface
4. **Performance**: Async-first, provider-based architecture
5. **Quality-first**: Optimized defaults with provider fallback chains
6. **Observability**: Comprehensive logging and streaming events via `EventEmitter`
7. **Database-first analytics**: Every search persisted via `SearchHelpers.persist_search_data()`
8. **Backwards compatibility**: Legacy adapter maintains API compatibility seamlessly

### SearchContext (`search/context.py`)

The `SearchContext` is the execution plan for a search request:

```python
class SearchContext(BaseModel):
    # Metadata
    request_id: str
    collection_id: UUID
    readable_collection_id: str
    stream: bool
    vector_size: int

    # Input query
    query: str

    # Operation instances (None if disabled)
    query_expansion: Optional[QueryExpansion]
    query_interpretation: Optional[QueryInterpretation]
    embed_query: EmbedQuery  # Always required
    user_filter: Optional[UserFilter]
    temporal_relevance: Optional[TemporalRelevance]
    retrieval: Retrieval  # Always required
    reranking: Optional[Reranking]
    generate_answer: Optional[GenerateAnswer]
```

The factory builds this context by:
1. Applying defaults from `defaults.yml`
2. Validating API keys for required providers
3. Creating provider instances for each enabled operation
4. Building operation instances with proper providers
5. Returning complete execution plan

## Module Structure

```
search/
├── service.py              # Main SearchService entry point
├── factory.py              # SearchFactory - builds SearchContext
├── orchestrator.py         # SearchOrchestrator - executes operations
├── context.py              # SearchContext - holds operation instances
├── emitter.py              # EventEmitter - streaming event publishing
├── helpers.py              # SearchHelpers - persistence and utilities
├── defaults.yml            # Configuration defaults and provider specs
├── legacy_adapter.py       # Converts between old and new schemas
├── utils.py                # Utility functions for filter handling
├── operations/             # Individual search operations
│   ├── _base.py            # SearchOperation abstract base
│   ├── query_expansion.py
│   ├── query_interpretation.py
│   ├── embed_query.py
│   ├── user_filter.py
│   ├── temporal_relevance.py
│   ├── retrieval.py
│   ├── reranking.py
│   └── generate_answer.py
├── providers/              # LLM provider implementations
│   ├── _base.py            # BaseProvider abstract class
│   ├── schemas.py          # Provider model specifications
│   ├── openai.py           # OpenAI provider
│   ├── groq.py             # Groq provider
│   └── cohere.py           # Cohere provider
└── prompts/                # System prompts for LLM operations
    ├── __init__.py
    ├── query_expansion.py
    ├── query_interpretation.py
    ├── reranking.py
    └── generate_answer.py
```

## Quick Reference

### Basic Search (New Schema)
```python
from airweave.schemas.search import SearchRequest
from airweave.search.service import service

# Minimal request - uses all defaults from defaults.yml
request = SearchRequest(query="find customer issues")

response = await service.search(
    request_id=ctx.request_id,
    readable_collection_id="my-collection",
    search_request=request,
    stream=False,
    db=db,
    ctx=ctx,
)

# Access results
results = response.results
completion = response.completion  # AI answer if generate_answer=True
```

### Advanced Search with Options
```python
request = SearchRequest(
    query="deployment failures last week",
    retrieval_strategy=RetrievalStrategy.HYBRID,
    temporal_relevance=0.7,  # Strong recency bias
    expand_query=True,
    interpret_filters=True,  # Extract "last week" as date filter
    rerank=True,
    generate_answer=False,  # Just return results
    limit=50,
    offset=0,
    filter={  # Manual Qdrant filter
        "must": [
            {"key": "source_name", "match": {"value": "github"}}
        ]
    }
)

response = await service.search(...)
```

### Streaming Search
```python
# In endpoint - same SearchRequest, set stream=True
response = await service.search(
    request_id=ctx.request_id,
    readable_collection_id="my-collection",
    search_request=request,
    stream=True,  # Enables event emission
    db=db,
    ctx=ctx,
)

# Events are automatically emitted to Redis channel: search:{request_id}
# Frontend subscribes via POST /collections/{id}/search/stream
```

### Legacy Request Conversion
```python
from airweave.schemas.search_legacy import LegacySearchRequest
from airweave.search.legacy_adapter import convert_legacy_request_to_new

# Old schema
legacy = LegacySearchRequest(
    query="test",
    response_type=ResponseType.COMPLETION,
    expansion_strategy=QueryExpansionStrategy.AUTO,
    enable_reranking=True,
)

# Convert to new schema
new_request = convert_legacy_request_to_new(legacy)
# Result: SearchRequest(query="test", generate_answer=True, expand_query=True, rerank=True)
```

### Common Patterns

**Simple search with defaults:**
```python
# Relies on defaults.yml configuration
request = SearchRequest(query="customer feedback")
response = await service.search(ctx.request_id, "collection-id", request, False, db, ctx)
```

**Search with custom temporal weighting:**
```python
# Prioritize recent documents strongly
request = SearchRequest(
    query="recent updates",
    temporal_relevance=0.9,  # Very strong recency bias
)
```

**Search with manual filtering only:**
```python
# Disable query interpretation, provide exact filter
request = SearchRequest(
    query="bugs",
    interpret_filters=False,  # Don't auto-extract filters
    filter={"must": [{"key": "status", "match": {"value": "open"}}]},
)
```

**Fast search (skip expensive operations):**
```python
# Disable reranking for speed
request = SearchRequest(
    query="quick lookup",
    rerank=False,  # Skip ~10s reranking step
    generate_answer=False,  # Skip completion generation
)
```