Compare commits

...
8 Commits
Author SHA1 Message Date
jpmschweitzerandClaude Opus 4.5 a1832e3245 refactor: consolidate Ollama model configuration
Build and Push / build (release) Successful in 27s
- Add OLLAMA_EMBEDDING_MODEL for embeddings (nomic-embed-text)
- OLLAMA_MODEL now used for all LLM operations (mistral-nemo-large:latest)
- Remove separate reranker_model setting
- Update WikiPageWriter to use settings instead of hardcoded model
- Improves VRAM efficiency by keeping one model hot

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-22 11:15:27 +01:00
jpmschweitzerandClaude Opus 4.5 5be31a5a00 docs: add release flow section to AGENTS.md
🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-18 18:51:49 +01:00
jpmschweitzerandClaude Opus 4.5 376284f90e fix: add content_extractor to smart-create endpoint
Build and Push / build (release) Successful in 30s
The POST /wiki/pages/smart-create endpoint was failing with 500
Internal Server Error because HybridRAGService.__init__() was
missing the required content_extractor parameter.

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-16 09:31:23 +01:00
jpmschweitzerandClaude Opus 4.5 f095de1162 docs: add HybridRAG architecture documentation
Documents two-stage RRF, configuration options, and notes
potential vector search noise improvements for future reference.

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-15 17:54:38 +01:00
jpmschweitzerandClaude Opus 4.5 c359fcbcd8 feat: two-stage RRF for fair wiki vs web ranking
Build and Push / build (release) Successful in 28s
- Merge vector+graph into single wiki source before RRF with web
- Wiki pages no longer get 2x advantage from dual retrieval
- Add vector similarity threshold (0.7 default)
- Skip synonyms in graph search to reduce noise
- Fix duplicate entity links bug in graph search

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-15 17:49:31 +01:00
jpmschweitzerandClaude Opus 4.5 464ec5380c fix: add content_extractor to hybrid_rag router dependency
Build and Push / build (release) Successful in 29s
The router had its own local get_hybrid_rag_service factory that was
missing the new content_extractor parameter, causing 500 errors.

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-15 17:00:38 +01:00
jpmschweitzerandClaude Opus 4.5 61863ff597 feat: add RAG search endpoint with content extraction
Build and Push / build (release) Successful in 1m2s
- 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>
2025-12-15 15:50:48 +01:00
jpmschweitzerandClaude Opus 4.5 16b86a7c1b feat: add Watchtower update trigger to build workflow
Build and Push / build (release) Successful in 30s
Automatically triggers Watchtower to update containers after successful build and push.

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-14 12:59:49 +01:00
25 changed files with 1994 additions and 99 deletions
+6
View File
@@ -25,3 +25,9 @@ jobs:
tags: |
git.schweitz.internal/jpmschweitzer/library-desk:latest
git.schweitz.internal/jpmschweitzer/library-desk:${{ github.ref_name }}
- name: Trigger Watchtower update
if: success()
run: |
curl -sf -H "Authorization: Bearer ${{ secrets.WATCHTOWER_TOKEN }}" \
http://watchtower:8080/v1/update
+26
View File
@@ -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
+86
View File
@@ -5,6 +5,92 @@ 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.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
- Watchtower update trigger in Gitea workflow after successful build
## [1.1.2] - 2025-12-14
### Fixed
+2 -1
View File
@@ -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
+58
View File
@@ -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
View File
@@ -1,6 +1,6 @@
[project]
name = "library-desk"
version = "1.1.2"
version = "1.3.2"
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")
+13 -3
View File
@@ -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."""
+56 -1
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__)
@@ -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,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")
+3 -1
View File
@@ -16,7 +16,7 @@ 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 +32,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 +49,7 @@ def get_hybrid_rag_service(
graph_service=graph_service,
searxng_client=searxng_client,
ollama_client=ollama_client,
content_extractor=content_extractor,
settings=settings
)
+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")
+4 -2
View File
@@ -24,7 +24,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 +180,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 +226,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(
+2 -2
View File
@@ -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(
@@ -465,7 +465,7 @@ JSON:"""
# Call Ollama for analysis
response = await self.ollama.generate_text(
prompt=prompt,
model=self.settings.reranker_model, # Use mistral-nemo
model=self.settings.ollama_model,
stream=False
)
+12 -2
View File
@@ -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 []
+159 -45
View File
@@ -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
@@ -284,7 +294,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 +320,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 +353,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 +362,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 +409,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 +545,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
+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
+4 -3
View File
@@ -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,
+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"
+50 -37
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
)
@@ -207,53 +219,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:
+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"