diff --git a/app/ai-service/api/routes.py b/app/ai-service/api/routes.py index 2281f5e5..99223a74 100644 --- a/app/ai-service/api/routes.py +++ b/app/ai-service/api/routes.py @@ -76,6 +76,7 @@ async def process_ocr( metrics.logger.info(f"OCR Inference completed in {inference_latency:.4f}s") processing_time_ms = int((time.time() - start_time) * 1000) + model_version = settings.groq_model if settings.get_active_provider() == "groq" else settings.openai_model return OCRResponse( success=True, @@ -88,12 +89,14 @@ async def process_ocr( processing_time_ms=processing_time_ms, ), processing_time_ms=processing_time_ms, + model_version=model_version, ) except HTTPException: raise except Exception as e: processing_time_ms = int((time.time() - start_time) * 1000) + model_version = settings.groq_model if settings.get_active_provider() == "groq" else settings.openai_model return OCRResponse( success=False, error={ @@ -101,4 +104,6 @@ async def process_ocr( "message": str(e), }, processing_time_ms=processing_time_ms, + model_version=model_version, ) + diff --git a/app/ai-service/api/v1/anonymize.py b/app/ai-service/api/v1/anonymize.py index 3ed1e779..ee39c18c 100644 --- a/app/ai-service/api/v1/anonymize.py +++ b/app/ai-service/api/v1/anonymize.py @@ -22,7 +22,11 @@ async def anonymize_text(request: AnonymizeRequest): try: result = _main.pii_scrubber_service.anonymize(request.text) - return AnonymizeResponse(success=True, **result) + from config import settings + active_p = settings.get_active_provider() + model_version = settings.groq_model if active_p == "groq" else settings.openai_model + return AnonymizeResponse(success=True, model_version=model_version, **result) except Exception as e: logger.error(f"Anonymization failed: {str(e)}", exc_info=True) raise HTTPException(status_code=500, detail="Failed to anonymize text") + diff --git a/app/ai-service/api/v1/humanitarian.py b/app/ai-service/api/v1/humanitarian.py index 8d4dac2c..eb8a72bb 100644 --- a/app/ai-service/api/v1/humanitarian.py +++ b/app/ai-service/api/v1/humanitarian.py @@ -44,7 +44,22 @@ async def verify_humanitarian_claim(request: HumanitarianVerificationRequest): ) else: raise exc - return HumanitarianVerificationResponse(success=True, **result) + from config import settings + provider = result.get("provider") + if provider == "openai": + model_version = settings.openai_model + elif provider == "groq": + model_version = settings.groq_model + else: + # Default fallback to active provider model or settings.openai_model + active_p = settings.get_active_provider() + model_version = settings.groq_model if active_p == "groq" else settings.openai_model + + return HumanitarianVerificationResponse(success=True, model_version=model_version, **result) except Exception as e: logger.error("Humanitarian verification failed: %s", str(e), exc_info=True) - return HumanitarianVerificationResponse(success=False, error=str(e)) + from config import settings + active_p = settings.get_active_provider() + model_version = settings.groq_model if active_p == "groq" else settings.openai_model + return HumanitarianVerificationResponse(success=False, error=str(e), model_version=model_version) + diff --git a/app/ai-service/api/v1/ocr.py b/app/ai-service/api/v1/ocr.py index 5fd170ae..08059bf6 100644 --- a/app/ai-service/api/v1/ocr.py +++ b/app/ai-service/api/v1/ocr.py @@ -87,6 +87,7 @@ async def process_ocr( metrics.logger.info(f"OCR Inference completed in {inference_latency:.4f}s") processing_time_ms = int((time.time() - start_time) * 1000) + model_version = settings.groq_model if settings.get_active_provider() == "groq" else settings.openai_model return OCRResponse( success=True, @@ -99,12 +100,14 @@ async def process_ocr( processing_time_ms=processing_time_ms, ), processing_time_ms=processing_time_ms, + model_version=model_version, ) except HTTPException: raise except Exception as e: processing_time_ms = int((time.time() - start_time) * 1000) + model_version = settings.groq_model if settings.get_active_provider() == "groq" else settings.openai_model return OCRResponse( success=False, error={ @@ -112,4 +115,6 @@ async def process_ocr( "message": str(e), }, processing_time_ms=processing_time_ms, + model_version=model_version, ) + diff --git a/app/ai-service/schemas/anonymization.py b/app/ai-service/schemas/anonymization.py index a9bb9c19..2f7ddf40 100644 --- a/app/ai-service/schemas/anonymization.py +++ b/app/ai-service/schemas/anonymization.py @@ -1,4 +1,4 @@ -from typing import Dict +from typing import Dict, Optional from pydantic import BaseModel, Field @@ -20,3 +20,4 @@ class AnonymizeResponse(BaseModel): original_length: int pii_summary: PIISummary token_counts: Dict[str, int] = Field(default_factory=dict) + model_version: Optional[str] = None diff --git a/app/ai-service/schemas/humanitarian.py b/app/ai-service/schemas/humanitarian.py index 1e8dc493..3fbc1d04 100644 --- a/app/ai-service/schemas/humanitarian.py +++ b/app/ai-service/schemas/humanitarian.py @@ -1,19 +1,20 @@ -from typing import Any, Dict, List, Literal, Optional -from pydantic import BaseModel, Field - - -class HumanitarianVerificationRequest(BaseModel): - aid_claim: str = Field(min_length=10, description="Aid claim to verify") - supporting_evidence: List[str] = Field(default_factory=list) - context_factors: Dict[str, Any] = Field(default_factory=dict) - provider_preference: Literal["auto", "test", "openai", "groq"] = "auto" - timeout: Optional[float] = Field(default=None, description="Request-level timeout in seconds for provider call") - - -class HumanitarianVerificationResponse(BaseModel): - success: bool - provider: Optional[str] = None - model: Optional[str] = None - prompt_variant: Optional[str] = None - verification: Optional[Dict[str, Any]] = None - error: Optional[str] = None \ No newline at end of file +from typing import Any, Dict, List, Literal, Optional +from pydantic import BaseModel, Field + + +class HumanitarianVerificationRequest(BaseModel): + aid_claim: str = Field(min_length=10, description="Aid claim to verify") + supporting_evidence: List[str] = Field(default_factory=list) + context_factors: Dict[str, Any] = Field(default_factory=dict) + provider_preference: Literal["auto", "test", "openai", "groq"] = "auto" + timeout: Optional[float] = Field(default=None, description="Request-level timeout in seconds for provider call") + + +class HumanitarianVerificationResponse(BaseModel): + success: bool + provider: Optional[str] = None + model: Optional[str] = None + prompt_variant: Optional[str] = None + verification: Optional[Dict[str, Any]] = None + error: Optional[str] = None + model_version: Optional[str] = None \ No newline at end of file diff --git a/app/ai-service/schemas/humanitarian_verification_v2.py b/app/ai-service/schemas/humanitarian_verification_v2.py new file mode 100644 index 00000000..fd98f472 --- /dev/null +++ b/app/ai-service/schemas/humanitarian_verification_v2.py @@ -0,0 +1,24 @@ +from typing import Any, Dict, List, Literal, Optional +from pydantic import BaseModel, Field + + +class HumanitarianVerificationDetailsV2(BaseModel): + verdict: Literal["credible", "partially_credible", "inconclusive", "not_credible"] = Field(description="The verification verdict") + confidence: float = Field(ge=0.0, le=1.0, description="Confidence score between 0.0 and 1.0") + summary: str = Field(description="Neutral summary of the claim verification") + criteria_assessment: Optional[List[Dict[str, Any]]] = Field(default=None, description="Detailed assessment mapping to Sphere criteria") + risk_flags: Optional[List[str]] = Field(default=None, description="Identified risk flags") + missing_information: Optional[List[str]] = Field(default=None, description="List of missing information needed") + recommended_next_steps: Optional[List[str]] = Field(default=None, description="Recommended next steps") + + +class HumanitarianVerificationResponseV2(BaseModel): + success: bool + provider: Optional[str] = None + model: Optional[str] = None + prompt_variant: Optional[str] = None + verification: Optional[HumanitarianVerificationDetailsV2] = None + error: Optional[str] = None + model_version: Optional[str] = None + stamp: Optional[Dict[str, str]] = None + diff --git a/app/ai-service/schemas/ocr.py b/app/ai-service/schemas/ocr.py index 5432d868..f0ecf4f7 100644 --- a/app/ai-service/schemas/ocr.py +++ b/app/ai-service/schemas/ocr.py @@ -1,3 +1,4 @@ +from typing import Optional from pydantic import BaseModel, Field @@ -17,3 +18,5 @@ class OCRResponse(BaseModel): data: OCRData | None = None error: dict[str, str] | None = None processing_time_ms: int + model_version: Optional[str] = None + diff --git a/app/ai-service/services/humanitarian_verification.py b/app/ai-service/services/humanitarian_verification.py index 1c3c595b..65eede6f 100644 --- a/app/ai-service/services/humanitarian_verification.py +++ b/app/ai-service/services/humanitarian_verification.py @@ -1,284 +1,343 @@ -"""Humanitarian claim verification service with model/provider fallbacks.""" - -import json -import logging -from typing import Any, Dict, List, Optional -import time -import metrics - -import httpx - -from config import settings -from services.humanitarian_prompt import HumanitarianPromptEngine -from services.circuit_breaker import CircuitBreaker -from services.test_provider import TestProvider -from exceptions import AIServiceError - -logger = logging.getLogger(__name__) - - -class HumanitarianVerificationService: - """Runs humanitarian verification against configured LLM providers.""" - - def __init__(self): - self.prompt_engine = HumanitarianPromptEngine() - self.test_provider = TestProvider() - self.breakers = { - "openai": CircuitBreaker( - name="openai", - failure_threshold=settings.circuit_breaker_failure_threshold, - recovery_timeout=settings.circuit_breaker_recovery_timeout_seconds, - ), - "groq": CircuitBreaker( - name="groq", - failure_threshold=settings.circuit_breaker_failure_threshold, - recovery_timeout=settings.circuit_breaker_recovery_timeout_seconds, - ), - } - - def verify_claim( - self, - aid_claim: str, - supporting_evidence: Optional[List[str]] = None, - context_factors: Optional[Dict[str, Any]] = None, - provider_preference: str = "auto", - timeout: Optional[float] = None, - ) -> Dict[str, Any]: - start_time = time.time() - try: - evidence = supporting_evidence or [] - context = context_factors or {} - - primary_prompt = self.prompt_engine.build_primary_prompt( - aid_claim=aid_claim, - supporting_evidence=evidence, - context_factors=context, - ) - fallback_prompt = self.prompt_engine.build_fallback_prompt( - aid_claim=aid_claim, - supporting_evidence=evidence, - context_factors=context, - ) - - providers = self._provider_attempt_order(provider_preference) - if not providers: - raise RuntimeError("No LLM providers configured for humanitarian verification") - - errors: List[str] = [] - - for provider in providers: - breaker = self.breakers.get(provider) - if breaker and not breaker.allow_request(): - logger.warning("Circuit breaker is OPEN for provider=%s. Skipping.", provider) - errors.append(f"provider={provider}, error=Circuit breaker is OPEN") - continue - - model = self._get_model_for_provider(provider) - for prompt_variant, prompt in (("primary", primary_prompt), ("fallback", fallback_prompt)): - try: - logger.info( - "Attempting humanitarian verification with provider=%s model=%s prompt=%s", - provider, - model, - prompt_variant, - ) - raw_content = self._call_provider( - provider=provider, - model=model, - system_prompt=prompt["system"], - user_prompt=prompt["user"], - timeout=timeout, - ) - parsed = self._parse_json_response(raw_content) - if breaker: - breaker.record_success() - return { - "provider": provider, - "model": model, - "prompt_variant": prompt_variant, - "verification": parsed, - "raw_response": raw_content, - } - except Exception as exc: - if breaker: - breaker.record_failure() - err = f"provider={provider}, model={model}, prompt={prompt_variant}, error={exc}" - errors.append(err) - logger.warning("Humanitarian verification attempt failed: %s", err) - - raise RuntimeError("All humanitarian verification attempts failed: " + " | ".join(errors)) - finally: - latency = time.time() - start_time - metrics.PIPELINE_STEP_LATENCY.labels(step_name='verify').observe(latency) - - def _provider_attempt_order(self, provider_preference: str) -> List[str]: - available: List[str] = [] - if settings.test_provider_mode: - available.append("test") - if settings.openai_api_key: - available.append("openai") - if settings.groq_api_key: - available.append("groq") - - preference = (provider_preference or "auto").lower() - if preference == "test" and settings.test_provider_mode: - return [preference] - if preference in ("openai", "groq", "test") and preference in available: - return [preference] + [provider for provider in available if provider != preference] - return available - - def _get_model_for_provider(self, provider: str) -> str: - if provider == "test": - return "test-provider/fixture" - if provider == "openai": - return settings.openai_model - if provider == "groq": - return settings.groq_model - raise ValueError(f"Unsupported provider: {provider}") - - def _call_provider( - self, - provider: str, - model: str, - system_prompt: str, - user_prompt: str, - timeout: Optional[float] = None, - ) -> str: - if provider == "test": - return self._call_test(model, system_prompt, user_prompt) - if provider == "openai": - return self._call_openai(model, system_prompt, user_prompt, timeout) - if provider == "groq": - return self._call_groq(model, system_prompt, user_prompt, timeout) - raise ValueError(f"Unsupported provider: {provider}") - - def _call_openai( - self, - model: str, - system_prompt: str, - user_prompt: str, - timeout: Optional[float] = None, - ) -> str: - if not settings.openai_api_key: - raise RuntimeError("OpenAI API key is not configured") - return self._call_chat_completion_api( - base_url="https://api.openai.com/v1/chat/completions", - api_key=settings.openai_api_key, - model=model, - system_prompt=system_prompt, - user_prompt=user_prompt, - timeout=timeout, - ) - - def _call_groq( - self, - model: str, - system_prompt: str, - user_prompt: str, - timeout: Optional[float] = None, - ) -> str: - if not settings.groq_api_key: - raise RuntimeError("Groq API key is not configured") - return self._call_chat_completion_api( - base_url="https://api.groq.com/openai/v1/chat/completions", - api_key=settings.groq_api_key, - model=model, - system_prompt=system_prompt, - user_prompt=user_prompt, - timeout=timeout, - ) - - def _call_chat_completion_api( - self, - base_url: str, - api_key: str, - model: str, - system_prompt: str, - user_prompt: str, - timeout: Optional[float] = None, - ) -> str: - if settings.ai_deterministic_mode: - logger.info("Deterministic AI mode enabled: returning stable response") - return self._get_deterministic_response(model, system_prompt, user_prompt) - - payload = { - "model": model, - "temperature": 0.1, - "messages": [ - {"role": "system", "content": system_prompt}, - {"role": "user", "content": user_prompt}, - ], - } - headers = { - "Authorization": f"Bearer {api_key}", - "Content-Type": "application/json", - } - - req_timeout = timeout if timeout is not None else float(settings.llm_timeout_seconds) - provider_name = "openai" if "openai" in base_url else "groq" - - try: - with httpx.Client(timeout=req_timeout) as client: - response = client.post(base_url, json=payload, headers=headers) - response.raise_for_status() - data = response.json() - except httpx.TimeoutException as exc: - logger.error("LLM provider %s request timed out after %s seconds", provider_name, req_timeout) - raise AIServiceError( - message=f"LLM request timed out after {req_timeout}s", - code="AI_TIMEOUT", - details={"provider": provider_name, "timeout_seconds": req_timeout}, - ) from exc - except httpx.HTTPStatusError as exc: - logger.error("LLM provider %s returned status %s: %s", provider_name, exc.response.status_code, exc.response.text) - raise AIServiceError( - message=f"LLM request failed with status {exc.response.status_code}", - code="AI_PROVIDER_ERROR", - details={"provider": provider_name, "status_code": exc.response.status_code}, - ) from exc - except Exception as exc: - logger.error("LLM provider %s connection or unexpected error: %s", provider_name, str(exc)) - raise AIServiceError( - message=f"LLM connection error: {str(exc)}", - code="AI_CONNECTION_ERROR", - details={"provider": provider_name}, - ) from exc - - try: - content = data["choices"][0]["message"]["content"] - except (KeyError, IndexError, TypeError) as exc: - raise RuntimeError(f"Unexpected LLM response format: {data}") from exc - - if not content: - raise RuntimeError("LLM returned empty content") - - return str(content) - - def _call_test(self, model: str, system_prompt: str, user_prompt: str) -> str: - response = self.test_provider.get_response( - endpoint="humanitarian", - request_data={ - "system_prompt": system_prompt, - "user_prompt": user_prompt, - }, - ) - return json.dumps(response, separators=(",", ":"), sort_keys=True) - - def _get_deterministic_response(self, model: str, system_prompt: str, user_prompt: str) -> str: - stable_response = { - "verdict": "credible", - "confidence": 0.74, - "summary": "Deterministic verification output for testing", - } - return json.dumps(stable_response, separators=(",", ":"), sort_keys=True) - - def _parse_json_response(self, content: str) -> Dict[str, Any]: - normalized = content.strip() - if normalized.startswith("```"): - normalized = normalized.strip("`") - if normalized.startswith("json"): - normalized = normalized[4:].strip() - parsed = json.loads(normalized) - if not isinstance(parsed, dict): - raise RuntimeError("LLM response must be a JSON object") - return parsed \ No newline at end of file +"""Humanitarian claim verification service with model/provider fallbacks.""" + +import json +import logging +from typing import Any, Dict, List, Optional +import time +import metrics + +import httpx + +from config import settings +from services.humanitarian_prompt import HumanitarianPromptEngine +from services.circuit_breaker import CircuitBreaker +from services.test_provider import TestProvider +from exceptions import AIServiceError + +logger = logging.getLogger(__name__) + + +class HumanitarianVerificationService: + """Runs humanitarian verification against configured LLM providers.""" + + def __init__(self): + self.prompt_engine = HumanitarianPromptEngine() + self.test_provider = TestProvider() + self.breakers = { + "openai": CircuitBreaker( + name="openai", + failure_threshold=settings.circuit_breaker_failure_threshold, + recovery_timeout=settings.circuit_breaker_recovery_timeout_seconds, + ), + "groq": CircuitBreaker( + name="groq", + failure_threshold=settings.circuit_breaker_failure_threshold, + recovery_timeout=settings.circuit_breaker_recovery_timeout_seconds, + ), + } + + def verify_claim( + self, + aid_claim: str, + supporting_evidence: Optional[List[str]] = None, + context_factors: Optional[Dict[str, Any]] = None, + provider_preference: str = "auto", + timeout: Optional[float] = None, + ) -> Dict[str, Any]: + start_time = time.time() + try: + evidence = supporting_evidence or [] + context = context_factors or {} + + primary_prompt = self.prompt_engine.build_primary_prompt( + aid_claim=aid_claim, + supporting_evidence=evidence, + context_factors=context, + ) + fallback_prompt = self.prompt_engine.build_fallback_prompt( + aid_claim=aid_claim, + supporting_evidence=evidence, + context_factors=context, + ) + + providers = self._provider_attempt_order(provider_preference) + if not providers: + raise RuntimeError("No LLM providers configured for humanitarian verification") + + errors: List[str] = [] + + for provider in providers: + breaker = self.breakers.get(provider) + if breaker and not breaker.allow_request(): + logger.warning("Circuit breaker is OPEN for provider=%s. Skipping.", provider) + errors.append(f"provider={provider}, error=Circuit breaker is OPEN") + continue + + model = self._get_model_for_provider(provider) + for prompt_variant, prompt in (("primary", primary_prompt), ("fallback", fallback_prompt)): + try: + logger.info( + "Attempting humanitarian verification with provider=%s model=%s prompt=%s", + provider, + model, + prompt_variant, + ) + raw_content = self._call_provider( + provider=provider, + model=model, + system_prompt=prompt["system"], + user_prompt=prompt["user"], + timeout=timeout, + ) + parsed = parse_verification_response(provider, raw_content) + if breaker: + breaker.record_success() + return { + "provider": provider, + "model": model, + "prompt_variant": prompt_variant, + "verification": parsed, + "raw_response": raw_content, + "stamp": { + "provider": provider, + "model": model, + "prompt_variant": prompt_variant, + } + } + except Exception as exc: + if breaker: + breaker.record_failure() + err = f"provider={provider}, model={model}, prompt={prompt_variant}, error={exc}" + errors.append(err) + logger.warning("Humanitarian verification attempt failed: %s", err) + + raise RuntimeError("All humanitarian verification attempts failed: " + " | ".join(errors)) + finally: + latency = time.time() - start_time + metrics.PIPELINE_STEP_LATENCY.labels(step_name='verify').observe(latency) + + def _provider_attempt_order(self, provider_preference: str) -> List[str]: + available: List[str] = [] + if settings.test_provider_mode: + available.append("test") + if settings.openai_api_key: + available.append("openai") + if settings.groq_api_key: + available.append("groq") + + preference = (provider_preference or "auto").lower() + if preference == "test" and settings.test_provider_mode: + return [preference] + if preference in ("openai", "groq", "test") and preference in available: + return [preference] + [provider for provider in available if provider != preference] + return available + + def _get_model_for_provider(self, provider: str) -> str: + if provider == "test": + return "test-provider/fixture" + if provider == "openai": + return settings.openai_model + if provider == "groq": + return settings.groq_model + raise ValueError(f"Unsupported provider: {provider}") + + def _call_provider( + self, + provider: str, + model: str, + system_prompt: str, + user_prompt: str, + timeout: Optional[float] = None, + ) -> str: + if provider == "test": + return self._call_test(model, system_prompt, user_prompt) + if provider == "openai": + return self._call_openai(model, system_prompt, user_prompt, timeout) + if provider == "groq": + return self._call_groq(model, system_prompt, user_prompt, timeout) + raise ValueError(f"Unsupported provider: {provider}") + + def _call_openai( + self, + model: str, + system_prompt: str, + user_prompt: str, + timeout: Optional[float] = None, + ) -> str: + if not settings.openai_api_key: + raise RuntimeError("OpenAI API key is not configured") + return self._call_chat_completion_api( + base_url="https://api.openai.com/v1/chat/completions", + api_key=settings.openai_api_key, + model=model, + system_prompt=system_prompt, + user_prompt=user_prompt, + timeout=timeout, + ) + + def _call_groq( + self, + model: str, + system_prompt: str, + user_prompt: str, + timeout: Optional[float] = None, + ) -> str: + if not settings.groq_api_key: + raise RuntimeError("Groq API key is not configured") + return self._call_chat_completion_api( + base_url="https://api.groq.com/openai/v1/chat/completions", + api_key=settings.groq_api_key, + model=model, + system_prompt=system_prompt, + user_prompt=user_prompt, + timeout=timeout, + ) + + def _call_chat_completion_api( + self, + base_url: str, + api_key: str, + model: str, + system_prompt: str, + user_prompt: str, + timeout: Optional[float] = None, + ) -> str: + if settings.ai_deterministic_mode: + logger.info("Deterministic AI mode enabled: returning stable response") + return self._get_deterministic_response(model, system_prompt, user_prompt) + + payload = { + "model": model, + "temperature": 0.1, + "messages": [ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": user_prompt}, + ], + } + headers = { + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + } + + req_timeout = timeout if timeout is not None else float(settings.llm_timeout_seconds) + provider_name = "openai" if "openai" in base_url else "groq" + + try: + with httpx.Client(timeout=req_timeout) as client: + response = client.post(base_url, json=payload, headers=headers) + response.raise_for_status() + data = response.json() + except httpx.TimeoutException as exc: + logger.error("LLM provider %s request timed out after %s seconds", provider_name, req_timeout) + raise AIServiceError( + message=f"LLM request timed out after {req_timeout}s", + code="AI_TIMEOUT", + details={"provider": provider_name, "timeout_seconds": req_timeout}, + ) from exc + except httpx.HTTPStatusError as exc: + logger.error("LLM provider %s returned status %s: %s", provider_name, exc.response.status_code, exc.response.text) + raise AIServiceError( + message=f"LLM request failed with status {exc.response.status_code}", + code="AI_PROVIDER_ERROR", + details={"provider": provider_name, "status_code": exc.response.status_code}, + ) from exc + except Exception as exc: + logger.error("LLM provider %s connection or unexpected error: %s", provider_name, str(exc)) + raise AIServiceError( + message=f"LLM connection error: {str(exc)}", + code="AI_CONNECTION_ERROR", + details={"provider": provider_name}, + ) from exc + + try: + content = data["choices"][0]["message"]["content"] + except (KeyError, IndexError, TypeError) as exc: + raise RuntimeError(f"Unexpected LLM response format: {data}") from exc + + if not content: + raise RuntimeError("LLM returned empty content") + + return str(content) + + def _call_test(self, model: str, system_prompt: str, user_prompt: str) -> str: + response = self.test_provider.get_response( + endpoint="humanitarian", + request_data={ + "system_prompt": system_prompt, + "user_prompt": user_prompt, + }, + ) + return json.dumps(response, separators=(",", ":"), sort_keys=True) + + def _get_deterministic_response(self, model: str, system_prompt: str, user_prompt: str) -> str: + stable_response = { + "verdict": "credible", + "confidence": 0.74, + "summary": "Deterministic verification output for testing", + } + return json.dumps(stable_response, separators=(",", ":"), sort_keys=True) + + def _parse_json_response(self, content: str) -> Dict[str, Any]: + return parse_verification_response("auto", content) + + +def parse_verification_response(provider_name: str, raw_content: str) -> Dict[str, Any]: + """Parses raw verification response, handling JSON markdown blocks and potential truncations.""" + normalized = raw_content.strip() + if normalized.startswith("```"): + normalized = normalized.strip("`") + if normalized.startswith("json"): + normalized = normalized[4:].strip() + + try: + parsed = json.loads(normalized) + if isinstance(parsed, dict): + return _normalize_verification_dict(parsed) + except json.JSONDecodeError: + pass + + # Recovery parsing in case of truncation + import re + verdict_match = re.search(r'"verdict"\s*:\s*"([^"]+)"', normalized) + confidence_match = re.search(r'"confidence"\s*:\s*([0-9.]+)', normalized) + summary_match = re.search(r'"summary"\s*:\s*"([^"]*)"', normalized) + + verdict = verdict_match.group(1) if verdict_match else "inconclusive" + confidence = float(confidence_match.group(1)) if confidence_match else 0.0 + summary = summary_match.group(1) if summary_match else "Truncated response parsed via recovery" + + if verdict not in ["credible", "partially_credible", "inconclusive", "not_credible"]: + verdict = "inconclusive" + + return { + "verdict": verdict, + "confidence": confidence, + "summary": summary, + "criteria_assessment": None, + "risk_flags": None, + "missing_information": None, + "recommended_next_steps": None, + } + + +def _normalize_verification_dict(parsed: Dict[str, Any]) -> Dict[str, Any]: + """Ensures a parsed dict strictly matches HumanitarianVerificationDetailsV2 structure.""" + verdict = parsed.get("verdict", "inconclusive") + if verdict not in ["credible", "partially_credible", "inconclusive", "not_credible"]: + verdict = "inconclusive" + + confidence = parsed.get("confidence", 0.0) + try: + confidence = float(confidence) + except (ValueError, TypeError): + confidence = 0.0 + + return { + "verdict": verdict, + "confidence": confidence, + "summary": str(parsed.get("summary", "")), + "criteria_assessment": parsed.get("criteria_assessment"), + "risk_flags": parsed.get("risk_flags"), + "missing_information": parsed.get("missing_information"), + "recommended_next_steps": parsed.get("recommended_next_steps"), + } \ No newline at end of file diff --git a/app/ai-service/tests/test_humanitarian_verification.py b/app/ai-service/tests/test_humanitarian_verification.py index 4a5a8e1e..fb366540 100644 --- a/app/ai-service/tests/test_humanitarian_verification.py +++ b/app/ai-service/tests/test_humanitarian_verification.py @@ -85,6 +85,10 @@ def test_verify_claim_returns_deterministic_response_when_enabled(self, monkeypa "confidence": 0.74, "summary": "Deterministic verification output for testing", "verdict": "credible", + "criteria_assessment": None, + "risk_flags": None, + "missing_information": None, + "recommended_next_steps": None, } def test_deterministic_verify_claim_outputs_remain_stable_across_runs(self, monkeypatch): @@ -220,3 +224,69 @@ def test_test_provider_does_not_require_api_keys(self, monkeypatch): assert result["provider"] == "test" assert result["verification"]["verdict"] in {"credible", "inconclusive", "not_credible"} + + def test_all_providers_shape_identity(self, monkeypatch): + from schemas.humanitarian_verification_v2 import HumanitarianVerificationResponseV2 + monkeypatch.setattr(settings, "test_provider_mode", True) + monkeypatch.setattr(settings, "openai_api_key", "mock-openai-key") + monkeypatch.setattr(settings, "groq_api_key", "mock-groq-key") + + openai_mock_response = '{"verdict": "credible", "confidence": 0.88, "summary": "mocked openai verification", "criteria_assessment": [], "risk_flags": [], "missing_information": [], "recommended_next_steps": []}' + groq_mock_response = '{"verdict": "not_credible", "confidence": 0.12, "summary": "mocked groq verification", "criteria_assessment": [], "risk_flags": [], "missing_information": [], "recommended_next_steps": []}' + + def fake_call_openai(model, system_prompt, user_prompt, timeout=None): + return openai_mock_response + + def fake_call_groq(model, system_prompt, user_prompt, timeout=None): + return groq_mock_response + + monkeypatch.setattr(self.service, "_call_openai", fake_call_openai) + monkeypatch.setattr(self.service, "_call_groq", fake_call_groq) + + # 1. Run test provider + res_test = self.service.verify_claim( + aid_claim="Food distribution completed in all sectors.", + supporting_evidence=["receipt"], + context_factors={}, + provider_preference="test" + ) + + # 2. Run openai provider + res_openai = self.service.verify_claim( + aid_claim="Food distribution completed in all sectors.", + supporting_evidence=["receipt"], + context_factors={}, + provider_preference="openai" + ) + + # 3. Run groq provider + res_groq = self.service.verify_claim( + aid_claim="Food distribution completed in all sectors.", + supporting_evidence=["receipt"], + context_factors={}, + provider_preference="groq" + ) + + for result in [res_test, res_openai, res_groq]: + v2_obj = HumanitarianVerificationResponseV2(success=True, **result) + assert v2_obj.success is True + assert v2_obj.provider is not None + assert v2_obj.model is not None + assert v2_obj.prompt_variant is not None + assert v2_obj.verification is not None + assert v2_obj.stamp is not None + assert v2_obj.stamp["provider"] == v2_obj.provider + assert v2_obj.stamp["model"] == v2_obj.model + assert v2_obj.stamp["prompt_variant"] == v2_obj.prompt_variant + + verif_dict = result["verification"] + assert set(verif_dict.keys()) == { + "verdict", + "confidence", + "summary", + "criteria_assessment", + "risk_flags", + "missing_information", + "recommended_next_steps" + } + diff --git a/app/ai-service/tests/test_schemas.py b/app/ai-service/tests/test_schemas.py index 0c5d3cfd..d92ff40e 100644 --- a/app/ai-service/tests/test_schemas.py +++ b/app/ai-service/tests/test_schemas.py @@ -66,3 +66,103 @@ def test_valid_error_detail(self): response = OCRResponse(success=False, error=error, processing_time_ms=0) assert response.error == error assert response.error["code"] == "test_error" + + +class TestSchemaModelVersionSnapshots: + """Pins model_version in expected output schema snapshots.""" + + def test_humanitarian_verification_response_snapshot(self): + from schemas.humanitarian import HumanitarianVerificationResponse + resp = HumanitarianVerificationResponse( + success=True, + provider="openai", + model="gpt-4o-mini", + prompt_variant="primary", + verification={"verdict": "credible", "confidence": 0.95}, + model_version="gpt-4o-mini" + ) + snapshot = resp.model_dump() + assert snapshot["model_version"] == "gpt-4o-mini" + assert snapshot["success"] is True + assert snapshot["provider"] == "openai" + + def test_humanitarian_verification_response_v2_snapshot(self): + from schemas.humanitarian_verification_v2 import HumanitarianVerificationResponseV2, HumanitarianVerificationDetailsV2 + resp = HumanitarianVerificationResponseV2( + success=True, + provider="openai", + model="gpt-4o-mini", + prompt_variant="primary", + verification=HumanitarianVerificationDetailsV2( + verdict="credible", + confidence=0.95, + summary="Clear test summary", + criteria_assessment=[ + {"criterion": "water_supply", "status": "met", "reason": "enough water"} + ], + risk_flags=[], + missing_information=[], + recommended_next_steps=[] + ), + model_version="gpt-4o-mini", + stamp={ + "provider": "openai", + "model": "gpt-4o-mini", + "prompt_variant": "primary" + } + ) + snapshot = resp.model_dump() + assert snapshot["model_version"] == "gpt-4o-mini" + assert snapshot["success"] is True + assert snapshot["provider"] == "openai" + assert snapshot["verification"]["verdict"] == "credible" + assert snapshot["verification"]["confidence"] == 0.95 + assert snapshot["stamp"]["provider"] == "openai" + assert snapshot["stamp"]["model"] == "gpt-4o-mini" + assert snapshot["stamp"]["prompt_variant"] == "primary" + + # Explicitly check json schema pinning + schema = HumanitarianVerificationResponseV2.model_json_schema() + assert "success" in schema["properties"] + assert "provider" in schema["properties"] + assert "model" in schema["properties"] + assert "prompt_variant" in schema["properties"] + assert "verification" in schema["properties"] + assert "stamp" in schema["properties"] + + # Details schema validation + details_schema = schema["$defs"]["HumanitarianVerificationDetailsV2"]["properties"] + assert "verdict" in details_schema + assert "confidence" in details_schema + assert "summary" in details_schema + + def test_anonymize_response_snapshot(self): + from schemas.anonymization import AnonymizeResponse, PIISummary + resp = AnonymizeResponse( + success=True, + anonymized_text="Hello [RECIPIENT_NAME]", + original_length=15, + pii_summary=PIISummary(names=1, locations=0, dates=0, total=1), + model_version="gpt-4o-mini" + ) + snapshot = resp.model_dump() + assert snapshot["model_version"] == "gpt-4o-mini" + assert snapshot["success"] is True + + def test_ocr_response_snapshot(self): + from schemas.ocr import OCRResponse, OCRData, OCRFieldResult + resp = OCRResponse( + success=True, + data=OCRData( + fields={"name": OCRFieldResult(value="John", confidence=0.9)}, + raw_text="Name: John", + processing_time_ms=120 + ), + processing_time_ms=120, + model_version="gpt-4o-mini" + ) + snapshot = resp.model_dump() + assert snapshot["model_version"] == "gpt-4o-mini" + assert snapshot["success"] is True + + diff --git a/app/ai-service/tests/test_versioned_routes.py b/app/ai-service/tests/test_versioned_routes.py index ce5898ed..cb2b05a9 100644 --- a/app/ai-service/tests/test_versioned_routes.py +++ b/app/ai-service/tests/test_versioned_routes.py @@ -173,7 +173,10 @@ def test_v1_ocr_processing_time_present(self, client): files={"image": ("img.png", buf.getvalue(), "image/png")}, ) assert response.status_code == 200 - assert "processing_time_ms" in response.json() + data = response.json() + assert "processing_time_ms" in data + assert data["model_version"] == "gpt-4o-mini" + # --------------------------------------------------------------------------- @@ -282,6 +285,8 @@ def test_v1_anonymize_success(self, following_client): data = response.json() assert data["success"] is True assert "anonymized_text" in data + assert data["model_version"] == "gpt-4o-mini" + def test_v1_anonymize_empty_text_returns_422(self, following_client): response = following_client.post("/v1/ai/anonymize", json={"text": ""}) @@ -325,6 +330,8 @@ def fake_verify( data = response.json() assert data["success"] is True assert data["verification"]["verdict"] == "credible" + assert data["model_version"] == "gpt-4o-mini" + def test_v1_humanitarian_verify_failure_path(self, following_client, monkeypatch): def fake_verify( diff --git a/app/backend/src/app.module.ts b/app/backend/src/app.module.ts index 3935b65e..68dd89c3 100644 --- a/app/backend/src/app.module.ts +++ b/app/backend/src/app.module.ts @@ -3,8 +3,7 @@ import { ConfigModule, ConfigService } from '@nestjs/config'; import { BullModule } from '@nestjs/bullmq'; import { APP_FILTER, APP_INTERCEPTOR } from '@nestjs/core'; import { ScheduleModule } from '@nestjs/schedule'; -import { existsSync } from 'node:fs'; -import { join } from 'node:path'; +import { loadEnv } from './common/utils/env-loader'; import { AppController } from './app.controller'; import { AppService } from './app.service'; @@ -50,16 +49,7 @@ import { SandboxModule } from './sandbox/sandbox.module'; imports: [ ConfigModule.forRoot({ isGlobal: true, - envFilePath: (() => { - const candidates = [ - join(__dirname, '..', '.env'), - join(process.cwd(), '.env'), - join(process.cwd(), 'app', 'backend', '.env'), - ]; - - const existing = candidates.filter(p => existsSync(p)); - return existing.length > 0 ? existing : candidates; - })(), + envFilePath: loadEnv(), }), BullModule.forRootAsync({ diff --git a/app/backend/src/common/utils/__tests__/env-loader.spec.ts b/app/backend/src/common/utils/__tests__/env-loader.spec.ts new file mode 100644 index 00000000..9d529502 --- /dev/null +++ b/app/backend/src/common/utils/__tests__/env-loader.spec.ts @@ -0,0 +1,120 @@ +import * as fs from 'node:fs'; +import { sep } from 'node:path'; +import { ConfigModule, ConfigService } from '@nestjs/config'; +import { Test } from '@nestjs/testing'; +import { loadEnv, getEnvCandidates } from '../env-loader'; + +// Mock fs module to control file existence and contents +jest.mock('node:fs', () => { + const original = jest.requireActual('node:fs'); + return { + ...original, + existsSync: jest.fn(), + readFileSync: jest.fn(), + }; +}); + +describe('Unified Env Loader', () => { + const originalEnv = { ...process.env }; + const mockExistsSync = fs.existsSync as jest.Mock; + const mockReadFileSync = fs.readFileSync as jest.Mock; + + beforeEach(() => { + jest.resetModules(); + jest.clearAllMocks(); + process.env = { ...originalEnv }; + }); + + afterAll(() => { + process.env = originalEnv; + }); + + it('should respect the canonical order of precedence when resolving env files', () => { + const candidates = getEnvCandidates(); + expect(candidates).toHaveLength(3); + // 1st candidate: process.cwd()/.env + expect(candidates[0]).toContain('.env'); + // 2nd candidate: process.cwd()/app/backend/.env + expect(candidates[1]).toContain(joinParts('app', 'backend', '.env')); + }); + + it('should override OS environment variables with dotenv variables (honest precedence)', () => { + const candidates = getEnvCandidates(); + const targetFile = candidates[0]; + + // Pre-set OS env var + process.env.TEST_VAR = 'os_value'; + + // Mock targetFile existence and content + mockExistsSync.mockImplementation((path: string) => path === targetFile); + mockReadFileSync.mockImplementation((path: string) => { + if (path === targetFile) { + return 'TEST_VAR=dotenv_value'; + } + return ''; + }); + + loadEnv(); + + // Verify dotenv value override OS env value + expect(process.env.TEST_VAR).toBe('dotenv_value'); + }); + + it('should ensure both direct loadEnv and ConfigModule call paths agree on final env state', async () => { + const candidates = getEnvCandidates(); + const rootEnvPath = candidates[0]; + const backendEnvPath = candidates[1]; + + // Set mock existence + mockExistsSync.mockImplementation((path: string) => { + return path === rootEnvPath || path === backendEnvPath; + }); + + // Mock contents: Root env should take precedence over backend env + mockReadFileSync.mockImplementation((path: string) => { + if (path === rootEnvPath) { + return 'COMMON_VAR=root_val\nROOT_ONLY=root_only'; + } + if (path === backendEnvPath) { + return 'COMMON_VAR=backend_val\nBACKEND_ONLY=backend_only'; + } + return ''; + }); + + // 1. Direct path (like main.ts) + const envFilePaths = loadEnv(); + expect(envFilePaths).toContain(rootEnvPath); + expect(envFilePaths).toContain(backendEnvPath); + + const directCommon = process.env.COMMON_VAR; + const directRootOnly = process.env.ROOT_ONLY; + const directBackendOnly = process.env.BACKEND_ONLY; + + // Clear process.env to test NestJS ConfigModule path cleanly + process.env = { ...originalEnv }; + + // 2. Nest ConfigModule path (like app.module.ts) + const moduleRef = await Test.createTestingModule({ + imports: [ + ConfigModule.forRoot({ + isGlobal: true, + envFilePath: envFilePaths, + }), + ], + }).compile(); + + const configService = moduleRef.get(ConfigService); + + // Assert agreement + expect(configService.get('COMMON_VAR')).toBe(directCommon); + expect(configService.get('ROOT_ONLY')).toBe(directRootOnly); + expect(configService.get('BACKEND_ONLY')).toBe(directBackendOnly); + + // Verify that the first candidate (rootEnvPath) successfully won over backendEnvPath + expect(configService.get('COMMON_VAR')).toBe('root_val'); + }); +}); + +function joinParts(...parts: string[]): string { + return parts.join(sep); +} diff --git a/app/backend/src/common/utils/env-loader.ts b/app/backend/src/common/utils/env-loader.ts new file mode 100644 index 00000000..a737de6d --- /dev/null +++ b/app/backend/src/common/utils/env-loader.ts @@ -0,0 +1,54 @@ +import { existsSync } from 'node:fs'; +import { join } from 'node:path'; +import { config as dotenvConfig } from 'dotenv'; + +/** + * Resolves the candidate .env paths in a unified order. + * Order of precedence (first one wins / first candidate takes priority): + * 1. process.cwd()/.env (Root .env) + * 2. process.cwd()/app/backend/.env (Backend .env) + * 3. __dirname-relative .env (equivalent to join(__dirname, '..', '.env') in main.ts / app.module.ts) + */ +export function getEnvCandidates(): string[] { + // If __dirname is inside src/common/utils (which it is for this file), + // we go up 3 levels to reach the parent of src/dist. + // Otherwise, we default to 1 level up. + const isNested = __dirname.includes(join('common', 'utils')) || __dirname.replace(/\\/g, '/').includes('common/utils'); + const relativeParent = isNested ? join(__dirname, '..', '..', '..') : join(__dirname, '..'); + + return [ + join(process.cwd(), '.env'), + join(process.cwd(), 'app', 'backend', '.env'), + join(relativeParent, '.env'), + ]; +} + +/** + * Loads environment variables from the candidate .env files. + * Precedence Rule: + * - dotenv variables ALWAYS win over existing OS environment variables (override: true). + * - The first existing candidate file in the list takes highest precedence. + * + * Both main.ts and app.module.ts call this helper. + * Returns the candidate files list to be used by NestJS ConfigModule. + */ +export function loadEnv(): string[] { + const candidates = getEnvCandidates(); + const existing = candidates.filter(p => existsSync(p)); + + const pathsToLoad = existing.length > 0 ? existing : candidates; + + // Precedence helper: + // To ensure that the first candidate in the list takes precedence, + // we load them in reverse order using dotenv with `override: true`. + // This way, the first candidate is loaded last and overrides any keys loaded by others. + // Also, since we use `override: true`, dotenv variables will override existing OS environment variables. + for (let i = pathsToLoad.length - 1; i >= 0; i--) { + const p = pathsToLoad[i]; + if (existsSync(p)) { + dotenvConfig({ path: p, override: true }); + } + } + + return pathsToLoad; +} diff --git a/app/backend/src/main.ts b/app/backend/src/main.ts index 9fa25572..d4ecbd0a 100644 --- a/app/backend/src/main.ts +++ b/app/backend/src/main.ts @@ -6,9 +6,7 @@ import { AppModule } from './app.module'; import { buildSwaggerConfig } from './swagger.config'; import { LoggerService } from './logger/logger.service'; import { LoggingInterceptor } from './interceptors/logging.interceptor'; -import { config as loadEnv } from 'dotenv'; -import { existsSync } from 'node:fs'; -import { join } from 'node:path'; +import { loadEnv } from './common/utils/env-loader'; import compression from 'compression'; import { RequestIdInterceptor } from './common/interceptors/request-id.interceptor'; @@ -21,16 +19,7 @@ import { async function bootstrap() { // Load environment variables - const candidates = [ - join(process.cwd(), '.env'), - join(process.cwd(), 'app', 'backend', '.env'), - join(__dirname, '..', '.env'), - ]; - - const envPath = candidates.find(p => existsSync(p)); - if (envPath) { - loadEnv({ path: envPath }); - } + loadEnv(); const app = await NestFactory.create(AppModule);