Spaces:
Running on Zero
Running on Zero
File size: 3,668 Bytes
712c393 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 | from __future__ import annotations
from typing import Any, Dict, List, Literal, Optional
from pydantic import BaseModel, Field, field_validator
class QueryRequest(BaseModel):
query: str = Field(..., min_length=1, max_length=2000, description="User question or research query.")
document_ids: Optional[List[str]] = Field(
default=None, max_length=50, description="Optional list of document IDs to scope search to."
)
mode: Optional[Literal["fast", "balanced", "deep", "adaptive"]] = Field(
default=None, description="Optional execution mode: adaptive, fast, balanced, or deep."
)
@field_validator("query")
@classmethod
def validate_non_empty_query(cls, v: str) -> str:
cleaned = v.strip()
if not cleaned:
raise ValueError("Query cannot be empty or whitespace.")
return cleaned
@field_validator("mode", mode="before")
@classmethod
def normalize_mode(cls, v: Optional[str]) -> Optional[str]:
if v is None or (isinstance(v, str) and not v.strip()):
return None
return v.strip().lower() if isinstance(v, str) else v
class SourceReference(BaseModel):
citation_id: int
document_id: str
filename: str
chunk_id: str
page: Optional[int] = None
section: Optional[str] = None
source_locator: Optional[str] = None
snippet: str = ""
class QueryResponse(BaseModel):
query: str
answer: str
confidence: float
abstained: bool = False
abstention_reason: Optional[str] = None
query_type: str = "general"
execution_mode: str = "balanced"
sources: List[SourceReference] = Field(default_factory=list)
retrieval: Dict[str, Any] = Field(default_factory=dict)
citation_validation: Dict[str, Any] = Field(default_factory=dict)
latency_ms: float = 0.0
class SearchRequest(BaseModel):
query: str = Field(..., min_length=1, max_length=2000)
top_k: int = Field(default=10, ge=1, le=50)
document_ids: Optional[List[str]] = Field(default=None, max_length=50)
strategy: Literal["hybrid", "dense", "keyword"] = "hybrid"
metadata_filter: Optional[Dict[str, Any]] = None
@field_validator("query")
@classmethod
def validate_non_empty_search_query(cls, v: str) -> str:
cleaned = v.strip()
if not cleaned:
raise ValueError("Query cannot be empty or whitespace.")
return cleaned
class SearchItem(BaseModel):
chunk_id: str
document_id: str
text: str
score: float
rank: int
source_locator: Optional[str] = None
page: Optional[int] = None
section: Optional[str] = None
retriever: str = ""
metadata: Dict[str, Any] = Field(default_factory=dict)
class SearchResponse(BaseModel):
query: str
total_results: int
results: List[SearchItem] = Field(default_factory=list)
latency_ms: float = 0.0
class DocumentItem(BaseModel):
document_id: str
filename: str
sha256: str
status: str
chunk_count: int = 0
size_bytes: int = 0
indexed_at: Optional[str] = None
metadata: Dict[str, Any] = Field(default_factory=dict)
class IngestRequest(BaseModel):
directory: Optional[str] = None
force_reindex: bool = False
class IngestResponse(BaseModel):
documents_indexed: int
documents_skipped: int
chunks_indexed: int
duration_ms: float
errors: List[Dict[str, str]] = Field(default_factory=list)
class IngestJobResponse(BaseModel):
job_id: str
status: str
message: str
class HealthResponse(BaseModel):
status: str
version: str = "1.1.0"
qdrant_healthy: bool
document_count: int
chunk_count: int
|