diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index a7a1de1..2d3b29d 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -212,10 +212,12 @@ All FastAPI route handlers use Pydantic models for request/response validation: from pydantic import BaseModel from fastapi import FastAPI + class UserRequest(BaseModel): name: str age: int + @app.post("/users") async def create_user(user: UserRequest): # Pydantic validates name is string, age is int diff --git a/DSL/CronManager/script/store_secrets_in_vault.sh b/DSL/CronManager/script/store_secrets_in_vault.sh index d977f1e..b47b8a0 100644 --- a/DSL/CronManager/script/store_secrets_in_vault.sh +++ b/DSL/CronManager/script/store_secrets_in_vault.sh @@ -165,7 +165,14 @@ setup_python_environment() { echo "[$(date '+%Y-%m-%d %H:%M:%S')] ERROR: Failed to install loguru" >&2 return 1 } - + + # Install requests, required by LokiLogger which decrypt_vault_secrets.py imports + echo "[$(date '+%Y-%m-%d %H:%M:%S')] Installing requests library..." + "$UV_BIN" pip install --python "$VENV_PATH/bin/python3" "requests>=2.32" 2>&1 || { + echo "[$(date '+%Y-%m-%d %H:%M:%S')] ERROR: Failed to install requests" >&2 + return 1 + } + # Mark setup as complete touch "$VENV_PATH/.setup_complete" @@ -196,6 +203,9 @@ if ! setup_python_environment; then exit 1 fi +# Set Python path +export PYTHONPATH="/app:/app/src:/app/src/vector_indexer:$PYTHONPATH" + # Function to determine platform name get_platform_name() { case "$llmPlatform" in diff --git a/docs/API_TOOL_CALLING.md b/docs/API_TOOL_CALLING.md index ce01a35..e7467cf 100644 --- a/docs/API_TOOL_CALLING.md +++ b/docs/API_TOOL_CALLING.md @@ -209,21 +209,21 @@ A `PointStruct` is built: ```python PointStruct( - id = endpoint_id, # UUID used directly as Qdrant point ID - vector = { - "dense": [v1, v2, ..., v3072], + id=endpoint_id, # UUID used directly as Qdrant point ID + vector={ + "dense": [v1, v2, ..., v3072], "sparse": {"indices": [...], "values": [...]}, }, - payload = { # Stored metadata — no extra DB lookup needed - "endpoint_id": "...", - "name": "get_national_holidays", - "description": "...", - "url": "https://openholidaysapi.org/PublicHolidays", - "method": "GET", - "params": [...], + payload={ # Stored metadata — no extra DB lookup needed + "endpoint_id": "...", + "name": "get_national_holidays", + "description": "...", + "url": "https://openholidaysapi.org/PublicHolidays", + "method": "GET", + "params": [...], "enriched_context": "...", - "service_id": "...", - } + "service_id": "...", + }, ) ``` @@ -306,9 +306,9 @@ Instantiated once in `ToolClassifier.__init__()` and reuses the shared Qdrant `h ```python APISemanticSearcher( - embedding_service=orchestration_service, # generates dense embeddings - qdrant_client=self._qdrant_client, # shared connection pool - disambiguator=None, # optional: inject for testing + embedding_service=orchestration_service, # generates dense embeddings + qdrant_client=self._qdrant_client, # shared connection pool + disambiguator=None, # optional: inject for testing ) ``` @@ -861,7 +861,7 @@ Results are **deduplicated by endpoint name** (a single endpoint matched by two ```python class ExecutionMode(str, Enum): - SINGLE = "single" + SINGLE = "single" PARALLEL = "parallel" ``` @@ -947,7 +947,7 @@ await multi_loop.stream_run_turn( chat_id=chat_id, user_message=request.message, conversation_history=conversation_history, - endpoint_states=session.parallel_endpoints, # list[EndpointSessionState] + endpoint_states=session.parallel_endpoints, # list[EndpointSessionState] turn_count=session.turn_count, awaiting_continuation=session.awaiting_continuation, session_language=effective_session_language, @@ -979,7 +979,9 @@ call_payloads = [ {**state.endpoint, "call_params": state.collected_params} for state in parallel_endpoints ] -multi_result = await MultiAPICaller(api_caller).call_all(call_payloads, language=detected_language) +multi_result = await MultiAPICaller(api_caller).call_all( + call_payloads, language=detected_language +) ``` --- @@ -1402,7 +1404,7 @@ endpoint A sends a message that strongly matches endpoint B), both the session a L2 key are cleaned up: ```python -await session_store.delete(request.chatId) # existing behaviour +await session_store.delete(request.chatId) # existing behaviour if FeatureFlags.ATC_RESPONSE_CACHE_ENABLED: await ATCCacheStore().invalidate_l2(request.chatId) ``` diff --git a/docs/CONTEXT_WORKFLOW_GREETING_DETECTION.md b/docs/CONTEXT_WORKFLOW_GREETING_DETECTION.md index dd6990d..66d8860 100644 --- a/docs/CONTEXT_WORKFLOW_GREETING_DETECTION.md +++ b/docs/CONTEXT_WORKFLOW_GREETING_DETECTION.md @@ -223,17 +223,17 @@ In **Phase 2**, `ContextWorkflowExecutor` calls `get_greeting_response(greeting_ ```python GREETINGS_ET = { - "hello": "Tere! Kuidas ma saan sind aidata?", + "hello": "Tere! Kuidas ma saan sind aidata?", "goodbye": "Nägemist! Head päeva!", - "thanks": "Palun! Kui on veel küsimusi, küsi julgelt.", - "casual": "Tere! Mida ma saan sinu jaoks teha?", + "thanks": "Palun! Kui on veel küsimusi, küsi julgelt.", + "casual": "Tere! Mida ma saan sinu jaoks teha?", } GREETINGS_EN = { - "hello": "Hello! How can I help you?", + "hello": "Hello! How can I help you?", "goodbye": "Goodbye! Have a great day!", - "thanks": "You're welcome! Feel free to ask if you have more questions.", - "casual": "Hey! What can I do for you?", + "thanks": "You're welcome! Feel free to ask if you have more questions.", + "casual": "Hey! What can I do for you?", } ``` @@ -358,29 +358,31 @@ Key log entries emitted during a request: ```python class ConversationRound(BaseModel): - user_message: str # The user's message text - bot_message: str # The bot's response text - timestamp: float # Unix timestamp of the round + user_message: str # The user's message text + bot_message: str # The bot's response text + timestamp: float # Unix timestamp of the round ``` ### `ConversationHistoryState` (Redis fetch result) ```python class ConversationHistoryState(BaseModel): - chat_id: str # Unique conversation identifier - rounds: list[ConversationRound] # Ordered rounds (newest last), capped at 10 - summary: Optional[str] # Incremental summary of evicted older rounds + chat_id: str # Unique conversation identifier + rounds: list[ConversationRound] # Ordered rounds (newest last), capped at 10 + summary: Optional[str] # Incremental summary of evicted older rounds ``` ### `ContextDetectionResult` (Phase 1 output) ```python class ContextDetectionResult(BaseModel): - is_greeting: bool # True if query is a greeting - greeting_type: str # "hello" | "goodbye" | "thanks" | "casual" - can_answer_from_context: bool # True if query can be answered from history or summary - reasoning: str # LLM's brief explanation - answered_from_summary: bool # True when answer derived from summary path + is_greeting: bool # True if query is a greeting + greeting_type: str # "hello" | "goodbye" | "thanks" | "casual" + can_answer_from_context: ( + bool # True if query can be answered from history or summary + ) + reasoning: str # LLM's brief explanation + answered_from_summary: bool # True when answer derived from summary path context_snippet: Optional[str] # Relevant excerpt for Phase 2 generation, or None ``` diff --git a/docs/HYBRID_SEARCH_CLASSIFICATION.md b/docs/HYBRID_SEARCH_CLASSIFICATION.md index 1de3f7f..a4521e2 100644 --- a/docs/HYBRID_SEARCH_CLASSIFICATION.md +++ b/docs/HYBRID_SEARCH_CLASSIFICATION.md @@ -115,9 +115,7 @@ tokens = re.findall(r"\w+", text.lower()) # ["mis", "suhe", "on", "euro", ...] ```python # Collection: "intent_collections" -vectors_config = { - "dense": VectorParams(size=3072, distance=Distance.COSINE) -} +vectors_config = {"dense": VectorParams(size=3072, distance=Distance.COSINE)} sparse_vectors_config = { "sparse": SparseVectorParams(index=SparseIndexParams(on_disk=False)) } @@ -197,12 +195,12 @@ Queries Qdrant using only the dense vector to get **actual cosine similarity sco ```python # classifier.py → _dense_search() -POST /collections/intent_collections/points/query +POST / collections / intent_collections / points / query { "query": [0.023, -0.041, ...], # 3072-dim dense vector "using": "dense", - "limit": 6, # DENSE_SEARCH_TOP_K * 2 (3 * 2 = 6, allows dedup) - "with_payload": true + "limit": 6, # DENSE_SEARCH_TOP_K * 2 (3 * 2 = 6, allows dedup) + "with_payload": true, } ``` @@ -219,15 +217,15 @@ Sparse prefetch is only included if the query produces a non-empty sparse vector ```python # classifier.py → _hybrid_search() # First checks collection exists and has data (points_count > 0) -POST /collections/intent_collections/points/query +POST / collections / intent_collections / points / query { "prefetch": [ {"query": dense_vector, "using": "dense", "limit": 10}, - {"query": {"indices": [...], "values": [...]}, "using": "sparse", "limit": 10} + {"query": {"indices": [...], "values": [...]}, "using": "sparse", "limit": 10}, ], "query": {"fusion": "rrf"}, "limit": 5, - "with_payload": true + "with_payload": true, } ``` diff --git a/docs/REDIS_SESSION_STORE.md b/docs/REDIS_SESSION_STORE.md index 502f24f..0008242 100644 --- a/docs/REDIS_SESSION_STORE.md +++ b/docs/REDIS_SESSION_STORE.md @@ -65,9 +65,9 @@ if session is None: # No active session → this is a fresh conversation ... else: - print(session.state) # "collecting_params" - print(session.collected_params) # {"city": "Tallinn"} - print(session.turn_count) # 2 + print(session.state) # "collecting_params" + print(session.collected_params) # {"city": "Tallinn"} + print(session.turn_count) # 2 ``` --- @@ -129,12 +129,14 @@ session_store = request.app.state.session_store session = await session_store.get(request.chatId) if session is None: detected_endpoint = ... # endpoint detected from user query - await session_store.save(APIToolSession( - chat_id=request.chatId, - state="collecting_params", - selected_endpoint=detected_endpoint, - turn_count=1, - )) + await session_store.save( + APIToolSession( + chat_id=request.chatId, + state="collecting_params", + selected_endpoint=detected_endpoint, + turn_count=1, + ) + ) return "Which city would you like weather for?" # --- Turn 2+ --- diff --git a/docs/TESTPRODUCTIONLLM_SERVICE_WORKFLOW.md b/docs/TESTPRODUCTIONLLM_SERVICE_WORKFLOW.md index 3e4ccfc..05c0211 100644 --- a/docs/TESTPRODUCTIONLLM_SERVICE_WORKFLOW.md +++ b/docs/TESTPRODUCTIONLLM_SERVICE_WORKFLOW.md @@ -212,6 +212,7 @@ orchestration_service = self.orchestration_service service_content = service_result["content"] service_buttons = service_result["buttons"] + async def service_stream() -> AsyncIterator[str]: yield orchestration_service.format_sse( chat_id, service_content, service_buttons or None @@ -219,6 +220,7 @@ async def service_stream() -> AsyncIterator[str]: yield orchestration_service.format_sse(chat_id, "END") orchestration_service.log_costs(costs_metric) + return service_stream() ``` @@ -229,12 +231,13 @@ return service_stream() **`llm_orchestration_service.py` → `format_sse()`** (line 1195): ```python -def format_sse(self, chat_id: str, content: str, - buttons: Optional[List[Dict[str, Any]]] = None) -> str: +def format_sse( + self, chat_id: str, content: str, buttons: Optional[List[Dict[str, Any]]] = None +) -> str: inner_payload: Dict[str, Any] = {"content": content} if buttons: inner_payload["buttons"] = buttons - + payload = { "chatId": chat_id, "payload": inner_payload, diff --git a/docs/TOOL_CLASSIFIER_AND_SERVICE_WORKFLOW.md b/docs/TOOL_CLASSIFIER_AND_SERVICE_WORKFLOW.md index ac92abb..46a3686 100644 --- a/docs/TOOL_CLASSIFIER_AND_SERVICE_WORKFLOW.md +++ b/docs/TOOL_CLASSIFIER_AND_SERVICE_WORKFLOW.md @@ -24,7 +24,9 @@ Layer 4: OOD → Out-of-domain fallback (polite rejection) ```python # Non-streaming mode classification = await classifier.classify(query, history, language) -response = await classifier.route_to_workflow(classification, request, is_streaming=False) +response = await classifier.route_to_workflow( + classification, request, is_streaming=False +) # Streaming mode classification = await classifier.classify(query, history, language) @@ -153,14 +155,13 @@ embedding = orchestration_service.create_embeddings_for_indexer([query]) # 2. Search Qdrant collection search_payload = { "vector": query_embedding, - "limit": 10, # Top 10 services (SEMANTIC_SEARCH_TOP_K) - "score_threshold": 0.2, # Minimum similarity (SEMANTIC_SEARCH_THRESHOLD) - "with_payload": True + "limit": 10, # Top 10 services (SEMANTIC_SEARCH_TOP_K) + "score_threshold": 0.2, # Minimum similarity (SEMANTIC_SEARCH_THRESHOLD) + "with_payload": True, } response = qdrant_client.post( - f"/collections/{QDRANT_COLLECTION}/points/search", - json=search_payload + f"/collections/{QDRANT_COLLECTION}/points/search", json=search_payload ) ``` @@ -182,12 +183,12 @@ Uses **DSPy + LLM** to intelligently match user query to a specific service and ```python class ServiceIntentDetector(dspy.Signature): # Inputs - user_query: str # "How much is 100 EUR in USD?" - available_services: str # JSON of service definitions - conversation_context: str # Recent 3 conversation turns - + user_query: str # "How much is 100 EUR in USD?" + available_services: str # JSON of service definitions + conversation_context: str # Recent 3 conversation turns + # Output - intent_result: str # JSON: {matched_service_id, confidence, entities, reasoning} + intent_result: str # JSON: {matched_service_id, confidence, entities, reasoning} ``` ### LLM Call Flow @@ -200,7 +201,7 @@ services_formatted = [ "name": "Currency Conversion", "description": "Convert EUR to other currencies", "required_entities": ["target_currency"], - "examples": ["How much is EUR in USD?", "Convert EUR to JPY"] # Top 3 examples + "examples": ["How much is EUR in USD?", "Convert EUR to JPY"], # Top 3 examples } ] @@ -216,7 +217,7 @@ with self.llm_manager.use_task_local(): intent_result = intent_module.forward( user_query="How much is 100 EUR in USD?", services=services_formatted, - conversation_history=conversation_history + conversation_history=conversation_history, ) ``` @@ -360,12 +361,12 @@ validation_errors = ["Entity 'target_currency' has empty value"] ```python { - "is_valid": True, # Always true (lenient validation) - "missing_entities": ["amount"], # Will send empty strings - "extra_entities": ["random_field"], # Will be ignored - "validation_errors": [ # Warnings only - "Entity 'amount' has empty value" - ] + "is_valid": True, # Always true (lenient validation) + "missing_entities": ["amount"], # Will send empty strings + "extra_entities": ["random_field"], # Will be ignored + "validation_errors": [ # Warnings only + "Entity 'amount' has empty value" + ], } ``` @@ -393,11 +394,7 @@ Ruuter services expect parameters in specific order: entities_schema = ["target_currency", "source_currency", "amount"] # LLM extraction (unordered dict) -entities_dict = { - "amount": "100", - "target_currency": "USD", - "source_currency": "EUR" -} +entities_dict = {"amount": "100", "target_currency": "USD", "source_currency": "EUR"} # Transform to ordered array entities_array = ["USD", "EUR", "100"] @@ -409,9 +406,7 @@ entities_array = ["USD", "EUR", "100"] ```python def _transform_entities_to_array( - self, - entities_dict: Dict[str, str], - entity_order: List[str] + self, entities_dict: Dict[str, str], entity_order: List[str] ) -> List[str]: """Transform entity dict to ordered array.""" if not entity_order: @@ -461,7 +456,7 @@ def _construct_service_endpoint(self, service_name: str, chat_id: str) -> str: payload = { "chatId": chat_id, "authorId": author_id, - "input": entities_array, # ["USD", "EUR", "100"] + "input": entities_array, # ["USD", "EUR", "100"] } ``` @@ -540,10 +535,10 @@ entities_dict = {"target_currency": "THB"} #### 4. Entity Validation ```python validation_result = { - "is_valid": True, - "missing_entities": [], - "extra_entities": [], - "validation_errors": [] + "is_valid": True, + "missing_entities": [], + "extra_entities": [], + "validation_errors": [], } ``` @@ -563,7 +558,7 @@ response = await _call_service_endpoint( http_method="POST", entities_array=["THB"], chat_id="...", - author_id="..." + author_id="...", ) # Returns content string from Ruuter response ``` @@ -632,25 +627,25 @@ RUUTER_SERVICE_BASE_URL = "http://ruuter-public:8086/services" RAG_SEARCH_RUUTER_PUBLIC = "http://ruuter-public:8086/rag-search" # Service call timeouts -SERVICE_CALL_TIMEOUT = 10 # seconds for external service calls -SERVICE_DISCOVERY_TIMEOUT = 10.0 # seconds for service discovery +SERVICE_CALL_TIMEOUT = 10 # seconds for external service calls +SERVICE_DISCOVERY_TIMEOUT = 10.0 # seconds for service discovery # Service selection thresholds -SERVICE_COUNT_THRESHOLD = 10 # Switch to semantic search if exceeded -MAX_SERVICES_FOR_LLM_CONTEXT = 50 # Max services to pass to LLM +SERVICE_COUNT_THRESHOLD = 10 # Switch to semantic search if exceeded +MAX_SERVICES_FOR_LLM_CONTEXT = 50 # Max services to pass to LLM # Semantic search QDRANT_COLLECTION = "intent_collections" -SEMANTIC_SEARCH_TOP_K = 10 # Top 10 relevant services -SEMANTIC_SEARCH_THRESHOLD = 0.2 # Minimum similarity score -QDRANT_TIMEOUT = 10.0 # seconds +SEMANTIC_SEARCH_TOP_K = 10 # Top 10 relevant services +SEMANTIC_SEARCH_THRESHOLD = 0.2 # Minimum similarity score +QDRANT_TIMEOUT = 10.0 # seconds # Hybrid search classification (see HYBRID_SEARCH_CLASSIFICATION.md) -DENSE_MIN_THRESHOLD = 0.38 # Minimum cosine to consider service match +DENSE_MIN_THRESHOLD = 0.38 # Minimum cosine to consider service match DENSE_HIGH_CONFIDENCE_THRESHOLD = 0.40 # Cosine for high-confidence path -DENSE_SCORE_GAP_THRESHOLD = 0.05 # Required gap between top two services -DENSE_SEARCH_TOP_K = 3 # Unique services from dense search -HYBRID_SEARCH_TOP_K = 5 # Results from hybrid RRF search +DENSE_SCORE_GAP_THRESHOLD = 0.05 # Required gap between top two services +DENSE_SEARCH_TOP_K = 3 # Unique services from dense search +HYBRID_SEARCH_TOP_K = 5 # Results from hybrid RRF search ``` --- diff --git a/grafana-configs/README.md b/grafana-configs/README.md index 6feba7b..dcdb3a9 100644 --- a/grafana-configs/README.md +++ b/grafana-configs/README.md @@ -87,12 +87,17 @@ from grafana_configs.loki_logger import LokiLogger logger = LokiLogger(service_name="model-deployment-orchestrator") # Log with model context -logger.info("Starting deployment", model_id="model123", - current_env="testing", target_env="production") +logger.info( + "Starting deployment", + model_id="model123", + current_env="testing", + target_env="production", +) # Log errors with extra context -logger.error("Deployment failed", model_id="model123", - error_code=500, step="model_loading") +logger.error( + "Deployment failed", model_id="model123", error_code=500, step="model_loading" +) ``` ### Accessing Grafana diff --git a/src/contextual_retrieval/contextual_retrieval.md b/src/contextual_retrieval/contextual_retrieval.md index ce3446c..3f7c4e7 100644 --- a/src/contextual_retrieval/contextual_retrieval.md +++ b/src/contextual_retrieval/contextual_retrieval.md @@ -134,7 +134,8 @@ async def retrieve_contextual_chunks( ```python class HTTPClientManager: """Centralized HTTP client with connection pooling and resource management""" - + + class ServiceResilienceManager: """Circuit breaker implementation for fault tolerance""" ``` @@ -155,12 +156,14 @@ When the LLM Orchestration Service receives multiple simultaneous requests, the class QdrantContextualSearch: def __init__(self): self.client = httpx.AsyncClient() # New client per instance - + + class SmartBM25Search: def __init__(self): self.client = httpx.AsyncClient() # Another new client -# Result: + +# Result: # - 100+ HTTP connections for 10 concurrent requests # - Connection exhaustion # - Resource leaks @@ -171,19 +174,20 @@ class SmartBM25Search: ```python # GOOD: Shared HTTP client with connection pooling class HTTPClientManager: - _instance: Optional['HTTPClientManager'] = None # Singleton - + _instance: Optional["HTTPClientManager"] = None # Singleton + async def get_client(self) -> httpx.AsyncClient: if self._client is None: self._client = httpx.AsyncClient( limits=httpx.Limits( - max_connections=100, # Total pool size - max_keepalive_connections=20 # Reuse connections + max_connections=100, # Total pool size + max_keepalive_connections=20, # Reuse connections ), - timeout=httpx.Timeout(30.0) + timeout=httpx.Timeout(30.0), ) return self._client + # Result: # - Single connection pool (100 connections max) # - Connection reuse across all components @@ -195,10 +199,10 @@ class HTTPClientManager: ```python class ServiceResilienceManager: def __init__(self, config): - self.failure_threshold = 3 # Open circuit after 3 failures - self.recovery_timeout = 60.0 # Try recovery after 60 seconds - self.state = "CLOSED" # CLOSED → OPEN → HALF_OPEN - + self.failure_threshold = 3 # Open circuit after 3 failures + self.recovery_timeout = 60.0 # Try recovery after 60 seconds + self.state = "CLOSED" # CLOSED → OPEN → HALF_OPEN + def can_execute(self) -> bool: """Prevents cascading failures during high load""" if self.state == "OPEN": @@ -217,16 +221,16 @@ class QdrantContextualSearch: def __init__(self, qdrant_url: str, config: ContextualRetrievalConfig): # Uses shared HTTP client manager self.http_manager = HTTPClientManager() - + async def search_contextual_embeddings(self, embedding, collections, limit): # All Qdrant API calls use managed HTTP client client = await self.http_manager.get_client() - + # Circuit breaker protects against Qdrant downtime response = await self.http_manager.execute_with_circuit_breaker( method="POST", url=f"{self.qdrant_url}/collections/{collection}/points/search", - json=search_payload + json=search_payload, ) ``` @@ -236,12 +240,10 @@ class QdrantContextualSearch: async def get_embedding_for_query(self, query: str): # Uses shared HTTP client for LLM Orchestration API calls client = await self.http_manager.get_client() - + # Resilient embedding generation response = await self.http_manager.execute_with_circuit_breaker( - method="POST", - url="/embeddings", - json={"inputs": [query]} + method="POST", url="/embeddings", json={"inputs": [query]} ) ``` @@ -275,28 +277,28 @@ async def retry_http_request( url: str, max_retries: int = 3, retry_delay: float = 1.0, - backoff_factor: float = 2.0 + backoff_factor: float = 2.0, ) -> Optional[httpx.Response]: """ Handles transient failures gracefully: - Network hiccups during high load - - Temporary service unavailability + - Temporary service unavailability - Rate limiting responses """ for attempt in range(max_retries + 1): try: response = await client.request(method, url, **kwargs) - + # Success - return immediately if response.status_code < 400: return response - + # 4xx errors (client errors) - don't retry if 400 <= response.status_code < 500: return response - + # 5xx errors (server errors) - retry with backoff - + except (httpx.ConnectError, httpx.TimeoutException) as e: if attempt < max_retries: await asyncio.sleep(retry_delay) @@ -312,11 +314,11 @@ def client_stats(self) -> Dict[str, Any]: """Monitor connection pool health during high load""" return { "status": "active", - "pool_connections": 45, # Currently active connections - "keepalive_connections": 15, # Reusable connections + "pool_connections": 45, # Currently active connections + "keepalive_connections": 15, # Reusable connections "circuit_breaker_state": "CLOSED", "total_requests": 1247, - "failed_requests": 3 + "failed_requests": 3, } ``` @@ -356,19 +358,19 @@ class ContextualRetriever: ```python def detect_optimal_collections(query: str) -> List[str]: collections = [] - + # Check Azure keywords if any(keyword in query.lower() for keyword in AZURE_KEYWORDS): collections.append("azure_contextual_collection") - - # Check AWS keywords + + # Check AWS keywords if any(keyword in query.lower() for keyword in AWS_KEYWORDS): collections.append("aws_contextual_collection") - + # Default fallback if not collections: collections = ["azure_contextual_collection", "aws_contextual_collection"] - + return collections ``` @@ -451,9 +453,7 @@ Where: ```python # 1. Initialize ContextualRetriever retriever = ContextualRetriever( - qdrant_url="http://qdrant:6333", - environment="production", - connection_id="user123" + qdrant_url="http://qdrant:6333", environment="production", connection_id="user123" ) # 2. Initialize components @@ -467,7 +467,7 @@ original_question = "How do I set up Azure authentication?" refined_questions = [ "What are the steps to configure Azure Active Directory authentication?", "How to implement OAuth2 with Azure AD?", - "Azure authentication setup guide" + "Azure authentication setup guide", ] ``` @@ -475,8 +475,7 @@ refined_questions = [ ```python # Dynamic provider detection collections = await provider_detection.detect_optimal_collections( - environment="production", - connection_id="user123" + environment="production", connection_id="user123" ) # Result: ["azure_contextual_collection"] (Azure keywords detected) ``` @@ -488,10 +487,8 @@ if config.enable_parallel_search: semantic_task = _semantic_search( original_question, refined_questions, collections, 40, env, conn_id ) - bm25_task = _bm25_search( - original_question, refined_questions, 40 - ) - + bm25_task = _bm25_search(original_question, refined_questions, 40) + semantic_results, bm25_results = await asyncio.gather( semantic_task, bm25_task, return_exceptions=True ) @@ -507,7 +504,7 @@ batch_embeddings = qdrant_search.get_embeddings_for_queries_batch( queries=all_queries, llm_service=cached_llm_service, environment="production", - connection_id="user123" + connection_id="user123", ) # Parallel search execution @@ -542,19 +539,21 @@ deduplicated_bm25 = deduplicate_bm25_results(bm25_results) # Dynamic Rank Fusion fused_results = rank_fusion.fuse_results( semantic_results=semantic_results, # 40 results - bm25_results=bm25_results, # 40 results - final_top_n=12 # Return top 12 + bm25_results=bm25_results, # 40 results + final_top_n=12, # Return top 12 ) # RRF calculation for each document for doc_id in all_document_ids: semantic_rank = get_rank_in_results(doc_id, semantic_results) bm25_rank = get_rank_in_results(doc_id, bm25_results) - + rrf_score = 0 - if semantic_rank: rrf_score += 1 / (60 + semantic_rank) - if bm25_rank: rrf_score += 1 / (60 + bm25_rank) - + if semantic_rank: + rrf_score += 1 / (60 + semantic_rank) + if bm25_rank: + rrf_score += 1 / (60 + bm25_rank) + doc_scores[doc_id] = rrf_score # Sort by RRF score and return top N @@ -574,10 +573,10 @@ for result in fused_results: "retrieval_type": "contextual", "semantic_score": result.get("normalized_score"), "bm25_score": result.get("normalized_bm25_score"), - "fused_score": result.get("fused_score") + "fused_score": result.get("fused_score"), }, "score": result.get("fused_score"), - "id": result.get("chunk_id") + "id": result.get("chunk_id"), } formatted_results.append(formatted_chunk) @@ -613,16 +612,14 @@ selected_collections = ["azure_contextual_collection"] # Batch embedding generation queries = [ "How do I set up Azure authentication?", - "What are the steps to configure Azure Active Directory authentication?", + "What are the steps to configure Azure Active Directory authentication?", "How to implement OAuth2 with Azure AD?", - "Azure authentication setup guide" + "Azure authentication setup guide", ] # LLM API call for batch embeddings embeddings = llm_service.create_embeddings_for_indexer( - texts=queries, - model="text-embedding-3-large", - environment="production" + texts=queries, model="text-embedding-3-large", environment="production" ) # Parallel search across queries @@ -632,7 +629,7 @@ semantic_results = [ "contextual_content": "This section covers Azure Active Directory authentication setup. To configure Azure AD authentication, you need to...", "score": 0.89, "document_url": "azure-auth-guide.pdf", - "source_query": "How do I set up Azure authentication?" + "source_query": "How do I set up Azure authentication?", }, # ... more results ] @@ -643,10 +640,10 @@ semantic_results = [ # BM25 lexical search bm25_results = [ { - "chunk_id": "azure_auth_002", + "chunk_id": "azure_auth_002", "contextual_content": "This guide explains Azure authentication implementation. Follow these steps to set up Azure AD...", "bm25_score": 8.42, - "document_url": "azure-implementation.md" + "document_url": "azure-implementation.md", }, # ... more results ] @@ -675,14 +672,14 @@ final_results = [ "text": "This section covers Azure Active Directory authentication setup. To configure Azure AD authentication, you need to register your application in the Azure portal, configure redirect URIs, and implement the OAuth2 flow...", "meta": { "source_file": "azure-auth-guide.pdf", - "chunk_id": "azure_auth_001", + "chunk_id": "azure_auth_001", "retrieval_type": "contextual", "semantic_score": 0.89, "bm25_score": 0.72, - "fused_score": 0.0323 + "fused_score": 0.0323, }, "score": 0.0323, - "id": "azure_auth_001" + "id": "azure_auth_001", } # ... 11 more chunks (final_top_n = 12) ] @@ -774,14 +771,12 @@ rank_fusion: def _initialize_contextual_retriever( self, environment: str, connection_id: Optional[str] ) -> ContextualRetriever: - qdrant_url = os.getenv('QDRANT_URL', 'http://qdrant:6333') - + qdrant_url = os.getenv("QDRANT_URL", "http://qdrant:6333") + contextual_retriever = ContextualRetriever( - qdrant_url=qdrant_url, - environment=environment, - connection_id=connection_id + qdrant_url=qdrant_url, environment=environment, connection_id=connection_id ) - + return contextual_retriever ``` @@ -791,14 +786,12 @@ def _initialize_contextual_retriever( def _execute_orchestration_pipeline(self, request, components, costs_metric): # Step 1: Refine user prompt refined_output = self._refine_user_prompt(...) - - # Step 2: Retrieve contextual chunks + + # Step 2: Retrieve contextual chunks relevant_chunks = self._safe_retrieve_contextual_chunks( - components["contextual_retriever"], - refined_output, - request + components["contextual_retriever"], refined_output, request ) - + # Step 3: Generate response with chunks response = self._generate_response_with_chunks( relevant_chunks, refined_output, request @@ -810,26 +803,26 @@ def _execute_orchestration_pipeline(self, request, components, costs_metric): def _safe_retrieve_contextual_chunks( self, contextual_retriever: Optional[ContextualRetriever], - refined_output: PromptRefinerOutput, + refined_output: PromptRefinerOutput, request: OrchestrationRequest, ) -> Optional[List[Dict]]: - + async def async_retrieve(): # Initialize if needed if not contextual_retriever.initialized: success = await contextual_retriever.initialize() if not success: return None - + # Retrieve chunks chunks = await contextual_retriever.retrieve_contextual_chunks( original_question=refined_output.original_question, refined_questions=refined_output.refined_questions, environment=request.environment, - connection_id=request.connection_id + connection_id=request.connection_id, ) return chunks - + # Run async in sync context return asyncio.run(async_retrieve()) ``` @@ -905,18 +898,18 @@ Circuit Breaker: CLOSED (no failures) { "total_pool_size": 100, "active_connections": { - "qdrant_searches": 35, # Vector searches - "llm_embeddings": 25, # Embedding generation - "bm25_operations": 10, # Lexical searches - "keepalive_reserved": 20, # Ready for reuse - "available": 10 # Unused capacity + "qdrant_searches": 35, # Vector searches + "llm_embeddings": 25, # Embedding generation + "bm25_operations": 10, # Lexical searches + "keepalive_reserved": 20, # Ready for reuse + "available": 10, # Unused capacity }, "efficiency_metrics": { "connection_reuse_rate": "85%", - "average_connection_lifetime": "45s", + "average_connection_lifetime": "45s", "failed_connections": 0, - "circuit_breaker_activations": 0 - } + "circuit_breaker_activations": 0, + }, } ``` @@ -945,24 +938,24 @@ Total System Downtime: 90 seconds ```python def handle_qdrant_failure_scenario(): """Real-world circuit breaker behavior""" - + # CLOSED → OPEN (after 3 failures) failures = [ "Request 1: Qdrant timeout (30s)", - "Request 2: Qdrant timeout (30s)", - "Request 3: Qdrant timeout (30s)" # Circuit opens here + "Request 2: Qdrant timeout (30s)", + "Request 3: Qdrant timeout (30s)", # Circuit opens here ] - + # OPEN state (60 seconds) blocked_requests = [ "Request 4-47: Immediate failure (0.1s each)", - "Total blocked: 44 requests in 4.4 seconds" + "Total blocked: 44 requests in 4.4 seconds", ] - + # HALF_OPEN → CLOSED (service recovery) recovery = [ "Request 48: Success (200ms) → Circuit CLOSED", - "Request 49-100: Normal operation resumed" + "Request 49-100: Normal operation resumed", ] ``` @@ -1004,14 +997,14 @@ def handle_qdrant_failure_scenario(): "original_question": "How do I set up Azure authentication?", "refined_questions": [ "What are the steps to configure Azure Active Directory authentication?", - "How to implement OAuth2 with Azure AD?", - "Azure authentication setup guide" + "How to implement OAuth2 with Azure AD?", + "Azure authentication setup guide", ], "environment": "production", "connection_id": "user123", - "topk_semantic": 40, # Optional - uses config default - "topk_bm25": 40, # Optional - uses config default - "final_top_n": 12 # Optional - uses config default + "topk_semantic": 40, # Optional - uses config default + "topk_bm25": 40, # Optional - uses config default + "final_top_n": 12, # Optional - uses config default } ``` @@ -1028,16 +1021,15 @@ def handle_qdrant_failure_scenario(): "retrieval_type": "contextual", "primary_source": "azure", "semantic_score": 0.89, - "bm25_score": 0.72, - "fused_score": 0.0323 + "bm25_score": 0.72, + "fused_score": 0.0323, }, - # Legacy compatibility fields "id": "azure_auth_001", "score": 0.0323, "content": "This section covers Azure Active Directory authentication setup...", "document_url": "azure-auth-guide.pdf", - "retrieval_type": "contextual" + "retrieval_type": "contextual", } # ... 11 more chunks ] @@ -1052,15 +1044,15 @@ refined_output = PromptRefinerOutput( original_question="How do I set up Azure authentication?", refined_questions=[...], is_off_topic=False, - reasoning="User asking about Azure authentication setup" + reasoning="User asking about Azure authentication setup", ) # OrchestrationRequest request = OrchestrationRequest( - message="How do I set up Azure authentication?", + message="How do I set up Azure authentication?", environment="production", connection_id="user123", - chatId="chat456" + chatId="chat456", ) ``` @@ -1070,8 +1062,8 @@ request = OrchestrationRequest( contextual_chunks = [ { "text": "contextual content...", # This is what ResponseGenerator uses - "meta": {...}, # Source information and scores - "score": 0.0323 # Final fused score + "meta": {...}, # Source information and scores + "score": 0.0323, # Final fused score } ] ``` @@ -1092,8 +1084,8 @@ class RateLimiter: #### 2. Enhanced Caching ```python class EmbeddingCache: - max_size: int = 1000 # LRU cache for embeddings - ttl_seconds: int = 3600 # 1 hour TTL + max_size: int = 1000 # LRU cache for embeddings + ttl_seconds: int = 3600 # 1 hour TTL ``` #### 3. Connection Pool Optimization diff --git a/src/guardrails/readme.md b/src/guardrails/readme.md index 7a69e93..17a97bb 100644 --- a/src/guardrails/readme.md +++ b/src/guardrails/readme.md @@ -118,12 +118,12 @@ Cost: $0.000156 (7 tokens) #### 3. **GuardrailCheckResult** (Pydantic Model) ```python class GuardrailCheckResult(BaseModel): - allowed: bool # True if content passes - verdict: str # "yes" = blocked, "no" = allowed - content: str # Response message + allowed: bool # True if content passes + verdict: str # "yes" = blocked, "no" = allowed + content: str # Response message blocked_by_rail: Optional[str] # Exception type if blocked - reason: Optional[str] # Explanation - error: Optional[str] # Error message if failed + reason: Optional[str] # Explanation + error: Optional[str] # Error message if failed usage: Dict[str, Union[float, int]] # Cost tracking ``` @@ -136,8 +136,8 @@ When `enable_rails_exceptions: true` in config: "role": "exception", "content": { "type": "InputRailException", - "message": "I'm not able to respond to that" - } + "message": "I'm not able to respond to that", + }, } ``` @@ -167,11 +167,11 @@ result.usage = usage_info # Contains: total_cost, tokens, num_calls **Usage Dictionary Structure**: ```python { - "total_cost": 0.000245, # USD + "total_cost": 0.000245, # USD "total_prompt_tokens": 8, "total_completion_tokens": 2, "total_tokens": 10, - "num_calls": 1 + "num_calls": 1, } ``` @@ -181,10 +181,10 @@ result.usage = usage_info # Contains: total_cost, tokens, num_calls ```python costs_metric = { - "input_guardrails": {...}, # Step 1 - "prompt_refiner": {...}, # Step 2 - "response_generator": {...}, # Step 4 - "output_guardrails": {...} # Step 5 + "input_guardrails": {...}, # Step 1 + "prompt_refiner": {...}, # Step 2 + "response_generator": {...}, # Step 4 + "output_guardrails": {...}, # Step 5 } # Step 3 (retrieval) has no LLM cost @@ -197,7 +197,7 @@ costs_metric = { if not input_result.allowed: return OrchestrationResponse( inputGuardFailed=True, - content=input_result.content # Refusal message + content=input_result.content, # Refusal message ) # Saves costs: no refinement, retrieval, or generation ``` diff --git a/src/vector_indexer/diff_identifier/DIFF_IDENTIFIER_FLOW.md b/src/vector_indexer/diff_identifier/DIFF_IDENTIFIER_FLOW.md index 57a48d2..34b2313 100644 --- a/src/vector_indexer/diff_identifier/DIFF_IDENTIFIER_FLOW.md +++ b/src/vector_indexer/diff_identifier/DIFF_IDENTIFIER_FLOW.md @@ -94,7 +94,7 @@ class VersionManager: "last_run_modified_files": 1, "last_run_deleted_files": 1, "last_cleanup_deleted_chunks": 15, - "last_run_timestamp": "2025-10-17T00:00:46Z" + "last_run_timestamp": "2025-10-17T00:00:46Z", }, "processed_files": { "sha256_hash": { @@ -103,9 +103,15 @@ class VersionManager: "file_size": 15234, "processed_at": "2025-10-17T00:00:46Z", "chunk_count": 5, # Track chunk count for validation - "chunk_ids": ["uuid1", "uuid2", "uuid3", "uuid4", "uuid5"] # Track exact chunks + "chunk_ids": [ + "uuid1", + "uuid2", + "uuid3", + "uuid4", + "uuid5", + ], # Track exact chunks } - } + }, } ``` @@ -138,27 +144,39 @@ class ProcessedFileInfo(BaseModel): chunk_count: int = 0 # NEW: Track number of chunks chunk_ids: List[str] = Field(default_factory=list) # NEW: Track chunk IDs + class DiffResult(BaseModel): # File change detection new_files: List[str] = Field(..., description="Files to process for first time") - modified_files: List[str] = Field(default_factory=list, description="Files with changed content") - deleted_files: List[str] = Field(default_factory=list, description="Files removed from dataset") - unchanged_files: List[str] = Field(default_factory=list, description="Files with same content") - + modified_files: List[str] = Field( + default_factory=list, description="Files with changed content" + ) + deleted_files: List[str] = Field( + default_factory=list, description="Files removed from dataset" + ) + unchanged_files: List[str] = Field( + default_factory=list, description="Files with same content" + ) + # Statistics total_files_scanned: int previously_processed_count: int is_first_run: bool - + # NEW: Cleanup metadata - chunks_to_delete: Dict[str, List[str]] = Field(default_factory=dict) # document_hash -> chunk_ids + chunks_to_delete: Dict[str, List[str]] = Field( + default_factory=dict + ) # document_hash -> chunk_ids estimated_cleanup_count: int = Field(default=0) # Total chunks to be removed + class VersionState(BaseModel): last_updated: str processed_files: Dict[str, ProcessedFileInfo] total_processed: int - processing_stats: Dict[str, Any] = Field(default_factory=dict) # NEW: Enhanced stats + processing_stats: Dict[str, Any] = Field( + default_factory=dict + ) # NEW: Enhanced stats ``` ## Enhanced Processing Flow @@ -247,8 +265,8 @@ if not files_to_process: ```python # NEW: Track chunk information in metadata await diff_detector.mark_files_processed( - processed_paths, - chunks_info=collected_chunk_information # Future enhancement + processed_paths, + chunks_info=collected_chunk_information, # Future enhancement ) ``` @@ -269,7 +287,7 @@ await diff_detector.mark_files_processed( # Efficient chunk identification for cleanup chunks_to_delete = { "document_hash_123": ["chunk_uuid_1", "chunk_uuid_2", "chunk_uuid_3"], - "document_hash_456": ["chunk_uuid_4", "chunk_uuid_5"] + "document_hash_456": ["chunk_uuid_4", "chunk_uuid_5"], } # Cleanup execution per collection @@ -395,7 +413,9 @@ diff_result = await diff_detector.get_changed_files() logger.info(f"Cleanup metadata: {diff_result.chunks_to_delete}") # Test cleanup operations -cleanup_count = await main_indexer._execute_cleanup_operations(qdrant_manager, diff_result) +cleanup_count = await main_indexer._execute_cleanup_operations( + qdrant_manager, diff_result +) logger.info(f"Total cleanup: {cleanup_count} chunks") ``` @@ -408,20 +428,22 @@ logger.info(f"Total cleanup: {cleanup_count} chunks") async def process_all_documents(self) -> ProcessingStats: # 1. Enhanced diff detection diff_result = await diff_detector.get_changed_files() - + # 2. NEW: Automatic cleanup execution if diff_result.chunks_to_delete: - cleanup_count = await self._execute_cleanup_operations(qdrant_manager, diff_result) - + cleanup_count = await self._execute_cleanup_operations( + qdrant_manager, diff_result + ) + # 3. Selective document processing files_to_process = diff_result.new_files + diff_result.modified_files if not files_to_process: return self.stats # Early exit - + # 4. Standard processing pipeline documents = self._filter_documents_by_paths(files_to_process) results = await self._process_documents(documents) - + # 5. Enhanced metadata update await diff_detector.mark_files_processed(processed_paths, chunks_info) ``` @@ -565,17 +587,18 @@ if is_first_run: current_files = version_manager.scan_current_files() # Returns: Dict[content_hash, file_path] for all discovered files + def scan_current_files(self) -> Dict[str, str]: file_hash_map = {} for root, _, files in os.walk(self.config.datasets_path): for file in files: file_path = os.path.join(root, file) relative_path = os.path.relpath(file_path, self.config.datasets_path) - + # Calculate content hash for change detection content_hash = self._calculate_file_hash(file_path) file_hash_map[content_hash] = relative_path - + return file_hash_map ``` @@ -592,23 +615,24 @@ def scan_current_files(self) -> Dict[str, str]: processed_metadata = await s3_ferry_client.download_metadata() # Downloads from: s3://rag-search/resources/datasets/processed-metadata.json + def download_metadata(self) -> Optional[Dict[str, Any]]: # Create temporary file for S3Ferry transfer - with tempfile.NamedTemporaryFile(suffix='.json', delete=False) as temp_file: + with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as temp_file: temp_file_path = temp_file.name - + # Transfer S3 → FS via S3Ferry API response = self._retry_with_backoff( lambda: self.s3_ferry.transfer_file( destinationFilePath=temp_file_path, - destinationStorageType="FS", + destinationStorageType="FS", sourceFilePath=self.config.metadata_s3_path, - sourceStorageType="S3" + sourceStorageType="S3", ) ) - + if response.status_code == 200: - with open(temp_file_path, 'r') as f: + with open(temp_file_path, "r") as f: return json.load(f) elif response.status_code == 404: return None # First run - no metadata exists yet @@ -624,19 +648,23 @@ def download_metadata(self) -> Optional[Dict[str, Any]]: #### Phase 5: Differential Analysis ```python # 6. Change Detection Algorithm (version_manager.py) -changed_files = version_manager.identify_changed_files(current_files, processed_metadata) +changed_files = version_manager.identify_changed_files( + current_files, processed_metadata +) -def identify_changed_files(self, current_files: Dict[str, str], - processed_state: Optional[Dict]) -> Set[str]: + +def identify_changed_files( + self, current_files: Dict[str, str], processed_state: Optional[Dict] +) -> Set[str]: if not processed_state: return set(current_files.values()) # All files are "new" - - processed_hashes = set(processed_state.get('processed_files', {}).keys()) + + processed_hashes = set(processed_state.get("processed_files", {}).keys()) current_hashes = set(current_files.keys()) - + # Identify new and modified files new_or_changed_hashes = current_hashes - processed_hashes - + # Convert hashes back to file paths return {current_files[hash_val] for hash_val in new_or_changed_hashes} ``` @@ -654,8 +682,8 @@ def identify_changed_files(self, current_files: Dict[str, str], return DiffResult( new_files=list(changed_files), total_files_scanned=len(current_files), - previously_processed_count=len(processed_state.get('processed_files', {})), - is_first_run=is_first_run + previously_processed_count=len(processed_state.get("processed_files", {})), + is_first_run=is_first_run, ) ``` @@ -706,11 +734,15 @@ Dataset Download → [shared-volume] → diff_identifier → [datasets mount] if diff_result.new_files: # Process only changed files documents = self._filter_documents_by_paths(diff_result.new_files) - logger.info(f"Processing {len(documents)} documents from {len(diff_result.new_files)} changed files") + logger.info( + f"Processing {len(documents)} documents from {len(diff_result.new_files)} changed files" + ) else: # No changes detected - skip processing entirely logger.info("No changes detected. Skipping processing phase.") - return ProcessingResult(processed_count=0, skipped_count=diff_result.total_files_scanned) + return ProcessingResult( + processed_count=0, skipped_count=diff_result.total_files_scanned + ) # Continue with existing vector generation pipeline... ``` @@ -727,25 +759,26 @@ else: async def mark_files_processed(self, file_paths: List[str]) -> bool: # Update processed files metadata new_metadata = self._create_updated_metadata(file_paths) - + # Upload to S3 via S3Ferry success = await self.s3_ferry_client.upload_metadata(new_metadata) - + # Commit DVC state (optional - for advanced versioning) if success: self.version_manager.commit_dvc_state(f"Processed {len(file_paths)} files") - + return success + def _create_updated_metadata(self, file_paths: List[str]) -> Dict[str, Any]: current_files = self.version_manager.scan_current_files() - + metadata = { "last_updated": datetime.utcnow().isoformat(), - "total_processed": len(file_paths), - "processed_files": {} + "total_processed": len(file_paths), + "processed_files": {}, } - + # Add file metadata for each processed file for file_path in file_paths: file_hash = self._get_file_hash(file_path) @@ -753,9 +786,9 @@ def _create_updated_metadata(self, file_paths: List[str]) -> Dict[str, Any]: content_hash=file_hash, original_path=file_path, file_size=os.path.getsize(file_path), - processed_at=datetime.utcnow().isoformat() + processed_at=datetime.utcnow().isoformat(), ).dict() - + return metadata ``` @@ -1067,14 +1100,14 @@ diff_detector = DiffDetector(diff_config) # Passes to main orchestrator # diff_detector.py - Configuration factory config = DiffConfig( - s3_ferry_url=s3_ferry_url, # → Used by S3FerryClient - metadata_s3_path=metadata_s3_path, # → Used for S3Ferry operations - datasets_path=datasets_path, # → Used for file scanning - metadata_filename=metadata_filename, # → Used to build paths - dvc_remote_url=dvc_remote_url, # → Used by DVC setup - s3_endpoint_url=str(s3_endpoint_url), # → Used by DVC S3 config - s3_access_key_id=str(s3_access_key_id), # → Used by DVC authentication - s3_secret_access_key=str(s3_secret_access_key) # → Used by DVC authentication + s3_ferry_url=s3_ferry_url, # → Used by S3FerryClient + metadata_s3_path=metadata_s3_path, # → Used for S3Ferry operations + datasets_path=datasets_path, # → Used for file scanning + metadata_filename=metadata_filename, # → Used to build paths + dvc_remote_url=dvc_remote_url, # → Used by DVC setup + s3_endpoint_url=str(s3_endpoint_url), # → Used by DVC S3 config + s3_access_key_id=str(s3_access_key_id), # → Used by DVC authentication + s3_secret_access_key=str(s3_secret_access_key), # → Used by DVC authentication ) ``` @@ -1118,7 +1151,7 @@ response = self.s3_ferry.transfer_file( destinationFilePath="resources/datasets/processed-metadata.json", destinationStorageType="S3", sourceFilePath="/tmp/tmpABC123.json", # Temporary file - sourceStorageType="FS" + sourceStorageType="FS", ) ``` @@ -1145,7 +1178,7 @@ response = self.s3_ferry.transfer_file( destinationFilePath="/tmp/tmpDEF456.json", # Temporary file destinationStorageType="FS", sourceFilePath="resources/datasets/processed-metadata.json", - sourceStorageType="S3" + sourceStorageType="S3", ) ``` @@ -1535,7 +1568,7 @@ DiffResult( new_files=["datasets/collection1/abc123/cleaned.txt"], total_files_scanned=100, previously_processed_count=99, - is_first_run=False + is_first_run=False, ) ``` diff --git a/src/vector_indexer/vector_indexer_integration.md b/src/vector_indexer/vector_indexer_integration.md index d6b10b2..a160c51 100644 --- a/src/vector_indexer/vector_indexer_integration.md +++ b/src/vector_indexer/vector_indexer_integration.md @@ -161,15 +161,18 @@ chunking: async def generate_context_batch(self, document_content: str, chunks: List[str]): # Level 1: Batch processing (context_batch_size = 5) for i in range(0, len(chunks), self.config.context_batch_size): - batch = chunks[i:i + self.config.context_batch_size] - + batch = chunks[i : i + self.config.context_batch_size] + # Level 2: Semaphore limiting (max_concurrent_chunks_per_doc = 5) semaphore = asyncio.Semaphore(self.config.max_concurrent_chunks_per_doc) - + # Process batch concurrently with controlled limits batch_contexts = await asyncio.gather( - *[self._generate_context_with_retry(document_content, chunk) for chunk in batch], - return_exceptions=True + *[ + self._generate_context_with_retry(document_content, chunk) + for chunk in batch + ], + return_exceptions=True, ) ``` @@ -220,15 +223,15 @@ graph LR # Configuration-Driven Batch Optimization async def _create_embeddings_in_batches(self, contextual_contents: List[str]): all_embeddings = [] - + # Process in configurable batches (embedding_batch_size = 10) for i in range(0, len(contextual_contents), self.config.embedding_batch_size): - batch = contextual_contents[i:i + self.config.embedding_batch_size] - + batch = contextual_contents[i : i + self.config.embedding_batch_size] + # API call with comprehensive error handling batch_response = await self.api_client.create_embeddings_batch(batch) all_embeddings.extend(batch_response["embeddings"]) - + # Configurable delay between batches if i + self.config.embedding_batch_size < len(contextual_contents): delay = self.config.processing.batch_delay_seconds # 0.1s @@ -298,9 +301,9 @@ graph TD ```python # Step 5: Add embeddings to chunks with full traceability for chunk, embedding in zip(contextual_chunks, embeddings_response["embeddings"]): - chunk.embedding = embedding # Vector data - chunk.embedding_model = embeddings_response["model_used"] # Model traceability - chunk.vector_dimensions = len(embedding) # Dimension validation + chunk.embedding = embedding # Vector data + chunk.embedding_model = embeddings_response["model_used"] # Model traceability + chunk.vector_dimensions = len(embedding) # Dimension validation # Provider automatically detected from model name ``` @@ -315,18 +318,18 @@ self.collections_config = { "contextual_chunks_azure": { "vector_size": 3072, # text-embedding-3-large (Azure) "distance": "Cosine", - "models": ["text-embedding-3-large", "text-embedding-ada-002"] + "models": ["text-embedding-3-large", "text-embedding-ada-002"], }, "contextual_chunks_aws": { "vector_size": 1024, # amazon.titan-embed-text-v2:0 - "distance": "Cosine", - "models": ["amazon.titan-embed-text-v2:0", "amazon.titan-embed-text-v1"] + "distance": "Cosine", + "models": ["amazon.titan-embed-text-v2:0", "amazon.titan-embed-text-v1"], }, "contextual_chunks_openai": { "vector_size": 1536, # text-embedding-3-small (Direct OpenAI) "distance": "Cosine", - "models": ["text-embedding-3-small", "text-embedding-ada-002"] - } + "models": ["text-embedding-3-small", "text-embedding-ada-002"], + }, } ``` @@ -336,9 +339,9 @@ self.collections_config = { point_id = str(uuid.uuid5(uuid.NAMESPACE_DNS, chunk.chunk_id)) point = { - "id": point_id, # Deterministic UUID - "vector": chunk.embedding, # Provider-specific dimensions - "payload": self._create_chunk_payload(chunk) # Rich metadata + "id": point_id, # Deterministic UUID + "vector": chunk.embedding, # Provider-specific dimensions + "payload": self._create_chunk_payload(chunk), # Rich metadata } ``` @@ -347,15 +350,15 @@ point = { # Production-Grade Batch Processing batch_size = 100 # Prevents request timeout issues for i in range(0, len(points), batch_size): - batch = points[i:i + batch_size] - + batch = points[i : i + batch_size] + # Comprehensive request logging for debugging logger.info(f"=== QDRANT HTTP REQUEST PAYLOAD DEBUG ===") logger.info(f"Batch size: {len(batch)} points") - + response = await self.client.put( f"{self.qdrant_url}/collections/{collection_name}/points", - json={"points": batch} + json={"points": batch}, ) ``` @@ -367,22 +370,19 @@ for i in range(0, len(points), batch_size): "document_hash": "2e9493512b7f01aecdc66bbca60b5b6b75d966f8", "chunk_index": 0, "total_chunks": 25, - # Anthropic Contextual Retrieval Content "original_content": "FAQ about supporting children and families...", "contextual_content": "Estonian family support policies context. FAQ about...", "context_only": "Estonian family support policies context.", - # Model & Processing Metadata - "embedding_model": "text-embedding-3-large", + "embedding_model": "text-embedding-3-large", "vector_dimensions": 3072, "processing_timestamp": "2025-10-09T12:00:00Z", "tokens_count": 150, - # Document Source Information "document_url": "https://sm.ee/en/faq-about-supporting-children-and-families", "dataset_collection": "sm_someuuid", - "file_type": "html_cleaned" + "file_type": "html_cleaned", } ``` @@ -486,9 +486,9 @@ The Vector Indexer leverages existing LLM configuration through API calls: ```python # Process chunks in batches of 5 with concurrent API calls for batch in chunks_batches(5): - contexts = await asyncio.gather(*[ - api_client.generate_context(document, chunk) for chunk in batch - ]) + contexts = await asyncio.gather( + *[api_client.generate_context(document, chunk) for chunk in batch] + ) ``` 5. **Contextual Chunk Creation** @@ -547,12 +547,12 @@ logs/ collections = { "contextual_chunks_azure": { "vectors": {"size": 1536, "distance": "Cosine"}, # text-embedding-3-large - "model": "text-embedding-3-large" + "model": "text-embedding-3-large", }, "contextual_chunks_aws": { "vectors": {"size": 1024, "distance": "Cosine"}, # amazon.titan-embed-text-v2:0 - "model": "amazon.titan-embed-text-v2:0" - } + "model": "amazon.titan-embed-text-v2:0", + }, } ``` @@ -572,7 +572,7 @@ collections = { "embedding_model": "text-embedding-3-large", "vector_dimensions": 1536, "processing_timestamp": "2025-10-08T12:00:00Z", - "tokens_count": 150 + "tokens_count": 150, } ``` @@ -692,9 +692,9 @@ vector_indexer: class ResourceOptimizedProcessor: def __init__(self): # Process in streaming fashion - never load all documents - self.max_memory_chunks = 100 # Chunk buffer limit - self.gc_frequency = 50 # Garbage collection interval - + self.max_memory_chunks = 100 # Chunk buffer limit + self.gc_frequency = 50 # Garbage collection interval + async def process_documents_streaming(self): """Memory-efficient document processing""" async for document_batch in self.stream_documents(): @@ -720,22 +720,22 @@ class ResourceOptimizedProcessor: "embeddings_created": 26834, "qdrant_points_stored": 26834, "processing_duration_minutes": 186.5, - "average_chunks_per_document": 21.6 + "average_chunks_per_document": 21.6, }, "performance_metrics": { "context_generation_rate_per_minute": 14.4, "embedding_creation_rate_per_minute": 187.3, "end_to_end_documents_per_hour": 10.1, "api_success_rate": 99.7, - "average_response_time_ms": 850 + "average_response_time_ms": 850, }, "error_analysis": { "api_timeouts": 2, "rate_limit_hits": 1, "embedding_dimension_mismatches": 0, "qdrant_storage_failures": 0, - "context_generation_failures": 2 - } + "context_generation_failures": 2, + }, } ``` @@ -773,7 +773,7 @@ logger.info( document_hash="2e9493512b7f01aecdc66bbca60b5b6b75d966f8", document_path="datasets/sm_someuuid/2e9493.../cleaned.txt", chunk_count=23, - processing_id="proc_20241009_120034_789" + processing_id="proc_20241009_120034_789", ) logger.info( @@ -782,7 +782,7 @@ logger.info( model_used="claude-3-haiku-20240307", context_tokens=75, generation_time_ms=1247, - cached_response=False + cached_response=False, ) ```