feat(library-desk): add core API routers for wiki, vector, and ingestion

Ingestion Router:
- POST /ingest/page - Index single page
- POST /ingest/batch - Batch indexing
- POST /ingest/all - Full knowledge base refresh
- Support vector and graph ingestion

Vector Router:
- POST /vector/search - Semantic search via Qdrant
- GET /vector/stats - Collection statistics
- DELETE /vector/page - Remove page embeddings

Wiki Router:
- GET /wiki/pages - List wiki pages
- GET /wiki/pages/{id} - Get page details
- PUT /wiki/pages/{id} - Update page
- POST /wiki/search - Search wiki content
- Full Wiki.js GraphQL integration
This commit is contained in:
2025-12-10 01:36:07 +01:00
parent 3ab44c2ab0
commit 9fe5faa3e7
3 changed files with 751 additions and 0 deletions
@@ -0,0 +1,169 @@
"""
Document Ingestion API Router
Endpoints for ingesting wiki pages into the knowledge base (vectors + graph).
"""
from fastapi import APIRouter, Depends, HTTPException, Query
from typing import Optional
from src.services.ingestion_service import IngestionService
from src.models.ingestion import (
IngestionRequest,
IngestionResult,
BatchIngestionRequest,
BatchIngestionResult
)
from src.core.dependencies import get_ingestion_service, verify_api_key
router = APIRouter(prefix="/ingest", tags=["Document Ingestion"])
@router.post("/page", response_model=IngestionResult)
async def ingest_page(
request: IngestionRequest,
ingestion: IngestionService = Depends(get_ingestion_service),
api_key: str = Depends(verify_api_key)
):
"""
Ingest a single wiki page into the knowledge base.
This endpoint:
1. Fetches page content from Wiki.js
2. Chunks content and generates embeddings (Qdrant)
3. Extracts entities and updates knowledge graph (Neo4j)
## Use Cases
- **After page creation**: Automatically called by consolidation service
- **Manual re-indexing**: Force refresh a page after manual edits
- **Selective ingestion**: Skip vectors or graph if only one is needed
## Performance
- Typical page: 1-3 seconds
- Large page (>5000 words): 5-10 seconds
- Vector and graph ingestion run in parallel
## Example
```bash
curl -X POST http://192.168.86.149:8089/ingest/page \
-H "Authorization: Bearer $API_KEY" \
-H "Content-Type: application/json" \
-d '{
"page_id": 19,
"user": "jpmschweitzer",
"force_refresh": false
}'
```
"""
result = await ingestion.ingest_page(
page_id=request.page_id,
user=request.user,
force_refresh=request.force_refresh,
skip_vectors=request.skip_vectors,
skip_graph=request.skip_graph
)
if not result.success:
raise HTTPException(
status_code=500,
detail=f"Ingestion failed: {result.error}"
)
return result
@router.post("/batch", response_model=BatchIngestionResult)
async def ingest_batch(
request: BatchIngestionRequest,
ingestion: IngestionService = Depends(get_ingestion_service),
api_key: str = Depends(verify_api_key)
):
"""
Ingest multiple wiki pages concurrently.
## Concurrency Control
The `max_concurrent` parameter controls how many pages are processed simultaneously:
- **Low (1-2)**: Safer for resource-constrained systems
- **Medium (3-5)**: Good balance of speed and stability
- **High (6-10)**: Maximum speed, requires good resources
## Batch Size Recommendations
- **Small batches (<10 pages)**: Use max_concurrent=3-5
- **Medium batches (10-50 pages)**: Use max_concurrent=3
- **Large batches (>50 pages)**: Use max_concurrent=2, consider splitting
## Example
```bash
curl -X POST http://192.168.86.149:8089/ingest/batch \
-H "Authorization: Bearer $API_KEY" \
-H "Content-Type: application/json" \
-d '{
"page_ids": [19, 20, 21, 22],
"user": "jpmschweitzer",
"max_concurrent": 3
}'
```
"""
result = await ingestion.ingest_batch(
page_ids=request.page_ids,
user=request.user,
force_refresh=request.force_refresh,
skip_vectors=request.skip_vectors,
skip_graph=request.skip_graph,
max_concurrent=request.max_concurrent
)
return result
@router.post("/all", response_model=BatchIngestionResult)
async def ingest_all_pages(
user: str = Query(default="jpmschweitzer", description="User identifier"),
path_prefix: Optional[str] = Query(None, description="Path prefix filter (e.g., 'users/jpmschweitzer/tech')"),
force_refresh: bool = Query(False, description="Force re-ingestion of all pages"),
max_concurrent: int = Query(3, ge=1, le=10, description="Maximum concurrent ingestion tasks"),
ingestion: IngestionService = Depends(get_ingestion_service),
api_key: str = Depends(verify_api_key)
):
"""
Ingest all wiki pages for a user (bulk re-indexing).
## Use Cases
- **Initial setup**: Index all existing pages
- **Full re-index**: After major schema changes
- **Path-specific**: Re-index a specific section (e.g., all tech docs)
## Performance
- **Small wiki (<50 pages)**: 2-5 minutes
- **Medium wiki (50-200 pages)**: 5-20 minutes
- **Large wiki (>200 pages)**: 20+ minutes
**Recommendation**: Run as background job for large wikis
## Example
```bash
# Ingest all pages for user
curl -X POST "http://192.168.86.149:8089/ingest/all?user=jpmschweitzer" \
-H "Authorization: Bearer $API_KEY"
# Ingest only tech docs
curl -X POST "http://192.168.86.149:8089/ingest/all?user=jpmschweitzer&path_prefix=users/jpmschweitzer/tech" \
-H "Authorization: Bearer $API_KEY"
```
"""
result = await ingestion.ingest_all_pages(
user=user,
path_prefix=path_prefix,
force_refresh=force_refresh,
max_concurrent=max_concurrent
)
return result
+173
View File
@@ -0,0 +1,173 @@
"""
Vector router for Library Desk API.
Endpoints for semantic search and vector operations.
"""
from fastapi import APIRouter, HTTPException, Depends, Query
from typing import Optional
import logging
from src.models.vector import (
SearchRequest, SearchResponse,
VectorUpdateRequest, VectorUpdateSummary,
CollectionListResponse,
DeletePageChunksRequest, DeletePageChunksResponse
)
from src.services.vector_service import VectorService
from src.clients.qdrant_client import QdrantClientWrapper
from src.clients.wikijs_client import WikiJSClient
from src.clients.ollama_client import OllamaClient
from src.core.dependencies import QdrantDep, WikiJSDep, OllamaDep, verify_api_key
from src.core.multi_tenancy import DEFAULT_USER
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/vector", tags=["Vector"])
# Dependency to get vector service
def get_vector_service(
qdrant_client: QdrantDep,
wiki_client: WikiJSDep,
ollama_client: OllamaDep
) -> VectorService:
"""Get vector service instance."""
return VectorService(qdrant_client, wiki_client, ollama_client)
@router.post("/search", response_model=SearchResponse)
async def semantic_search(
request: SearchRequest,
vector_service: VectorService = Depends(get_vector_service),
api_key: str = Depends(verify_api_key)
):
"""
Perform semantic search across user's documents.
Uses Ollama to generate query embedding, then searches Qdrant
for similar document chunks.
**Example Request:**
```json
{
"query": "how to configure docker",
"user": "jpmschweitzer",
"limit": 10,
"score_threshold": 0.5
}
```
**Returns:** List of matching chunks with similarity scores
"""
try:
return await vector_service.search(
query=request.query,
user=request.user,
limit=request.limit,
score_threshold=request.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")
@router.post("/update-from-page/{page_id}", response_model=VectorUpdateSummary)
async def update_vectors_from_page(
page_id: int,
user: str = Query(default=DEFAULT_USER, description="User identifier"),
force_refresh: bool = Query(default=False, description="Force re-embedding"),
vector_service: VectorService = Depends(get_vector_service),
api_key: str = Depends(verify_api_key)
):
"""
Update vector embeddings from a wiki page.
This endpoint:
1. Fetches the page from Wiki.js
2. Chunks the content (500 tokens with 50 token overlap)
3. Generates embeddings via Ollama
4. Upserts chunks to Qdrant with metadata
**Use Cases:**
- Called automatically after page creation/update (via BackgroundTasks)
- Called manually by user/Librarian to refresh vectors
- Called by Scheduler for batch processing
**Example:** `POST /vector/update-from-page/5?user=jpmschweitzer`
**Returns:** Summary with chunks created and processing time
"""
try:
summary = await vector_service.update_from_page(
page_id=page_id,
user=user,
force_refresh=force_refresh
)
if not summary.success:
raise HTTPException(
status_code=500,
detail=f"Vector update failed: {summary.error_message}"
)
return summary
except HTTPException:
raise
except Exception as e:
logger.error(f"Failed to update vectors from page {page_id}: {e}", exc_info=True)
raise HTTPException(status_code=500, detail="Vector update failed")
@router.delete("/pages/{page_id}", response_model=DeletePageChunksResponse)
async def delete_page_chunks(
page_id: int,
user: str = Query(default=DEFAULT_USER, description="User identifier"),
vector_service: VectorService = Depends(get_vector_service),
api_key: str = Depends(verify_api_key)
):
"""
Delete all vector chunks for a wiki page.
This is automatically called when a page is deleted from the wiki.
**Example:** `DELETE /vector/pages/5?user=jpmschweitzer`
"""
try:
deleted_count = await vector_service.delete_page_chunks(
page_id=page_id,
user=user
)
return DeletePageChunksResponse(
page_id=page_id,
chunks_deleted=deleted_count,
success=deleted_count > 0,
message=f"Deleted {deleted_count} chunks for page {page_id}"
)
except Exception as e:
logger.error(f"Failed to delete chunks for page {page_id}: {e}", exc_info=True)
raise HTTPException(status_code=500, detail="Failed to delete chunks")
@router.get("/collections", response_model=CollectionListResponse)
async def list_collections(
vector_service: VectorService = Depends(get_vector_service),
api_key: str = Depends(verify_api_key)
):
"""
List all Qdrant collections with statistics.
Returns collection names, vector counts, and point counts.
**Example:** `GET /vector/collections`
"""
try:
return await vector_service.list_collections()
except Exception as e:
logger.error(f"Failed to list collections: {e}", exc_info=True)
raise HTTPException(status_code=500, detail="Failed to list collections")
+409
View File
@@ -0,0 +1,409 @@
"""
Wiki router for Library Desk API.
Endpoints for wiki page and dossier management.
All operations are scoped to user namespaces for multi-tenancy.
"""
from fastapi import APIRouter, HTTPException, Depends, Query, Security, BackgroundTasks
from fastapi.security import HTTPAuthorizationCredentials
from typing import Optional
import logging
from src.models.wiki import (
WikiPage, WikiPageList, WikiPageCreate, WikiPageUpdate, WikiPageMove,
WikiOperationResponse, WikiSearchResponse,
DossierList, WikiSearchResult
)
from src.services.wiki_service import WikiService
from src.services.graph_service import GraphService
from src.services.vector_service import VectorService
from src.clients.wikijs_client import WikiJSClient
from src.clients.neo4j_client import Neo4jClient
from src.clients.qdrant_client import QdrantClientWrapper
from src.clients.ollama_client import OllamaClient
from src.core.dependencies import WikiJSDep, Neo4jDep, QdrantDep, OllamaDep, verify_api_key
from src.core.multi_tenancy import DEFAULT_USER
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/wiki", tags=["Wiki"])
# Dependency to get wiki service
def get_wiki_service(wiki_client: WikiJSDep) -> WikiService:
"""Get wiki service instance."""
return WikiService(wiki_client)
# Dependency to get graph service
def get_graph_service(neo4j_client: Neo4jDep, wiki_client: WikiJSDep) -> GraphService:
"""Get graph service instance."""
return GraphService(neo4j_client, wiki_client)
# Dependency to get vector service
def get_vector_service(
qdrant_client: QdrantDep,
wiki_client: WikiJSDep,
ollama_client: OllamaDep
) -> VectorService:
"""Get vector service instance."""
return VectorService(qdrant_client, wiki_client, ollama_client)
# Page operations
@router.get("/pages", response_model=WikiPageList)
async def list_pages(
user: str = Query(default=DEFAULT_USER, description="User identifier"),
tag: Optional[str] = Query(default=None, description="Filter by tag (dossier)"),
limit: int = Query(default=50, ge=1, le=200, description="Maximum pages to return"),
wiki_service: WikiService = Depends(get_wiki_service),
api_key: str = Depends(verify_api_key)
):
"""
List wiki pages for a user.
Optionally filter by tag (dossier). Pages are scoped to user's namespace.
**Example:** `/wiki/pages?user=jpmschweitzer&tag=projects&limit=20`
"""
try:
return await wiki_service.list_pages(user=user, tag=tag, limit=limit)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
logger.error(f"Failed to list pages: {e}", exc_info=True)
raise HTTPException(status_code=500, detail="Internal server error")
@router.get("/pages/{page_id}", response_model=WikiPage)
async def get_page(
page_id: int,
user: str = Query(default=DEFAULT_USER, description="User identifier"),
wiki_service: WikiService = Depends(get_wiki_service),
api_key: str = Depends(verify_api_key)
):
"""
Get a single wiki page by ID.
Access is restricted to pages within the user's namespace.
**Example:** `/wiki/pages/123?user=jpmschweitzer`
"""
try:
page = await wiki_service.get_page(page_id=page_id, user=user)
if not page:
raise HTTPException(status_code=404, detail=f"Page {page_id} not found or access denied")
return page
except HTTPException:
raise
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
logger.error(f"Failed to get page {page_id}: {e}", exc_info=True)
raise HTTPException(status_code=500, detail="Internal server error")
@router.post("/pages", response_model=WikiPage, status_code=201)
async def create_page(
page_data: WikiPageCreate,
background_tasks: BackgroundTasks,
wiki_service: WikiService = Depends(get_wiki_service),
graph_service: GraphService = Depends(get_graph_service),
vector_service: VectorService = Depends(get_vector_service),
api_key: str = Depends(verify_api_key)
):
"""
Create a new wiki page.
The page will be created in the user's namespace. If path doesn't start
with namespace, it will be automatically prefixed.
**Auto-updates knowledge graph and vector embeddings**: After creating
the page, the graph and vectors are automatically updated in the
background to extract entities/relationships and generate semantic embeddings.
**Example Request:**
```json
{
"title": "Library Desk Architecture",
"path": "/projects/library-desk/architecture",
"content": "# Architecture\\n\\nThis describes...",
"description": "Architecture documentation",
"tags": ["projects", "architecture"],
"user": "jpmschweitzer"
}
```
"""
try:
page = await wiki_service.create_page(page_data)
user = page_data.user or DEFAULT_USER
# Schedule BOTH graph and vector updates in background (non-blocking)
background_tasks.add_task(
graph_service.update_from_page,
page_id=page.id,
user=user
)
background_tasks.add_task(
vector_service.update_from_page,
page_id=page.id,
user=user
)
logger.info(f"Page {page.id} created, graph and vector updates scheduled")
return page
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
logger.error(f"Failed to create page: {e}", exc_info=True)
raise HTTPException(status_code=500, detail="Internal server error")
@router.put("/pages/{page_id}", response_model=WikiPage)
async def update_page(
page_id: int,
page_data: WikiPageUpdate,
background_tasks: BackgroundTasks,
user: str = Query(default=DEFAULT_USER, description="User identifier"),
wiki_service: WikiService = Depends(get_wiki_service),
graph_service: GraphService = Depends(get_graph_service),
vector_service: VectorService = Depends(get_vector_service),
api_key: str = Depends(verify_api_key)
):
"""
Update an existing wiki page.
Only pages within the user's namespace can be updated.
Partial updates are supported - only provided fields will be updated.
**Auto-updates knowledge graph and vector embeddings**: After updating
the page, the graph and vectors are automatically refreshed in the
background to reflect the changes.
**Example Request:**
```json
{
"title": "Updated Title",
"tags": ["projects", "updated"]
}
```
"""
try:
page = await wiki_service.update_page(
page_id=page_id,
page_data=page_data,
user=user
)
# Schedule BOTH graph and vector updates in background (non-blocking)
background_tasks.add_task(
graph_service.update_from_page,
page_id=page_id,
user=user,
force_refresh=True # Force refresh on updates
)
background_tasks.add_task(
vector_service.update_from_page,
page_id=page_id,
user=user,
force_refresh=True # Force refresh on updates
)
logger.info(f"Page {page_id} updated, graph and vector refresh scheduled")
return page
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
logger.error(f"Failed to update page {page_id}: {e}", exc_info=True)
raise HTTPException(status_code=500, detail="Internal server error")
@router.delete("/pages/{page_id}", response_model=WikiOperationResponse)
async def delete_page(
page_id: int,
background_tasks: BackgroundTasks,
user: str = Query(default=DEFAULT_USER, description="User identifier"),
wiki_service: WikiService = Depends(get_wiki_service),
vector_service: VectorService = Depends(get_vector_service),
graph_service: GraphService = Depends(get_graph_service),
api_key: str = Depends(verify_api_key)
):
"""
Delete a wiki page.
Only pages within the user's namespace can be deleted.
This operation cannot be undone.
**Auto-cleanup**: Vector chunks and graph nodes for this page are
automatically deleted in the background.
"""
try:
success = await wiki_service.delete_page(page_id=page_id, user=user)
# Schedule vector cleanup in background
background_tasks.add_task(
vector_service.delete_page_chunks,
page_id=page_id,
user=user
)
# Schedule graph cleanup in background
background_tasks.add_task(
graph_service.delete_page,
page_id=page_id,
user=user
)
logger.info(f"Page {page_id} deleted, vector and graph cleanup scheduled")
return WikiOperationResponse(
success=success,
message=f"Page {page_id} deleted successfully",
page_id=page_id
)
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
except Exception as e:
logger.error(f"Failed to delete page {page_id}: {e}", exc_info=True)
raise HTTPException(status_code=500, detail="Internal server error")
@router.post("/pages/{page_id}/move", response_model=WikiOperationResponse)
async def move_page(
page_id: int,
move_data: WikiPageMove,
user: str = Query(default=DEFAULT_USER, description="User identifier"),
wiki_service: WikiService = Depends(get_wiki_service),
api_key: str = Depends(verify_api_key)
):
"""
Move or rename a wiki page.
The new path must be within the user's namespace.
**Example Request:**
```json
{
"new_path": "/projects/library-desk/docs/architecture"
}
```
"""
try:
success = await wiki_service.move_page(
page_id=page_id,
new_path=move_data.new_path,
user=user
)
if not success:
raise HTTPException(status_code=500, detail="Failed to move page")
return WikiOperationResponse(
success=True,
message=f"Page {page_id} moved successfully",
page_id=page_id,
page_path=move_data.new_path
)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
except HTTPException:
raise
except Exception as e:
logger.error(f"Failed to move page {page_id}: {e}", exc_info=True)
raise HTTPException(status_code=500, detail="Internal server error")
# Search operations
@router.get("/search", response_model=WikiSearchResponse)
async def search_pages(
q: str = Query(..., min_length=1, description="Search query"),
user: str = Query(default=DEFAULT_USER, description="User identifier"),
limit: int = Query(default=20, ge=1, le=100, description="Maximum results"),
wiki_service: WikiService = Depends(get_wiki_service),
api_key: str = Depends(verify_api_key)
):
"""
Search wiki pages within user's namespace.
**Example:** `/wiki/search?q=architecture&user=jpmschweitzer&limit=10`
"""
try:
results = await wiki_service.search_pages(
query=q,
user=user,
limit=limit
)
return WikiSearchResponse(
results=[
WikiSearchResult(
id=r.id,
path=r.path,
title=r.title,
description=r.description,
relevance=None
)
for r in results
],
query=q,
total=len(results)
)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
logger.error(f"Search failed: {e}", exc_info=True)
raise HTTPException(status_code=500, detail="Internal server error")
# Dossier operations
@router.get("/dossiers", response_model=DossierList)
async def list_dossiers(
user: str = Query(default=DEFAULT_USER, description="User identifier"),
wiki_service: WikiService = Depends(get_wiki_service),
api_key: str = Depends(verify_api_key)
):
"""
List all dossiers (unique tags) for a user.
Dossiers are tag-based collections of pages.
**Example:** `/wiki/dossiers?user=jpmschweitzer`
"""
try:
return await wiki_service.list_dossiers(user=user)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
logger.error(f"Failed to list dossiers: {e}", exc_info=True)
raise HTTPException(status_code=500, detail="Internal server error")
@router.get("/dossiers/{dossier_name}/pages", response_model=WikiPageList)
async def get_dossier_pages(
dossier_name: str,
user: str = Query(default=DEFAULT_USER, description="User identifier"),
limit: int = Query(default=100, ge=1, le=500, description="Maximum pages"),
wiki_service: WikiService = Depends(get_wiki_service),
api_key: str = Depends(verify_api_key)
):
"""
Get all pages in a dossier.
Returns pages tagged with the dossier name.
**Example:** `/wiki/dossiers/projects/pages?user=jpmschweitzer`
"""
try:
return await wiki_service.get_dossier_pages(
dossier_name=dossier_name,
user=user,
limit=limit
)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
logger.error(f"Failed to get dossier pages: {e}", exc_info=True)
raise HTTPException(status_code=500, detail="Internal server error")