diff --git a/app/ai-service/.env.example b/app/ai-service/.env.example index 6cafe3f..c00d92f 100644 --- a/app/ai-service/.env.example +++ b/app/ai-service/.env.example @@ -4,6 +4,9 @@ # API Keys (at least one is required for AI features) OPENAI_API_KEY=your_openai_api_key_here GROQ_API_KEY=your_groq_api_key_here +OPENAI_MODEL=gpt-4o-mini +GROQ_MODEL=llama-3.3-70b-versatile +LLM_TIMEOUT_SECONDS=30 # Application Settings APP_ENV=development diff --git a/app/ai-service/README.md b/app/ai-service/README.md index 68bef29..1cb1a55 100644 --- a/app/ai-service/README.md +++ b/app/ai-service/README.md @@ -63,6 +63,40 @@ Response body: ### OCR Processing - **POST** `/ai/ocr` - Identity document OCR with field extraction +### Humanitarian Verification +- **POST** `/ai/humanitarian/verify` - Standardized humanitarian claim verification (Sphere criteria + context factors + provider fallback) + +Request body: + +```json +{ + "aid_claim": "Relief teams delivered hygiene kits to all registered households in Sector B.", + "supporting_evidence": ["Distribution list #B-17", "Field monitor report"], + "context_factors": { + "security_status": "stable", + "weather": "heavy_rain", + "displacement_level": "moderate" + }, + "provider_preference": "auto" +} +``` + +Response body: + +```json +{ + "success": true, + "provider": "openai", + "model": "gpt-4o-mini", + "prompt_variant": "primary", + "verification": { + "verdict": "credible", + "confidence": 0.86, + "summary": "Evidence aligns with claim across key criteria" + } +} +``` + ```bash curl -X POST "http://localhost:8000/ai/ocr" -F "image=@document.jpg" ``` diff --git a/app/ai-service/config.py b/app/ai-service/config.py index febe2c7..c63ae9a 100644 --- a/app/ai-service/config.py +++ b/app/ai-service/config.py @@ -3,7 +3,7 @@ Handles environment variables and API key management """ -from pydantic_settings import BaseSettings +from pydantic_settings import BaseSettings, SettingsConfigDict from typing import Optional import logging @@ -17,6 +17,9 @@ class Settings(BaseSettings): Environment Variables: OPENAI_API_KEY: OpenAI API key for AI model access GROQ_API_KEY: Groq API key for AI model access (alternative to OpenAI) + OPENAI_MODEL: Default OpenAI model for humanitarian verification + GROQ_MODEL: Default Groq model for humanitarian verification + LLM_TIMEOUT_SECONDS: Timeout for LLM API requests APP_ENV: Application environment (development, staging, production) LOG_LEVEL: Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL) HOST: Server host (default: 0.0.0.0) @@ -30,6 +33,9 @@ class Settings(BaseSettings): # API Keys openai_api_key: Optional[str] = None groq_api_key: Optional[str] = None + openai_model: str = "gpt-4o-mini" + groq_model: str = "llama-3.3-70b-versatile" + llm_timeout_seconds: int = 30 # Application settings app_env: str = "development" @@ -47,10 +53,11 @@ class Settings(BaseSettings): proof_of_life_confidence_threshold: float = 0.65 proof_of_life_min_face_size: int = 80 - class Config: - env_file = ".env" - env_file_encoding = "utf-8" - case_sensitive = False + model_config = SettingsConfigDict( + env_file=".env", + env_file_encoding="utf-8", + case_sensitive=False, + ) def validate_api_keys(self) -> bool: """ diff --git a/app/ai-service/main.py b/app/ai-service/main.py index f9f0a3e..464953d 100644 --- a/app/ai-service/main.py +++ b/app/ai-service/main.py @@ -21,6 +21,11 @@ from config import settings import tasks from proof_of_life import ProofOfLifeAnalyzer, ProofOfLifeConfig +from schemas.humanitarian import ( + HumanitarianVerificationRequest, + HumanitarianVerificationResponse, +) +from services.humanitarian_verification import HumanitarianVerificationService limiter = Limiter(key_func=get_remote_address) @@ -60,6 +65,7 @@ async def lifespan(app: FastAPI): min_face_size=settings.proof_of_life_min_face_size, ) ) +humanitarian_verification_service = HumanitarianVerificationService() # Request/Response models @@ -229,6 +235,24 @@ async def analyze_proof_of_life(request: ProofOfLifeRequest): ) +@app.post("/ai/humanitarian/verify", response_model=HumanitarianVerificationResponse) +async def verify_humanitarian_claim(request: HumanitarianVerificationRequest): + """Verify an aid claim against standardized humanitarian criteria.""" + logger.info("Processing humanitarian verification request") + + try: + result = humanitarian_verification_service.verify_claim( + aid_claim=request.aid_claim, + supporting_evidence=request.supporting_evidence, + context_factors=request.context_factors, + provider_preference=request.provider_preference, + ) + return HumanitarianVerificationResponse(success=True, **result) + except Exception as e: + logger.error("Humanitarian verification failed: %s", str(e), exc_info=True) + return HumanitarianVerificationResponse(success=False, error=str(e)) + + @app.get("/ai/status/{task_id}", response_model=TaskStatusResponse) async def get_task_status(task_id: str): """ diff --git a/app/ai-service/pytest.ini b/app/ai-service/pytest.ini new file mode 100644 index 0000000..c9b49aa --- /dev/null +++ b/app/ai-service/pytest.ini @@ -0,0 +1,3 @@ +[pytest] +filterwarnings = + ignore:.*pkgutil.find_loader.*:DeprecationWarning:pytesseract.pytesseract diff --git a/app/ai-service/schemas/humanitarian.py b/app/ai-service/schemas/humanitarian.py new file mode 100644 index 0000000..0ce1706 --- /dev/null +++ b/app/ai-service/schemas/humanitarian.py @@ -0,0 +1,19 @@ +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", "openai", "groq"] = "auto" + + +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 diff --git a/app/ai-service/services/humanitarian_prompt.py b/app/ai-service/services/humanitarian_prompt.py new file mode 100644 index 0000000..3c1af88 --- /dev/null +++ b/app/ai-service/services/humanitarian_prompt.py @@ -0,0 +1,133 @@ +""" +Prompt templating for humanitarian aid claim verification. + +This module standardizes prompt construction across providers and model families +(OpenAI/Groq-compatible APIs) to keep scoring objective and reproducible. +""" + +from typing import Any, Dict, List + + +SPHERE_HANDBOOK_CRITERIA: Dict[str, List[str]] = { + "water_supply_sanitation_hygiene": [ + "Minimum daily water access is sufficient and equitable.", + "Sanitation facilities are safe, accessible, and culturally appropriate.", + "Hygiene support (soap, menstrual hygiene, handwashing) is consistently available.", + ], + "food_security_nutrition": [ + "Food assistance is adequate in quantity, quality, and nutritional value.", + "Distribution is regular, impartial, and reaches vulnerable groups.", + "Nutrition-sensitive support addresses children, pregnant, and lactating women.", + ], + "shelter_settlement": [ + "Shelter provides safety, privacy, weather protection, and dignity.", + "Settlement planning reduces overcrowding and health risks.", + "Shelter materials and design align with local context and inclusion needs.", + ], + "health": [ + "Essential health services are accessible without discrimination.", + "Disease prevention and outbreak readiness are in place.", + "Referral pathways and continuity of care are functioning.", + ], + "protection_inclusion_accountability": [ + "Assistance is impartial and minimizes protection risks.", + "Affected people can provide feedback and raise complaints safely.", + "Data and decision-making include age, gender, disability, and risk context.", + ], +} + + +class HumanitarianPromptEngine: + """Builds standardized humanitarian verification prompts.""" + + def build_primary_prompt( + self, + aid_claim: str, + supporting_evidence: List[str], + context_factors: Dict[str, Any], + ) -> Dict[str, str]: + criteria_text = self._format_sphere_criteria() + evidence_text = self._format_evidence(supporting_evidence) + context_text = self._format_context_factors(context_factors) + + system_prompt = ( + "You are an objective humanitarian verification analyst. " + "Evaluate aid claims only from provided evidence and context. " + "Apply a Humanitarian Standard grounded in Sphere criteria. " + "Do not infer facts that are not explicitly present. " + "Return valid JSON only." + ) + + user_prompt = ( + "Humanitarian Standard Verification Task\n\n" + "Assess whether the aid claim is credible, partially credible, inconclusive, or not credible. " + "Your analysis must map to Sphere Handbook criteria and explain uncertainty.\n\n" + f"Sphere Criteria:\n{criteria_text}\n\n" + f"Aid Claim:\n{aid_claim}\n\n" + f"Supporting Evidence:\n{evidence_text}\n\n" + f"Context Factors (from backend):\n{context_text}\n\n" + "Output JSON schema exactly:\n" + "{\n" + " \"verdict\": \"credible|partially_credible|inconclusive|not_credible\",\n" + " \"confidence\": 0.0,\n" + " \"summary\": \"short neutral summary\",\n" + " \"criteria_assessment\": [\n" + " {\"criterion\": \"string\", \"status\": \"met|partially_met|not_met|unknown\", \"reason\": \"string\"}\n" + " ],\n" + " \"risk_flags\": [\"string\"],\n" + " \"missing_information\": [\"string\"],\n" + " \"recommended_next_steps\": [\"string\"]\n" + "}" + ) + + return {"system": system_prompt, "user": user_prompt} + + def build_fallback_prompt( + self, + aid_claim: str, + supporting_evidence: List[str], + context_factors: Dict[str, Any], + ) -> Dict[str, str]: + evidence_text = self._format_evidence(supporting_evidence) + context_text = self._format_context_factors(context_factors) + + system_prompt = ( + "You verify humanitarian aid claims conservatively. " + "Use only supplied inputs. Return strict JSON only." + ) + + user_prompt = ( + "Fallback Humanitarian Verification\n\n" + f"Claim: {aid_claim}\n" + f"Evidence: {evidence_text}\n" + f"Context: {context_text}\n\n" + "Respond with JSON only:\n" + "{\"verdict\":\"credible|partially_credible|inconclusive|not_credible\"," + "\"confidence\":0.0,\"summary\":\"\"," + "\"risk_flags\":[],\"missing_information\":[],\"recommended_next_steps\":[]}" + ) + + return {"system": system_prompt, "user": user_prompt} + + def _format_sphere_criteria(self) -> str: + lines: List[str] = [] + for section, items in SPHERE_HANDBOOK_CRITERIA.items(): + lines.append(f"- {section}:") + for item in items: + lines.append(f" * {item}") + return "\n".join(lines) + + def _format_evidence(self, supporting_evidence: List[str]) -> str: + if not supporting_evidence: + return "- No supporting evidence provided" + return "\n".join(f"- {entry}" for entry in supporting_evidence) + + def _format_context_factors(self, context_factors: Dict[str, Any]) -> str: + if not context_factors: + return "- No context factors provided" + + lines: List[str] = [] + for key in sorted(context_factors.keys()): + value = context_factors[key] + lines.append(f"- {key}: {value}") + return "\n".join(lines) diff --git a/app/ai-service/services/humanitarian_verification.py b/app/ai-service/services/humanitarian_verification.py new file mode 100644 index 0000000..f0fe04d --- /dev/null +++ b/app/ai-service/services/humanitarian_verification.py @@ -0,0 +1,179 @@ +"""Humanitarian claim verification service with model/provider fallbacks.""" + +import json +import logging +from typing import Any, Dict, List, Optional + +import httpx + +from config import settings +from services.humanitarian_prompt import HumanitarianPromptEngine + +logger = logging.getLogger(__name__) + + +class HumanitarianVerificationService: + """Runs humanitarian verification against configured LLM providers.""" + + def __init__(self): + self.prompt_engine = HumanitarianPromptEngine() + + def verify_claim( + self, + aid_claim: str, + supporting_evidence: Optional[List[str]] = None, + context_factors: Optional[Dict[str, Any]] = None, + provider_preference: str = "auto", + ) -> Dict[str, Any]: + 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: + 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"], + ) + parsed = self._parse_json_response(raw_content) + return { + "provider": provider, + "model": model, + "prompt_variant": prompt_variant, + "verification": parsed, + "raw_response": raw_content, + } + except Exception as exc: + 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)) + + def _provider_attempt_order(self, provider_preference: str) -> List[str]: + available: List[str] = [] + if settings.openai_api_key: + available.append("openai") + if settings.groq_api_key: + available.append("groq") + + preference = (provider_preference or "auto").lower() + if preference in ("openai", "groq") 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 == "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) -> str: + if provider == "openai": + return self._call_openai(model, system_prompt, user_prompt) + if provider == "groq": + return self._call_groq(model, system_prompt, user_prompt) + raise ValueError(f"Unsupported provider: {provider}") + + def _call_openai(self, model: str, system_prompt: str, user_prompt: str) -> 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, + ) + + def _call_groq(self, model: str, system_prompt: str, user_prompt: str) -> 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, + ) + + def _call_chat_completion_api( + self, + base_url: str, + api_key: str, + model: str, + system_prompt: str, + user_prompt: str, + ) -> str: + 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", + } + + timeout = float(settings.llm_timeout_seconds) + + with httpx.Client(timeout=timeout) as client: + response = client.post(base_url, json=payload, headers=headers) + response.raise_for_status() + data = response.json() + + 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 _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 diff --git a/app/ai-service/tasks.py b/app/ai-service/tasks.py index 86df0f6..bd77225 100644 --- a/app/ai-service/tasks.py +++ b/app/ai-service/tasks.py @@ -13,33 +13,65 @@ import metrics from config import settings +from services.humanitarian_verification import HumanitarianVerificationService # Configure logging logger = logging.getLogger(__name__) -# Initialize Celery app -celery_app = Celery( - 'soter_ai_service', - broker=settings.redis_url, - backend=settings.redis_url, - include=['tasks'] -) +# Lazy Celery app initialization - defers actual connection until needed +celery_app = None -# Celery configuration -celery_app.conf.update( - task_serializer='json', - accept_content=['json'], - result_serializer='json', - timezone='UTC', - enable_utc=True, - task_track_started=True, - task_time_limit=3600, # 1 hour max - task_soft_time_limit=1800, # 30 minutes soft limit - result_expires=86400, # Results expire after 24 hours -) +def get_celery_app() -> Celery: + """ + Get or initialize the Celery app. + Uses lazy initialization to avoid connection errors during startup. + """ + global celery_app + if celery_app is None: + try: + celery_app = Celery( + 'soter_ai_service', + broker=settings.redis_url, + backend=settings.redis_url, + include=['tasks'] + ) + + # Celery configuration + celery_app.conf.update( + task_serializer='json', + accept_content=['json'], + result_serializer='json', + timezone='UTC', + enable_utc=True, + task_track_started=True, + task_time_limit=3600, # 1 hour max + task_soft_time_limit=1800, # 30 minutes soft limit + result_expires=86400, # Results expire after 24 hours + ) + except Exception as e: + logger.warning(f"Failed to initialize Celery: {e}. Task processing disabled.") + # Return a dummy app that won't crash + celery_app = Celery('soter_ai_service') + + return celery_app + + +def get_process_heavy_inference_task(): + """ + Get the lazily-registered process_heavy_inference task. + This allows the task to be registered only when Celery is actually available. + """ + app = get_celery_app() + # Define and register the task with the app + @app.task(bind=True, name='process_heavy_inference') + def process_heavy_inference_task(self, task_id: str, payload: Dict[str, Any]) -> Dict[str, Any]: + return process_heavy_inference_impl(self, task_id, payload) + + return process_heavy_inference_task # Task status storage (in production, use Redis with proper TTL) task_results: Dict[str, Dict[str, Any]] = {} +humanitarian_verification_service = HumanitarianVerificationService() def update_task_status( @@ -112,8 +144,7 @@ def send_notification(): logger.error(f"Error setting up webhook notification: {e}") -@celery_app.task(bind=True, name='process_heavy_inference') -def process_heavy_inference(self, task_id: str, payload: Dict[str, Any]) -> Dict[str, Any]: +def process_heavy_inference_impl(self, task_id: str, payload: Dict[str, Any]) -> Dict[str, Any]: """ Process heavy AI inference tasks in background @@ -145,6 +176,8 @@ def process_heavy_inference(self, task_id: str, payload: Dict[str, Any]) -> Dict result = _process_image_analysis(payload) elif task_type == 'model_inference': result = _process_model_inference(payload) + elif task_type == 'humanitarian_verification': + result = _process_humanitarian_verification(payload) elif task_type == 'batch_processing': result = _process_batch(payload) else: @@ -284,6 +317,27 @@ def _process_default_inference(payload: Dict[str, Any]) -> Dict[str, Any]: } +def _process_humanitarian_verification(payload: Dict[str, Any]) -> Dict[str, Any]: + """Process humanitarian claim verification using standardized prompts.""" + data = payload.get('data', {}) + aid_claim = data.get('aid_claim') + if not aid_claim: + raise ValueError("'aid_claim' is required for humanitarian_verification tasks") + + verification = humanitarian_verification_service.verify_claim( + aid_claim=aid_claim, + supporting_evidence=data.get('supporting_evidence', []), + context_factors=data.get('context_factors', {}), + provider_preference=data.get('provider_preference', 'auto'), + ) + + return { + 'type': 'humanitarian_verification', + 'status': 'success', + 'result': verification, + } + + def get_task_status(task_id: str) -> Dict[str, Any]: """ Get the status of a background task @@ -346,11 +400,17 @@ def create_task(task_type: str, payload: Dict[str, Any]) -> str: # Initialize task status update_task_status(task_id, 'pending') - # Queue the task - process_heavy_inference.apply_async( - args=[task_id, {**payload, 'type': task_type}], - task_id=task_id - ) + try: + # Queue the task using the lazy-registered task + task = get_process_heavy_inference_task() + task.apply_async( + args=[task_id, {**payload, 'type': task_type}], + task_id=task_id + ) + except Exception as e: + logger.error(f"Failed to queue task {task_id}: {e}. Redis may not be available.") + update_task_status(task_id, 'failed', error=str(e)) + raise logger.info(f"Created task {task_id} of type {task_type}") diff --git a/app/ai-service/test_main.py b/app/ai-service/test_main.py index b31d08f..1d0a5c5 100644 --- a/app/ai-service/test_main.py +++ b/app/ai-service/test_main.py @@ -144,3 +144,62 @@ def test_proof_of_life_threshold_validation(client): }, ) assert response.status_code == 422 + + +def test_humanitarian_verification_success(client, monkeypatch): + """Test successful humanitarian verification response contract.""" + + def fake_verify_claim(aid_claim, supporting_evidence=None, context_factors=None, provider_preference="auto"): + return { + "provider": "openai", + "model": "gpt-4o-mini", + "prompt_variant": "primary", + "verification": { + "verdict": "credible", + "confidence": 0.86, + "summary": "Evidence aligns with key distribution records.", + }, + "raw_response": "{}", + } + + monkeypatch.setattr(main.humanitarian_verification_service, "verify_claim", fake_verify_claim) + + response = client.post( + "/ai/humanitarian/verify", + json={ + "aid_claim": "Relief teams delivered hygiene kits to all registered households in Sector B.", + "supporting_evidence": ["Distribution list #B-17"], + "context_factors": {"security_status": "stable"}, + "provider_preference": "auto", + }, + ) + + assert response.status_code == 200 + data = response.json() + assert data["success"] is True + assert data["provider"] == "openai" + assert data["verification"]["verdict"] == "credible" + + +def test_humanitarian_verification_failure(client, monkeypatch): + """Test humanitarian verification failure path.""" + + def fake_verify_claim(aid_claim, supporting_evidence=None, context_factors=None, provider_preference="auto"): + raise RuntimeError("all providers unavailable") + + monkeypatch.setattr(main.humanitarian_verification_service, "verify_claim", fake_verify_claim) + + response = client.post( + "/ai/humanitarian/verify", + json={ + "aid_claim": "Temporary clinics are fully operational in all camps.", + "supporting_evidence": [], + "context_factors": {}, + "provider_preference": "auto", + }, + ) + + assert response.status_code == 200 + data = response.json() + assert data["success"] is False + assert "all providers unavailable" in data["error"] diff --git a/app/ai-service/test_setup.py b/app/ai-service/test_setup.py index 0e9bd14..310a75d 100644 --- a/app/ai-service/test_setup.py +++ b/app/ai-service/test_setup.py @@ -7,8 +7,8 @@ import subprocess -def test_imports(): - """Test if all required packages can be imported""" +def _check_imports() -> bool: + """Check if all required packages can be imported.""" print("Testing imports...") try: @@ -49,8 +49,13 @@ def test_imports(): return True -def test_config(): - """Test if configuration loads properly""" +def test_imports(): + """Test if all required packages can be imported.""" + assert _check_imports() + + +def _check_config() -> bool: + """Check if configuration loads properly.""" print("\nTesting configuration...") try: @@ -74,8 +79,13 @@ def test_config(): return False -def test_app(): - """Test if the FastAPI app can be instantiated""" +def test_config(): + """Test if configuration loads properly.""" + assert _check_config() + + +def _check_app() -> bool: + """Check if the FastAPI app can be instantiated.""" print("\nTesting application...") try: @@ -96,6 +106,11 @@ def test_app(): return False +def test_app(): + """Test if the FastAPI app can be instantiated.""" + assert _check_app() + + def main(): """Run all tests""" print("=" * 60) @@ -105,13 +120,13 @@ def main(): results = [] # Test 1: Imports - results.append(("Imports", test_imports())) + results.append(("Imports", _check_imports())) # Test 2: Configuration - results.append(("Configuration", test_config())) + results.append(("Configuration", _check_config())) # Test 3: Application - results.append(("Application", test_app())) + results.append(("Application", _check_app())) # Summary print("\n" + "=" * 60) diff --git a/app/ai-service/tests/test_humanitarian_prompt.py b/app/ai-service/tests/test_humanitarian_prompt.py new file mode 100644 index 0000000..8c260a9 --- /dev/null +++ b/app/ai-service/tests/test_humanitarian_prompt.py @@ -0,0 +1,39 @@ +from services.humanitarian_prompt import HumanitarianPromptEngine + + +class TestHumanitarianPromptEngine: + def setup_method(self): + self.engine = HumanitarianPromptEngine() + + def test_primary_prompt_includes_sphere_criteria(self): + prompt = self.engine.build_primary_prompt( + aid_claim="Community reports potable water deliveries are insufficient.", + supporting_evidence=["Field report #22", "Distribution logs"], + context_factors={"region": "north", "season": "dry"}, + ) + + assert "Sphere Criteria" in prompt["user"] + assert "water_supply_sanitation_hygiene" in prompt["user"] + assert "food_security_nutrition" in prompt["user"] + + def test_primary_prompt_includes_context_factors(self): + prompt = self.engine.build_primary_prompt( + aid_claim="Temporary shelter distribution completed.", + supporting_evidence=[], + context_factors={"security_level": "high_risk", "displacement_status": "ongoing"}, + ) + + assert "Context Factors" in prompt["user"] + assert "security_level: high_risk" in prompt["user"] + assert "displacement_status: ongoing" in prompt["user"] + + def test_fallback_prompt_is_compact_and_structured(self): + prompt = self.engine.build_fallback_prompt( + aid_claim="Clinic stockout has been resolved.", + supporting_evidence=["Health cluster update"], + context_factors={"district": "A1"}, + ) + + assert "Fallback Humanitarian Verification" in prompt["user"] + assert "Respond with JSON only" in prompt["user"] + assert "verdict" in prompt["user"] diff --git a/app/ai-service/tests/test_humanitarian_verification.py b/app/ai-service/tests/test_humanitarian_verification.py new file mode 100644 index 0000000..0317fce --- /dev/null +++ b/app/ai-service/tests/test_humanitarian_verification.py @@ -0,0 +1,56 @@ +import pytest + +from services.humanitarian_verification import HumanitarianVerificationService + + +class TestHumanitarianVerificationService: + def setup_method(self): + self.service = HumanitarianVerificationService() + + def test_verify_claim_uses_fallback_prompt_after_primary_failure(self, monkeypatch): + calls = [] + + def fake_attempt_order(provider_preference): + return ["openai"] + + def fake_model(provider): + return "test-model" + + def fake_call_provider(provider, model, system_prompt, user_prompt): + calls.append((provider, model, system_prompt, user_prompt)) + if len(calls) == 1: + raise RuntimeError("primary model failure") + return '{"verdict":"inconclusive","confidence":0.4,"summary":"insufficient evidence"}' + + monkeypatch.setattr(self.service, "_provider_attempt_order", fake_attempt_order) + monkeypatch.setattr(self.service, "_get_model_for_provider", fake_model) + monkeypatch.setattr(self.service, "_call_provider", fake_call_provider) + + result = self.service.verify_claim( + aid_claim="Aid package reached all households.", + supporting_evidence=["monitoring sheet"], + context_factors={"weather": "flooding"}, + provider_preference="openai", + ) + + assert result["prompt_variant"] == "fallback" + assert result["provider"] == "openai" + assert result["verification"]["verdict"] == "inconclusive" + assert len(calls) == 2 + + def test_verify_claim_fails_when_no_provider_configured(self, monkeypatch): + monkeypatch.setattr(self.service, "_provider_attempt_order", lambda provider_preference: []) + + with pytest.raises(RuntimeError): + self.service.verify_claim( + aid_claim="Food distribution completed.", + supporting_evidence=[], + context_factors={}, + ) + + def test_parse_json_response_supports_markdown_block(self): + content = "```json\n{\"verdict\":\"credible\",\"confidence\":0.9}\n```" + parsed = self.service._parse_json_response(content) + + assert parsed["verdict"] == "credible" + assert parsed["confidence"] == 0.9 diff --git a/app/ai-service/tests/test_ocr.py b/app/ai-service/tests/test_ocr.py index 8c8c2c0..2ba15c1 100644 --- a/app/ai-service/tests/test_ocr.py +++ b/app/ai-service/tests/test_ocr.py @@ -76,9 +76,17 @@ class TestOCRService: def setup_method(self): self.ocr = OCRService() - def test_process_image_returns_result(self): + def test_process_image_returns_result(self, monkeypatch): from PIL import Image + def fake_run_tesseract(_image): + return { + "text": ["Name:", "John", "Doe", "ID", "AB123456"], + "conf": [90, 92, 91, 88, 95], + } + + monkeypatch.setattr(self.ocr, "_run_tesseract", fake_run_tesseract) + img = Image.new("RGB", (200, 100), color="white") result = self.ocr.process_image(img) assert isinstance(result, OCRResult)