Compare commits
@@ -23,6 +23,32 @@
|
||||
* **Update `CHANGELOG.md`** with every user-facing change.
|
||||
* Format: `## [Unreleased] - YYYY-MM-DD` followed by `### Added`, `### Changed`, or `### Fixed`.
|
||||
|
||||
### 🚀 Release Flow
|
||||
When changes are ready for deployment:
|
||||
|
||||
1. **Ask user if deploy cycle is desired **
|
||||
|
||||
2. **Update version** in `pyproject.toml`:
|
||||
- Bug fixes: bump patch version (1.8.3 → 1.8.4)
|
||||
- New features: bump minor version (1.8.4 → 1.9.0)
|
||||
|
||||
3. **Update CHANGELOG.md**:
|
||||
- Move items from `[Unreleased]` to new version section
|
||||
- Add release date: `## [1.8.4] - 2025-12-16`
|
||||
|
||||
4. **Commit and tag**:
|
||||
```bash
|
||||
git add -A
|
||||
git commit -m "fix: description of changes"
|
||||
git tag v1.8.4
|
||||
git push origin main --tags
|
||||
```
|
||||
|
||||
5. **CI/CD triggers automatically**:
|
||||
- Gitea CI builds Docker image on new tag
|
||||
- Watchtower pulls and deploys to production
|
||||
- Verify deployment: `curl http://192.168.86.149:8000/health`
|
||||
|
||||
---
|
||||
|
||||
## 2. FastAPI Architecture & Best Practices
|
||||
|
||||
+106
@@ -5,6 +5,112 @@ All notable changes to Library Desk will be documented in this file.
|
||||
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
|
||||
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
|
||||
|
||||
## [1.3.3] - 2025-12-23
|
||||
|
||||
### Added
|
||||
|
||||
- Temperature parameter to `OllamaClient.generate_text()` for controlling output determinism
|
||||
- `TODO.md` tracking remaining stub endpoints to implement
|
||||
- Wired `/query/semantic` endpoint to VectorService
|
||||
- Wired `/query/graph` endpoint to GraphService
|
||||
|
||||
### Changed
|
||||
|
||||
- **Improved LLM prompts** based on llm-findings.md recommendations:
|
||||
- Keyword extraction: temperature 0.0, negative constraints
|
||||
- LLM re-ranking: temperature 0.0, explicit rules
|
||||
- Conflict detection: temperature 0.0, analysis steps (CoT)
|
||||
- Wiki page creation: temperature 0.3, anti-hallucination constraints
|
||||
- Page reconstruction: temperature 0.2, preservation constraints
|
||||
- Web results analysis: temperature 0.0, conservative approach
|
||||
- Test fixtures now use configurable host (TEST_HOST) instead of Docker hostnames
|
||||
|
||||
### Removed
|
||||
|
||||
- Dead code: unused `get_default_user()` function
|
||||
- Unused imports from routers (wiki.py, graph.py, hybrid_rag.py)
|
||||
- Stub endpoints shadowed by real implementations (/stats, /ingest/document, /ingest/batch)
|
||||
|
||||
## [1.3.2] - 2025-12-22
|
||||
|
||||
### Changed
|
||||
|
||||
- **Consolidated Ollama model configuration** - All LLM operations now use single `OLLAMA_MODEL` environment variable
|
||||
- Removed separate `reranker_model` setting
|
||||
- HybridRAG re-ranking, consolidation analysis, and wiki page writing all use the same model
|
||||
- Improves VRAM efficiency by keeping one model hot
|
||||
- Added `OLLAMA_EMBEDDING_MODEL` environment variable for embedding model (previously overloaded `OLLAMA_MODEL`)
|
||||
- Updated WikiPageWriter to accept settings instead of hardcoded model name
|
||||
|
||||
## [1.3.1] - 2025-12-16
|
||||
|
||||
### Fixed
|
||||
|
||||
- Smart create endpoint missing `content_extractor` dependency causing 500 errors on `POST /wiki/pages/smart-create`
|
||||
|
||||
## [1.3.0] - 2025-12-15
|
||||
|
||||
### Changed
|
||||
|
||||
- **Two-Stage RRF Architecture** - Major refactor to level the playing field between wiki and web results
|
||||
- Stage 1: Vector and graph results merged into single "wiki" ranking using mini-RRF
|
||||
- Stage 2: Final RRF between wiki (single source) and web (single source)
|
||||
- Wiki pages no longer get 2x advantage from appearing in both vector and graph searches
|
||||
- Multi-source confirmation still determines wiki internal ranking
|
||||
|
||||
- **Skip synonyms in graph search** - LLM-generated synonyms (e.g., "author") no longer match unrelated graph entities (e.g., "author2000")
|
||||
- Vector search still uses synonyms for semantic similarity
|
||||
- Graph search uses only core keywords for exact entity matching
|
||||
|
||||
### Added
|
||||
|
||||
- `VECTOR_SIMILARITY_THRESHOLD` config setting (default: 0.7) to filter weak vector matches
|
||||
- Deduplication in graph search to prevent same document appearing multiple times
|
||||
|
||||
### Fixed
|
||||
|
||||
- Graph search duplicate entity bug where same document could appear twice if entity linked multiple times
|
||||
|
||||
## [1.2.1] - 2025-12-15
|
||||
|
||||
### Fixed
|
||||
|
||||
- HybridRAG router missing `content_extractor` dependency causing 500 errors on `/query/hybrid` endpoint
|
||||
|
||||
## [1.2.0] - 2025-12-15
|
||||
|
||||
### Added
|
||||
|
||||
- **RAG Search Endpoint** (`POST /rag/search`)
|
||||
- Web, news, and image search via SearXNG
|
||||
- Full content extraction using Trafilatura (F1 score 0.958)
|
||||
- Redis caching with configurable TTL
|
||||
- Markdown sources summary for LLM consumption
|
||||
- Returns both extracted content and original snippets
|
||||
|
||||
- **Content Extraction Endpoints** (`/content/*`)
|
||||
- `POST /content/extract` - Extract content from a single URL
|
||||
- `POST /content/extract/batch` - Batch extraction (up to 20 URLs)
|
||||
- Reusable ContentExtractor client for use across the codebase
|
||||
|
||||
- **HybridRAG Content Extraction Enhancement**
|
||||
- Web search results now include full extracted content via Trafilatura
|
||||
- Falls back to original snippets if extraction fails
|
||||
- Improves context quality for LLM re-ranking and consumption
|
||||
|
||||
### Changed
|
||||
|
||||
- Added new configuration options:
|
||||
- `SEARCH_CACHE_TTL` - Search cache TTL in seconds (default: 300)
|
||||
- `SEARCH_TIMEOUT` - SearXNG timeout (default: 10s)
|
||||
- `CONTENT_EXTRACTION_TIMEOUT` - Per-URL extraction timeout (default: 5s)
|
||||
- `CONTENT_MAX_LENGTH` - Max extracted content length (default: 2000)
|
||||
- `SEARCH_DEFAULT_LIMIT` - Default search results (default: 10)
|
||||
|
||||
### Dependencies
|
||||
|
||||
- Added `trafilatura~=1.12.0` for content extraction
|
||||
|
||||
## [1.1.3] - 2025-12-14
|
||||
|
||||
### Added
|
||||
|
||||
@@ -49,7 +49,8 @@ QDRANT_PORT=6333
|
||||
WIKIJS_URL=http://wiki:3000
|
||||
SEARXNG_URL=http://searxng:8080
|
||||
OLLAMA_URL=http://ollama:11434
|
||||
OLLAMA_MODEL=nomic-embed-text
|
||||
OLLAMA_MODEL=mistral-nemo-large:latest
|
||||
OLLAMA_EMBEDDING_MODEL=nomic-embed-text
|
||||
REDIS_HOST=redis-shared
|
||||
REDIS_PORT=6379
|
||||
REDIS_DB=2
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
# TODO
|
||||
|
||||
Outstanding work items for Library Desk.
|
||||
|
||||
## Stub Endpoints to Implement
|
||||
|
||||
The following endpoints in `src/main.py` return stub responses and need real implementations:
|
||||
|
||||
### Ingestion Status Endpoints
|
||||
|
||||
#### `POST /ingest/check-updates`
|
||||
Check which documents need updating based on content hashes. Used by Scheduler to determine what changed since last sync.
|
||||
|
||||
**Implementation needed:**
|
||||
1. Query existing documents by path
|
||||
2. Compare content hashes
|
||||
3. Return list of updates needed
|
||||
|
||||
#### `GET /ingest/status/{document_id}`
|
||||
Get processing status for a document.
|
||||
|
||||
**Implementation needed:**
|
||||
- Status tracking system (Redis or database)
|
||||
- Track ingestion progress per document
|
||||
|
||||
#### `GET /ingest/repo-status/{repository}`
|
||||
Get indexing status for an entire repository.
|
||||
|
||||
**Implementation needed:**
|
||||
- Repository-level statistics
|
||||
- Track which documents from a repo are indexed
|
||||
|
||||
### Deduplication
|
||||
|
||||
#### `POST /deduplicate/check`
|
||||
Check for duplicate or highly similar documents using vector similarity and graph analysis.
|
||||
|
||||
**Implementation needed:**
|
||||
1. Get document embedding from Qdrant
|
||||
2. Find similar vectors above threshold
|
||||
3. Check graph relationships
|
||||
4. Return candidates with similarity scores
|
||||
|
||||
## System Statistics
|
||||
|
||||
#### `GET /stats`
|
||||
Get system statistics (wiki pages, neo4j nodes, qdrant vectors).
|
||||
|
||||
**Implementation needed:**
|
||||
- Query Neo4j for node count
|
||||
- Query Qdrant for vector count
|
||||
- Query Wiki.js for page count
|
||||
@@ -0,0 +1,58 @@
|
||||
# HybridRAG Architecture
|
||||
|
||||
## Overview
|
||||
|
||||
HybridRAG combines three search sources to provide comprehensive results:
|
||||
- **Vector search** (Qdrant) - Semantic similarity via embeddings
|
||||
- **Graph search** (Neo4j) - Entity relationships in knowledge graph
|
||||
- **Web search** (SearXNG) - External web results via Trafilatura extraction
|
||||
|
||||
## Two-Stage RRF Fusion (v1.3.0+)
|
||||
|
||||
To ensure fair ranking between wiki and web results, we use a two-stage Reciprocal Rank Fusion:
|
||||
|
||||
```
|
||||
Stage 1: Wiki Merge
|
||||
vector results ─┬─→ Mini-RRF ─→ Unified wiki ranking
|
||||
graph results ─┘
|
||||
|
||||
Stage 2: Final RRF
|
||||
wiki (merged) ─┬─→ Final RRF ─→ Combined results
|
||||
web results ─┘
|
||||
```
|
||||
|
||||
**Why two stages?**
|
||||
|
||||
Previously, wiki pages found by BOTH vector and graph received double RRF contribution, giving them an unfair 2x advantage over web results. The two-stage approach:
|
||||
1. Merges vector+graph into a single "wiki" source
|
||||
2. Wiki's internal ranking still benefits from multi-source confirmation
|
||||
3. Wiki and web compete as equals in final ranking
|
||||
|
||||
## Configuration
|
||||
|
||||
| Setting | Default | Description |
|
||||
|---------|---------|-------------|
|
||||
| `VECTOR_SIMILARITY_THRESHOLD` | 0.7 | Minimum similarity score for vector results |
|
||||
| `HYBRID_RAG_VECTOR_LIMIT` | 10 | Max vector results |
|
||||
| `HYBRID_RAG_GRAPH_LIMIT` | 10 | Max graph results |
|
||||
| `HYBRID_RAG_WEB_LIMIT` | 5 | Max web results |
|
||||
|
||||
## Known Limitations & Future Improvements
|
||||
|
||||
### Vector Search Noise
|
||||
|
||||
**Status:** Open for improvement if needed after observation period.
|
||||
|
||||
Vector search may return generic category/index pages (e.g., "Reference", "Projects", "Places") with high similarity scores (~0.86). These pages often have similar boilerplate content leading to uniform scores.
|
||||
|
||||
**Potential solutions if this becomes problematic:**
|
||||
1. **Raise threshold** - Increase `VECTOR_SIMILARITY_THRESHOLD` to 0.85+
|
||||
2. **Page-type filtering** - Exclude pages tagged as category/index/stub
|
||||
3. **Content length signal** - Penalize pages with minimal content
|
||||
4. **Duplicate score detection** - Flag results with suspiciously identical scores
|
||||
|
||||
The LLM re-ranking phase typically demotes these low-quality results, so this may not require immediate action.
|
||||
|
||||
### Graph Search
|
||||
|
||||
Graph search uses only core keywords (no LLM-generated synonyms) to avoid false matches like "author" → "author2000". This is intentional - vector search handles semantic similarity via embeddings.
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "library-desk"
|
||||
version = "1.1.3"
|
||||
version = "1.3.3"
|
||||
description = "Coordination service for The Library system - HybridRAG queries, document ingestion, entity extraction, and knowledge consolidation"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.12"
|
||||
|
||||
@@ -25,6 +25,9 @@ python-multipart~=0.0.20
|
||||
# Utilities
|
||||
python-dateutil~=2.9.0
|
||||
|
||||
# Content Extraction
|
||||
trafilatura~=1.12.0
|
||||
|
||||
# Testing
|
||||
pytest~=8.3.0
|
||||
pytest-asyncio~=0.24.0
|
||||
|
||||
@@ -0,0 +1,310 @@
|
||||
"""
|
||||
Content extraction client for Library Desk.
|
||||
|
||||
A reusable Trafilatura wrapper that can be used throughout library-desk:
|
||||
- RAG search service (extract content from search results)
|
||||
- Ingestion service (extract content from URLs)
|
||||
- Standalone endpoint (ad-hoc content extraction)
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from typing import List, Optional
|
||||
|
||||
import trafilatura
|
||||
|
||||
from src.models.content import ContentExtractionResult
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class ContentExtractor:
|
||||
"""
|
||||
Generic content extraction client using Trafilatura.
|
||||
|
||||
Provides async wrappers around Trafilatura's synchronous extraction,
|
||||
with support for parallel batch processing and configurable timeouts.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
timeout: int = 5,
|
||||
max_length: int = 2000,
|
||||
max_workers: int = 10
|
||||
):
|
||||
"""
|
||||
Initialize ContentExtractor.
|
||||
|
||||
Args:
|
||||
timeout: Per-URL timeout in seconds
|
||||
max_length: Maximum content length to return (truncated if longer)
|
||||
max_workers: Max concurrent extractions for batch operations
|
||||
"""
|
||||
self.timeout = timeout
|
||||
self.max_length = max_length
|
||||
self._executor = ThreadPoolExecutor(max_workers=max_workers)
|
||||
logger.info(
|
||||
f"Initialized ContentExtractor: timeout={timeout}s, "
|
||||
f"max_length={max_length}, max_workers={max_workers}"
|
||||
)
|
||||
|
||||
def _extract_sync(
|
||||
self,
|
||||
url: str,
|
||||
include_metadata: bool = True,
|
||||
max_length: Optional[int] = None
|
||||
) -> ContentExtractionResult:
|
||||
"""
|
||||
Synchronous extraction (runs in thread pool).
|
||||
|
||||
Args:
|
||||
url: URL to extract content from
|
||||
include_metadata: Whether to extract title, author, date
|
||||
max_length: Override default max length
|
||||
|
||||
Returns:
|
||||
ContentExtractionResult with extracted content or error
|
||||
"""
|
||||
effective_max_length = max_length or self.max_length
|
||||
|
||||
try:
|
||||
# Fetch the URL
|
||||
downloaded = trafilatura.fetch_url(url)
|
||||
if not downloaded:
|
||||
return ContentExtractionResult(
|
||||
url=url,
|
||||
content="",
|
||||
success=False,
|
||||
error="Failed to fetch URL"
|
||||
)
|
||||
|
||||
# Extract content
|
||||
content = trafilatura.extract(
|
||||
downloaded,
|
||||
include_comments=False,
|
||||
include_tables=True,
|
||||
output_format='txt'
|
||||
)
|
||||
|
||||
if not content:
|
||||
return ContentExtractionResult(
|
||||
url=url,
|
||||
content="",
|
||||
success=False,
|
||||
error="No content extracted"
|
||||
)
|
||||
|
||||
# Truncate if needed
|
||||
if len(content) > effective_max_length:
|
||||
content = content[:effective_max_length] + "..."
|
||||
|
||||
# Extract metadata if requested
|
||||
title = None
|
||||
author = None
|
||||
date = None
|
||||
language = None
|
||||
|
||||
if include_metadata:
|
||||
metadata = trafilatura.extract(
|
||||
downloaded,
|
||||
output_format='xml',
|
||||
include_comments=False
|
||||
)
|
||||
# Parse metadata from XML if available
|
||||
# trafilatura.extract with output_format='xml' returns XML with metadata
|
||||
# For simplicity, we'll use bare_extraction which returns a dict
|
||||
try:
|
||||
meta_dict = trafilatura.bare_extraction(
|
||||
downloaded,
|
||||
include_comments=False
|
||||
)
|
||||
if meta_dict:
|
||||
title = meta_dict.get('title')
|
||||
author = meta_dict.get('author')
|
||||
date = meta_dict.get('date')
|
||||
language = meta_dict.get('language')
|
||||
except Exception as e:
|
||||
logger.debug(f"Metadata extraction failed for {url}: {e}")
|
||||
|
||||
return ContentExtractionResult(
|
||||
url=url,
|
||||
title=title,
|
||||
content=content,
|
||||
author=author,
|
||||
date=date,
|
||||
language=language,
|
||||
success=True,
|
||||
error=None
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Content extraction failed for {url}: {e}")
|
||||
return ContentExtractionResult(
|
||||
url=url,
|
||||
content="",
|
||||
success=False,
|
||||
error=str(e)
|
||||
)
|
||||
|
||||
async def extract(
|
||||
self,
|
||||
url: str,
|
||||
include_metadata: bool = True,
|
||||
max_length: Optional[int] = None
|
||||
) -> ContentExtractionResult:
|
||||
"""
|
||||
Extract content from a single URL asynchronously.
|
||||
|
||||
Args:
|
||||
url: URL to extract content from
|
||||
include_metadata: Whether to extract title, author, date
|
||||
max_length: Override default max length
|
||||
|
||||
Returns:
|
||||
ContentExtractionResult with extracted content or error
|
||||
"""
|
||||
loop = asyncio.get_event_loop()
|
||||
|
||||
try:
|
||||
result = await asyncio.wait_for(
|
||||
loop.run_in_executor(
|
||||
self._executor,
|
||||
self._extract_sync,
|
||||
url,
|
||||
include_metadata,
|
||||
max_length
|
||||
),
|
||||
timeout=self.timeout
|
||||
)
|
||||
return result
|
||||
except asyncio.TimeoutError:
|
||||
logger.warning(f"Content extraction timed out for {url}")
|
||||
return ContentExtractionResult(
|
||||
url=url,
|
||||
content="",
|
||||
success=False,
|
||||
error=f"Extraction timed out after {self.timeout}s"
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Unexpected error extracting {url}: {e}")
|
||||
return ContentExtractionResult(
|
||||
url=url,
|
||||
content="",
|
||||
success=False,
|
||||
error=str(e)
|
||||
)
|
||||
|
||||
async def extract_batch(
|
||||
self,
|
||||
urls: List[str],
|
||||
include_metadata: bool = True,
|
||||
max_length: Optional[int] = None
|
||||
) -> List[ContentExtractionResult]:
|
||||
"""
|
||||
Extract content from multiple URLs in parallel.
|
||||
|
||||
Args:
|
||||
urls: List of URLs to extract content from
|
||||
include_metadata: Whether to extract title, author, date
|
||||
max_length: Override default max length
|
||||
|
||||
Returns:
|
||||
List of ContentExtractionResult in same order as input URLs
|
||||
"""
|
||||
tasks = [
|
||||
self.extract(url, include_metadata, max_length)
|
||||
for url in urls
|
||||
]
|
||||
results = await asyncio.gather(*tasks)
|
||||
return list(results)
|
||||
|
||||
async def extract_from_html(
|
||||
self,
|
||||
html: str,
|
||||
url: str = "",
|
||||
include_metadata: bool = True,
|
||||
max_length: Optional[int] = None
|
||||
) -> ContentExtractionResult:
|
||||
"""
|
||||
Extract content from raw HTML string.
|
||||
|
||||
Args:
|
||||
html: Raw HTML content
|
||||
url: Optional URL for reference (not fetched)
|
||||
include_metadata: Whether to extract title, author, date
|
||||
max_length: Override default max length
|
||||
|
||||
Returns:
|
||||
ContentExtractionResult with extracted content or error
|
||||
"""
|
||||
effective_max_length = max_length or self.max_length
|
||||
|
||||
def _extract():
|
||||
try:
|
||||
content = trafilatura.extract(
|
||||
html,
|
||||
include_comments=False,
|
||||
include_tables=True,
|
||||
output_format='txt'
|
||||
)
|
||||
|
||||
if not content:
|
||||
return ContentExtractionResult(
|
||||
url=url,
|
||||
content="",
|
||||
success=False,
|
||||
error="No content extracted from HTML"
|
||||
)
|
||||
|
||||
# Truncate if needed
|
||||
if len(content) > effective_max_length:
|
||||
content = content[:effective_max_length] + "..."
|
||||
|
||||
# Extract metadata
|
||||
title = None
|
||||
author = None
|
||||
date = None
|
||||
language = None
|
||||
|
||||
if include_metadata:
|
||||
try:
|
||||
meta_dict = trafilatura.bare_extraction(
|
||||
html,
|
||||
include_comments=False
|
||||
)
|
||||
if meta_dict:
|
||||
title = meta_dict.get('title')
|
||||
author = meta_dict.get('author')
|
||||
date = meta_dict.get('date')
|
||||
language = meta_dict.get('language')
|
||||
except Exception as e:
|
||||
logger.debug(f"Metadata extraction failed: {e}")
|
||||
|
||||
return ContentExtractionResult(
|
||||
url=url,
|
||||
title=title,
|
||||
content=content,
|
||||
author=author,
|
||||
date=date,
|
||||
language=language,
|
||||
success=True,
|
||||
error=None
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"HTML content extraction failed: {e}")
|
||||
return ContentExtractionResult(
|
||||
url=url,
|
||||
content="",
|
||||
success=False,
|
||||
error=str(e)
|
||||
)
|
||||
|
||||
loop = asyncio.get_event_loop()
|
||||
return await loop.run_in_executor(self._executor, _extract)
|
||||
|
||||
async def close(self):
|
||||
"""Shutdown the thread pool executor."""
|
||||
self._executor.shutdown(wait=False)
|
||||
logger.info("ContentExtractor closed")
|
||||
@@ -249,7 +249,8 @@ class OllamaClient:
|
||||
self,
|
||||
prompt: str,
|
||||
model: Optional[str] = None,
|
||||
stream: bool = False
|
||||
stream: bool = False,
|
||||
temperature: Optional[float] = None
|
||||
) -> Optional[str]:
|
||||
"""
|
||||
Generate text completion (for non-embedding use cases).
|
||||
@@ -258,12 +259,15 @@ class OllamaClient:
|
||||
prompt: Input prompt
|
||||
model: Model name (defaults to self.model)
|
||||
stream: Enable streaming response
|
||||
temperature: Sampling temperature (0.0 = deterministic, higher = more creative)
|
||||
None uses model default (~0.7 for mistral-nemo)
|
||||
|
||||
Returns:
|
||||
Generated text or None on failure
|
||||
|
||||
Note: This is primarily for debugging/testing. Use specialized
|
||||
LLM services for production text generation.
|
||||
Note: Use temperature=0.0 for deterministic outputs like JSON parsing,
|
||||
ranking, and factual extraction. Use higher values (0.3-0.7) for
|
||||
creative content generation.
|
||||
"""
|
||||
try:
|
||||
payload = {
|
||||
@@ -272,6 +276,10 @@ class OllamaClient:
|
||||
"stream": stream
|
||||
}
|
||||
|
||||
# Add temperature to options if specified
|
||||
if temperature is not None:
|
||||
payload["options"] = {"temperature": temperature}
|
||||
|
||||
response = await self.client.post(
|
||||
self.generate_url,
|
||||
json=payload
|
||||
|
||||
+13
-3
@@ -62,16 +62,17 @@ class Settings(BaseSettings):
|
||||
# SearXNG Configuration
|
||||
searxng_url: str = Field(default="http://searxng:8080", description="SearXNG URL")
|
||||
|
||||
# Ollama Configuration (for embeddings)
|
||||
# Ollama Configuration
|
||||
ollama_url: str = Field(default="http://ollama:11434", description="Ollama URL")
|
||||
ollama_model: str = Field(default="nomic-embed-text", description="Ollama embedding model")
|
||||
ollama_model: str = Field(default="mistral-nemo-large:latest", description="Ollama LLM model")
|
||||
ollama_embedding_model: str = Field(default="nomic-embed-text", description="Ollama embedding model")
|
||||
|
||||
# HybridRAG Configuration
|
||||
reranker_model: str = Field(default="mistral-nemo", description="Model for LLM re-ranking")
|
||||
reranker_enabled: bool = Field(default=True, description="Enable LLM re-ranking")
|
||||
hybrid_rag_vector_limit: int = Field(default=10, ge=1, le=50, description="Vector search limit")
|
||||
hybrid_rag_graph_limit: int = Field(default=10, ge=1, le=50, description="Graph search limit")
|
||||
hybrid_rag_web_limit: int = Field(default=5, ge=1, le=20, description="Web search limit")
|
||||
vector_similarity_threshold: float = Field(default=0.7, ge=0.0, le=1.0, description="Minimum similarity score for vector results")
|
||||
|
||||
# Entity Linking Fuzzy Matching Configuration
|
||||
entity_linking_min_confidence: float = Field(default=0.70, ge=0.0, le=1.0, description="Minimum confidence for entity-document matching")
|
||||
@@ -89,6 +90,15 @@ class Settings(BaseSettings):
|
||||
app_version: str = Field(default=__version__, description="Application version")
|
||||
debug: bool = Field(default=False, description="Debug mode")
|
||||
|
||||
# RAG Search Configuration
|
||||
search_cache_ttl: int = Field(default=300, ge=0, le=3600, description="Search cache TTL in seconds")
|
||||
search_timeout: int = Field(default=10, ge=1, le=60, description="SearXNG timeout in seconds")
|
||||
search_default_limit: int = Field(default=10, ge=1, le=20, description="Default number of search results")
|
||||
|
||||
# Content Extraction Configuration
|
||||
content_extraction_timeout: int = Field(default=5, ge=1, le=30, description="Trafilatura per-URL timeout in seconds")
|
||||
content_max_length: int = Field(default=2000, ge=500, le=10000, description="Max extracted content length per result")
|
||||
|
||||
@property
|
||||
def qdrant_url(self) -> str:
|
||||
"""Computed Qdrant URL."""
|
||||
|
||||
+54
-11
@@ -13,12 +13,15 @@ from typing import Annotated
|
||||
from fastapi import Depends
|
||||
import logging
|
||||
|
||||
import redis.asyncio as aioredis
|
||||
|
||||
from src.config import Settings, get_settings
|
||||
from src.clients.neo4j_client import Neo4jClient
|
||||
from src.clients.qdrant_client import QdrantClientWrapper
|
||||
from src.clients.wikijs_client import WikiJSClient
|
||||
from src.clients.searxng_client import SearXNGClient
|
||||
from src.clients.ollama_client import OllamaClient
|
||||
from src.clients.content_extractor import ContentExtractor
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -110,12 +113,49 @@ def get_ollama_client() -> OllamaClient:
|
||||
settings = get_settings()
|
||||
client = OllamaClient(
|
||||
base_url=settings.ollama_url,
|
||||
model=settings.ollama_model
|
||||
model=settings.ollama_embedding_model
|
||||
)
|
||||
logger.debug("Created Ollama client instance")
|
||||
return client
|
||||
|
||||
|
||||
@lru_cache
|
||||
def get_redis_client() -> aioredis.Redis:
|
||||
"""
|
||||
Get Redis client singleton for caching.
|
||||
|
||||
Returns:
|
||||
Async Redis client connected to the configured database
|
||||
|
||||
Note: Uses Redis DB 4 (configured for library-desk)
|
||||
"""
|
||||
settings = get_settings()
|
||||
client = aioredis.from_url(
|
||||
settings.redis_url,
|
||||
encoding="utf-8",
|
||||
decode_responses=True
|
||||
)
|
||||
logger.debug(f"Created Redis client: {settings.redis_url}")
|
||||
return client
|
||||
|
||||
|
||||
@lru_cache
|
||||
def get_content_extractor() -> ContentExtractor:
|
||||
"""
|
||||
Get ContentExtractor singleton.
|
||||
|
||||
Returns:
|
||||
Initialized content extraction client using Trafilatura
|
||||
"""
|
||||
settings = get_settings()
|
||||
extractor = ContentExtractor(
|
||||
timeout=settings.content_extraction_timeout,
|
||||
max_length=settings.content_max_length
|
||||
)
|
||||
logger.debug("Created ContentExtractor instance")
|
||||
return extractor
|
||||
|
||||
|
||||
# Type aliases for FastAPI endpoint dependencies
|
||||
# Usage: def my_endpoint(neo4j: Neo4jDep):
|
||||
Neo4jDep = Annotated[Neo4jClient, Depends(get_neo4j_client)]
|
||||
@@ -123,6 +163,8 @@ QdrantDep = Annotated[QdrantClientWrapper, Depends(get_qdrant_client)]
|
||||
WikiJSDep = Annotated[WikiJSClient, Depends(get_wikijs_client)]
|
||||
SearXNGDep = Annotated[SearXNGClient, Depends(get_searxng_client)]
|
||||
OllamaDep = Annotated[OllamaClient, Depends(get_ollama_client)]
|
||||
RedisDep = Annotated[aioredis.Redis, Depends(get_redis_client)]
|
||||
ContentExtractorDep = Annotated[ContentExtractor, Depends(get_content_extractor)]
|
||||
|
||||
|
||||
# Lifecycle management functions
|
||||
@@ -336,20 +378,21 @@ def get_hybrid_rag_service() -> "HybridRAGService":
|
||||
graph_service=get_graph_service(),
|
||||
searxng_client=get_searxng_client(),
|
||||
ollama_client=get_ollama_client(),
|
||||
content_extractor=get_content_extractor(),
|
||||
settings=get_settings()
|
||||
)
|
||||
|
||||
|
||||
# Utility: Get default user from settings or multi_tenancy
|
||||
def get_default_user() -> str:
|
||||
"""
|
||||
Get default user for operations.
|
||||
|
||||
Returns:
|
||||
Default user identifier
|
||||
"""
|
||||
from src.core.multi_tenancy import DEFAULT_USER
|
||||
return DEFAULT_USER
|
||||
@lru_cache
|
||||
def get_rag_search_service() -> "RAGSearchService":
|
||||
"""Get RAGSearchService singleton."""
|
||||
from src.services.rag_search_service import RAGSearchService
|
||||
return RAGSearchService(
|
||||
searxng_client=get_searxng_client(),
|
||||
content_extractor=get_content_extractor(),
|
||||
redis_client=get_redis_client(),
|
||||
settings=get_settings()
|
||||
)
|
||||
|
||||
|
||||
# Authentication
|
||||
|
||||
+75
-101
@@ -8,7 +8,7 @@ Following best practices:
|
||||
- OpenAPI documentation
|
||||
"""
|
||||
|
||||
from fastapi import FastAPI, HTTPException, Depends
|
||||
from fastapi import FastAPI, HTTPException, Depends, Query
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
from pydantic import BaseModel
|
||||
@@ -17,7 +17,10 @@ import logging
|
||||
from pathlib import Path
|
||||
|
||||
from src.config import Settings, get_settings, __version__
|
||||
from src.core.dependencies import verify_api_key
|
||||
from src.core.dependencies import (
|
||||
verify_api_key, QdrantDep, WikiJSDep, OllamaDep, Neo4jDep
|
||||
)
|
||||
from src.core.multi_tenancy import DEFAULT_USER
|
||||
|
||||
# Configure logging
|
||||
logging.basicConfig(
|
||||
@@ -45,7 +48,10 @@ app.add_middleware(
|
||||
)
|
||||
|
||||
# Register routers
|
||||
from src.routers import wiki, tools, graph, vector, hybrid_rag, consolidation, ingestion, entity_linking, webhooks
|
||||
from src.routers import (
|
||||
wiki, tools, graph, vector, hybrid_rag, consolidation,
|
||||
ingestion, entity_linking, webhooks, rag_search, content
|
||||
)
|
||||
|
||||
app.include_router(wiki.router)
|
||||
app.include_router(tools.router)
|
||||
@@ -56,6 +62,8 @@ app.include_router(consolidation.router)
|
||||
app.include_router(ingestion.router)
|
||||
app.include_router(entity_linking.router)
|
||||
app.include_router(webhooks.router)
|
||||
app.include_router(rag_search.router)
|
||||
app.include_router(content.router)
|
||||
|
||||
# Mount static files directory for Wiki.js integration scripts
|
||||
static_dir = Path(__file__).parent.parent / "static"
|
||||
@@ -73,13 +81,6 @@ class HealthResponse(BaseModel):
|
||||
services: Dict[str, Any]
|
||||
|
||||
|
||||
class StatsResponse(BaseModel):
|
||||
"""Statistics response model."""
|
||||
wiki_pages: int
|
||||
neo4j_nodes: int
|
||||
qdrant_vectors: int
|
||||
|
||||
|
||||
# Routes
|
||||
@app.get("/", tags=["Root"])
|
||||
async def root() -> Dict[str, str]:
|
||||
@@ -136,77 +137,6 @@ async def health(settings: Settings = Depends(get_settings)) -> HealthResponse:
|
||||
)
|
||||
|
||||
|
||||
@app.get("/stats", response_model=StatsResponse, tags=["System"])
|
||||
async def stats(
|
||||
api_key: str = Depends(verify_api_key)
|
||||
) -> StatsResponse:
|
||||
"""
|
||||
Get system statistics.
|
||||
Protected endpoint - requires API key.
|
||||
|
||||
TODO: Implement actual stats gathering from:
|
||||
- Neo4j (node count)
|
||||
- Qdrant (vector count)
|
||||
- Wiki.js (page count)
|
||||
"""
|
||||
return StatsResponse(
|
||||
wiki_pages=0,
|
||||
neo4j_nodes=0,
|
||||
qdrant_vectors=0
|
||||
)
|
||||
|
||||
|
||||
# Ingestion endpoints (for Scheduler integration)
|
||||
@app.post("/ingest/document", tags=["Ingestion"])
|
||||
async def ingest_document(
|
||||
document: Dict[str, Any],
|
||||
api_key: str = Depends(verify_api_key)
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Ingest a single document for indexing.
|
||||
Used by The Scheduler to add mirrored documentation to the knowledge base.
|
||||
|
||||
Expected fields:
|
||||
- source: str (e.g., "github", "gitea")
|
||||
- repository: str (e.g., "anthropic-cookbook")
|
||||
- path: str (file path)
|
||||
- content: str (document content)
|
||||
- metadata: dict (commit, author, tags, etc.)
|
||||
|
||||
TODO: Implement document ingestion pipeline:
|
||||
1. Chunk content
|
||||
2. Generate embeddings (Ollama)
|
||||
3. Extract entities (NLP)
|
||||
4. Index in Qdrant
|
||||
5. Create graph nodes/relationships in Neo4j
|
||||
"""
|
||||
return {
|
||||
"message": "Document ingestion not yet implemented",
|
||||
"document_id": f"doc_{document.get('path', 'unknown')}",
|
||||
"status": "stub"
|
||||
}
|
||||
|
||||
|
||||
@app.post("/ingest/batch", tags=["Ingestion"])
|
||||
async def batch_ingest(
|
||||
batch: Dict[str, Any],
|
||||
api_key: str = Depends(verify_api_key)
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Ingest multiple documents in a batch.
|
||||
More efficient than individual ingestion for large syncs.
|
||||
|
||||
TODO: Implement batch processing with task queue
|
||||
"""
|
||||
document_count = len(batch.get("documents", []))
|
||||
return {
|
||||
"message": "Batch ingestion not yet implemented",
|
||||
"batch_id": "batch_stub",
|
||||
"total_documents": document_count,
|
||||
"status": "stub"
|
||||
}
|
||||
|
||||
|
||||
@app.post("/ingest/check-updates", tags=["Ingestion"])
|
||||
async def check_updates(
|
||||
documents: Dict[str, Any],
|
||||
@@ -264,41 +194,85 @@ async def get_repo_status(
|
||||
}
|
||||
|
||||
|
||||
# Query endpoints (stubs for future implementation)
|
||||
# NOTE: /query/hybrid is now implemented in routers/hybrid_rag.py
|
||||
# Query endpoints
|
||||
# NOTE: /query/hybrid is implemented in routers/hybrid_rag.py
|
||||
|
||||
@app.post("/query/semantic", tags=["Query"])
|
||||
async def semantic_query(
|
||||
query: Dict[str, Any],
|
||||
query: str = Query(..., min_length=1, description="Search query text"),
|
||||
user: str = Query(default=DEFAULT_USER, description="User identifier"),
|
||||
limit: int = Query(default=10, ge=1, le=100, description="Maximum results"),
|
||||
score_threshold: float = Query(default=0.5, ge=0.0, le=1.0, description="Minimum similarity score"),
|
||||
qdrant_client: QdrantDep = None,
|
||||
wiki_client: WikiJSDep = None,
|
||||
ollama_client: OllamaDep = None,
|
||||
api_key: str = Depends(verify_api_key)
|
||||
) -> Dict[str, Any]:
|
||||
):
|
||||
"""
|
||||
Semantic search via Qdrant.
|
||||
Pure vector similarity search.
|
||||
Semantic search via Qdrant vector similarity.
|
||||
|
||||
TODO: Implement semantic search
|
||||
Searches document chunks using embedding similarity. Returns matching
|
||||
chunks with relevance scores, page titles, and paths.
|
||||
|
||||
**Example:**
|
||||
```
|
||||
POST /query/semantic?query=docker%20configuration&user=jpmschweitzer&limit=10
|
||||
```
|
||||
|
||||
**Returns:** List of matching chunks with similarity scores (0-1)
|
||||
"""
|
||||
return {
|
||||
"message": "Semantic search not yet implemented",
|
||||
"query": query
|
||||
}
|
||||
from src.services.vector_service import VectorService
|
||||
|
||||
vector_service = VectorService(qdrant_client, wiki_client, ollama_client)
|
||||
try:
|
||||
return await vector_service.search(
|
||||
query=query,
|
||||
user=user,
|
||||
limit=limit,
|
||||
score_threshold=score_threshold
|
||||
)
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=400, detail=str(e))
|
||||
except Exception as e:
|
||||
logger.error(f"Semantic search failed: {e}", exc_info=True)
|
||||
raise HTTPException(status_code=500, detail="Search failed")
|
||||
|
||||
|
||||
@app.post("/query/graph", tags=["Query"])
|
||||
async def graph_query(
|
||||
query: Dict[str, Any],
|
||||
query: str = Query(..., description="Cypher query to execute"),
|
||||
user: str = Query(default=DEFAULT_USER, description="User for scoping (auto-filters results)"),
|
||||
neo4j_client: Neo4jDep = None,
|
||||
wiki_client: WikiJSDep = None,
|
||||
api_key: str = Depends(verify_api_key)
|
||||
) -> Dict[str, Any]:
|
||||
):
|
||||
"""
|
||||
Graph traversal via Neo4j.
|
||||
Execute Cypher queries.
|
||||
Execute a Cypher query against the Neo4j knowledge graph.
|
||||
|
||||
TODO: Implement graph queries
|
||||
Queries are automatically scoped to the user's data for security.
|
||||
Use this for custom graph traversals beyond what /graph/nodes provides.
|
||||
|
||||
**Example:**
|
||||
```
|
||||
POST /query/graph?query=MATCH%20(d:Document)-[:MENTIONS]->(p:Person)%20RETURN%20d,p&user=jpmschweitzer
|
||||
```
|
||||
|
||||
**Security:** All queries are user-scoped to prevent cross-user data access.
|
||||
"""
|
||||
return {
|
||||
"message": "Graph query not yet implemented",
|
||||
"query": query
|
||||
}
|
||||
from src.services.graph_service import GraphService
|
||||
|
||||
graph_service = GraphService(neo4j_client, wiki_client)
|
||||
try:
|
||||
return await graph_service.execute_query(
|
||||
query=query,
|
||||
parameters={},
|
||||
user=user
|
||||
)
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=400, detail=str(e))
|
||||
except Exception as e:
|
||||
logger.error(f"Graph query failed: {e}", exc_info=True)
|
||||
raise HTTPException(status_code=500, detail="Query execution failed")
|
||||
|
||||
|
||||
# Deduplication endpoints
|
||||
|
||||
@@ -0,0 +1,69 @@
|
||||
"""
|
||||
Content extraction models for Library Desk.
|
||||
|
||||
Pydantic models for content extraction requests and responses.
|
||||
"""
|
||||
|
||||
from typing import Optional, List
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
|
||||
class ContentExtractionResult(BaseModel):
|
||||
"""Result of extracting content from a single URL."""
|
||||
|
||||
url: str = Field(..., description="The URL that was processed")
|
||||
title: Optional[str] = Field(None, description="Page title if extracted")
|
||||
content: str = Field("", description="Extracted main text content")
|
||||
author: Optional[str] = Field(None, description="Author if available")
|
||||
date: Optional[str] = Field(None, description="Publication date if available (ISO format)")
|
||||
language: Optional[str] = Field(None, description="Detected language code")
|
||||
success: bool = Field(..., description="Whether extraction succeeded")
|
||||
error: Optional[str] = Field(None, description="Error message if extraction failed")
|
||||
|
||||
|
||||
class ContentExtractionRequest(BaseModel):
|
||||
"""Request to extract content from a single URL."""
|
||||
|
||||
url: str = Field(..., min_length=1, description="URL to extract content from")
|
||||
include_metadata: bool = Field(default=True, description="Include title, author, date metadata")
|
||||
max_length: Optional[int] = Field(
|
||||
None,
|
||||
ge=100,
|
||||
le=50000,
|
||||
description="Override default max content length"
|
||||
)
|
||||
|
||||
|
||||
class ContentExtractionResponse(BaseModel):
|
||||
"""Response for single URL extraction."""
|
||||
|
||||
result: ContentExtractionResult
|
||||
extraction_time_ms: int = Field(..., ge=0, description="Time taken to extract content")
|
||||
|
||||
|
||||
class BatchContentExtractionRequest(BaseModel):
|
||||
"""Request to extract content from multiple URLs."""
|
||||
|
||||
urls: List[str] = Field(
|
||||
...,
|
||||
min_length=1,
|
||||
max_length=20,
|
||||
description="URLs to extract content from (max 20)"
|
||||
)
|
||||
include_metadata: bool = Field(default=True, description="Include title, author, date metadata")
|
||||
max_length: Optional[int] = Field(
|
||||
None,
|
||||
ge=100,
|
||||
le=50000,
|
||||
description="Override default max content length"
|
||||
)
|
||||
|
||||
|
||||
class BatchContentExtractionResponse(BaseModel):
|
||||
"""Response for batch URL extraction."""
|
||||
|
||||
results: List[ContentExtractionResult]
|
||||
total_urls: int = Field(..., ge=0, description="Total number of URLs processed")
|
||||
successful: int = Field(..., ge=0, description="Number of successful extractions")
|
||||
failed: int = Field(..., ge=0, description="Number of failed extractions")
|
||||
extraction_time_ms: int = Field(..., ge=0, description="Total time for batch extraction")
|
||||
@@ -0,0 +1,88 @@
|
||||
"""
|
||||
RAG search models for Library Desk.
|
||||
|
||||
Pydantic models for web/news/image search requests and responses.
|
||||
"""
|
||||
|
||||
from enum import Enum
|
||||
from typing import Optional, List
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from src.core.multi_tenancy import DEFAULT_USER
|
||||
|
||||
|
||||
class SearchType(str, Enum):
|
||||
"""Supported search types."""
|
||||
WEB = "web"
|
||||
NEWS = "news"
|
||||
IMAGES = "images"
|
||||
|
||||
|
||||
class RAGSearchRequest(BaseModel):
|
||||
"""Request for RAG search endpoint."""
|
||||
|
||||
query: str = Field(
|
||||
...,
|
||||
min_length=1,
|
||||
max_length=500,
|
||||
description="The search query"
|
||||
)
|
||||
search_type: SearchType = Field(
|
||||
default=SearchType.WEB,
|
||||
description="Type of search: web, news, or images"
|
||||
)
|
||||
limit: int = Field(
|
||||
default=10,
|
||||
ge=1,
|
||||
le=20,
|
||||
description="Maximum number of results (1-20)"
|
||||
)
|
||||
user: str = Field(
|
||||
default=DEFAULT_USER,
|
||||
description="User identifier for rate limiting/personalization"
|
||||
)
|
||||
|
||||
|
||||
class RAGSearchResult(BaseModel):
|
||||
"""A single search result with extracted content."""
|
||||
|
||||
title: str = Field(..., description="Title of the result")
|
||||
url: str = Field(..., description="URL of the source")
|
||||
content: str = Field(
|
||||
"",
|
||||
description="Full extracted text via Trafilatura (max ~2000 chars)"
|
||||
)
|
||||
snippet: str = Field(
|
||||
"",
|
||||
description="Original search engine snippet (150-300 chars)"
|
||||
)
|
||||
source: str = Field(..., description="Domain name of the source")
|
||||
published_date: Optional[str] = Field(
|
||||
None,
|
||||
description="Publication date in ISO format if available"
|
||||
)
|
||||
|
||||
|
||||
class RAGSearchResponse(BaseModel):
|
||||
"""Response from RAG search endpoint."""
|
||||
|
||||
query: str = Field(..., description="Echo of the original query")
|
||||
search_type: SearchType = Field(..., description="Type of search performed")
|
||||
results: List[RAGSearchResult] = Field(
|
||||
default_factory=list,
|
||||
description="List of search results with extracted content"
|
||||
)
|
||||
total_results: int = Field(
|
||||
...,
|
||||
ge=0,
|
||||
description="Number of results returned"
|
||||
)
|
||||
search_time_ms: int = Field(
|
||||
...,
|
||||
ge=0,
|
||||
description="Total time for search and content extraction"
|
||||
)
|
||||
sources_summary: str = Field(
|
||||
"",
|
||||
description="Markdown-formatted list of all source URLs"
|
||||
)
|
||||
@@ -0,0 +1,132 @@
|
||||
"""
|
||||
Content extraction router for Library Desk API.
|
||||
|
||||
Endpoints for extracting main content from web URLs using Trafilatura.
|
||||
"""
|
||||
|
||||
import time
|
||||
from fastapi import APIRouter, HTTPException, Depends
|
||||
import logging
|
||||
|
||||
from src.models.content import (
|
||||
ContentExtractionRequest,
|
||||
ContentExtractionResponse,
|
||||
BatchContentExtractionRequest,
|
||||
BatchContentExtractionResponse,
|
||||
)
|
||||
from src.clients.content_extractor import ContentExtractor
|
||||
from src.core.dependencies import verify_api_key
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter(prefix="/content", tags=["Content Extraction"])
|
||||
|
||||
|
||||
# Lazy import to avoid circular dependency
|
||||
def get_content_extractor() -> ContentExtractor:
|
||||
"""Get content extractor instance."""
|
||||
from src.core.dependencies import get_content_extractor as _get_extractor
|
||||
return _get_extractor()
|
||||
|
||||
|
||||
@router.post("/extract", response_model=ContentExtractionResponse)
|
||||
async def extract_content(
|
||||
request: ContentExtractionRequest,
|
||||
api_key: str = Depends(verify_api_key)
|
||||
):
|
||||
"""
|
||||
Extract main content from a single URL.
|
||||
|
||||
Uses Trafilatura to fetch the URL and extract the main text content,
|
||||
removing navigation, ads, and other boilerplate.
|
||||
|
||||
**Example Request:**
|
||||
```json
|
||||
{
|
||||
"url": "https://example.com/article",
|
||||
"include_metadata": true,
|
||||
"max_length": 2000
|
||||
}
|
||||
```
|
||||
|
||||
**Returns:** Extracted content with optional metadata (title, author, date)
|
||||
"""
|
||||
start_time = time.time()
|
||||
|
||||
try:
|
||||
extractor = get_content_extractor()
|
||||
result = await extractor.extract(
|
||||
url=request.url,
|
||||
include_metadata=request.include_metadata,
|
||||
max_length=request.max_length
|
||||
)
|
||||
|
||||
extraction_time_ms = int((time.time() - start_time) * 1000)
|
||||
|
||||
return ContentExtractionResponse(
|
||||
result=result,
|
||||
extraction_time_ms=extraction_time_ms
|
||||
)
|
||||
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=400, detail=str(e))
|
||||
except Exception as e:
|
||||
logger.error(f"Content extraction failed: {e}", exc_info=True)
|
||||
raise HTTPException(status_code=500, detail="Content extraction failed")
|
||||
|
||||
|
||||
@router.post("/extract/batch", response_model=BatchContentExtractionResponse)
|
||||
async def extract_content_batch(
|
||||
request: BatchContentExtractionRequest,
|
||||
api_key: str = Depends(verify_api_key)
|
||||
):
|
||||
"""
|
||||
Extract content from multiple URLs in parallel.
|
||||
|
||||
Processes up to 20 URLs concurrently with per-URL timeouts.
|
||||
Failed extractions are included in results with success=false.
|
||||
|
||||
**Example Request:**
|
||||
```json
|
||||
{
|
||||
"urls": [
|
||||
"https://example.com/article1",
|
||||
"https://example.com/article2"
|
||||
],
|
||||
"include_metadata": true,
|
||||
"max_length": 2000
|
||||
}
|
||||
```
|
||||
|
||||
**Returns:** List of extraction results with success/failure counts
|
||||
"""
|
||||
start_time = time.time()
|
||||
|
||||
if not request.urls:
|
||||
raise HTTPException(status_code=400, detail="URLs list cannot be empty")
|
||||
|
||||
try:
|
||||
extractor = get_content_extractor()
|
||||
results = await extractor.extract_batch(
|
||||
urls=request.urls,
|
||||
include_metadata=request.include_metadata,
|
||||
max_length=request.max_length
|
||||
)
|
||||
|
||||
extraction_time_ms = int((time.time() - start_time) * 1000)
|
||||
successful = sum(1 for r in results if r.success)
|
||||
failed = len(results) - successful
|
||||
|
||||
return BatchContentExtractionResponse(
|
||||
results=results,
|
||||
total_urls=len(request.urls),
|
||||
successful=successful,
|
||||
failed=failed,
|
||||
extraction_time_ms=extraction_time_ms
|
||||
)
|
||||
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=400, detail=str(e))
|
||||
except Exception as e:
|
||||
logger.error(f"Batch content extraction failed: {e}", exc_info=True)
|
||||
raise HTTPException(status_code=500, detail="Batch extraction failed")
|
||||
@@ -15,8 +15,6 @@ from src.models.graph import (
|
||||
MindMapResponse
|
||||
)
|
||||
from src.services.graph_service import GraphService
|
||||
from src.clients.neo4j_client import Neo4jClient
|
||||
from src.clients.wikijs_client import WikiJSClient
|
||||
from src.core.dependencies import Neo4jDep, WikiJSDep, verify_api_key
|
||||
from src.core.multi_tenancy import DEFAULT_USER
|
||||
|
||||
|
||||
@@ -10,13 +10,9 @@ import logging
|
||||
|
||||
from src.models.hybrid_rag import HybridRAGRequest, HybridRAGResponse
|
||||
from src.services.hybrid_rag_service import HybridRAGService
|
||||
from src.services.vector_service import VectorService
|
||||
from src.services.graph_service import GraphService
|
||||
from src.clients.searxng_client import SearXNGClient
|
||||
from src.clients.ollama_client import OllamaClient
|
||||
from src.core.dependencies import (
|
||||
Neo4jDep, WikiJSDep, QdrantDep, OllamaDep,
|
||||
SearXNGDep, verify_api_key, get_settings
|
||||
SearXNGDep, ContentExtractorDep, verify_api_key, get_settings
|
||||
)
|
||||
from src.config import Settings
|
||||
|
||||
@@ -32,6 +28,7 @@ def get_hybrid_rag_service(
|
||||
qdrant_client: QdrantDep,
|
||||
ollama_client: OllamaDep,
|
||||
searxng_client: SearXNGDep,
|
||||
content_extractor: ContentExtractorDep,
|
||||
settings: Settings = Depends(get_settings)
|
||||
) -> HybridRAGService:
|
||||
"""Get HybridRAG service instance with all dependencies."""
|
||||
@@ -48,6 +45,7 @@ def get_hybrid_rag_service(
|
||||
graph_service=graph_service,
|
||||
searxng_client=searxng_client,
|
||||
ollama_client=ollama_client,
|
||||
content_extractor=content_extractor,
|
||||
settings=settings
|
||||
)
|
||||
|
||||
|
||||
@@ -0,0 +1,87 @@
|
||||
"""
|
||||
RAG search router for Library Desk API.
|
||||
|
||||
Endpoints for web, news, and image search with content extraction.
|
||||
"""
|
||||
|
||||
import httpx
|
||||
from fastapi import APIRouter, HTTPException, Depends
|
||||
import logging
|
||||
|
||||
from src.models.rag_search import RAGSearchRequest, RAGSearchResponse
|
||||
from src.services.rag_search_service import RAGSearchService
|
||||
from src.core.dependencies import verify_api_key
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter(prefix="/rag", tags=["RAG Search"])
|
||||
|
||||
|
||||
# Lazy import to avoid circular dependency
|
||||
def get_rag_search_service() -> RAGSearchService:
|
||||
"""Get RAG search service instance."""
|
||||
from src.core.dependencies import get_rag_search_service as _get_service
|
||||
return _get_service()
|
||||
|
||||
|
||||
@router.post("/search", response_model=RAGSearchResponse)
|
||||
async def search(
|
||||
request: RAGSearchRequest,
|
||||
api_key: str = Depends(verify_api_key)
|
||||
):
|
||||
"""
|
||||
Execute RAG-optimized web search with content extraction.
|
||||
|
||||
Searches via SearXNG and extracts full content from results using
|
||||
Trafilatura. Results are cached in Redis for efficiency.
|
||||
|
||||
**Search Types:**
|
||||
- `web`: General web search (default)
|
||||
- `news`: News articles with recency filtering
|
||||
- `images`: Image search results
|
||||
|
||||
**Example Request:**
|
||||
```json
|
||||
{
|
||||
"query": "Python async programming best practices",
|
||||
"search_type": "web",
|
||||
"limit": 10,
|
||||
"user": "default"
|
||||
}
|
||||
```
|
||||
|
||||
**Response includes:**
|
||||
- Full extracted text content per result
|
||||
- Original search snippets
|
||||
- Source domain names
|
||||
- Markdown sources summary for LLM consumption
|
||||
|
||||
**Error Codes:**
|
||||
- 400: Invalid query (empty or too long)
|
||||
- 502: Search provider (SearXNG) error
|
||||
- 504: Search timeout
|
||||
"""
|
||||
try:
|
||||
service = get_rag_search_service()
|
||||
response = await service.search(
|
||||
query=request.query,
|
||||
search_type=request.search_type,
|
||||
limit=request.limit,
|
||||
user=request.user
|
||||
)
|
||||
return response
|
||||
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=400, detail=str(e))
|
||||
|
||||
except httpx.TimeoutException:
|
||||
logger.error(f"Search timed out for query: {request.query}")
|
||||
raise HTTPException(status_code=504, detail="Search timed out")
|
||||
|
||||
except httpx.HTTPError as e:
|
||||
logger.error(f"Search provider error: {e}")
|
||||
raise HTTPException(status_code=502, detail="Search provider error")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"RAG search failed: {e}", exc_info=True)
|
||||
raise HTTPException(status_code=500, detail="Search failed")
|
||||
+5
-4
@@ -5,8 +5,7 @@ Endpoints for wiki page and dossier management.
|
||||
All operations are scoped to user namespaces for multi-tenancy.
|
||||
"""
|
||||
|
||||
from fastapi import APIRouter, HTTPException, Depends, Query, Security, BackgroundTasks
|
||||
from fastapi.security import HTTPAuthorizationCredentials
|
||||
from fastapi import APIRouter, HTTPException, Depends, Query, BackgroundTasks
|
||||
from typing import Optional
|
||||
import logging
|
||||
|
||||
@@ -24,7 +23,7 @@ from src.clients.neo4j_client import Neo4jClient
|
||||
from src.clients.qdrant_client import QdrantClientWrapper
|
||||
from src.clients.ollama_client import OllamaClient
|
||||
from src.core.dependencies import (
|
||||
WikiJSDep, Neo4jDep, QdrantDep, OllamaDep, SearXNGDep,
|
||||
WikiJSDep, Neo4jDep, QdrantDep, OllamaDep, SearXNGDep, ContentExtractorDep,
|
||||
verify_api_key, get_settings, get_hybrid_rag_service, get_ingestion_service
|
||||
)
|
||||
from src.core.multi_tenancy import DEFAULT_USER
|
||||
@@ -180,6 +179,7 @@ async def smart_create_page(
|
||||
qdrant_client: QdrantDep,
|
||||
ollama_client: OllamaDep,
|
||||
searxng_client: SearXNGDep,
|
||||
content_extractor: ContentExtractorDep,
|
||||
settings: Settings = Depends(get_settings),
|
||||
api_key: str = Depends(verify_api_key)
|
||||
):
|
||||
@@ -225,9 +225,10 @@ async def smart_create_page(
|
||||
graph_service=graph_service,
|
||||
searxng_client=searxng_client,
|
||||
ollama_client=ollama_client,
|
||||
content_extractor=content_extractor,
|
||||
settings=settings
|
||||
)
|
||||
wiki_page_writer = WikiPageWriter(ollama_client=ollama_client)
|
||||
wiki_page_writer = WikiPageWriter(ollama_client=ollama_client, settings=settings)
|
||||
|
||||
# Step 1-5: Research + Generate + Create page
|
||||
page, research_data = await wiki_service.smart_create_page(
|
||||
|
||||
@@ -47,7 +47,7 @@ class ConsolidationService:
|
||||
self.ollama = ollama
|
||||
self.wiki = wiki
|
||||
self.settings = settings
|
||||
self.wiki_page_writer = WikiPageWriter(ollama_client=ollama)
|
||||
self.wiki_page_writer = WikiPageWriter(ollama_client=ollama, settings=settings)
|
||||
self.ingestion_service = ingestion_service # Optional to avoid circular dependency
|
||||
|
||||
async def consolidate_knowledge(
|
||||
@@ -410,13 +410,19 @@ This is a PERSONAL knowledge base using Schema.org-aligned taxonomy that capture
|
||||
- Projects: Work projects, personal projects (Schema.org: Project)
|
||||
- Reference: General knowledge, how-tos (Custom extension)
|
||||
|
||||
Identify information worth documenting:
|
||||
1. New topics/people/things that deserve their own wiki page
|
||||
2. Facts that could enhance existing pages
|
||||
3. Entities (people, places, things, concepts) for the knowledge graph
|
||||
ANALYSIS STEPS:
|
||||
1. Read each web result carefully for substantive, factual content
|
||||
2. Identify genuinely novel information not likely already known
|
||||
3. Match topics to appropriate taxonomy categories
|
||||
4. Generate valid paths following the exact format below
|
||||
|
||||
Be INCLUSIVE - if someone searched for it, it's likely worth documenting.
|
||||
Personal information is just as valuable as technical information.
|
||||
RULES:
|
||||
- Do NOT suggest pages for topics with insufficient information in results
|
||||
- Do NOT invent entities not explicitly mentioned in results
|
||||
- Do NOT suggest paths that don't match the taxonomy exactly
|
||||
- Do NOT suggest generic or vague page topics
|
||||
- Be CONSERVATIVE - fewer high-quality suggestions is better than many low-quality ones
|
||||
- ONLY suggest documentation for substantive, specific information
|
||||
|
||||
**CRITICAL: Use ONLY these Schema.org-aligned path prefixes (case-sensitive):**
|
||||
|
||||
@@ -462,11 +468,12 @@ Return ONLY valid JSON:
|
||||
JSON:"""
|
||||
|
||||
try:
|
||||
# Call Ollama for analysis
|
||||
# Call Ollama for analysis (temperature=0.0 for consistent classification)
|
||||
response = await self.ollama.generate_text(
|
||||
prompt=prompt,
|
||||
model=self.settings.reranker_model, # Use mistral-nemo
|
||||
stream=False
|
||||
model=self.settings.ollama_model,
|
||||
stream=False,
|
||||
temperature=0.0
|
||||
)
|
||||
|
||||
if not response:
|
||||
|
||||
@@ -1077,8 +1077,18 @@ Feel free to expand it with more details!
|
||||
search_query,
|
||||
{"terms": all_terms, "limit": limit}
|
||||
)
|
||||
logger.info(f"Graph search found {len(results)} documents")
|
||||
return results
|
||||
|
||||
# Deduplicate by page_id (safety net for any edge cases)
|
||||
seen_page_ids = set()
|
||||
unique_results = []
|
||||
for r in results:
|
||||
page_id = r.get("page_id")
|
||||
if page_id and page_id not in seen_page_ids:
|
||||
seen_page_ids.add(page_id)
|
||||
unique_results.append(r)
|
||||
|
||||
logger.info(f"Graph search found {len(unique_results)} unique documents (raw: {len(results)})")
|
||||
return unique_results
|
||||
except Exception as e:
|
||||
logger.error(f"Graph document search failed: {e}", exc_info=True)
|
||||
return []
|
||||
|
||||
@@ -6,7 +6,7 @@ HybridRAG service combining vector, graph, and web search.
|
||||
1. Parallel Retrieval - Vector + Graph + Web search
|
||||
2. RRF Fusion - Merge results with Reciprocal Rank Fusion
|
||||
3. Enrichment - Add related dossiers via graph
|
||||
4. LLM Re-ranking - Re-rank with mistral-nemo
|
||||
4. LLM Re-ranking - Re-rank with configured Ollama model
|
||||
5. Context Formatting - Format for LLM consumption
|
||||
6. Persistence - Store for Librarian processing
|
||||
"""
|
||||
@@ -22,6 +22,7 @@ from src.services.vector_service import VectorService
|
||||
from src.services.graph_service import GraphService
|
||||
from src.clients.searxng_client import SearXNGClient
|
||||
from src.clients.ollama_client import OllamaClient
|
||||
from src.clients.content_extractor import ContentExtractor
|
||||
from src.config import Settings
|
||||
from src.models.hybrid_rag import (
|
||||
HybridRAGConfig, HybridRAGRequest, HybridRAGResponse,
|
||||
@@ -44,6 +45,7 @@ class HybridRAGService:
|
||||
graph_service: GraphService,
|
||||
searxng_client: SearXNGClient,
|
||||
ollama_client: OllamaClient,
|
||||
content_extractor: ContentExtractor,
|
||||
settings: Settings
|
||||
):
|
||||
"""
|
||||
@@ -54,14 +56,16 @@ class HybridRAGService:
|
||||
graph_service: Service for Neo4j graph search
|
||||
searxng_client: Client for web search
|
||||
ollama_client: Client for LLM (keyword extraction, re-ranking)
|
||||
content_extractor: Client for extracting full content from URLs
|
||||
settings: Application settings
|
||||
"""
|
||||
self.vector = vector_service
|
||||
self.graph = graph_service
|
||||
self.searxng = searxng_client
|
||||
self.ollama = ollama_client
|
||||
self.content_extractor = content_extractor
|
||||
self.settings = settings
|
||||
self.reranker_model = settings.reranker_model
|
||||
self.reranker_model = settings.ollama_model
|
||||
|
||||
async def search(
|
||||
self,
|
||||
@@ -101,14 +105,20 @@ class HybridRAGService:
|
||||
timing["graph_ms"] = raw_results.get("timing", {}).get("graph_ms", 0)
|
||||
timing["web_ms"] = raw_results.get("timing", {}).get("web_ms", 0)
|
||||
|
||||
# Phase 2: RRF Fusion
|
||||
# Phase 2: Two-Stage RRF Fusion
|
||||
phase2_start = time.time()
|
||||
|
||||
# Stage 1: Merge wiki sources (vector + graph) into single ranking
|
||||
wiki_merged = self._merge_wiki_sources(
|
||||
vector_results=raw_results.get("vector", []),
|
||||
graph_results=raw_results.get("graph", []),
|
||||
k=config.rrf_k
|
||||
)
|
||||
|
||||
# Stage 2: Final RRF between wiki and web (equal footing)
|
||||
fused_results = self._reciprocal_rank_fusion(
|
||||
results_by_source={
|
||||
"vector": raw_results.get("vector", []),
|
||||
"graph": raw_results.get("graph", []),
|
||||
"web": raw_results.get("web", [])
|
||||
},
|
||||
wiki_results=wiki_merged,
|
||||
web_results=raw_results.get("web", []),
|
||||
k=config.rrf_k
|
||||
)
|
||||
timing["fusion_ms"] = (time.time() - phase2_start) * 1000
|
||||
@@ -185,25 +195,21 @@ class HybridRAGService:
|
||||
Returns:
|
||||
Dictionary with keywords, entities, synonyms, expansions
|
||||
"""
|
||||
prompt = f"""Extract search terms from this query. For each important word, provide synonyms and expansions.
|
||||
prompt = f"""Extract search terms from this query.
|
||||
|
||||
Query: "{query}"
|
||||
|
||||
Return ONLY valid JSON:
|
||||
{{
|
||||
"core_keywords": ["key", "words", "from", "query"],
|
||||
"synonyms": {{
|
||||
"word": ["alternative", "terms"]
|
||||
}}
|
||||
}}
|
||||
RULES:
|
||||
- Extract ONLY keywords explicitly present or directly implied in the query
|
||||
- Do NOT invent terms, concepts, or synonyms not clearly related
|
||||
- Do NOT add general knowledge or associations
|
||||
- Provide synonyms ONLY for technical terms with well-known alternatives
|
||||
- Return valid JSON only, no commentary
|
||||
|
||||
Example for "Docker container hosting":
|
||||
Return format:
|
||||
{{
|
||||
"core_keywords": ["docker", "container", "hosting"],
|
||||
"synonyms": {{
|
||||
"docker": ["containerization", "container runtime"],
|
||||
"hosting": ["server", "infrastructure"]
|
||||
}}
|
||||
"core_keywords": ["words", "from", "query"],
|
||||
"synonyms": {{"term": ["direct", "alternatives"]}}
|
||||
}}
|
||||
|
||||
JSON:"""
|
||||
@@ -211,7 +217,8 @@ JSON:"""
|
||||
try:
|
||||
response = await self.ollama.generate_text(
|
||||
prompt=prompt,
|
||||
model=self.reranker_model
|
||||
model=self.reranker_model,
|
||||
temperature=0.0 # Deterministic for consistent extraction
|
||||
)
|
||||
|
||||
# Parse JSON response (handle potential extra text)
|
||||
@@ -284,7 +291,8 @@ JSON:"""
|
||||
response = await self.vector.search(
|
||||
query=query,
|
||||
user=user,
|
||||
limit=config.vector_limit
|
||||
limit=config.vector_limit,
|
||||
score_threshold=self.settings.vector_similarity_threshold
|
||||
)
|
||||
results = [
|
||||
{
|
||||
@@ -309,11 +317,19 @@ JSON:"""
|
||||
async def graph_search():
|
||||
start = time.time()
|
||||
try:
|
||||
# Skip synonyms for graph search - only use core keywords
|
||||
# Synonyms like "author" can match unrelated entities like "author2000"
|
||||
graph_keywords = {
|
||||
"core_keywords": keywords_data.get("core_keywords", []),
|
||||
"entities": keywords_data.get("entities", []),
|
||||
"synonyms": {}, # No synonyms for exact entity matching
|
||||
"expansions": {}
|
||||
}
|
||||
results = await self.graph.search_documents(
|
||||
query=query,
|
||||
user=user,
|
||||
limit=config.graph_limit,
|
||||
keywords_data=keywords_data
|
||||
keywords_data=graph_keywords
|
||||
)
|
||||
formatted = [
|
||||
{
|
||||
@@ -334,7 +350,7 @@ JSON:"""
|
||||
|
||||
tasks["graph"] = graph_search()
|
||||
|
||||
# Web search
|
||||
# Web search with content extraction
|
||||
if config.enable_web:
|
||||
async def web_search():
|
||||
start = time.time()
|
||||
@@ -343,11 +359,24 @@ JSON:"""
|
||||
query=query,
|
||||
limit=config.web_limit
|
||||
)
|
||||
|
||||
# Extract full content from URLs using Trafilatura
|
||||
urls = [r.get("url") for r in results if r.get("url")]
|
||||
extraction_results = await self.content_extractor.extract_batch(urls)
|
||||
|
||||
# Map extracted content back to results by URL
|
||||
url_to_content = {
|
||||
ext.url: ext.content
|
||||
for ext in extraction_results
|
||||
if ext.success and ext.content
|
||||
}
|
||||
|
||||
formatted = [
|
||||
{
|
||||
"url": r.get("url"),
|
||||
"title": r.get("title", ""),
|
||||
"content": r.get("content", ""),
|
||||
"content": url_to_content.get(r.get("url"), r.get("content", "")),
|
||||
"snippet": r.get("content", ""), # Keep original snippet
|
||||
"engine": r.get("engine", ""),
|
||||
"source": "web"
|
||||
}
|
||||
@@ -377,52 +406,134 @@ JSON:"""
|
||||
|
||||
return output
|
||||
|
||||
def _reciprocal_rank_fusion(
|
||||
def _merge_wiki_sources(
|
||||
self,
|
||||
results_by_source: Dict[str, List],
|
||||
vector_results: List[Dict],
|
||||
graph_results: List[Dict],
|
||||
k: int = 60
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Phase 2: Merge results using Reciprocal Rank Fusion.
|
||||
Stage 1: Merge vector and graph into single wiki ranking using RRF.
|
||||
|
||||
RRF formula: score = sum(1 / (k + rank)) for each source
|
||||
Both sources search the same wiki pool, so we combine them before
|
||||
final RRF with web to avoid double-counting wiki pages.
|
||||
|
||||
Args:
|
||||
results_by_source: Results from each source
|
||||
vector_results: Results from vector search
|
||||
graph_results: Results from graph search
|
||||
k: RRF constant (default 60)
|
||||
|
||||
Returns:
|
||||
Merged and sorted results
|
||||
Merged wiki results sorted by wiki RRF score
|
||||
"""
|
||||
wiki_scores = {}
|
||||
|
||||
# Process vector results
|
||||
for rank, result in enumerate(vector_results, start=1):
|
||||
page_id = result.get("page_id")
|
||||
if not page_id:
|
||||
continue
|
||||
result_id = f"page_{page_id}"
|
||||
|
||||
if result_id not in wiki_scores:
|
||||
wiki_scores[result_id] = {
|
||||
"result": dict(result), # Copy to avoid mutation
|
||||
"wiki_rrf_score": 0.0,
|
||||
"found_by": []
|
||||
}
|
||||
|
||||
wiki_scores[result_id]["wiki_rrf_score"] += 1 / (k + rank)
|
||||
wiki_scores[result_id]["found_by"].append("vector")
|
||||
|
||||
# Process graph results
|
||||
for rank, result in enumerate(graph_results, start=1):
|
||||
page_id = result.get("page_id")
|
||||
if not page_id:
|
||||
continue
|
||||
result_id = f"page_{page_id}"
|
||||
|
||||
if result_id not in wiki_scores:
|
||||
wiki_scores[result_id] = {
|
||||
"result": dict(result),
|
||||
"wiki_rrf_score": 0.0,
|
||||
"found_by": []
|
||||
}
|
||||
|
||||
wiki_scores[result_id]["wiki_rrf_score"] += 1 / (k + rank)
|
||||
wiki_scores[result_id]["found_by"].append("graph")
|
||||
|
||||
# Add graph metadata to existing result
|
||||
wiki_scores[result_id]["result"]["entity_matches"] = result.get("entity_matches")
|
||||
wiki_scores[result_id]["result"]["matched_entities"] = result.get("matched_entities")
|
||||
|
||||
# Sort by wiki RRF score
|
||||
sorted_wiki = sorted(
|
||||
wiki_scores.values(),
|
||||
key=lambda x: x["wiki_rrf_score"],
|
||||
reverse=True
|
||||
)
|
||||
|
||||
# Return merged results with wiki ranking
|
||||
merged = []
|
||||
for wiki_rank, item in enumerate(sorted_wiki, start=1):
|
||||
merged.append({
|
||||
**item["result"],
|
||||
"wiki_rank": wiki_rank,
|
||||
"wiki_rrf_score": item["wiki_rrf_score"],
|
||||
"found_by": item["found_by"],
|
||||
"source": "wiki"
|
||||
})
|
||||
|
||||
logger.info(f"Wiki merge: {len(merged)} unique pages from vector+graph")
|
||||
return merged
|
||||
|
||||
def _reciprocal_rank_fusion(
|
||||
self,
|
||||
wiki_results: List[Dict],
|
||||
web_results: List[Dict],
|
||||
k: int = 60
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Stage 2: Final RRF between wiki (single source) and web.
|
||||
|
||||
Wiki results are pre-merged from vector+graph, so wiki and web
|
||||
now compete on equal footing.
|
||||
|
||||
Args:
|
||||
wiki_results: Pre-merged wiki results from _merge_wiki_sources()
|
||||
web_results: Results from web search
|
||||
k: RRF constant (default 60)
|
||||
|
||||
Returns:
|
||||
Final merged and sorted results
|
||||
"""
|
||||
rrf_scores = {}
|
||||
|
||||
for source, results in results_by_source.items():
|
||||
for rank, result in enumerate(results, start=1):
|
||||
# Use page_id for wiki results, url hash for web results
|
||||
if result.get("page_id"):
|
||||
result_id = f"page_{result['page_id']}"
|
||||
elif result.get("url"):
|
||||
result_id = f"url_{hash(result['url'])}"
|
||||
else:
|
||||
continue # Skip results without ID
|
||||
# Wiki results (single source, already merged)
|
||||
for rank, result in enumerate(wiki_results, start=1):
|
||||
page_id = result.get("page_id")
|
||||
if not page_id:
|
||||
continue
|
||||
result_id = f"page_{page_id}"
|
||||
rrf_scores[result_id] = {
|
||||
"result": result,
|
||||
"rrf_score": 1 / (k + rank),
|
||||
"sources": result.get("found_by", ["wiki"]),
|
||||
"source_type": "wiki"
|
||||
}
|
||||
|
||||
if result_id not in rrf_scores:
|
||||
rrf_scores[result_id] = {
|
||||
"result": result,
|
||||
"rrf_score": 0.0,
|
||||
"sources": [],
|
||||
"source_type": source
|
||||
}
|
||||
|
||||
# RRF formula: sum of 1/(k + rank) across sources
|
||||
rrf_scores[result_id]["rrf_score"] += 1 / (k + rank)
|
||||
rrf_scores[result_id]["sources"].append(source)
|
||||
|
||||
# If result appears in multiple sources, update source_type
|
||||
if len(rrf_scores[result_id]["sources"]) > 1:
|
||||
rrf_scores[result_id]["source_type"] = "+".join(
|
||||
sorted(set(rrf_scores[result_id]["sources"]))
|
||||
)
|
||||
# Web results (single source)
|
||||
for rank, result in enumerate(web_results, start=1):
|
||||
url = result.get("url")
|
||||
if not url:
|
||||
continue
|
||||
result_id = f"url_{hash(url)}"
|
||||
rrf_scores[result_id] = {
|
||||
"result": result,
|
||||
"rrf_score": 1 / (k + rank),
|
||||
"sources": ["web"],
|
||||
"source_type": "web"
|
||||
}
|
||||
|
||||
# Sort by RRF score descending
|
||||
sorted_results = sorted(
|
||||
@@ -431,7 +542,7 @@ JSON:"""
|
||||
reverse=True
|
||||
)
|
||||
|
||||
logger.info(f"RRF fusion: {len(sorted_results)} unique results from {len(results_by_source)} sources")
|
||||
logger.info(f"Final RRF: {len(sorted_results)} results (wiki + web)")
|
||||
|
||||
return sorted_results
|
||||
|
||||
@@ -509,21 +620,27 @@ JSON:"""
|
||||
for i, r in enumerate(results)
|
||||
])
|
||||
|
||||
prompt = f"""Given this search query and documents, rank them by relevance.
|
||||
prompt = f"""Rank these documents by relevance to the query.
|
||||
|
||||
Query: {query}
|
||||
|
||||
Documents:
|
||||
{docs_text}
|
||||
|
||||
Return only the numbers in order of relevance (most relevant first).
|
||||
Example: 3,1,5,2,4
|
||||
RULES:
|
||||
- Rank ONLY by how well content answers the query
|
||||
- Do NOT consider document length, formatting, or style
|
||||
- Do NOT add explanation or commentary
|
||||
- Return ONLY comma-separated numbers, most relevant first
|
||||
|
||||
Example output: 3,1,5,2,4
|
||||
|
||||
Ranking:"""
|
||||
|
||||
response = await self.ollama.generate_text(
|
||||
prompt=prompt,
|
||||
model=self.reranker_model
|
||||
model=self.reranker_model,
|
||||
temperature=0.0 # Deterministic for consistent rankings
|
||||
)
|
||||
|
||||
# Parse response: "3,1,5,2,4" → [2, 0, 4, 1, 3] (0-indexed)
|
||||
|
||||
@@ -0,0 +1,265 @@
|
||||
"""
|
||||
RAG Search service for Library Desk.
|
||||
|
||||
Provides web, news, and image search with content extraction:
|
||||
- Uses SearXNG for search queries
|
||||
- Uses Trafilatura for content extraction
|
||||
- Caches results in Redis
|
||||
"""
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import logging
|
||||
import time
|
||||
from typing import List, Optional
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import redis.asyncio as aioredis
|
||||
|
||||
from src.clients.searxng_client import SearXNGClient
|
||||
from src.clients.content_extractor import ContentExtractor
|
||||
from src.config import Settings
|
||||
from src.models.rag_search import (
|
||||
SearchType,
|
||||
RAGSearchRequest,
|
||||
RAGSearchResult,
|
||||
RAGSearchResponse,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def extract_domain(url: str) -> str:
|
||||
"""Extract domain name from URL, removing 'www.' prefix."""
|
||||
try:
|
||||
parsed = urlparse(url)
|
||||
domain = parsed.netloc
|
||||
return domain.removeprefix("www.")
|
||||
except Exception:
|
||||
return url
|
||||
|
||||
|
||||
class RAGSearchService:
|
||||
"""
|
||||
Service for RAG-optimized web search with content extraction.
|
||||
|
||||
Combines SearXNG search with Trafilatura content extraction
|
||||
and Redis caching for efficient RAG pipeline integration.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
searxng_client: SearXNGClient,
|
||||
content_extractor: ContentExtractor,
|
||||
redis_client: aioredis.Redis,
|
||||
settings: Settings
|
||||
):
|
||||
"""
|
||||
Initialize RAG search service.
|
||||
|
||||
Args:
|
||||
searxng_client: SearXNG search client
|
||||
content_extractor: Trafilatura content extractor
|
||||
redis_client: Async Redis client for caching
|
||||
settings: Application settings
|
||||
"""
|
||||
self.searxng = searxng_client
|
||||
self.extractor = content_extractor
|
||||
self.redis = redis_client
|
||||
self.settings = settings
|
||||
|
||||
self.cache_ttl = settings.search_cache_ttl
|
||||
self.default_limit = settings.search_default_limit
|
||||
|
||||
logger.info(
|
||||
f"Initialized RAGSearchService: cache_ttl={self.cache_ttl}s, "
|
||||
f"default_limit={self.default_limit}"
|
||||
)
|
||||
|
||||
def _cache_key(self, query: str, search_type: str, limit: int) -> str:
|
||||
"""Generate cache key from search parameters."""
|
||||
key_data = f"{query}:{search_type}:{limit}"
|
||||
key_hash = hashlib.md5(key_data.encode()).hexdigest()
|
||||
return f"rag_search:{key_hash}"
|
||||
|
||||
async def _get_cached_result(self, cache_key: str) -> Optional[RAGSearchResponse]:
|
||||
"""Try to get cached search result."""
|
||||
try:
|
||||
cached = await self.redis.get(cache_key)
|
||||
if cached:
|
||||
data = json.loads(cached)
|
||||
logger.debug(f"Cache hit: {cache_key}")
|
||||
return RAGSearchResponse(**data)
|
||||
except Exception as e:
|
||||
logger.warning(f"Cache read failed: {e}")
|
||||
return None
|
||||
|
||||
async def _set_cached_result(self, cache_key: str, result: RAGSearchResponse):
|
||||
"""Cache search result."""
|
||||
try:
|
||||
await self.redis.setex(
|
||||
cache_key,
|
||||
self.cache_ttl,
|
||||
result.model_dump_json()
|
||||
)
|
||||
logger.debug(f"Cached result: {cache_key} (TTL={self.cache_ttl}s)")
|
||||
except Exception as e:
|
||||
logger.warning(f"Cache write failed: {e}")
|
||||
|
||||
async def _search_searxng(
|
||||
self,
|
||||
query: str,
|
||||
search_type: SearchType,
|
||||
limit: int
|
||||
) -> List[dict]:
|
||||
"""Execute search via SearXNG based on search type."""
|
||||
try:
|
||||
if search_type == SearchType.WEB:
|
||||
results = await self.searxng.search_general(
|
||||
query=query,
|
||||
limit=limit
|
||||
)
|
||||
elif search_type == SearchType.NEWS:
|
||||
results = await self.searxng.search_news(
|
||||
query=query,
|
||||
limit=limit
|
||||
)
|
||||
elif search_type == SearchType.IMAGES:
|
||||
results = await self.searxng.search_images(
|
||||
query=query,
|
||||
limit=limit
|
||||
)
|
||||
else:
|
||||
results = await self.searxng.search_general(
|
||||
query=query,
|
||||
limit=limit
|
||||
)
|
||||
|
||||
return results
|
||||
except Exception as e:
|
||||
logger.error(f"SearXNG search failed: {e}")
|
||||
raise
|
||||
|
||||
async def _extract_content_for_results(
|
||||
self,
|
||||
results: List[dict]
|
||||
) -> List[RAGSearchResult]:
|
||||
"""Extract full content from search result URLs."""
|
||||
# Get URLs for extraction
|
||||
urls = [r.get("url", "") for r in results if r.get("url")]
|
||||
|
||||
# Extract content in parallel
|
||||
extraction_results = await self.extractor.extract_batch(urls)
|
||||
|
||||
# Build result objects
|
||||
search_results = []
|
||||
for i, raw_result in enumerate(results):
|
||||
url = raw_result.get("url", "")
|
||||
|
||||
# Find matching extraction result
|
||||
extracted_content = ""
|
||||
for ext_result in extraction_results:
|
||||
if ext_result.url == url and ext_result.success:
|
||||
extracted_content = ext_result.content
|
||||
break
|
||||
|
||||
# Get original snippet
|
||||
snippet = raw_result.get("content", "")
|
||||
if len(snippet) > 300:
|
||||
snippet = snippet[:300] + "..."
|
||||
|
||||
# Build result
|
||||
search_results.append(RAGSearchResult(
|
||||
title=raw_result.get("title", ""),
|
||||
url=url,
|
||||
content=extracted_content,
|
||||
snippet=snippet,
|
||||
source=extract_domain(url),
|
||||
published_date=raw_result.get("publishedDate")
|
||||
))
|
||||
|
||||
return search_results
|
||||
|
||||
def _generate_sources_summary(self, results: List[RAGSearchResult]) -> str:
|
||||
"""Generate markdown list of source URLs."""
|
||||
if not results:
|
||||
return ""
|
||||
|
||||
lines = ["## Sources"]
|
||||
for i, r in enumerate(results, 1):
|
||||
lines.append(f"{i}. [{r.title}]({r.url})")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
async def search(
|
||||
self,
|
||||
query: str,
|
||||
search_type: SearchType = SearchType.WEB,
|
||||
limit: Optional[int] = None,
|
||||
user: str = "default"
|
||||
) -> RAGSearchResponse:
|
||||
"""
|
||||
Execute RAG-optimized search.
|
||||
|
||||
Args:
|
||||
query: Search query string
|
||||
search_type: Type of search (web, news, images)
|
||||
limit: Maximum results to return (default from settings)
|
||||
user: User identifier for logging/rate limiting
|
||||
|
||||
Returns:
|
||||
RAGSearchResponse with extracted content and sources
|
||||
|
||||
Raises:
|
||||
ValueError: If query is empty
|
||||
Exception: If search fails
|
||||
"""
|
||||
start_time = time.time()
|
||||
|
||||
if not query or not query.strip():
|
||||
raise ValueError("Query cannot be empty")
|
||||
|
||||
effective_limit = limit or self.default_limit
|
||||
|
||||
# Check cache
|
||||
cache_key = self._cache_key(query, search_type.value, effective_limit)
|
||||
cached = await self._get_cached_result(cache_key)
|
||||
if cached:
|
||||
return cached
|
||||
|
||||
logger.info(
|
||||
f"RAG search: '{query}' type={search_type.value} "
|
||||
f"limit={effective_limit} user={user}"
|
||||
)
|
||||
|
||||
# Execute search
|
||||
raw_results = await self._search_searxng(query, search_type, effective_limit)
|
||||
|
||||
# Extract content from results
|
||||
search_results = await self._extract_content_for_results(raw_results)
|
||||
|
||||
# Generate sources summary
|
||||
sources_summary = self._generate_sources_summary(search_results)
|
||||
|
||||
# Calculate timing
|
||||
search_time_ms = int((time.time() - start_time) * 1000)
|
||||
|
||||
# Build response
|
||||
response = RAGSearchResponse(
|
||||
query=query,
|
||||
search_type=search_type,
|
||||
results=search_results,
|
||||
total_results=len(search_results),
|
||||
search_time_ms=search_time_ms,
|
||||
sources_summary=sources_summary
|
||||
)
|
||||
|
||||
# Cache result
|
||||
await self._set_cached_result(cache_key, response)
|
||||
|
||||
logger.info(
|
||||
f"RAG search completed: {len(search_results)} results "
|
||||
f"in {search_time_ms}ms"
|
||||
)
|
||||
|
||||
return response
|
||||
@@ -1,7 +1,7 @@
|
||||
"""
|
||||
Intelligent Wiki Page Writer Service
|
||||
|
||||
Uses LLM (mistral-nemo) to create and reconstruct wiki pages with:
|
||||
Uses LLM to create and reconstruct wiki pages with:
|
||||
- Holistic content restructuring
|
||||
- Zero fact loss (unless superseded)
|
||||
- Conflict detection and flagging
|
||||
@@ -25,15 +25,16 @@ class WikiPageWriter:
|
||||
Intelligent wiki page writer using LLM for content generation and restructuring.
|
||||
"""
|
||||
|
||||
def __init__(self, ollama_client):
|
||||
def __init__(self, ollama_client, settings):
|
||||
"""
|
||||
Initialize wiki page writer.
|
||||
|
||||
Args:
|
||||
ollama_client: OllamaClient for LLM operations
|
||||
settings: Application settings
|
||||
"""
|
||||
self.ollama = ollama_client
|
||||
self.model = "mistral-nemo" # Default model for writing
|
||||
self.model = settings.ollama_model
|
||||
|
||||
async def create_page(
|
||||
self,
|
||||
@@ -130,8 +131,8 @@ class WikiPageWriter:
|
||||
conflicts=conflicts
|
||||
)
|
||||
|
||||
# Reconstruct with LLM
|
||||
reconstructed = await self._call_llm(prompt)
|
||||
# Reconstruct with LLM (lower temperature for precise merging)
|
||||
reconstructed = await self._call_llm(prompt, temperature=0.2)
|
||||
|
||||
# Ensure standard sections are present
|
||||
reconstructed = self._ensure_standard_sections(
|
||||
@@ -154,7 +155,7 @@ class WikiPageWriter:
|
||||
Returns:
|
||||
List of conflicts with: {fact_a, fact_b, confidence, context}
|
||||
"""
|
||||
prompt = f"""Analyze these two pieces of content for factual conflicts.
|
||||
prompt = f"""Analyze these contents for direct factual conflicts.
|
||||
|
||||
EXISTING CONTENT:
|
||||
{existing_content[:2000]}
|
||||
@@ -162,25 +163,26 @@ EXISTING CONTENT:
|
||||
NEW INFORMATION:
|
||||
{new_information[:2000]}
|
||||
|
||||
Identify any facts that contradict each other. For each conflict, provide:
|
||||
1. The fact from existing content
|
||||
2. The contradicting fact from new information
|
||||
3. Confidence level (low/medium/high)
|
||||
4. Context/explanation
|
||||
ANALYSIS STEPS:
|
||||
1. Identify specific factual claims in existing content (dates, numbers, names, states)
|
||||
2. Identify specific factual claims in new content
|
||||
3. Compare ONLY for direct contradictions (X says A, Y says not-A)
|
||||
|
||||
Return ONLY valid JSON:
|
||||
RULES:
|
||||
- Do NOT flag differences in wording or phrasing as conflicts
|
||||
- Do NOT flag new/additional information as conflicts
|
||||
- Do NOT flag opinion differences as conflicts
|
||||
- ONLY flag direct factual contradictions
|
||||
- Return valid JSON only, no commentary
|
||||
|
||||
Return format:
|
||||
{{
|
||||
"conflicts": [
|
||||
{{
|
||||
"existing_fact": "fact from old content",
|
||||
"new_fact": "contradicting fact",
|
||||
"confidence": "medium",
|
||||
"context": "explanation of why these conflict"
|
||||
}}
|
||||
{{"existing_fact": "...", "new_fact": "...", "confidence": "low/medium/high", "context": "..."}}
|
||||
]
|
||||
}}
|
||||
|
||||
If no conflicts, return: {{"conflicts": []}}
|
||||
If no conflicts: {{"conflicts": []}}
|
||||
|
||||
JSON:"""
|
||||
|
||||
@@ -188,7 +190,8 @@ JSON:"""
|
||||
response = await self.ollama.generate_text(
|
||||
prompt=prompt,
|
||||
model=self.model,
|
||||
stream=False
|
||||
stream=False,
|
||||
temperature=0.0 # Deterministic for consistent conflict detection
|
||||
)
|
||||
|
||||
# Extract JSON
|
||||
@@ -347,6 +350,13 @@ FORMATTING RULES:
|
||||
- Keep sections focused and scannable
|
||||
- Adapt structure to content - not all sections apply to all topics
|
||||
|
||||
CRITICAL CONSTRAINTS:
|
||||
- Do NOT invent facts not present in the source information above
|
||||
- Do NOT add speculative information or assumptions
|
||||
- Do NOT fill sections with placeholder text or generic statements
|
||||
- If information for a section is not available, OMIT the section entirely
|
||||
- Base ALL content strictly on provided source information
|
||||
|
||||
Generate ONLY the markdown content (do not include Sources, Knowledge Graph, or Mind Map sections - those are added automatically).
|
||||
|
||||
MARKDOWN:"""
|
||||
@@ -402,6 +412,13 @@ FORMATTING RULES:
|
||||
- Bold important terms
|
||||
- Add subsections (###) where it improves clarity
|
||||
|
||||
CRITICAL CONSTRAINTS:
|
||||
- Do NOT rephrase facts in ways that change their meaning
|
||||
- Do NOT remove ANY information unless explicitly superseded by newer facts
|
||||
- Do NOT add information not present in existing content or new information
|
||||
- Preserve exact quotes, dates, numbers, and names verbatim
|
||||
- Do NOT fill gaps with assumptions or general knowledge
|
||||
|
||||
OUTPUT INSTRUCTIONS:
|
||||
- Return complete page content (do not include Sources, Knowledge Graph, Mind Map - those are added automatically)
|
||||
- Include updated "Changes & Updates" section noting what was changed today
|
||||
@@ -409,13 +426,21 @@ OUTPUT INSTRUCTIONS:
|
||||
|
||||
RECONSTRUCTED MARKDOWN:"""
|
||||
|
||||
async def _call_llm(self, prompt: str) -> str:
|
||||
"""Call LLM with prompt and return response."""
|
||||
async def _call_llm(self, prompt: str, temperature: float = 0.3) -> str:
|
||||
"""
|
||||
Call LLM with prompt and return response.
|
||||
|
||||
Args:
|
||||
prompt: The prompt text
|
||||
temperature: Sampling temperature (0.0=deterministic, higher=creative)
|
||||
Default 0.3 for controlled but natural content generation
|
||||
"""
|
||||
try:
|
||||
response = await self.ollama.generate_text(
|
||||
prompt=prompt,
|
||||
model=self.model,
|
||||
stream=False
|
||||
stream=False,
|
||||
temperature=temperature
|
||||
)
|
||||
|
||||
if not response:
|
||||
|
||||
+20
-9
@@ -1,5 +1,6 @@
|
||||
"""Pytest configuration and shared fixtures for Library Desk tests."""
|
||||
|
||||
import os
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from typing import AsyncGenerator
|
||||
@@ -7,6 +8,9 @@ from typing import AsyncGenerator
|
||||
# Test configuration
|
||||
pytest_plugins = ("pytest_asyncio",)
|
||||
|
||||
# Use real host for tests (services available at this IP)
|
||||
TEST_HOST = os.environ.get("TEST_HOST", "192.168.86.149")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_user() -> str:
|
||||
@@ -17,49 +21,56 @@ def test_user() -> str:
|
||||
@pytest.fixture
|
||||
def neo4j_test_uri() -> str:
|
||||
"""Test Neo4j URI."""
|
||||
return "bolt://neo4j:7687"
|
||||
return f"bolt://{TEST_HOST}:7687"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def neo4j_test_auth() -> tuple:
|
||||
"""Test Neo4j authentication."""
|
||||
return ("neo4j", "test_password")
|
||||
from src.config import get_settings
|
||||
settings = get_settings()
|
||||
return ("neo4j", settings.neo4j_password)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def qdrant_test_url() -> str:
|
||||
"""Test Qdrant URL."""
|
||||
return "http://qdrant:6333"
|
||||
return f"http://{TEST_HOST}:6333"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def wikijs_test_config() -> dict:
|
||||
"""Test Wiki.js configuration."""
|
||||
from src.config import get_settings
|
||||
settings = get_settings()
|
||||
return {
|
||||
"base_url": "http://wiki:3000",
|
||||
"api_key": "test_api_key"
|
||||
"base_url": f"http://{TEST_HOST}:3000",
|
||||
"username": settings.wikijs_username,
|
||||
"password": settings.wikijs_password
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def searxng_test_url() -> str:
|
||||
"""Test SearXNG URL."""
|
||||
return "http://searxng:8080"
|
||||
return f"http://{TEST_HOST}:8080"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def ollama_test_config() -> dict:
|
||||
"""Test Ollama configuration."""
|
||||
from src.config import get_settings
|
||||
settings = get_settings()
|
||||
return {
|
||||
"base_url": "http://ollama:11434",
|
||||
"model": "nomic-embed-text"
|
||||
"base_url": f"http://{TEST_HOST}:11434",
|
||||
"model": settings.ollama_embedding_model
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def redis_test_url() -> str:
|
||||
"""Test Redis URL."""
|
||||
return "redis://redis-shared:6379/4"
|
||||
return f"redis://{TEST_HOST}:6379/4"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
||||
@@ -0,0 +1,183 @@
|
||||
"""Tests for ContentExtractor client."""
|
||||
|
||||
import pytest
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
from src.clients.content_extractor import ContentExtractor
|
||||
from src.models.content import ContentExtractionResult
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def content_extractor():
|
||||
"""Create ContentExtractor with test configuration."""
|
||||
return ContentExtractor(timeout=5, max_length=2000)
|
||||
|
||||
|
||||
class TestContentExtractor:
|
||||
"""Tests for ContentExtractor client."""
|
||||
|
||||
def test_init(self, content_extractor):
|
||||
"""Test ContentExtractor initialization."""
|
||||
assert content_extractor.timeout == 5
|
||||
assert content_extractor.max_length == 2000
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_extract_success(self, content_extractor):
|
||||
"""Test successful content extraction."""
|
||||
test_url = "https://example.com/article"
|
||||
test_content = "This is the extracted article content."
|
||||
|
||||
with patch('src.clients.content_extractor.trafilatura') as mock_traf:
|
||||
mock_traf.fetch_url.return_value = "<html><body>Test</body></html>"
|
||||
mock_traf.extract.return_value = test_content
|
||||
mock_traf.bare_extraction.return_value = {
|
||||
"title": "Test Article",
|
||||
"author": "John Doe",
|
||||
"date": "2024-01-15",
|
||||
"language": "en"
|
||||
}
|
||||
|
||||
result = await content_extractor.extract(test_url)
|
||||
|
||||
assert result.success is True
|
||||
assert result.url == test_url
|
||||
assert result.content == test_content
|
||||
assert result.error is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_extract_fetch_failure(self, content_extractor):
|
||||
"""Test extraction when URL fetch fails."""
|
||||
test_url = "https://example.com/nonexistent"
|
||||
|
||||
with patch('src.clients.content_extractor.trafilatura') as mock_traf:
|
||||
mock_traf.fetch_url.return_value = None
|
||||
|
||||
result = await content_extractor.extract(test_url)
|
||||
|
||||
assert result.success is False
|
||||
assert result.url == test_url
|
||||
assert result.content == ""
|
||||
assert "Failed to fetch URL" in result.error
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_extract_no_content(self, content_extractor):
|
||||
"""Test extraction when page has no extractable content."""
|
||||
test_url = "https://example.com/empty"
|
||||
|
||||
with patch('src.clients.content_extractor.trafilatura') as mock_traf:
|
||||
mock_traf.fetch_url.return_value = "<html><body></body></html>"
|
||||
mock_traf.extract.return_value = None
|
||||
|
||||
result = await content_extractor.extract(test_url)
|
||||
|
||||
assert result.success is False
|
||||
assert "No content extracted" in result.error
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_extract_max_length_truncation(self, content_extractor):
|
||||
"""Test that content is truncated to max length."""
|
||||
test_url = "https://example.com/long-article"
|
||||
# Content longer than max_length (2000)
|
||||
long_content = "x" * 3000
|
||||
|
||||
with patch('src.clients.content_extractor.trafilatura') as mock_traf:
|
||||
mock_traf.fetch_url.return_value = "<html><body>Test</body></html>"
|
||||
mock_traf.extract.return_value = long_content
|
||||
mock_traf.bare_extraction.return_value = {}
|
||||
|
||||
result = await content_extractor.extract(test_url)
|
||||
|
||||
assert result.success is True
|
||||
assert len(result.content) <= content_extractor.max_length + 3 # +3 for "..."
|
||||
assert result.content.endswith("...")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_extract_batch(self, content_extractor):
|
||||
"""Test batch extraction of multiple URLs."""
|
||||
test_urls = [
|
||||
"https://example.com/article1",
|
||||
"https://example.com/article2",
|
||||
"https://example.com/article3"
|
||||
]
|
||||
|
||||
with patch('src.clients.content_extractor.trafilatura') as mock_traf:
|
||||
mock_traf.fetch_url.return_value = "<html><body>Test</body></html>"
|
||||
mock_traf.extract.return_value = "Extracted content"
|
||||
mock_traf.bare_extraction.return_value = {}
|
||||
|
||||
results = await content_extractor.extract_batch(test_urls)
|
||||
|
||||
assert len(results) == 3
|
||||
for i, result in enumerate(results):
|
||||
assert result.url == test_urls[i]
|
||||
assert result.success is True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_extract_timeout(self):
|
||||
"""Test extraction timeout handling."""
|
||||
import time
|
||||
|
||||
test_url = "https://example.com/slow"
|
||||
|
||||
# Create an extractor with very short timeout
|
||||
fast_extractor = ContentExtractor(timeout=0.001, max_length=2000)
|
||||
|
||||
def slow_fetch(url):
|
||||
time.sleep(1) # Sleep synchronously (this runs in thread pool)
|
||||
return "<html></html>"
|
||||
|
||||
with patch('src.clients.content_extractor.trafilatura') as mock_traf:
|
||||
mock_traf.fetch_url = slow_fetch
|
||||
|
||||
result = await fast_extractor.extract(test_url)
|
||||
|
||||
assert result.success is False
|
||||
assert "timed out" in result.error.lower()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_extract_from_html(self, content_extractor):
|
||||
"""Test extraction from raw HTML."""
|
||||
test_html = "<html><body><article>Article content here.</article></body></html>"
|
||||
|
||||
with patch('src.clients.content_extractor.trafilatura') as mock_traf:
|
||||
mock_traf.extract.return_value = "Article content here."
|
||||
mock_traf.bare_extraction.return_value = {"title": "Test"}
|
||||
|
||||
result = await content_extractor.extract_from_html(test_html, url="https://example.com")
|
||||
|
||||
assert result.success is True
|
||||
assert result.content == "Article content here."
|
||||
|
||||
|
||||
class TestContentExtractionResult:
|
||||
"""Tests for ContentExtractionResult model."""
|
||||
|
||||
def test_success_result(self):
|
||||
"""Test creating a successful result."""
|
||||
result = ContentExtractionResult(
|
||||
url="https://example.com",
|
||||
title="Test Article",
|
||||
content="Article content",
|
||||
author="John Doe",
|
||||
date="2024-01-15",
|
||||
language="en",
|
||||
success=True,
|
||||
error=None
|
||||
)
|
||||
|
||||
assert result.url == "https://example.com"
|
||||
assert result.success is True
|
||||
assert result.error is None
|
||||
|
||||
def test_failure_result(self):
|
||||
"""Test creating a failure result."""
|
||||
result = ContentExtractionResult(
|
||||
url="https://example.com/error",
|
||||
content="",
|
||||
success=False,
|
||||
error="Failed to fetch URL"
|
||||
)
|
||||
|
||||
assert result.url == "https://example.com/error"
|
||||
assert result.success is False
|
||||
assert result.error == "Failed to fetch URL"
|
||||
@@ -38,10 +38,10 @@ def settings():
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def neo4j_client(settings) -> AsyncGenerator[Neo4jClient, None]:
|
||||
async def neo4j_client(settings, neo4j_test_uri) -> AsyncGenerator[Neo4jClient, None]:
|
||||
"""Get connected Neo4j client."""
|
||||
client = Neo4jClient(
|
||||
uri=settings.neo4j_uri,
|
||||
uri=neo4j_test_uri,
|
||||
user=settings.neo4j_user,
|
||||
password=settings.neo4j_password
|
||||
)
|
||||
@@ -51,12 +51,12 @@ async def neo4j_client(settings) -> AsyncGenerator[Neo4jClient, None]:
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def wiki_client(settings) -> AsyncGenerator[WikiJSClient, None]:
|
||||
async def wiki_client(wikijs_test_config) -> AsyncGenerator[WikiJSClient, None]:
|
||||
"""Get Wiki.js client."""
|
||||
client = WikiJSClient(
|
||||
base_url=settings.wikijs_url,
|
||||
username=settings.wikijs_username,
|
||||
password=settings.wikijs_password
|
||||
base_url=wikijs_test_config["base_url"],
|
||||
username=wikijs_test_config["username"],
|
||||
password=wikijs_test_config["password"]
|
||||
)
|
||||
yield client
|
||||
|
||||
|
||||
+65
-49
@@ -25,6 +25,7 @@ from src.clients.qdrant_client import QdrantClientWrapper
|
||||
from src.clients.wikijs_client import WikiJSClient
|
||||
from src.clients.searxng_client import SearXNGClient
|
||||
from src.clients.ollama_client import OllamaClient
|
||||
from src.clients.content_extractor import ContentExtractor
|
||||
from src.services.hybrid_rag_service import HybridRAGService
|
||||
from src.services.vector_service import VectorService
|
||||
from src.services.graph_service import GraphService
|
||||
@@ -42,10 +43,10 @@ def settings():
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def neo4j_client(settings) -> AsyncGenerator[Neo4jClient, None]:
|
||||
async def neo4j_client(settings, neo4j_test_uri) -> AsyncGenerator[Neo4jClient, None]:
|
||||
"""Get connected Neo4j client."""
|
||||
client = Neo4jClient(
|
||||
uri=settings.neo4j_uri,
|
||||
uri=neo4j_test_uri,
|
||||
user=settings.neo4j_user,
|
||||
password=settings.neo4j_password
|
||||
)
|
||||
@@ -55,32 +56,44 @@ async def neo4j_client(settings) -> AsyncGenerator[Neo4jClient, None]:
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def qdrant_client(settings) -> QdrantClientWrapper:
|
||||
def qdrant_client(qdrant_test_url) -> QdrantClientWrapper:
|
||||
"""Get Qdrant client."""
|
||||
return QdrantClientWrapper(url=settings.qdrant_url)
|
||||
return QdrantClientWrapper(url=qdrant_test_url)
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def wiki_client(settings) -> AsyncGenerator[WikiJSClient, None]:
|
||||
async def wiki_client(wikijs_test_config) -> AsyncGenerator[WikiJSClient, None]:
|
||||
"""Get Wiki.js client."""
|
||||
client = WikiJSClient(
|
||||
base_url=settings.wikijs_url,
|
||||
username=settings.wikijs_username,
|
||||
password=settings.wikijs_password
|
||||
base_url=wikijs_test_config["base_url"],
|
||||
username=wikijs_test_config["username"],
|
||||
password=wikijs_test_config["password"]
|
||||
)
|
||||
yield client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def searxng_client(settings) -> SearXNGClient:
|
||||
def searxng_client(searxng_test_url) -> SearXNGClient:
|
||||
"""Get SearXNG client."""
|
||||
return SearXNGClient(base_url=settings.searxng_url)
|
||||
return SearXNGClient(base_url=searxng_test_url)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def ollama_client(settings) -> OllamaClient:
|
||||
def ollama_client(ollama_test_config) -> OllamaClient:
|
||||
"""Get Ollama client."""
|
||||
return OllamaClient(base_url=settings.ollama_url)
|
||||
return OllamaClient(
|
||||
base_url=ollama_test_config["base_url"],
|
||||
model=ollama_test_config["model"]
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def content_extractor(settings) -> ContentExtractor:
|
||||
"""Get ContentExtractor client."""
|
||||
return ContentExtractor(
|
||||
timeout=settings.content_extraction_timeout,
|
||||
max_length=settings.content_max_length
|
||||
)
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
@@ -101,6 +114,7 @@ async def hybrid_rag_service(
|
||||
graph_service,
|
||||
searxng_client,
|
||||
ollama_client,
|
||||
content_extractor,
|
||||
settings
|
||||
):
|
||||
"""Get HybridRAGService instance."""
|
||||
@@ -109,6 +123,7 @@ async def hybrid_rag_service(
|
||||
graph_service=graph_service,
|
||||
searxng_client=searxng_client,
|
||||
ollama_client=ollama_client,
|
||||
content_extractor=content_extractor,
|
||||
settings=settings
|
||||
)
|
||||
|
||||
@@ -207,53 +222,54 @@ async def test_vector_data(vector_service, test_wiki_page):
|
||||
# ============================================================================
|
||||
|
||||
class TestRRFFusion:
|
||||
"""Test Reciprocal Rank Fusion algorithm."""
|
||||
"""Test two-stage Reciprocal Rank Fusion algorithm."""
|
||||
|
||||
def test_rrf_single_source(self, hybrid_rag_service):
|
||||
"""Test RRF with single source."""
|
||||
results_by_source = {
|
||||
"vector": [
|
||||
{"page_id": 1, "title": "Doc 1", "content": "test"},
|
||||
{"page_id": 2, "title": "Doc 2", "content": "test"}
|
||||
]
|
||||
}
|
||||
def test_wiki_merge_single_source(self, hybrid_rag_service):
|
||||
"""Test wiki merge with single source (vector only)."""
|
||||
vector_results = [
|
||||
{"page_id": 1, "title": "Doc 1", "content": "test"},
|
||||
{"page_id": 2, "title": "Doc 2", "content": "test"}
|
||||
]
|
||||
|
||||
fused = hybrid_rag_service._reciprocal_rank_fusion(results_by_source, k=60)
|
||||
merged = hybrid_rag_service._merge_wiki_sources(vector_results, [], k=60)
|
||||
|
||||
assert len(fused) == 2
|
||||
assert fused[0]["rrf_score"] > fused[1]["rrf_score"] # Rank 1 > Rank 2
|
||||
assert fused[0]["sources"] == ["vector"]
|
||||
assert len(merged) == 2
|
||||
assert merged[0]["wiki_rrf_score"] > merged[1]["wiki_rrf_score"] # Rank 1 > Rank 2
|
||||
assert merged[0]["found_by"] == ["vector"]
|
||||
|
||||
def test_rrf_multiple_sources_same_doc(self, hybrid_rag_service):
|
||||
"""Test RRF with same document from multiple sources."""
|
||||
results_by_source = {
|
||||
"vector": [{"page_id": 1, "title": "Doc 1", "content": "test"}],
|
||||
"graph": [{"page_id": 1, "title": "Doc 1", "content": ""}],
|
||||
}
|
||||
def test_wiki_merge_multiple_sources_same_doc(self, hybrid_rag_service):
|
||||
"""Test wiki merge with same document from vector and graph."""
|
||||
vector_results = [{"page_id": 1, "title": "Doc 1", "content": "test"}]
|
||||
graph_results = [{"page_id": 1, "title": "Doc 1", "content": ""}]
|
||||
|
||||
fused = hybrid_rag_service._reciprocal_rank_fusion(results_by_source, k=60)
|
||||
merged = hybrid_rag_service._merge_wiki_sources(vector_results, graph_results, k=60)
|
||||
|
||||
assert len(fused) == 1 # Deduplicated
|
||||
assert len(fused[0]["sources"]) == 2 # Both sources
|
||||
assert "vector" in fused[0]["sources"]
|
||||
assert "graph" in fused[0]["sources"]
|
||||
# RRF score should be sum: 1/(60+1) + 1/(60+1)
|
||||
assert len(merged) == 1 # Deduplicated
|
||||
assert len(merged[0]["found_by"]) == 2 # Both sources
|
||||
assert "vector" in merged[0]["found_by"]
|
||||
assert "graph" in merged[0]["found_by"]
|
||||
# Wiki RRF score should be sum: 1/(60+1) + 1/(60+1)
|
||||
expected_score = 1/61 + 1/61
|
||||
assert abs(fused[0]["rrf_score"] - expected_score) < 0.001
|
||||
assert abs(merged[0]["wiki_rrf_score"] - expected_score) < 0.001
|
||||
|
||||
def test_rrf_web_results(self, hybrid_rag_service):
|
||||
"""Test RRF with web results (URL-based)."""
|
||||
results_by_source = {
|
||||
"web": [
|
||||
{"url": "https://example.com/1", "title": "Web 1", "content": "test"},
|
||||
{"url": "https://example.com/2", "title": "Web 2", "content": "test"}
|
||||
]
|
||||
}
|
||||
def test_final_rrf_wiki_and_web(self, hybrid_rag_service):
|
||||
"""Test final RRF between wiki and web results."""
|
||||
# Pre-merged wiki results
|
||||
wiki_results = [
|
||||
{"page_id": 1, "title": "Wiki 1", "content": "test", "found_by": ["vector"]}
|
||||
]
|
||||
web_results = [
|
||||
{"url": "https://example.com/1", "title": "Web 1", "content": "test"},
|
||||
{"url": "https://example.com/2", "title": "Web 2", "content": "test"}
|
||||
]
|
||||
|
||||
fused = hybrid_rag_service._reciprocal_rank_fusion(results_by_source, k=60)
|
||||
fused = hybrid_rag_service._reciprocal_rank_fusion(wiki_results, web_results, k=60)
|
||||
|
||||
assert len(fused) == 2
|
||||
assert fused[0]["result"]["url"] == "https://example.com/1"
|
||||
assert len(fused) == 3
|
||||
# Wiki rank 1 and web rank 1 should have same RRF score
|
||||
wiki_score = next(r["rrf_score"] for r in fused if r["source_type"] == "wiki")
|
||||
web_score = next(r["rrf_score"] for r in fused if r["source_type"] == "web")
|
||||
assert abs(wiki_score - web_score) < 0.001 # Equal footing
|
||||
|
||||
|
||||
class TestContextFormatting:
|
||||
|
||||
+16
-15
@@ -30,10 +30,10 @@ def settings():
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def neo4j_client(settings) -> AsyncGenerator[Neo4jClient, None]:
|
||||
async def neo4j_client(settings, neo4j_test_uri) -> AsyncGenerator[Neo4jClient, None]:
|
||||
"""Get connected Neo4j client."""
|
||||
client = Neo4jClient(
|
||||
uri=settings.neo4j_uri,
|
||||
uri=neo4j_test_uri,
|
||||
user=settings.neo4j_user,
|
||||
password=settings.neo4j_password
|
||||
)
|
||||
@@ -43,45 +43,46 @@ async def neo4j_client(settings) -> AsyncGenerator[Neo4jClient, None]:
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def qdrant_client(settings) -> QdrantClientWrapper:
|
||||
def qdrant_client(settings, qdrant_test_url) -> QdrantClientWrapper:
|
||||
"""Get Qdrant client."""
|
||||
return QdrantClientWrapper(url=settings.qdrant_url)
|
||||
return QdrantClientWrapper(url=qdrant_test_url)
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def wikijs_client(settings) -> AsyncGenerator[WikiJSClient, None]:
|
||||
async def wikijs_client(wikijs_test_config) -> AsyncGenerator[WikiJSClient, None]:
|
||||
"""Get Wiki.js client."""
|
||||
client = WikiJSClient(
|
||||
base_url=settings.wikijs_url,
|
||||
api_key=settings.wikijs_api_key
|
||||
base_url=wikijs_test_config["base_url"],
|
||||
username=wikijs_test_config["username"],
|
||||
password=wikijs_test_config["password"]
|
||||
)
|
||||
yield client
|
||||
await client.close()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def searxng_client(settings) -> AsyncGenerator[SearXNGClient, None]:
|
||||
async def searxng_client(searxng_test_url) -> AsyncGenerator[SearXNGClient, None]:
|
||||
"""Get SearXNG client."""
|
||||
client = SearXNGClient(base_url=settings.searxng_url)
|
||||
client = SearXNGClient(base_url=searxng_test_url)
|
||||
yield client
|
||||
await client.close()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def ollama_client(settings) -> AsyncGenerator[OllamaClient, None]:
|
||||
"""Get Ollama client."""
|
||||
async def ollama_client(ollama_test_config) -> AsyncGenerator[OllamaClient, None]:
|
||||
"""Get Ollama client for embeddings."""
|
||||
client = OllamaClient(
|
||||
base_url=settings.ollama_url,
|
||||
model=settings.ollama_model
|
||||
base_url=ollama_test_config["base_url"],
|
||||
model=ollama_test_config["model"]
|
||||
)
|
||||
yield client
|
||||
await client.close()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def job_manager(settings) -> AsyncGenerator[JobManager, None]:
|
||||
async def job_manager(redis_test_url) -> AsyncGenerator[JobManager, None]:
|
||||
"""Get job manager."""
|
||||
manager = JobManager(redis_url=settings.redis_url)
|
||||
manager = JobManager(redis_url=redis_test_url)
|
||||
await manager.connect()
|
||||
yield manager
|
||||
await manager.close()
|
||||
|
||||
@@ -0,0 +1,369 @@
|
||||
"""Tests for RAG search service and endpoints."""
|
||||
|
||||
import pytest
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
from src.models.rag_search import (
|
||||
SearchType,
|
||||
RAGSearchRequest,
|
||||
RAGSearchResult,
|
||||
RAGSearchResponse,
|
||||
)
|
||||
from src.services.rag_search_service import RAGSearchService, extract_domain
|
||||
|
||||
|
||||
class TestExtractDomain:
|
||||
"""Tests for domain extraction utility."""
|
||||
|
||||
def test_extract_simple_domain(self):
|
||||
"""Test extracting domain from simple URL."""
|
||||
assert extract_domain("https://example.com/page") == "example.com"
|
||||
|
||||
def test_extract_domain_with_www(self):
|
||||
"""Test extracting domain removes www prefix."""
|
||||
assert extract_domain("https://www.example.com/page") == "example.com"
|
||||
|
||||
def test_extract_domain_with_subdomain(self):
|
||||
"""Test extracting domain preserves subdomains."""
|
||||
assert extract_domain("https://blog.example.com/post") == "blog.example.com"
|
||||
|
||||
def test_extract_domain_invalid_url(self):
|
||||
"""Test extracting domain from invalid URL returns empty string."""
|
||||
# urlparse returns empty netloc for invalid URLs
|
||||
assert extract_domain("not-a-url") == ""
|
||||
|
||||
|
||||
class TestRAGSearchModels:
|
||||
"""Tests for RAG search Pydantic models."""
|
||||
|
||||
def test_search_request_defaults(self):
|
||||
"""Test RAGSearchRequest with default values."""
|
||||
request = RAGSearchRequest(query="test query")
|
||||
|
||||
assert request.query == "test query"
|
||||
assert request.search_type == SearchType.WEB
|
||||
assert request.limit == 10
|
||||
|
||||
def test_search_request_custom_values(self):
|
||||
"""Test RAGSearchRequest with custom values."""
|
||||
request = RAGSearchRequest(
|
||||
query="news about AI",
|
||||
search_type=SearchType.NEWS,
|
||||
limit=5,
|
||||
user="custom_user"
|
||||
)
|
||||
|
||||
assert request.query == "news about AI"
|
||||
assert request.search_type == SearchType.NEWS
|
||||
assert request.limit == 5
|
||||
assert request.user == "custom_user"
|
||||
|
||||
def test_search_result(self):
|
||||
"""Test RAGSearchResult model."""
|
||||
result = RAGSearchResult(
|
||||
title="Test Article",
|
||||
url="https://example.com/article",
|
||||
content="Full article content",
|
||||
snippet="Article snippet...",
|
||||
source="example.com",
|
||||
published_date="2024-01-15"
|
||||
)
|
||||
|
||||
assert result.title == "Test Article"
|
||||
assert result.source == "example.com"
|
||||
assert result.published_date == "2024-01-15"
|
||||
|
||||
def test_search_response(self):
|
||||
"""Test RAGSearchResponse model."""
|
||||
response = RAGSearchResponse(
|
||||
query="test",
|
||||
search_type=SearchType.WEB,
|
||||
results=[],
|
||||
total_results=0,
|
||||
search_time_ms=100,
|
||||
sources_summary=""
|
||||
)
|
||||
|
||||
assert response.query == "test"
|
||||
assert response.total_results == 0
|
||||
assert response.search_time_ms == 100
|
||||
|
||||
|
||||
class TestRAGSearchService:
|
||||
"""Tests for RAGSearchService."""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_searxng_client(self):
|
||||
"""Create mock SearXNG client."""
|
||||
client = MagicMock()
|
||||
client.search_general = AsyncMock(return_value=[
|
||||
{
|
||||
"title": "Test Result 1",
|
||||
"url": "https://example.com/1",
|
||||
"content": "Snippet 1",
|
||||
"publishedDate": "2024-01-15"
|
||||
},
|
||||
{
|
||||
"title": "Test Result 2",
|
||||
"url": "https://example.com/2",
|
||||
"content": "Snippet 2",
|
||||
"publishedDate": None
|
||||
}
|
||||
])
|
||||
client.search_news = AsyncMock(return_value=[])
|
||||
client.search_images = AsyncMock(return_value=[])
|
||||
return client
|
||||
|
||||
@pytest.fixture
|
||||
def mock_content_extractor(self):
|
||||
"""Create mock ContentExtractor."""
|
||||
from src.models.content import ContentExtractionResult
|
||||
|
||||
extractor = MagicMock()
|
||||
extractor.extract_batch = AsyncMock(return_value=[
|
||||
ContentExtractionResult(
|
||||
url="https://example.com/1",
|
||||
content="Full extracted content 1",
|
||||
success=True
|
||||
),
|
||||
ContentExtractionResult(
|
||||
url="https://example.com/2",
|
||||
content="Full extracted content 2",
|
||||
success=True
|
||||
)
|
||||
])
|
||||
return extractor
|
||||
|
||||
@pytest.fixture
|
||||
def mock_redis_client(self):
|
||||
"""Create mock Redis client."""
|
||||
redis = MagicMock()
|
||||
redis.get = AsyncMock(return_value=None) # No cache hit
|
||||
redis.setex = AsyncMock()
|
||||
return redis
|
||||
|
||||
@pytest.fixture
|
||||
def mock_settings(self):
|
||||
"""Create mock settings."""
|
||||
settings = MagicMock()
|
||||
settings.search_cache_ttl = 300
|
||||
settings.search_default_limit = 10
|
||||
return settings
|
||||
|
||||
@pytest.fixture
|
||||
def rag_search_service(
|
||||
self,
|
||||
mock_searxng_client,
|
||||
mock_content_extractor,
|
||||
mock_redis_client,
|
||||
mock_settings
|
||||
):
|
||||
"""Create RAGSearchService with mocked dependencies."""
|
||||
return RAGSearchService(
|
||||
searxng_client=mock_searxng_client,
|
||||
content_extractor=mock_content_extractor,
|
||||
redis_client=mock_redis_client,
|
||||
settings=mock_settings
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_basic(self, rag_search_service, mock_searxng_client):
|
||||
"""Test basic web search."""
|
||||
response = await rag_search_service.search(
|
||||
query="test query",
|
||||
search_type=SearchType.WEB,
|
||||
limit=10
|
||||
)
|
||||
|
||||
assert response.query == "test query"
|
||||
assert response.search_type == SearchType.WEB
|
||||
assert len(response.results) == 2
|
||||
assert response.total_results == 2
|
||||
assert response.search_time_ms >= 0
|
||||
|
||||
mock_searxng_client.search_general.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_news(self, rag_search_service, mock_searxng_client):
|
||||
"""Test news search type."""
|
||||
mock_searxng_client.search_news.return_value = [
|
||||
{"title": "News", "url": "https://news.com/1", "content": "News content"}
|
||||
]
|
||||
|
||||
response = await rag_search_service.search(
|
||||
query="latest news",
|
||||
search_type=SearchType.NEWS
|
||||
)
|
||||
|
||||
assert response.search_type == SearchType.NEWS
|
||||
mock_searxng_client.search_news.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_images(self, rag_search_service, mock_searxng_client):
|
||||
"""Test image search type."""
|
||||
mock_searxng_client.search_images.return_value = [
|
||||
{"title": "Image", "url": "https://images.com/1.jpg", "content": ""}
|
||||
]
|
||||
|
||||
response = await rag_search_service.search(
|
||||
query="cat photos",
|
||||
search_type=SearchType.IMAGES
|
||||
)
|
||||
|
||||
assert response.search_type == SearchType.IMAGES
|
||||
mock_searxng_client.search_images.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_empty_query(self, rag_search_service):
|
||||
"""Test search with empty query raises ValueError."""
|
||||
with pytest.raises(ValueError, match="Query cannot be empty"):
|
||||
await rag_search_service.search(query="", search_type=SearchType.WEB)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_caching_miss(
|
||||
self,
|
||||
rag_search_service,
|
||||
mock_redis_client,
|
||||
mock_searxng_client
|
||||
):
|
||||
"""Test search caches results on cache miss."""
|
||||
mock_redis_client.get.return_value = None # Cache miss
|
||||
|
||||
await rag_search_service.search(query="test", search_type=SearchType.WEB)
|
||||
|
||||
# Should call SearXNG (cache miss)
|
||||
mock_searxng_client.search_general.assert_called_once()
|
||||
# Should cache result
|
||||
mock_redis_client.setex.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_caching_hit(
|
||||
self,
|
||||
rag_search_service,
|
||||
mock_redis_client,
|
||||
mock_searxng_client
|
||||
):
|
||||
"""Test search returns cached results on cache hit."""
|
||||
# Simulate cache hit
|
||||
cached_response = RAGSearchResponse(
|
||||
query="test",
|
||||
search_type=SearchType.WEB,
|
||||
results=[],
|
||||
total_results=0,
|
||||
search_time_ms=50,
|
||||
sources_summary=""
|
||||
)
|
||||
mock_redis_client.get.return_value = cached_response.model_dump_json()
|
||||
|
||||
response = await rag_search_service.search(query="test", search_type=SearchType.WEB)
|
||||
|
||||
# Should NOT call SearXNG (cache hit)
|
||||
mock_searxng_client.search_general.assert_not_called()
|
||||
assert response.query == "test"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_content_extraction(
|
||||
self,
|
||||
rag_search_service,
|
||||
mock_content_extractor
|
||||
):
|
||||
"""Test search extracts content from result URLs."""
|
||||
response = await rag_search_service.search(
|
||||
query="test",
|
||||
search_type=SearchType.WEB
|
||||
)
|
||||
|
||||
# Should have called content extractor
|
||||
mock_content_extractor.extract_batch.assert_called_once()
|
||||
|
||||
# Results should have extracted content
|
||||
for result in response.results:
|
||||
assert result.content # Content should be populated
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_sources_summary(self, rag_search_service):
|
||||
"""Test search generates sources summary."""
|
||||
response = await rag_search_service.search(
|
||||
query="test",
|
||||
search_type=SearchType.WEB
|
||||
)
|
||||
|
||||
assert response.sources_summary
|
||||
assert "## Sources" in response.sources_summary
|
||||
assert "[Test Result 1]" in response.sources_summary
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_limit(self, rag_search_service, mock_searxng_client):
|
||||
"""Test search respects limit parameter."""
|
||||
await rag_search_service.search(
|
||||
query="test",
|
||||
search_type=SearchType.WEB,
|
||||
limit=5
|
||||
)
|
||||
|
||||
# Check limit was passed to SearXNG
|
||||
mock_searxng_client.search_general.assert_called_once_with(
|
||||
query="test",
|
||||
limit=5
|
||||
)
|
||||
|
||||
|
||||
class TestRAGSearchServiceIntegration:
|
||||
"""Integration-style tests (still mocked but test more of the flow)."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_full_search_flow(self):
|
||||
"""Test full search flow with all components mocked."""
|
||||
from src.models.content import ContentExtractionResult
|
||||
|
||||
# Setup mocks
|
||||
mock_searxng = MagicMock()
|
||||
mock_searxng.search_general = AsyncMock(return_value=[
|
||||
{
|
||||
"title": "Python Tutorial",
|
||||
"url": "https://python.org/tutorial",
|
||||
"content": "Learn Python programming",
|
||||
"publishedDate": "2024-01-10"
|
||||
}
|
||||
])
|
||||
|
||||
mock_extractor = MagicMock()
|
||||
mock_extractor.extract_batch = AsyncMock(return_value=[
|
||||
ContentExtractionResult(
|
||||
url="https://python.org/tutorial",
|
||||
title="Python Tutorial",
|
||||
content="This is a comprehensive Python tutorial covering basics to advanced topics.",
|
||||
success=True
|
||||
)
|
||||
])
|
||||
|
||||
mock_redis = MagicMock()
|
||||
mock_redis.get = AsyncMock(return_value=None)
|
||||
mock_redis.setex = AsyncMock()
|
||||
|
||||
mock_settings = MagicMock()
|
||||
mock_settings.search_cache_ttl = 300
|
||||
mock_settings.search_default_limit = 10
|
||||
|
||||
# Create service and execute search
|
||||
service = RAGSearchService(
|
||||
searxng_client=mock_searxng,
|
||||
content_extractor=mock_extractor,
|
||||
redis_client=mock_redis,
|
||||
settings=mock_settings
|
||||
)
|
||||
|
||||
response = await service.search(
|
||||
query="python tutorial",
|
||||
search_type=SearchType.WEB,
|
||||
limit=10,
|
||||
user="test_user"
|
||||
)
|
||||
|
||||
# Verify response
|
||||
assert response.query == "python tutorial"
|
||||
assert len(response.results) == 1
|
||||
assert response.results[0].title == "Python Tutorial"
|
||||
assert response.results[0].source == "python.org"
|
||||
assert "comprehensive Python tutorial" in response.results[0].content
|
||||
assert response.results[0].snippet == "Learn Python programming"
|
||||
Reference in New Issue
Block a user