feat: add RAG search endpoint with content extraction

- Add /rag/search endpoint for web, news, and image search via SearXNG
- Add /content/extract and /content/extract/batch endpoints
- Add ContentExtractor client using Trafilatura for content extraction
- Enhance HybridRAG web search with full content extraction
- Add Redis caching for search results
- Add new configuration options for search and extraction timeouts

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
2025-12-15 15:50:48 +01:00
co-authored by Claude Opus 4.5
parent 16b86a7c1b
commit 4f03c9c2ef
16 changed files with 1642 additions and 4 deletions
+34
View File
@@ -5,6 +5,40 @@ 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.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
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "library-desk"
version = "1.1.3"
version = "1.2.0"
description = "Coordination service for The Library system - HybridRAG queries, document ingestion, entity extraction, and knowledge consolidation"
readme = "README.md"
requires-python = ">=3.12"
+3
View File
@@ -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
+310
View File
@@ -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")
+9
View File
@@ -89,6 +89,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."""
+55
View File
@@ -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__)
@@ -116,6 +119,43 @@ def get_ollama_client() -> OllamaClient:
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,6 +378,19 @@ 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()
)
@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()
)
+6 -1
View File
@@ -45,7 +45,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 +59,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"
+69
View File
@@ -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")
+88
View File
@@ -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"
)
+132
View File
@@ -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")
+87
View File
@@ -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")
+19 -2
View File
@@ -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,12 +56,14 @@ 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
@@ -334,7 +338,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 +347,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"
}
+265
View File
@@ -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
+183
View File
@@ -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"
+12
View File
@@ -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
@@ -83,6 +84,15 @@ def ollama_client(settings) -> OllamaClient:
return OllamaClient(base_url=settings.ollama_url)
@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
async def vector_service(qdrant_client, wiki_client, ollama_client):
"""Get VectorService instance."""
@@ -101,6 +111,7 @@ async def hybrid_rag_service(
graph_service,
searxng_client,
ollama_client,
content_extractor,
settings
):
"""Get HybridRAGService instance."""
@@ -109,6 +120,7 @@ async def hybrid_rag_service(
graph_service=graph_service,
searxng_client=searxng_client,
ollama_client=ollama_client,
content_extractor=content_extractor,
settings=settings
)
+369
View File
@@ -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"