"""Abstractions for search functionality."""
import uuid
from typing import Any, Dict, List, Optional, Tuple
from pydantic import BaseModel, Field
from .llm import GenerationConfig
class VectorSearchRequest(BaseModel):
"""Request for a search operation."""
query: str
limit: int
filters: Optional[dict[str, Any]] = None
class VectorSearchResult(BaseModel):
"""Result of a search operation."""
id: uuid.UUID
score: float
metadata: dict[str, Any]
def __str__(self) -> str:
return f"VectorSearchResult(id={self.id}, score={self.score}, metadata={self.metadata})"
def __repr__(self) -> str:
return f"VectorSearchResult(id={self.id}, score={self.score}, metadata={self.metadata})"
def dict(self) -> dict:
return {
"id": self.id,
"score": self.score,
"metadata": self.metadata,
}
class KGSearchRequest(BaseModel):
"""Request for a knowledge graph search operation."""
query: str
# [query, ...]
KGSearchResult = List[Tuple[str, List[Dict[str, Any]]]]
class AggregateSearchResult(BaseModel):
"""Result of an aggregate search operation."""
vector_search_results: Optional[List[VectorSearchResult]]
kg_search_results: Optional[KGSearchResult] = None
def __str__(self) -> str:
return f"AggregateSearchResult(vector_search_results={self.vector_search_results}, kg_search_results={self.kg_search_results})"
def __repr__(self) -> str:
return f"AggregateSearchResult(vector_search_results={self.vector_search_results}, kg_search_results={self.kg_search_results})"
def dict(self) -> dict:
return {
"vector_search_results": (
[result.dict() for result in self.vector_search_results]
if self.vector_search_results
else []
),
"kg_search_results": self.kg_search_results or [],
}
class VectorSearchSettings(BaseModel):
use_vector_search: bool = True
search_filters: dict[str, Any] = Field(default_factory=dict)
search_limit: int = 10
do_hybrid_search: bool = False
class KGSearchSettings(BaseModel):
use_kg_search: bool = False
agent_generation_config: Optional[GenerationConfig] = Field(
default_factory=GenerationConfig
)