diff --git a/CHANGELOG.md b/CHANGELOG.md index 02d2a3d..0187f6b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/pyproject.toml b/pyproject.toml index c63a4b7..f47edf0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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" diff --git a/requirements.txt b/requirements.txt index d547598..5013f1d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -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 diff --git a/src/clients/content_extractor.py b/src/clients/content_extractor.py new file mode 100644 index 0000000..9460eac --- /dev/null +++ b/src/clients/content_extractor.py @@ -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") diff --git a/src/config.py b/src/config.py index 0b4a291..1a69c60 100644 --- a/src/config.py +++ b/src/config.py @@ -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.""" diff --git a/src/core/dependencies.py b/src/core/dependencies.py index 1aff0b5..eb3fe7d 100644 --- a/src/core/dependencies.py +++ b/src/core/dependencies.py @@ -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() ) diff --git a/src/main.py b/src/main.py index 18f4166..65bc223 100644 --- a/src/main.py +++ b/src/main.py @@ -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" diff --git a/src/models/content.py b/src/models/content.py new file mode 100644 index 0000000..cd38d92 --- /dev/null +++ b/src/models/content.py @@ -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") diff --git a/src/models/rag_search.py b/src/models/rag_search.py new file mode 100644 index 0000000..8cdfbf8 --- /dev/null +++ b/src/models/rag_search.py @@ -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" + ) diff --git a/src/routers/content.py b/src/routers/content.py new file mode 100644 index 0000000..404e0f2 --- /dev/null +++ b/src/routers/content.py @@ -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") diff --git a/src/routers/rag_search.py b/src/routers/rag_search.py new file mode 100644 index 0000000..56ee745 --- /dev/null +++ b/src/routers/rag_search.py @@ -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") diff --git a/src/services/hybrid_rag_service.py b/src/services/hybrid_rag_service.py index 41478be..13beb75 100644 --- a/src/services/hybrid_rag_service.py +++ b/src/services/hybrid_rag_service.py @@ -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" } diff --git a/src/services/rag_search_service.py b/src/services/rag_search_service.py new file mode 100644 index 0000000..4d8dc98 --- /dev/null +++ b/src/services/rag_search_service.py @@ -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 diff --git a/tests/test_content_extractor.py b/tests/test_content_extractor.py new file mode 100644 index 0000000..24f55f5 --- /dev/null +++ b/tests/test_content_extractor.py @@ -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 = "
Test" + 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 = "" + 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 = "Test" + 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 = "Test" + 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 "" + + 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 = "