Compare commits

..
15 Commits
Author SHA1 Message Date
jpmschweitzerandClaude Opus 4.5 318636d33d release: v1.3.3 - LLM prompt improvements and dead code cleanup
Build and Push / build (release) Successful in 28s
- Improved LLM prompts with temperature control and negative constraints
- Removed dead code and unused imports
- Wired /query/semantic and /query/graph to real implementations
- Updated test fixtures for external service access

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-23 17:23:44 +01:00
jpmschweitzerandClaude Opus 4.5 b976da0092 refactor: improve web results analysis prompt
Apply llm-findings.md recommendations:

Web results analysis (temp 0.0):
- Add ANALYSIS STEPS for chain-of-thought reasoning
- Strict RULES section with negative constraints:
  - "Do NOT suggest pages with insufficient info"
  - "Do NOT invent entities not mentioned"
  - "Do NOT suggest paths outside taxonomy"
- Conservative approach: quality over quantity
- Removed "be INCLUSIVE" guidance (caused over-suggestion)

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-23 17:22:12 +01:00
jpmschweitzerandClaude Opus 4.5 edfe11f0fb refactor: improve wiki page writer prompts
Apply llm-findings.md recommendations:

Conflict detection (temp 0.0):
- Add explicit analysis steps (CoT)
- Strict rules: only flag direct contradictions
- Negative constraints for false positives

Page creation (temp 0.3):
- Add CRITICAL CONSTRAINTS section
- "Do NOT invent facts not in source"
- "Do NOT fill sections with placeholders"
- Omit sections if information unavailable

Page reconstruction (temp 0.2):
- Add preservation constraints
- "Do NOT rephrase facts changing meaning"
- "Preserve exact quotes, dates, numbers verbatim"

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-23 17:21:58 +01:00
jpmschweitzerandClaude Opus 4.5 8e003bb9e8 refactor: improve keyword extraction and re-ranking prompts
Apply llm-findings.md recommendations:

Keyword extraction (temp 0.0):
- Add negative constraints: "Do NOT invent terms"
- Simplify output format
- Remove verbose example

LLM re-ranking (temp 0.0):
- Add explicit rules section
- Negative constraints: "Do NOT consider document length"
- Clearer output format specification

Both prompts now use temperature=0.0 for deterministic,
consistent outputs.

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-23 17:21:41 +01:00
jpmschweitzerandClaude Opus 4.5 dfd1f19bf9 feat: add temperature parameter to Ollama generate_text
Add temperature control for LLM text generation:
- temperature=0.0 for deterministic outputs (JSON, rankings)
- temperature=0.3-0.5 for controlled creative content
- None uses model default (~0.7 for mistral-nemo)

Based on llm-findings.md recommendations for improving
mistral-nemo output consistency.

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-23 17:21:18 +01:00
jpmschweitzerandClaude Opus 4.5 1aca286703 test: update fixtures to use real service host
- Add TEST_HOST config (default: 192.168.86.149) in conftest.py
- Update all test fixtures to use configurable host instead of
  docker hostnames (neo4j, qdrant, wiki, etc.)
- Fix test_integration.py WikiJS client to use username/password auth
- Fix ollama_client fixture to use ollama_embedding_model setting

This allows tests to run against real services from outside Docker.

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-23 17:21:02 +01:00
jpmschweitzerandClaude Opus 4.5 262a58b0d2 refactor: wire query endpoints and remove stub endpoints
- Wire /query/semantic to VectorService.search()
- Wire /query/graph to GraphService.execute_query()
- Remove stub endpoints:
  - /stats (returns zeros)
  - /ingest/document (shadowed by router)
  - /ingest/batch (shadowed by router)
- Remove unused StatsResponse model
- Add TODO.md tracking remaining stubs to implement:
  - /ingest/check-updates
  - /ingest/status/{document_id}
  - /ingest/repo-status/{repository}
  - /deduplicate/check

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-23 17:20:45 +01:00
jpmschweitzerandClaude Opus 4.5 02d728ac5b refactor: remove dead code and unused imports
- Remove unused get_default_user() from dependencies.py
- Remove unused imports from routers:
  - wiki.py: HTTPAuthorizationCredentials, Security
  - graph.py: Neo4jClient, WikiJSClient
  - hybrid_rag.py: VectorService, GraphService (duplicates)

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-23 17:20:27 +01:00
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
30 changed files with 2277 additions and 309 deletions
+26
View File
@@ -23,6 +23,32 @@
* **Update `CHANGELOG.md`** with every user-facing change. * **Update `CHANGELOG.md`** with every user-facing change.
* Format: `## [Unreleased] - YYYY-MM-DD` followed by `### Added`, `### Changed`, or `### Fixed`. * 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 ## 2. FastAPI Architecture & Best Practices
+106
View File
@@ -5,6 +5,112 @@ All notable changes to Library Desk will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), 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). and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
## [1.3.3] - 2025-12-23
### Added
- Temperature parameter to `OllamaClient.generate_text()` for controlling output determinism
- `TODO.md` tracking remaining stub endpoints to implement
- Wired `/query/semantic` endpoint to VectorService
- Wired `/query/graph` endpoint to GraphService
### Changed
- **Improved LLM prompts** based on llm-findings.md recommendations:
- Keyword extraction: temperature 0.0, negative constraints
- LLM re-ranking: temperature 0.0, explicit rules
- Conflict detection: temperature 0.0, analysis steps (CoT)
- Wiki page creation: temperature 0.3, anti-hallucination constraints
- Page reconstruction: temperature 0.2, preservation constraints
- Web results analysis: temperature 0.0, conservative approach
- Test fixtures now use configurable host (TEST_HOST) instead of Docker hostnames
### Removed
- Dead code: unused `get_default_user()` function
- Unused imports from routers (wiki.py, graph.py, hybrid_rag.py)
- Stub endpoints shadowed by real implementations (/stats, /ingest/document, /ingest/batch)
## [1.3.2] - 2025-12-22
### Changed
- **Consolidated Ollama model configuration** - All LLM operations now use single `OLLAMA_MODEL` environment variable
- Removed separate `reranker_model` setting
- HybridRAG re-ranking, consolidation analysis, and wiki page writing all use the same model
- Improves VRAM efficiency by keeping one model hot
- Added `OLLAMA_EMBEDDING_MODEL` environment variable for embedding model (previously overloaded `OLLAMA_MODEL`)
- Updated WikiPageWriter to accept settings instead of hardcoded model name
## [1.3.1] - 2025-12-16
### Fixed
- Smart create endpoint missing `content_extractor` dependency causing 500 errors on `POST /wiki/pages/smart-create`
## [1.3.0] - 2025-12-15
### Changed
- **Two-Stage RRF Architecture** - Major refactor to level the playing field between wiki and web results
- Stage 1: Vector and graph results merged into single "wiki" ranking using mini-RRF
- Stage 2: Final RRF between wiki (single source) and web (single source)
- Wiki pages no longer get 2x advantage from appearing in both vector and graph searches
- Multi-source confirmation still determines wiki internal ranking
- **Skip synonyms in graph search** - LLM-generated synonyms (e.g., "author") no longer match unrelated graph entities (e.g., "author2000")
- Vector search still uses synonyms for semantic similarity
- Graph search uses only core keywords for exact entity matching
### Added
- `VECTOR_SIMILARITY_THRESHOLD` config setting (default: 0.7) to filter weak vector matches
- Deduplication in graph search to prevent same document appearing multiple times
### Fixed
- Graph search duplicate entity bug where same document could appear twice if entity linked multiple times
## [1.2.1] - 2025-12-15
### Fixed
- HybridRAG router missing `content_extractor` dependency causing 500 errors on `/query/hybrid` endpoint
## [1.2.0] - 2025-12-15
### Added
- **RAG Search Endpoint** (`POST /rag/search`)
- Web, news, and image search via SearXNG
- Full content extraction using Trafilatura (F1 score 0.958)
- Redis caching with configurable TTL
- Markdown sources summary for LLM consumption
- Returns both extracted content and original snippets
- **Content Extraction Endpoints** (`/content/*`)
- `POST /content/extract` - Extract content from a single URL
- `POST /content/extract/batch` - Batch extraction (up to 20 URLs)
- Reusable ContentExtractor client for use across the codebase
- **HybridRAG Content Extraction Enhancement**
- Web search results now include full extracted content via Trafilatura
- Falls back to original snippets if extraction fails
- Improves context quality for LLM re-ranking and consumption
### Changed
- Added new configuration options:
- `SEARCH_CACHE_TTL` - Search cache TTL in seconds (default: 300)
- `SEARCH_TIMEOUT` - SearXNG timeout (default: 10s)
- `CONTENT_EXTRACTION_TIMEOUT` - Per-URL extraction timeout (default: 5s)
- `CONTENT_MAX_LENGTH` - Max extracted content length (default: 2000)
- `SEARCH_DEFAULT_LIMIT` - Default search results (default: 10)
### Dependencies
- Added `trafilatura~=1.12.0` for content extraction
## [1.1.3] - 2025-12-14 ## [1.1.3] - 2025-12-14
### Added ### Added
+2 -1
View File
@@ -49,7 +49,8 @@ QDRANT_PORT=6333
WIKIJS_URL=http://wiki:3000 WIKIJS_URL=http://wiki:3000
SEARXNG_URL=http://searxng:8080 SEARXNG_URL=http://searxng:8080
OLLAMA_URL=http://ollama:11434 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_HOST=redis-shared
REDIS_PORT=6379 REDIS_PORT=6379
REDIS_DB=2 REDIS_DB=2
+52
View File
@@ -0,0 +1,52 @@
# TODO
Outstanding work items for Library Desk.
## Stub Endpoints to Implement
The following endpoints in `src/main.py` return stub responses and need real implementations:
### Ingestion Status Endpoints
#### `POST /ingest/check-updates`
Check which documents need updating based on content hashes. Used by Scheduler to determine what changed since last sync.
**Implementation needed:**
1. Query existing documents by path
2. Compare content hashes
3. Return list of updates needed
#### `GET /ingest/status/{document_id}`
Get processing status for a document.
**Implementation needed:**
- Status tracking system (Redis or database)
- Track ingestion progress per document
#### `GET /ingest/repo-status/{repository}`
Get indexing status for an entire repository.
**Implementation needed:**
- Repository-level statistics
- Track which documents from a repo are indexed
### Deduplication
#### `POST /deduplicate/check`
Check for duplicate or highly similar documents using vector similarity and graph analysis.
**Implementation needed:**
1. Get document embedding from Qdrant
2. Find similar vectors above threshold
3. Check graph relationships
4. Return candidates with similarity scores
## System Statistics
#### `GET /stats`
Get system statistics (wiki pages, neo4j nodes, qdrant vectors).
**Implementation needed:**
- Query Neo4j for node count
- Query Qdrant for vector count
- Query Wiki.js for page count
+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] [project]
name = "library-desk" name = "library-desk"
version = "1.1.3" version = "1.3.3"
description = "Coordination service for The Library system - HybridRAG queries, document ingestion, entity extraction, and knowledge consolidation" description = "Coordination service for The Library system - HybridRAG queries, document ingestion, entity extraction, and knowledge consolidation"
readme = "README.md" readme = "README.md"
requires-python = ">=3.12" requires-python = ">=3.12"
+3
View File
@@ -25,6 +25,9 @@ python-multipart~=0.0.20
# Utilities # Utilities
python-dateutil~=2.9.0 python-dateutil~=2.9.0
# Content Extraction
trafilatura~=1.12.0
# Testing # Testing
pytest~=8.3.0 pytest~=8.3.0
pytest-asyncio~=0.24.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")
+11 -3
View File
@@ -249,7 +249,8 @@ class OllamaClient:
self, self,
prompt: str, prompt: str,
model: Optional[str] = None, model: Optional[str] = None,
stream: bool = False stream: bool = False,
temperature: Optional[float] = None
) -> Optional[str]: ) -> Optional[str]:
""" """
Generate text completion (for non-embedding use cases). Generate text completion (for non-embedding use cases).
@@ -258,12 +259,15 @@ class OllamaClient:
prompt: Input prompt prompt: Input prompt
model: Model name (defaults to self.model) model: Model name (defaults to self.model)
stream: Enable streaming response stream: Enable streaming response
temperature: Sampling temperature (0.0 = deterministic, higher = more creative)
None uses model default (~0.7 for mistral-nemo)
Returns: Returns:
Generated text or None on failure Generated text or None on failure
Note: This is primarily for debugging/testing. Use specialized Note: Use temperature=0.0 for deterministic outputs like JSON parsing,
LLM services for production text generation. ranking, and factual extraction. Use higher values (0.3-0.7) for
creative content generation.
""" """
try: try:
payload = { payload = {
@@ -272,6 +276,10 @@ class OllamaClient:
"stream": stream "stream": stream
} }
# Add temperature to options if specified
if temperature is not None:
payload["options"] = {"temperature": temperature}
response = await self.client.post( response = await self.client.post(
self.generate_url, self.generate_url,
json=payload json=payload
+13 -3
View File
@@ -62,16 +62,17 @@ class Settings(BaseSettings):
# SearXNG Configuration # SearXNG Configuration
searxng_url: str = Field(default="http://searxng:8080", description="SearXNG URL") 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_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 # 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") 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_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_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") 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 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") 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") app_version: str = Field(default=__version__, description="Application version")
debug: bool = Field(default=False, description="Debug mode") 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 @property
def qdrant_url(self) -> str: def qdrant_url(self) -> str:
"""Computed Qdrant URL.""" """Computed Qdrant URL."""
+54 -11
View File
@@ -13,12 +13,15 @@ from typing import Annotated
from fastapi import Depends from fastapi import Depends
import logging import logging
import redis.asyncio as aioredis
from src.config import Settings, get_settings from src.config import Settings, get_settings
from src.clients.neo4j_client import Neo4jClient from src.clients.neo4j_client import Neo4jClient
from src.clients.qdrant_client import QdrantClientWrapper from src.clients.qdrant_client import QdrantClientWrapper
from src.clients.wikijs_client import WikiJSClient from src.clients.wikijs_client import WikiJSClient
from src.clients.searxng_client import SearXNGClient from src.clients.searxng_client import SearXNGClient
from src.clients.ollama_client import OllamaClient from src.clients.ollama_client import OllamaClient
from src.clients.content_extractor import ContentExtractor
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -110,12 +113,49 @@ def get_ollama_client() -> OllamaClient:
settings = get_settings() settings = get_settings()
client = OllamaClient( client = OllamaClient(
base_url=settings.ollama_url, base_url=settings.ollama_url,
model=settings.ollama_model model=settings.ollama_embedding_model
) )
logger.debug("Created Ollama client instance") logger.debug("Created Ollama client instance")
return client 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 # Type aliases for FastAPI endpoint dependencies
# Usage: def my_endpoint(neo4j: Neo4jDep): # Usage: def my_endpoint(neo4j: Neo4jDep):
Neo4jDep = Annotated[Neo4jClient, Depends(get_neo4j_client)] 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)] WikiJSDep = Annotated[WikiJSClient, Depends(get_wikijs_client)]
SearXNGDep = Annotated[SearXNGClient, Depends(get_searxng_client)] SearXNGDep = Annotated[SearXNGClient, Depends(get_searxng_client)]
OllamaDep = Annotated[OllamaClient, Depends(get_ollama_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 # Lifecycle management functions
@@ -336,20 +378,21 @@ def get_hybrid_rag_service() -> "HybridRAGService":
graph_service=get_graph_service(), graph_service=get_graph_service(),
searxng_client=get_searxng_client(), searxng_client=get_searxng_client(),
ollama_client=get_ollama_client(), ollama_client=get_ollama_client(),
content_extractor=get_content_extractor(),
settings=get_settings() settings=get_settings()
) )
# Utility: Get default user from settings or multi_tenancy @lru_cache
def get_default_user() -> str: def get_rag_search_service() -> "RAGSearchService":
""" """Get RAGSearchService singleton."""
Get default user for operations. from src.services.rag_search_service import RAGSearchService
return RAGSearchService(
Returns: searxng_client=get_searxng_client(),
Default user identifier content_extractor=get_content_extractor(),
""" redis_client=get_redis_client(),
from src.core.multi_tenancy import DEFAULT_USER settings=get_settings()
return DEFAULT_USER )
# Authentication # Authentication
+75 -101
View File
@@ -8,7 +8,7 @@ Following best practices:
- OpenAPI documentation - OpenAPI documentation
""" """
from fastapi import FastAPI, HTTPException, Depends from fastapi import FastAPI, HTTPException, Depends, Query
from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel from pydantic import BaseModel
@@ -17,7 +17,10 @@ import logging
from pathlib import Path from pathlib import Path
from src.config import Settings, get_settings, __version__ from src.config import Settings, get_settings, __version__
from src.core.dependencies import verify_api_key from src.core.dependencies import (
verify_api_key, QdrantDep, WikiJSDep, OllamaDep, Neo4jDep
)
from src.core.multi_tenancy import DEFAULT_USER
# Configure logging # Configure logging
logging.basicConfig( logging.basicConfig(
@@ -45,7 +48,10 @@ app.add_middleware(
) )
# Register routers # 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(wiki.router)
app.include_router(tools.router) app.include_router(tools.router)
@@ -56,6 +62,8 @@ app.include_router(consolidation.router)
app.include_router(ingestion.router) app.include_router(ingestion.router)
app.include_router(entity_linking.router) app.include_router(entity_linking.router)
app.include_router(webhooks.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 # Mount static files directory for Wiki.js integration scripts
static_dir = Path(__file__).parent.parent / "static" static_dir = Path(__file__).parent.parent / "static"
@@ -73,13 +81,6 @@ class HealthResponse(BaseModel):
services: Dict[str, Any] services: Dict[str, Any]
class StatsResponse(BaseModel):
"""Statistics response model."""
wiki_pages: int
neo4j_nodes: int
qdrant_vectors: int
# Routes # Routes
@app.get("/", tags=["Root"]) @app.get("/", tags=["Root"])
async def root() -> Dict[str, str]: async def root() -> Dict[str, str]:
@@ -136,77 +137,6 @@ async def health(settings: Settings = Depends(get_settings)) -> HealthResponse:
) )
@app.get("/stats", response_model=StatsResponse, tags=["System"])
async def stats(
api_key: str = Depends(verify_api_key)
) -> StatsResponse:
"""
Get system statistics.
Protected endpoint - requires API key.
TODO: Implement actual stats gathering from:
- Neo4j (node count)
- Qdrant (vector count)
- Wiki.js (page count)
"""
return StatsResponse(
wiki_pages=0,
neo4j_nodes=0,
qdrant_vectors=0
)
# Ingestion endpoints (for Scheduler integration)
@app.post("/ingest/document", tags=["Ingestion"])
async def ingest_document(
document: Dict[str, Any],
api_key: str = Depends(verify_api_key)
) -> Dict[str, Any]:
"""
Ingest a single document for indexing.
Used by The Scheduler to add mirrored documentation to the knowledge base.
Expected fields:
- source: str (e.g., "github", "gitea")
- repository: str (e.g., "anthropic-cookbook")
- path: str (file path)
- content: str (document content)
- metadata: dict (commit, author, tags, etc.)
TODO: Implement document ingestion pipeline:
1. Chunk content
2. Generate embeddings (Ollama)
3. Extract entities (NLP)
4. Index in Qdrant
5. Create graph nodes/relationships in Neo4j
"""
return {
"message": "Document ingestion not yet implemented",
"document_id": f"doc_{document.get('path', 'unknown')}",
"status": "stub"
}
@app.post("/ingest/batch", tags=["Ingestion"])
async def batch_ingest(
batch: Dict[str, Any],
api_key: str = Depends(verify_api_key)
) -> Dict[str, Any]:
"""
Ingest multiple documents in a batch.
More efficient than individual ingestion for large syncs.
TODO: Implement batch processing with task queue
"""
document_count = len(batch.get("documents", []))
return {
"message": "Batch ingestion not yet implemented",
"batch_id": "batch_stub",
"total_documents": document_count,
"status": "stub"
}
@app.post("/ingest/check-updates", tags=["Ingestion"]) @app.post("/ingest/check-updates", tags=["Ingestion"])
async def check_updates( async def check_updates(
documents: Dict[str, Any], documents: Dict[str, Any],
@@ -264,41 +194,85 @@ async def get_repo_status(
} }
# Query endpoints (stubs for future implementation) # Query endpoints
# NOTE: /query/hybrid is now implemented in routers/hybrid_rag.py # NOTE: /query/hybrid is implemented in routers/hybrid_rag.py
@app.post("/query/semantic", tags=["Query"]) @app.post("/query/semantic", tags=["Query"])
async def semantic_query( async def semantic_query(
query: Dict[str, Any], query: str = Query(..., min_length=1, description="Search query text"),
user: str = Query(default=DEFAULT_USER, description="User identifier"),
limit: int = Query(default=10, ge=1, le=100, description="Maximum results"),
score_threshold: float = Query(default=0.5, ge=0.0, le=1.0, description="Minimum similarity score"),
qdrant_client: QdrantDep = None,
wiki_client: WikiJSDep = None,
ollama_client: OllamaDep = None,
api_key: str = Depends(verify_api_key) api_key: str = Depends(verify_api_key)
) -> Dict[str, Any]: ):
""" """
Semantic search via Qdrant. Semantic search via Qdrant vector similarity.
Pure vector similarity search.
TODO: Implement semantic search Searches document chunks using embedding similarity. Returns matching
chunks with relevance scores, page titles, and paths.
**Example:**
```
POST /query/semantic?query=docker%20configuration&user=jpmschweitzer&limit=10
```
**Returns:** List of matching chunks with similarity scores (0-1)
""" """
return { from src.services.vector_service import VectorService
"message": "Semantic search not yet implemented",
"query": query vector_service = VectorService(qdrant_client, wiki_client, ollama_client)
} try:
return await vector_service.search(
query=query,
user=user,
limit=limit,
score_threshold=score_threshold
)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
logger.error(f"Semantic search failed: {e}", exc_info=True)
raise HTTPException(status_code=500, detail="Search failed")
@app.post("/query/graph", tags=["Query"]) @app.post("/query/graph", tags=["Query"])
async def graph_query( async def graph_query(
query: Dict[str, Any], query: str = Query(..., description="Cypher query to execute"),
user: str = Query(default=DEFAULT_USER, description="User for scoping (auto-filters results)"),
neo4j_client: Neo4jDep = None,
wiki_client: WikiJSDep = None,
api_key: str = Depends(verify_api_key) api_key: str = Depends(verify_api_key)
) -> Dict[str, Any]: ):
""" """
Graph traversal via Neo4j. Execute a Cypher query against the Neo4j knowledge graph.
Execute Cypher queries.
TODO: Implement graph queries Queries are automatically scoped to the user's data for security.
Use this for custom graph traversals beyond what /graph/nodes provides.
**Example:**
```
POST /query/graph?query=MATCH%20(d:Document)-[:MENTIONS]->(p:Person)%20RETURN%20d,p&user=jpmschweitzer
```
**Security:** All queries are user-scoped to prevent cross-user data access.
""" """
return { from src.services.graph_service import GraphService
"message": "Graph query not yet implemented",
"query": query graph_service = GraphService(neo4j_client, wiki_client)
} try:
return await graph_service.execute_query(
query=query,
parameters={},
user=user
)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
logger.error(f"Graph query failed: {e}", exc_info=True)
raise HTTPException(status_code=500, detail="Query execution failed")
# Deduplication endpoints # Deduplication endpoints
+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")
-2
View File
@@ -15,8 +15,6 @@ from src.models.graph import (
MindMapResponse MindMapResponse
) )
from src.services.graph_service import GraphService from src.services.graph_service import GraphService
from src.clients.neo4j_client import Neo4jClient
from src.clients.wikijs_client import WikiJSClient
from src.core.dependencies import Neo4jDep, WikiJSDep, verify_api_key from src.core.dependencies import Neo4jDep, WikiJSDep, verify_api_key
from src.core.multi_tenancy import DEFAULT_USER from src.core.multi_tenancy import DEFAULT_USER
+3 -5
View File
@@ -10,13 +10,9 @@ import logging
from src.models.hybrid_rag import HybridRAGRequest, HybridRAGResponse from src.models.hybrid_rag import HybridRAGRequest, HybridRAGResponse
from src.services.hybrid_rag_service import HybridRAGService from src.services.hybrid_rag_service import HybridRAGService
from src.services.vector_service import VectorService
from src.services.graph_service import GraphService
from src.clients.searxng_client import SearXNGClient
from src.clients.ollama_client import OllamaClient
from src.core.dependencies import ( from src.core.dependencies import (
Neo4jDep, WikiJSDep, QdrantDep, OllamaDep, Neo4jDep, WikiJSDep, QdrantDep, OllamaDep,
SearXNGDep, verify_api_key, get_settings SearXNGDep, ContentExtractorDep, verify_api_key, get_settings
) )
from src.config import Settings from src.config import Settings
@@ -32,6 +28,7 @@ def get_hybrid_rag_service(
qdrant_client: QdrantDep, qdrant_client: QdrantDep,
ollama_client: OllamaDep, ollama_client: OllamaDep,
searxng_client: SearXNGDep, searxng_client: SearXNGDep,
content_extractor: ContentExtractorDep,
settings: Settings = Depends(get_settings) settings: Settings = Depends(get_settings)
) -> HybridRAGService: ) -> HybridRAGService:
"""Get HybridRAG service instance with all dependencies.""" """Get HybridRAG service instance with all dependencies."""
@@ -48,6 +45,7 @@ def get_hybrid_rag_service(
graph_service=graph_service, graph_service=graph_service,
searxng_client=searxng_client, searxng_client=searxng_client,
ollama_client=ollama_client, ollama_client=ollama_client,
content_extractor=content_extractor,
settings=settings 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")
+5 -4
View File
@@ -5,8 +5,7 @@ Endpoints for wiki page and dossier management.
All operations are scoped to user namespaces for multi-tenancy. All operations are scoped to user namespaces for multi-tenancy.
""" """
from fastapi import APIRouter, HTTPException, Depends, Query, Security, BackgroundTasks from fastapi import APIRouter, HTTPException, Depends, Query, BackgroundTasks
from fastapi.security import HTTPAuthorizationCredentials
from typing import Optional from typing import Optional
import logging import logging
@@ -24,7 +23,7 @@ from src.clients.neo4j_client import Neo4jClient
from src.clients.qdrant_client import QdrantClientWrapper from src.clients.qdrant_client import QdrantClientWrapper
from src.clients.ollama_client import OllamaClient from src.clients.ollama_client import OllamaClient
from src.core.dependencies import ( 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 verify_api_key, get_settings, get_hybrid_rag_service, get_ingestion_service
) )
from src.core.multi_tenancy import DEFAULT_USER from src.core.multi_tenancy import DEFAULT_USER
@@ -180,6 +179,7 @@ async def smart_create_page(
qdrant_client: QdrantDep, qdrant_client: QdrantDep,
ollama_client: OllamaDep, ollama_client: OllamaDep,
searxng_client: SearXNGDep, searxng_client: SearXNGDep,
content_extractor: ContentExtractorDep,
settings: Settings = Depends(get_settings), settings: Settings = Depends(get_settings),
api_key: str = Depends(verify_api_key) api_key: str = Depends(verify_api_key)
): ):
@@ -225,9 +225,10 @@ async def smart_create_page(
graph_service=graph_service, graph_service=graph_service,
searxng_client=searxng_client, searxng_client=searxng_client,
ollama_client=ollama_client, ollama_client=ollama_client,
content_extractor=content_extractor,
settings=settings 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 # Step 1-5: Research + Generate + Create page
page, research_data = await wiki_service.smart_create_page( page, research_data = await wiki_service.smart_create_page(
+17 -10
View File
@@ -47,7 +47,7 @@ class ConsolidationService:
self.ollama = ollama self.ollama = ollama
self.wiki = wiki self.wiki = wiki
self.settings = settings 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 self.ingestion_service = ingestion_service # Optional to avoid circular dependency
async def consolidate_knowledge( async def consolidate_knowledge(
@@ -410,13 +410,19 @@ This is a PERSONAL knowledge base using Schema.org-aligned taxonomy that capture
- Projects: Work projects, personal projects (Schema.org: Project) - Projects: Work projects, personal projects (Schema.org: Project)
- Reference: General knowledge, how-tos (Custom extension) - Reference: General knowledge, how-tos (Custom extension)
Identify information worth documenting: ANALYSIS STEPS:
1. New topics/people/things that deserve their own wiki page 1. Read each web result carefully for substantive, factual content
2. Facts that could enhance existing pages 2. Identify genuinely novel information not likely already known
3. Entities (people, places, things, concepts) for the knowledge graph 3. Match topics to appropriate taxonomy categories
4. Generate valid paths following the exact format below
Be INCLUSIVE - if someone searched for it, it's likely worth documenting. RULES:
Personal information is just as valuable as technical information. - Do NOT suggest pages for topics with insufficient information in results
- Do NOT invent entities not explicitly mentioned in results
- Do NOT suggest paths that don't match the taxonomy exactly
- Do NOT suggest generic or vague page topics
- Be CONSERVATIVE - fewer high-quality suggestions is better than many low-quality ones
- ONLY suggest documentation for substantive, specific information
**CRITICAL: Use ONLY these Schema.org-aligned path prefixes (case-sensitive):** **CRITICAL: Use ONLY these Schema.org-aligned path prefixes (case-sensitive):**
@@ -462,11 +468,12 @@ Return ONLY valid JSON:
JSON:""" JSON:"""
try: try:
# Call Ollama for analysis # Call Ollama for analysis (temperature=0.0 for consistent classification)
response = await self.ollama.generate_text( response = await self.ollama.generate_text(
prompt=prompt, prompt=prompt,
model=self.settings.reranker_model, # Use mistral-nemo model=self.settings.ollama_model,
stream=False stream=False,
temperature=0.0
) )
if not response: if not response:
+12 -2
View File
@@ -1077,8 +1077,18 @@ Feel free to expand it with more details!
search_query, search_query,
{"terms": all_terms, "limit": limit} {"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: except Exception as e:
logger.error(f"Graph document search failed: {e}", exc_info=True) logger.error(f"Graph document search failed: {e}", exc_info=True)
return [] return []
+181 -64
View File
@@ -6,7 +6,7 @@ HybridRAG service combining vector, graph, and web search.
1. Parallel Retrieval - Vector + Graph + Web search 1. Parallel Retrieval - Vector + Graph + Web search
2. RRF Fusion - Merge results with Reciprocal Rank Fusion 2. RRF Fusion - Merge results with Reciprocal Rank Fusion
3. Enrichment - Add related dossiers via graph 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 5. Context Formatting - Format for LLM consumption
6. Persistence - Store for Librarian processing 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.services.graph_service import GraphService
from src.clients.searxng_client import SearXNGClient from src.clients.searxng_client import SearXNGClient
from src.clients.ollama_client import OllamaClient from src.clients.ollama_client import OllamaClient
from src.clients.content_extractor import ContentExtractor
from src.config import Settings from src.config import Settings
from src.models.hybrid_rag import ( from src.models.hybrid_rag import (
HybridRAGConfig, HybridRAGRequest, HybridRAGResponse, HybridRAGConfig, HybridRAGRequest, HybridRAGResponse,
@@ -44,6 +45,7 @@ class HybridRAGService:
graph_service: GraphService, graph_service: GraphService,
searxng_client: SearXNGClient, searxng_client: SearXNGClient,
ollama_client: OllamaClient, ollama_client: OllamaClient,
content_extractor: ContentExtractor,
settings: Settings settings: Settings
): ):
""" """
@@ -54,14 +56,16 @@ class HybridRAGService:
graph_service: Service for Neo4j graph search graph_service: Service for Neo4j graph search
searxng_client: Client for web search searxng_client: Client for web search
ollama_client: Client for LLM (keyword extraction, re-ranking) ollama_client: Client for LLM (keyword extraction, re-ranking)
content_extractor: Client for extracting full content from URLs
settings: Application settings settings: Application settings
""" """
self.vector = vector_service self.vector = vector_service
self.graph = graph_service self.graph = graph_service
self.searxng = searxng_client self.searxng = searxng_client
self.ollama = ollama_client self.ollama = ollama_client
self.content_extractor = content_extractor
self.settings = settings self.settings = settings
self.reranker_model = settings.reranker_model self.reranker_model = settings.ollama_model
async def search( async def search(
self, self,
@@ -101,14 +105,20 @@ class HybridRAGService:
timing["graph_ms"] = raw_results.get("timing", {}).get("graph_ms", 0) timing["graph_ms"] = raw_results.get("timing", {}).get("graph_ms", 0)
timing["web_ms"] = raw_results.get("timing", {}).get("web_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() 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( fused_results = self._reciprocal_rank_fusion(
results_by_source={ wiki_results=wiki_merged,
"vector": raw_results.get("vector", []), web_results=raw_results.get("web", []),
"graph": raw_results.get("graph", []),
"web": raw_results.get("web", [])
},
k=config.rrf_k k=config.rrf_k
) )
timing["fusion_ms"] = (time.time() - phase2_start) * 1000 timing["fusion_ms"] = (time.time() - phase2_start) * 1000
@@ -185,25 +195,21 @@ class HybridRAGService:
Returns: Returns:
Dictionary with keywords, entities, synonyms, expansions Dictionary with keywords, entities, synonyms, expansions
""" """
prompt = f"""Extract search terms from this query. For each important word, provide synonyms and expansions. prompt = f"""Extract search terms from this query.
Query: "{query}" Query: "{query}"
Return ONLY valid JSON: RULES:
{{ - Extract ONLY keywords explicitly present or directly implied in the query
"core_keywords": ["key", "words", "from", "query"], - Do NOT invent terms, concepts, or synonyms not clearly related
"synonyms": {{ - Do NOT add general knowledge or associations
"word": ["alternative", "terms"] - Provide synonyms ONLY for technical terms with well-known alternatives
}} - Return valid JSON only, no commentary
}}
Example for "Docker container hosting": Return format:
{{ {{
"core_keywords": ["docker", "container", "hosting"], "core_keywords": ["words", "from", "query"],
"synonyms": {{ "synonyms": {{"term": ["direct", "alternatives"]}}
"docker": ["containerization", "container runtime"],
"hosting": ["server", "infrastructure"]
}}
}} }}
JSON:""" JSON:"""
@@ -211,7 +217,8 @@ JSON:"""
try: try:
response = await self.ollama.generate_text( response = await self.ollama.generate_text(
prompt=prompt, prompt=prompt,
model=self.reranker_model model=self.reranker_model,
temperature=0.0 # Deterministic for consistent extraction
) )
# Parse JSON response (handle potential extra text) # Parse JSON response (handle potential extra text)
@@ -284,7 +291,8 @@ JSON:"""
response = await self.vector.search( response = await self.vector.search(
query=query, query=query,
user=user, user=user,
limit=config.vector_limit limit=config.vector_limit,
score_threshold=self.settings.vector_similarity_threshold
) )
results = [ results = [
{ {
@@ -309,11 +317,19 @@ JSON:"""
async def graph_search(): async def graph_search():
start = time.time() start = time.time()
try: 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( results = await self.graph.search_documents(
query=query, query=query,
user=user, user=user,
limit=config.graph_limit, limit=config.graph_limit,
keywords_data=keywords_data keywords_data=graph_keywords
) )
formatted = [ formatted = [
{ {
@@ -334,7 +350,7 @@ JSON:"""
tasks["graph"] = graph_search() tasks["graph"] = graph_search()
# Web search # Web search with content extraction
if config.enable_web: if config.enable_web:
async def web_search(): async def web_search():
start = time.time() start = time.time()
@@ -343,11 +359,24 @@ JSON:"""
query=query, query=query,
limit=config.web_limit 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 = [ formatted = [
{ {
"url": r.get("url"), "url": r.get("url"),
"title": r.get("title", ""), "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", ""), "engine": r.get("engine", ""),
"source": "web" "source": "web"
} }
@@ -377,52 +406,134 @@ JSON:"""
return output return output
def _reciprocal_rank_fusion( def _merge_wiki_sources(
self, self,
results_by_source: Dict[str, List], vector_results: List[Dict],
graph_results: List[Dict],
k: int = 60 k: int = 60
) -> List[Dict[str, Any]]: ) -> 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: 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) k: RRF constant (default 60)
Returns: 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 = {} rrf_scores = {}
for source, results in results_by_source.items(): # Wiki results (single source, already merged)
for rank, result in enumerate(results, start=1): for rank, result in enumerate(wiki_results, start=1):
# Use page_id for wiki results, url hash for web results page_id = result.get("page_id")
if result.get("page_id"): if not page_id:
result_id = f"page_{result['page_id']}" continue
elif result.get("url"): result_id = f"page_{page_id}"
result_id = f"url_{hash(result['url'])}" rrf_scores[result_id] = {
else: "result": result,
continue # Skip results without ID "rrf_score": 1 / (k + rank),
"sources": result.get("found_by", ["wiki"]),
"source_type": "wiki"
}
if result_id not in rrf_scores: # Web results (single source)
rrf_scores[result_id] = { for rank, result in enumerate(web_results, start=1):
"result": result, url = result.get("url")
"rrf_score": 0.0, if not url:
"sources": [], continue
"source_type": source result_id = f"url_{hash(url)}"
} rrf_scores[result_id] = {
"result": result,
# RRF formula: sum of 1/(k + rank) across sources "rrf_score": 1 / (k + rank),
rrf_scores[result_id]["rrf_score"] += 1 / (k + rank) "sources": ["web"],
rrf_scores[result_id]["sources"].append(source) "source_type": "web"
}
# 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"]))
)
# Sort by RRF score descending # Sort by RRF score descending
sorted_results = sorted( sorted_results = sorted(
@@ -431,7 +542,7 @@ JSON:"""
reverse=True 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 return sorted_results
@@ -509,21 +620,27 @@ JSON:"""
for i, r in enumerate(results) for i, r in enumerate(results)
]) ])
prompt = f"""Given this search query and documents, rank them by relevance. prompt = f"""Rank these documents by relevance to the query.
Query: {query} Query: {query}
Documents: Documents:
{docs_text} {docs_text}
Return only the numbers in order of relevance (most relevant first). RULES:
Example: 3,1,5,2,4 - Rank ONLY by how well content answers the query
- Do NOT consider document length, formatting, or style
- Do NOT add explanation or commentary
- Return ONLY comma-separated numbers, most relevant first
Example output: 3,1,5,2,4
Ranking:""" Ranking:"""
response = await self.ollama.generate_text( response = await self.ollama.generate_text(
prompt=prompt, prompt=prompt,
model=self.reranker_model model=self.reranker_model,
temperature=0.0 # Deterministic for consistent rankings
) )
# Parse response: "3,1,5,2,4" → [2, 0, 4, 1, 3] (0-indexed) # Parse response: "3,1,5,2,4" → [2, 0, 4, 1, 3] (0-indexed)
+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
+48 -23
View File
@@ -1,7 +1,7 @@
""" """
Intelligent Wiki Page Writer Service 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 - Holistic content restructuring
- Zero fact loss (unless superseded) - Zero fact loss (unless superseded)
- Conflict detection and flagging - Conflict detection and flagging
@@ -25,15 +25,16 @@ class WikiPageWriter:
Intelligent wiki page writer using LLM for content generation and restructuring. 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. Initialize wiki page writer.
Args: Args:
ollama_client: OllamaClient for LLM operations ollama_client: OllamaClient for LLM operations
settings: Application settings
""" """
self.ollama = ollama_client self.ollama = ollama_client
self.model = "mistral-nemo" # Default model for writing self.model = settings.ollama_model
async def create_page( async def create_page(
self, self,
@@ -130,8 +131,8 @@ class WikiPageWriter:
conflicts=conflicts conflicts=conflicts
) )
# Reconstruct with LLM # Reconstruct with LLM (lower temperature for precise merging)
reconstructed = await self._call_llm(prompt) reconstructed = await self._call_llm(prompt, temperature=0.2)
# Ensure standard sections are present # Ensure standard sections are present
reconstructed = self._ensure_standard_sections( reconstructed = self._ensure_standard_sections(
@@ -154,7 +155,7 @@ class WikiPageWriter:
Returns: Returns:
List of conflicts with: {fact_a, fact_b, confidence, context} List of conflicts with: {fact_a, fact_b, confidence, context}
""" """
prompt = f"""Analyze these two pieces of content for factual conflicts. prompt = f"""Analyze these contents for direct factual conflicts.
EXISTING CONTENT: EXISTING CONTENT:
{existing_content[:2000]} {existing_content[:2000]}
@@ -162,25 +163,26 @@ EXISTING CONTENT:
NEW INFORMATION: NEW INFORMATION:
{new_information[:2000]} {new_information[:2000]}
Identify any facts that contradict each other. For each conflict, provide: ANALYSIS STEPS:
1. The fact from existing content 1. Identify specific factual claims in existing content (dates, numbers, names, states)
2. The contradicting fact from new information 2. Identify specific factual claims in new content
3. Confidence level (low/medium/high) 3. Compare ONLY for direct contradictions (X says A, Y says not-A)
4. Context/explanation
Return ONLY valid JSON: RULES:
- Do NOT flag differences in wording or phrasing as conflicts
- Do NOT flag new/additional information as conflicts
- Do NOT flag opinion differences as conflicts
- ONLY flag direct factual contradictions
- Return valid JSON only, no commentary
Return format:
{{ {{
"conflicts": [ "conflicts": [
{{ {{"existing_fact": "...", "new_fact": "...", "confidence": "low/medium/high", "context": "..."}}
"existing_fact": "fact from old content",
"new_fact": "contradicting fact",
"confidence": "medium",
"context": "explanation of why these conflict"
}}
] ]
}} }}
If no conflicts, return: {{"conflicts": []}} If no conflicts: {{"conflicts": []}}
JSON:""" JSON:"""
@@ -188,7 +190,8 @@ JSON:"""
response = await self.ollama.generate_text( response = await self.ollama.generate_text(
prompt=prompt, prompt=prompt,
model=self.model, model=self.model,
stream=False stream=False,
temperature=0.0 # Deterministic for consistent conflict detection
) )
# Extract JSON # Extract JSON
@@ -347,6 +350,13 @@ FORMATTING RULES:
- Keep sections focused and scannable - Keep sections focused and scannable
- Adapt structure to content - not all sections apply to all topics - Adapt structure to content - not all sections apply to all topics
CRITICAL CONSTRAINTS:
- Do NOT invent facts not present in the source information above
- Do NOT add speculative information or assumptions
- Do NOT fill sections with placeholder text or generic statements
- If information for a section is not available, OMIT the section entirely
- Base ALL content strictly on provided source information
Generate ONLY the markdown content (do not include Sources, Knowledge Graph, or Mind Map sections - those are added automatically). Generate ONLY the markdown content (do not include Sources, Knowledge Graph, or Mind Map sections - those are added automatically).
MARKDOWN:""" MARKDOWN:"""
@@ -402,6 +412,13 @@ FORMATTING RULES:
- Bold important terms - Bold important terms
- Add subsections (###) where it improves clarity - Add subsections (###) where it improves clarity
CRITICAL CONSTRAINTS:
- Do NOT rephrase facts in ways that change their meaning
- Do NOT remove ANY information unless explicitly superseded by newer facts
- Do NOT add information not present in existing content or new information
- Preserve exact quotes, dates, numbers, and names verbatim
- Do NOT fill gaps with assumptions or general knowledge
OUTPUT INSTRUCTIONS: OUTPUT INSTRUCTIONS:
- Return complete page content (do not include Sources, Knowledge Graph, Mind Map - those are added automatically) - Return complete page content (do not include Sources, Knowledge Graph, Mind Map - those are added automatically)
- Include updated "Changes & Updates" section noting what was changed today - Include updated "Changes & Updates" section noting what was changed today
@@ -409,13 +426,21 @@ OUTPUT INSTRUCTIONS:
RECONSTRUCTED MARKDOWN:""" RECONSTRUCTED MARKDOWN:"""
async def _call_llm(self, prompt: str) -> str: async def _call_llm(self, prompt: str, temperature: float = 0.3) -> str:
"""Call LLM with prompt and return response.""" """
Call LLM with prompt and return response.
Args:
prompt: The prompt text
temperature: Sampling temperature (0.0=deterministic, higher=creative)
Default 0.3 for controlled but natural content generation
"""
try: try:
response = await self.ollama.generate_text( response = await self.ollama.generate_text(
prompt=prompt, prompt=prompt,
model=self.model, model=self.model,
stream=False stream=False,
temperature=temperature
) )
if not response: if not response:
+20 -9
View File
@@ -1,5 +1,6 @@
"""Pytest configuration and shared fixtures for Library Desk tests.""" """Pytest configuration and shared fixtures for Library Desk tests."""
import os
import pytest import pytest
import pytest_asyncio import pytest_asyncio
from typing import AsyncGenerator from typing import AsyncGenerator
@@ -7,6 +8,9 @@ from typing import AsyncGenerator
# Test configuration # Test configuration
pytest_plugins = ("pytest_asyncio",) pytest_plugins = ("pytest_asyncio",)
# Use real host for tests (services available at this IP)
TEST_HOST = os.environ.get("TEST_HOST", "192.168.86.149")
@pytest.fixture @pytest.fixture
def test_user() -> str: def test_user() -> str:
@@ -17,49 +21,56 @@ def test_user() -> str:
@pytest.fixture @pytest.fixture
def neo4j_test_uri() -> str: def neo4j_test_uri() -> str:
"""Test Neo4j URI.""" """Test Neo4j URI."""
return "bolt://neo4j:7687" return f"bolt://{TEST_HOST}:7687"
@pytest.fixture @pytest.fixture
def neo4j_test_auth() -> tuple: def neo4j_test_auth() -> tuple:
"""Test Neo4j authentication.""" """Test Neo4j authentication."""
return ("neo4j", "test_password") from src.config import get_settings
settings = get_settings()
return ("neo4j", settings.neo4j_password)
@pytest.fixture @pytest.fixture
def qdrant_test_url() -> str: def qdrant_test_url() -> str:
"""Test Qdrant URL.""" """Test Qdrant URL."""
return "http://qdrant:6333" return f"http://{TEST_HOST}:6333"
@pytest.fixture @pytest.fixture
def wikijs_test_config() -> dict: def wikijs_test_config() -> dict:
"""Test Wiki.js configuration.""" """Test Wiki.js configuration."""
from src.config import get_settings
settings = get_settings()
return { return {
"base_url": "http://wiki:3000", "base_url": f"http://{TEST_HOST}:3000",
"api_key": "test_api_key" "username": settings.wikijs_username,
"password": settings.wikijs_password
} }
@pytest.fixture @pytest.fixture
def searxng_test_url() -> str: def searxng_test_url() -> str:
"""Test SearXNG URL.""" """Test SearXNG URL."""
return "http://searxng:8080" return f"http://{TEST_HOST}:8080"
@pytest.fixture @pytest.fixture
def ollama_test_config() -> dict: def ollama_test_config() -> dict:
"""Test Ollama configuration.""" """Test Ollama configuration."""
from src.config import get_settings
settings = get_settings()
return { return {
"base_url": "http://ollama:11434", "base_url": f"http://{TEST_HOST}:11434",
"model": "nomic-embed-text" "model": settings.ollama_embedding_model
} }
@pytest.fixture @pytest.fixture
def redis_test_url() -> str: def redis_test_url() -> str:
"""Test Redis URL.""" """Test Redis URL."""
return "redis://redis-shared:6379/4" return f"redis://{TEST_HOST}:6379/4"
@pytest.fixture @pytest.fixture
+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"
+6 -6
View File
@@ -38,10 +38,10 @@ def settings():
@pytest_asyncio.fixture @pytest_asyncio.fixture
async def neo4j_client(settings) -> AsyncGenerator[Neo4jClient, None]: async def neo4j_client(settings, neo4j_test_uri) -> AsyncGenerator[Neo4jClient, None]:
"""Get connected Neo4j client.""" """Get connected Neo4j client."""
client = Neo4jClient( client = Neo4jClient(
uri=settings.neo4j_uri, uri=neo4j_test_uri,
user=settings.neo4j_user, user=settings.neo4j_user,
password=settings.neo4j_password password=settings.neo4j_password
) )
@@ -51,12 +51,12 @@ async def neo4j_client(settings) -> AsyncGenerator[Neo4jClient, None]:
@pytest_asyncio.fixture @pytest_asyncio.fixture
async def wiki_client(settings) -> AsyncGenerator[WikiJSClient, None]: async def wiki_client(wikijs_test_config) -> AsyncGenerator[WikiJSClient, None]:
"""Get Wiki.js client.""" """Get Wiki.js client."""
client = WikiJSClient( client = WikiJSClient(
base_url=settings.wikijs_url, base_url=wikijs_test_config["base_url"],
username=settings.wikijs_username, username=wikijs_test_config["username"],
password=settings.wikijs_password password=wikijs_test_config["password"]
) )
yield client yield client
+65 -49
View File
@@ -25,6 +25,7 @@ from src.clients.qdrant_client import QdrantClientWrapper
from src.clients.wikijs_client import WikiJSClient from src.clients.wikijs_client import WikiJSClient
from src.clients.searxng_client import SearXNGClient from src.clients.searxng_client import SearXNGClient
from src.clients.ollama_client import OllamaClient 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.hybrid_rag_service import HybridRAGService
from src.services.vector_service import VectorService from src.services.vector_service import VectorService
from src.services.graph_service import GraphService from src.services.graph_service import GraphService
@@ -42,10 +43,10 @@ def settings():
@pytest_asyncio.fixture @pytest_asyncio.fixture
async def neo4j_client(settings) -> AsyncGenerator[Neo4jClient, None]: async def neo4j_client(settings, neo4j_test_uri) -> AsyncGenerator[Neo4jClient, None]:
"""Get connected Neo4j client.""" """Get connected Neo4j client."""
client = Neo4jClient( client = Neo4jClient(
uri=settings.neo4j_uri, uri=neo4j_test_uri,
user=settings.neo4j_user, user=settings.neo4j_user,
password=settings.neo4j_password password=settings.neo4j_password
) )
@@ -55,32 +56,44 @@ async def neo4j_client(settings) -> AsyncGenerator[Neo4jClient, None]:
@pytest.fixture @pytest.fixture
def qdrant_client(settings) -> QdrantClientWrapper: def qdrant_client(qdrant_test_url) -> QdrantClientWrapper:
"""Get Qdrant client.""" """Get Qdrant client."""
return QdrantClientWrapper(url=settings.qdrant_url) return QdrantClientWrapper(url=qdrant_test_url)
@pytest_asyncio.fixture @pytest_asyncio.fixture
async def wiki_client(settings) -> AsyncGenerator[WikiJSClient, None]: async def wiki_client(wikijs_test_config) -> AsyncGenerator[WikiJSClient, None]:
"""Get Wiki.js client.""" """Get Wiki.js client."""
client = WikiJSClient( client = WikiJSClient(
base_url=settings.wikijs_url, base_url=wikijs_test_config["base_url"],
username=settings.wikijs_username, username=wikijs_test_config["username"],
password=settings.wikijs_password password=wikijs_test_config["password"]
) )
yield client yield client
@pytest.fixture @pytest.fixture
def searxng_client(settings) -> SearXNGClient: def searxng_client(searxng_test_url) -> SearXNGClient:
"""Get SearXNG client.""" """Get SearXNG client."""
return SearXNGClient(base_url=settings.searxng_url) return SearXNGClient(base_url=searxng_test_url)
@pytest.fixture @pytest.fixture
def ollama_client(settings) -> OllamaClient: def ollama_client(ollama_test_config) -> OllamaClient:
"""Get Ollama client.""" """Get Ollama client."""
return OllamaClient(base_url=settings.ollama_url) return OllamaClient(
base_url=ollama_test_config["base_url"],
model=ollama_test_config["model"]
)
@pytest.fixture
def content_extractor(settings) -> ContentExtractor:
"""Get ContentExtractor client."""
return ContentExtractor(
timeout=settings.content_extraction_timeout,
max_length=settings.content_max_length
)
@pytest_asyncio.fixture @pytest_asyncio.fixture
@@ -101,6 +114,7 @@ async def hybrid_rag_service(
graph_service, graph_service,
searxng_client, searxng_client,
ollama_client, ollama_client,
content_extractor,
settings settings
): ):
"""Get HybridRAGService instance.""" """Get HybridRAGService instance."""
@@ -109,6 +123,7 @@ async def hybrid_rag_service(
graph_service=graph_service, graph_service=graph_service,
searxng_client=searxng_client, searxng_client=searxng_client,
ollama_client=ollama_client, ollama_client=ollama_client,
content_extractor=content_extractor,
settings=settings settings=settings
) )
@@ -207,53 +222,54 @@ async def test_vector_data(vector_service, test_wiki_page):
# ============================================================================ # ============================================================================
class TestRRFFusion: class TestRRFFusion:
"""Test Reciprocal Rank Fusion algorithm.""" """Test two-stage Reciprocal Rank Fusion algorithm."""
def test_rrf_single_source(self, hybrid_rag_service): def test_wiki_merge_single_source(self, hybrid_rag_service):
"""Test RRF with single source.""" """Test wiki merge with single source (vector only)."""
results_by_source = { vector_results = [
"vector": [ {"page_id": 1, "title": "Doc 1", "content": "test"},
{"page_id": 1, "title": "Doc 1", "content": "test"}, {"page_id": 2, "title": "Doc 2", "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 len(merged) == 2
assert fused[0]["rrf_score"] > fused[1]["rrf_score"] # Rank 1 > Rank 2 assert merged[0]["wiki_rrf_score"] > merged[1]["wiki_rrf_score"] # Rank 1 > Rank 2
assert fused[0]["sources"] == ["vector"] assert merged[0]["found_by"] == ["vector"]
def test_rrf_multiple_sources_same_doc(self, hybrid_rag_service): def test_wiki_merge_multiple_sources_same_doc(self, hybrid_rag_service):
"""Test RRF with same document from multiple sources.""" """Test wiki merge with same document from vector and graph."""
results_by_source = { vector_results = [{"page_id": 1, "title": "Doc 1", "content": "test"}]
"vector": [{"page_id": 1, "title": "Doc 1", "content": "test"}], graph_results = [{"page_id": 1, "title": "Doc 1", "content": ""}]
"graph": [{"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(merged) == 1 # Deduplicated
assert len(fused[0]["sources"]) == 2 # Both sources assert len(merged[0]["found_by"]) == 2 # Both sources
assert "vector" in fused[0]["sources"] assert "vector" in merged[0]["found_by"]
assert "graph" in fused[0]["sources"] assert "graph" in merged[0]["found_by"]
# RRF score should be sum: 1/(60+1) + 1/(60+1) # Wiki RRF score should be sum: 1/(60+1) + 1/(60+1)
expected_score = 1/61 + 1/61 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): def test_final_rrf_wiki_and_web(self, hybrid_rag_service):
"""Test RRF with web results (URL-based).""" """Test final RRF between wiki and web results."""
results_by_source = { # Pre-merged wiki results
"web": [ wiki_results = [
{"url": "https://example.com/1", "title": "Web 1", "content": "test"}, {"page_id": 1, "title": "Wiki 1", "content": "test", "found_by": ["vector"]}
{"url": "https://example.com/2", "title": "Web 2", "content": "test"} ]
] 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 len(fused) == 3
assert fused[0]["result"]["url"] == "https://example.com/1" # 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: class TestContextFormatting:
+16 -15
View File
@@ -30,10 +30,10 @@ def settings():
@pytest_asyncio.fixture @pytest_asyncio.fixture
async def neo4j_client(settings) -> AsyncGenerator[Neo4jClient, None]: async def neo4j_client(settings, neo4j_test_uri) -> AsyncGenerator[Neo4jClient, None]:
"""Get connected Neo4j client.""" """Get connected Neo4j client."""
client = Neo4jClient( client = Neo4jClient(
uri=settings.neo4j_uri, uri=neo4j_test_uri,
user=settings.neo4j_user, user=settings.neo4j_user,
password=settings.neo4j_password password=settings.neo4j_password
) )
@@ -43,45 +43,46 @@ async def neo4j_client(settings) -> AsyncGenerator[Neo4jClient, None]:
@pytest.fixture @pytest.fixture
def qdrant_client(settings) -> QdrantClientWrapper: def qdrant_client(settings, qdrant_test_url) -> QdrantClientWrapper:
"""Get Qdrant client.""" """Get Qdrant client."""
return QdrantClientWrapper(url=settings.qdrant_url) return QdrantClientWrapper(url=qdrant_test_url)
@pytest_asyncio.fixture @pytest_asyncio.fixture
async def wikijs_client(settings) -> AsyncGenerator[WikiJSClient, None]: async def wikijs_client(wikijs_test_config) -> AsyncGenerator[WikiJSClient, None]:
"""Get Wiki.js client.""" """Get Wiki.js client."""
client = WikiJSClient( client = WikiJSClient(
base_url=settings.wikijs_url, base_url=wikijs_test_config["base_url"],
api_key=settings.wikijs_api_key username=wikijs_test_config["username"],
password=wikijs_test_config["password"]
) )
yield client yield client
await client.close() await client.close()
@pytest_asyncio.fixture @pytest_asyncio.fixture
async def searxng_client(settings) -> AsyncGenerator[SearXNGClient, None]: async def searxng_client(searxng_test_url) -> AsyncGenerator[SearXNGClient, None]:
"""Get SearXNG client.""" """Get SearXNG client."""
client = SearXNGClient(base_url=settings.searxng_url) client = SearXNGClient(base_url=searxng_test_url)
yield client yield client
await client.close() await client.close()
@pytest_asyncio.fixture @pytest_asyncio.fixture
async def ollama_client(settings) -> AsyncGenerator[OllamaClient, None]: async def ollama_client(ollama_test_config) -> AsyncGenerator[OllamaClient, None]:
"""Get Ollama client.""" """Get Ollama client for embeddings."""
client = OllamaClient( client = OllamaClient(
base_url=settings.ollama_url, base_url=ollama_test_config["base_url"],
model=settings.ollama_model model=ollama_test_config["model"]
) )
yield client yield client
await client.close() await client.close()
@pytest_asyncio.fixture @pytest_asyncio.fixture
async def job_manager(settings) -> AsyncGenerator[JobManager, None]: async def job_manager(redis_test_url) -> AsyncGenerator[JobManager, None]:
"""Get job manager.""" """Get job manager."""
manager = JobManager(redis_url=settings.redis_url) manager = JobManager(redis_url=redis_test_url)
await manager.connect() await manager.connect()
yield manager yield manager
await manager.close() await manager.close()
+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"