diff --git a/docker-compose.yml b/docker-compose.yml index 6ff7c3f57..a9a8faf89 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -33,6 +33,8 @@ x-server: &base_server_setup CACHE_TEST_REDIS_URL: ${CACHE_TEST_REDIS_URL:-redis://redis:6379/11} CACHE_MIDDLEWARE_SECONDS: ${CACHE_MIDDLEWARE_SECONDS:-600} ELASTIC_SEARCH_HOST: ${ELASTIC_SEARCH_HOST:-elasticsearch://elasticsearch:9200} + # LLM + USE_DUMMY_LLM_CLIENT: ${USE_DUMMY_LLM_CLIENT:-true} # Appeal API APPEALS_USER: ${APPEALS_USER:-} APPEALS_PASS: ${APPEALS_PASS:-} diff --git a/dref/admin.py b/dref/admin.py index e5849fb98..390b4c4d9 100644 --- a/dref/admin.py +++ b/dref/admin.py @@ -330,7 +330,7 @@ class ProposedActionAdmin(ReadOnlyMixin, admin.ModelAdmin): @admin.register(DrefSummary) class DrefSummaryAdmin(TranslationAdmin, admin.ModelAdmin): - list_display = ("dref", "status", "source_model_name", "source_id", "created_at", "updated_at") + list_display = ("dref", "status", "source", "source_id", "created_at", "updated_at") list_filter = ("status",) search_fields = ("dref__title", "dref__appeal_code") readonly_fields = ("source_hash", "created_at", "updated_at") @@ -351,15 +351,10 @@ class DrefSummaryAdmin(TranslationAdmin, admin.ModelAdmin): @admin.action(description="Regenerate summary for selected DREFs") def regenerate_summary(self, request, queryset): - """Re-trigger summary generation, replaying the source the summary was last built from.""" + """Re-trigger summary generation from the DREF's current latest approved source.""" + queryset.filter(status=DrefSummary.SummaryStatus.PROCESSING).update(status=DrefSummary.SummaryStatus.FAILED) for summary in queryset: - generate_dref_summary.delay( - # Fall back to the Dref itself for rows generated before the - # source was tracked. - source_model_name=summary.source_model_name or DrefSummary.SourceModel.DREF, - source_id=summary.source_id or summary.dref_id, - overwrite=True, - ) + generate_dref_summary.delay(dref_id=summary.dref_id, overwrite=True) self.message_user( request, f"Queued summary regeneration for {queryset.count()} DREF(s).", diff --git a/dref/factories/dref.py b/dref/factories/dref.py index 4ebb6d509..4cc61d27a 100644 --- a/dref/factories/dref.py +++ b/dref/factories/dref.py @@ -207,7 +207,7 @@ class Meta: dref = factory.SubFactory(DrefFactory) source_hash = factory.Sequence(lambda n: f"{n:064d}") - source_model_name = DrefSummary.SourceModel.DREF + source = DrefSummary.SourceModel.DREF source_id = factory.SelfAttribute("dref.id") status = DrefSummary.SummaryStatus.SUCCESS situational_overview = fuzzy.FuzzyText(length=100) diff --git a/dref/migrations/0090_drefsummary.py b/dref/migrations/0090_drefsummary.py index 401009b54..6d9b171ef 100644 --- a/dref/migrations/0090_drefsummary.py +++ b/dref/migrations/0090_drefsummary.py @@ -17,7 +17,7 @@ class Migration(migrations.Migration): ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), ('created_at', models.DateTimeField(auto_now_add=True)), ('updated_at', models.DateTimeField(auto_now=True)), - ('source_model_name', models.IntegerField(choices=[(100, 'Dref'), (200, 'Dref Operational Update'), (300, 'Dref Final Report')], verbose_name='source model name')), + ('source', models.IntegerField(choices=[(100, 'Dref'), (200, 'Dref Operational Update'), (300, 'Dref Final Report')], help_text='The model this summary was generated from.', verbose_name='source')), ('source_id', models.PositiveBigIntegerField(verbose_name='source id')), ('source_hash', models.CharField(max_length=64, unique=True)), ('situational_overview', models.TextField(blank=True, null=True)), diff --git a/dref/models.py b/dref/models.py index 2715ebaa2..84f63b7fd 100644 --- a/dref/models.py +++ b/dref/models.py @@ -1661,6 +1661,9 @@ class DrefFinalReport(models.Model): ) __financial_report_id = None + # TYPING + id: int + class Meta: verbose_name = _("Dref Final Report") verbose_name_plural = _("Dref Final Reports") @@ -1728,11 +1731,9 @@ class SourceModel(models.IntegerChoices): created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) - # The source document the summary was last generated from. Stored so a - # retrigger (e.g. from the admin) can replay the same source instead of - # re-resolving the latest approved document. - source_model_name = models.IntegerField( - verbose_name=_("source model name"), + source = models.IntegerField( + verbose_name=_("source"), + help_text=_("The model this summary was generated from."), choices=SourceModel.choices, ) source_id = models.PositiveBigIntegerField( diff --git a/dref/summary.py b/dref/summary.py index 013b3f21c..4fae765ca 100644 --- a/dref/summary.py +++ b/dref/summary.py @@ -5,15 +5,14 @@ import hashlib import json import logging -from typing import Callable, Dict, List, Optional +from typing import Callable, Dict, List, Optional, Union import tiktoken -from django.conf import settings -from django.utils.functional import cached_property -from openai import AzureOpenAI +from django.db.models import F from api.utils import get_model_name -from dref.models import Dref, DrefFinalReport, DrefOperationalUpdate +from dref.models import Dref, DrefFinalReport, DrefOperationalUpdate, DrefSummary +from main.llm import get_dref_summary_llm_client logger = logging.getLogger(__name__) @@ -22,6 +21,16 @@ MAX_OUTPUT_CHARS_PER_FIELD = 1500 MAX_INPUT_TOKENS = 10000 + +# The models a DrefSummary can be generated from. +DrefSummarySource = Union[Dref, DrefOperationalUpdate, DrefFinalReport] + +SOURCE_BY_MODEL: Dict[type, DrefSummary.SourceModel] = { + Dref: DrefSummary.SourceModel.DREF, + DrefOperationalUpdate: DrefSummary.SourceModel.DREF_OPERATIONAL_UPDATE, + DrefFinalReport: DrefSummary.SourceModel.DREF_FINAL_REPORT, +} + # DrefSummary fields — order is the iteration order for prompt assembly. SUMMARY_FIELDS: List[str] = [ "situational_overview", @@ -121,28 +130,6 @@ def _build_lessons_learned_prompt(**kwargs) -> str: ) -class DrefSummaryLLMClient: - @cached_property - def client(self): - return AzureOpenAI( - azure_endpoint=settings.AZURE_OPENAI_ENDPOINT, - api_key=settings.AZURE_OPENAI_KEY, - api_version="2023-05-15", - ) - - def get_response(self, messages) -> Optional[str]: - try: - response = self.client.chat.completions.create( - model=settings.AZURE_OPENAI_DEPLOYMENT_NAME, - messages=messages, - temperature=0.5, - ) - return response.choices[0].message.content - except Exception as e: - logger.error(f"Error while generating DREF summary response: {e}", exc_info=True) - return None - - def count_tokens(text: str) -> int: try: return len(tiktoken.get_encoding(ENCODING_NAME).encode(text)) @@ -184,8 +171,8 @@ def _extract_fields(obj, field_names: List[str]) -> dict: class DrefSummaryGenerator: """Assembles per-section prompts from a source document and produces all summaries.""" - def __init__(self, client: Optional[DrefSummaryLLMClient] = None): - self.client = client or DrefSummaryLLMClient() + def __init__(self): + self.client = get_dref_summary_llm_client() # Shared helpers — called by multiple extractors @@ -302,8 +289,13 @@ def _extract_dref_final_kwargs(cls, final) -> Dict[str, dict]: return kwargs @classmethod - def _get_section_kwargs(cls, source_doc) -> Dict[str, dict]: - """Dispatch to the model-specific extractor.""" + def get_section_kwargs(cls, source_doc: DrefSummarySource) -> Dict[str, dict]: + """Dispatch to the model-specific extractor. + + Public so callers that need both the hash and the generated summary + for the same ``source_doc`` (see ``compute_source_hash``/``generate_all``) + can compute this once and pass it to both instead of extracting twice. + """ if isinstance(source_doc, Dref): return cls._extract_dref_kwargs(source_doc) if isinstance(source_doc, DrefOperationalUpdate): @@ -312,16 +304,47 @@ def _get_section_kwargs(cls, source_doc) -> Dict[str, dict]: return cls._extract_dref_final_kwargs(source_doc) return {} + @staticmethod + def get_latest_approved_source(dref: Dref) -> Optional[tuple[DrefSummary.SourceModel, DrefSummarySource]]: + """Most authoritative (source, source object) pair for ``dref``. + + Priority: Final Report > latest Operational Update > Dref itself. + """ + final_report = ( + DrefFinalReport.objects.select_related("country", "disaster_type") + .filter(dref=dref, status=Dref.Status.APPROVED) + .order_by("-created_at") + .first() + ) + if final_report: + return SOURCE_BY_MODEL[DrefFinalReport], final_report + + latest_ops_update = ( + DrefOperationalUpdate.objects.select_related("country", "disaster_type") + .filter(dref=dref, status=Dref.Status.APPROVED) + .order_by(F("operational_update_number").desc(nulls_last=True), "-created_at") + .first() + ) + if latest_ops_update: + return SOURCE_BY_MODEL[DrefOperationalUpdate], latest_ops_update + + if dref.status == Dref.Status.APPROVED: + return SOURCE_BY_MODEL[Dref], dref + + return None + @classmethod - def build_source_text(cls, source_doc) -> str: + def build_source_text(cls, source_doc: DrefSummarySource) -> str: """JSON representation of all section kwargs for inspection and debugging.""" - return json.dumps(cls._get_section_kwargs(source_doc), indent=2, ensure_ascii=False, default=str) + return json.dumps(cls.get_section_kwargs(source_doc), indent=2, ensure_ascii=False, default=str) @classmethod - def compute_source_hash(cls, source_doc) -> str: + def compute_source_hash(cls, source_doc: DrefSummarySource, section_kwargs: Optional[Dict[str, dict]] = None) -> str: """Hash of all source content feeding the summary, for change detection.""" model_label = get_model_name(type(source_doc)) - payload = {"model": model_label, "id": source_doc.id, "source": cls._get_section_kwargs(source_doc)} + if section_kwargs is None: + section_kwargs = cls.get_section_kwargs(source_doc) + payload = {"model": model_label, "id": source_doc.id, "source": section_kwargs} content = json.dumps(payload, sort_keys=True, ensure_ascii=False, default=str) return hashlib.sha256(content.encode("utf-8")).hexdigest() @@ -337,16 +360,19 @@ def _parse_response(response: str) -> Dict[str, str]: raise ValueError("Expected a JSON object of summary fields") return data - def generate_all(self, source_doc) -> Dict[str, str]: + def generate_all(self, source_doc: DrefSummarySource, section_kwargs: Optional[Dict[str, dict]] = None) -> Dict[str, str]: """Generate every summary for ``source_doc`` in a single LLM call. Always returns a value for every ``SUMMARY_FIELDS`` key (empty string when a section has no content), so callers overwrite stale values - instead of leaving them in place on regeneration. + instead of leaving them in place on regeneration. ``section_kwargs`` + can be passed in by a caller that already computed it (e.g. via + ``compute_source_hash``) to avoid re-extracting it from ``source_doc``. """ empty_results: Dict[str, str] = {field_name: "" for field_name in SUMMARY_FIELDS} - section_kwargs = self._get_section_kwargs(source_doc) + if section_kwargs is None: + section_kwargs = self.get_section_kwargs(source_doc) if not section_kwargs: logger.info(f"No source content for ({type(source_doc).__name__}) ({source_doc.id}) summary; skipping generation.") return empty_results diff --git a/dref/tasks.py b/dref/tasks.py index d4c2ae2cd..fde4ef745 100644 --- a/dref/tasks.py +++ b/dref/tasks.py @@ -1,12 +1,17 @@ import logging +from datetime import timedelta +from enum import Enum +from typing import Optional from celery import shared_task from django.apps import apps from django.db import transaction from django.template.loader import render_to_string +from django.utils import timezone from api.utils import get_model_name from lang.tasks import translate_model_fields +from main.lock import RedisLockKey, redis_lock from main.translation import TRANSLATOR_ORIGINAL_LANGUAGE_FIELD_NAME from main.utils import logger_context from notifications.notification import send_notification @@ -14,8 +19,6 @@ from .models import ( Dref, DrefFile, - DrefFinalReport, - DrefOperationalUpdate, DrefSummary, IdentifiedNeed, NationalSocietyAction, @@ -31,6 +34,22 @@ logger = logging.getLogger(__name__) +# The PROCESSING status has no TTL of its own (see comment in generate_dref_summary +# below), so a row can get stuck PROCESSING forever if a worker dies mid-generation. +# Treat it as stale past this, so a later trigger can take over. +PROCESSING_STALE_AFTER = timedelta(minutes=10) + + +class DrefSummaryGenerationResult(str, Enum): + """Outcome of a ``generate_dref_summary`` run.""" + + SUCCESS = "success" + SOURCE_NOT_FOUND = "source_not_found" + ALREADY_IN_PROGRESS = "already_in_progress" + UP_TO_DATE = "up_to_date" + FAILED = "failed" + SUPERSEDED = "superseded" + @shared_task def send_dref_email(dref_id, users_emails, new_or_updated=""): @@ -63,88 +82,110 @@ def send_dref_email(dref_id, users_emails, new_or_updated=""): ] -def _resolve_source_model(source_model_name): - """Map a ``DrefSummary.SourceModel`` choice to its model class.""" - if source_model_name == DrefSummary.SourceModel.DREF: - return Dref - elif source_model_name == DrefSummary.SourceModel.DREF_OPERATIONAL_UPDATE: - return DrefOperationalUpdate - elif source_model_name == DrefSummary.SourceModel.DREF_FINAL_REPORT: - return DrefFinalReport - return None - - -@shared_task -def generate_dref_summary(source_model_name, source_id, overwrite=False): +@shared_task(soft_time_limit=600, time_limit=630) +def generate_dref_summary(dref_id: int, overwrite: bool = False) -> DrefSummaryGenerationResult: """Generate and store the AI-assisted summaries for a DREF. - The DREF is derived from the source document, so the source (a Dref, - DrefOperationalUpdate or DrefFinalReport) is the single source of truth for - what the summary is built from. + Always (re)generates from whichever approved source is currently latest + for the DREF (see ``get_latest_approved_source``), rather than a + specific source passed in, so it self-corrects no matter which + approval triggered it or the order concurrent runs execute in. """ - source_model = _resolve_source_model(source_model_name) - if source_model is None: - logger.error( - "Could not resolve source model for DREF summary", - extra=logger_context({"source_model_name": source_model_name, "source_id": source_id}), - ) - return False - source_doc = source_model.objects.filter(id=source_id).first() - if not source_doc: - logger.error( - "Source document not found for DREF summary", - extra=logger_context({"source_model_name": source_model_name, "source_id": source_id}), - ) - return False - - # Derive the DREF from the source: the Dref is its own source, while - # DrefOperationalUpdate and DrefFinalReport both point back via ``dref``. - dref = source_doc if isinstance(source_doc, Dref) else source_doc.dref + dref = Dref.objects.select_related("country", "disaster_type").filter(id=dref_id).first() if not dref: - logger.error( - "DREF not found for summary generation", - extra=logger_context({"source_model_name": source_model_name, "source_id": source_id}), - ) - return False - dref_id = dref.id - source_id = source_doc.id - - generator = DrefSummaryGenerator() - source_hash = generator.compute_source_hash(source_doc) - - summary_instance = DrefSummary.objects.filter(dref=dref).first() - if ( - summary_instance - and not overwrite - and summary_instance.source_hash == source_hash - and summary_instance.status == DrefSummary.SummaryStatus.SUCCESS - ): - logger.info(f"DREF summary up to date for DREF ({dref_id}); skipping generation.") - return True - - if summary_instance is None: - summary_instance = DrefSummary(dref=dref) - summary_instance.source_hash = source_hash - summary_instance.source_model_name = source_model_name - summary_instance.source_id = source_id - summary_instance.status = DrefSummary.SummaryStatus.PROCESSING - summary_instance.save() - + logger.error("Dref not found for summary generation", extra=logger_context({"dref_id": dref_id})) + return DrefSummaryGenerationResult.SOURCE_NOT_FOUND + + # The Redis lock only needs to guard the brief read-check-mark-PROCESSING + # section below against a concurrent trigger for the same DREF (double- + # approve, admin retrigger, ...); it is released before the LLM call. + # The PROCESSING status written inside the lock is what actually blocks + # a second run for as long as generation takes - unlike the lock, it has + # no TTL, so it still holds even if generation outlives lock_expire. + with redis_lock(key=RedisLockKey.DREF_SUMMARY, id=dref_id) as acquired: + if not acquired: + logger.warning(f"DREF summary generation already in progress for DREF ({dref_id}); skipping.") + return DrefSummaryGenerationResult.ALREADY_IN_PROGRESS + + latest_source = DrefSummaryGenerator.get_latest_approved_source(dref) + if not latest_source: + logger.error(f"No approved source found for DREF ({dref_id}) summary") + return DrefSummaryGenerationResult.SOURCE_NOT_FOUND + source_type, source_obj = latest_source + + section_kwargs = DrefSummaryGenerator.get_section_kwargs(source_obj) + source_hash = DrefSummaryGenerator.compute_source_hash(source_obj, section_kwargs=section_kwargs) + summary_instance: Optional[DrefSummary] = DrefSummary.objects.filter(dref=dref).first() + + if ( + summary_instance + and summary_instance.status == DrefSummary.SummaryStatus.PROCESSING + and timezone.now() - summary_instance.updated_at < PROCESSING_STALE_AFTER + and summary_instance.source == source_type + and summary_instance.source_id == source_obj.id + ): + # Same source already in flight; a different one falls through to regenerate below. + logger.warning(f"DREF summary already in progress for DREF ({dref_id}); skipping.") + return DrefSummaryGenerationResult.ALREADY_IN_PROGRESS + + if ( + summary_instance + and not overwrite + and summary_instance.source_hash == source_hash + and summary_instance.status == DrefSummary.SummaryStatus.SUCCESS + ): + logger.info(f"DREF summary up to date for DREF ({dref_id}); skipping generation.") + return DrefSummaryGenerationResult.UP_TO_DATE + + if summary_instance is None: + summary_instance = DrefSummary(dref=dref) + + try: + summary_instance.source_hash = source_hash + summary_instance.source = source_type + summary_instance.source_id = source_obj.id + summary_instance.status = DrefSummary.SummaryStatus.PROCESSING + summary_instance.save() + except Exception: + logger.warning(f"Failed to mark DREF summary as processing for DREF ({dref_id})", exc_info=True) + if summary_instance.pk: + summary_instance.status = DrefSummary.SummaryStatus.FAILED + summary_instance.save() + return DrefSummaryGenerationResult.FAILED + + # Lock released. The (possibly slow) LLM call runs unlocked; the + # PROCESSING status set above is what a concurrent trigger checks. + own_marker = { + "pk": summary_instance.pk, + "source": source_type, + "source_id": source_obj.id, + "status": DrefSummary.SummaryStatus.PROCESSING, + } try: - logger.info(f"Generating DREF summaries for DREF ({dref_id}) from ({source_model_name}) ({source_id})") - results = generator.generate_all(source_doc) - for field_name, value in results.items(): - setattr(summary_instance, field_name, value) - summary_instance.status = DrefSummary.SummaryStatus.SUCCESS - summary_instance.save() + logger.info(f"Generating DREF summaries for DREF ({dref_id}) from ({source_type.label}) ({source_obj.id})") + results = DrefSummaryGenerator().generate_all(source_obj, section_kwargs=section_kwargs) + with transaction.atomic(): + latest_summary_instance = DrefSummary.objects.select_for_update().filter(**own_marker).first() + if latest_summary_instance is None: + logger.warning(f"DREF summary run for DREF ({dref_id}) was superseded by a newer trigger; discarding result.") + return DrefSummaryGenerationResult.SUPERSEDED + for field_name, value in results.items(): + setattr(latest_summary_instance, field_name, value) + latest_summary_instance.status = DrefSummary.SummaryStatus.SUCCESS + latest_summary_instance.save() logger.info(f"Successfully generated DREF summaries for DREF ({dref_id})") - transaction.on_commit(lambda: translate_model_fields.delay(get_model_name(DrefSummary), summary_instance.pk)) - return True + transaction.on_commit(lambda: translate_model_fields.delay(get_model_name(DrefSummary), latest_summary_instance.pk)) + return DrefSummaryGenerationResult.SUCCESS except Exception: - summary_instance.status = DrefSummary.SummaryStatus.FAILED - summary_instance.save() - logger.warning(f"DREF summary generation failed for DREF ({dref_id})", exc_info=True) - return False + with transaction.atomic(): + latest_summary_instance = DrefSummary.objects.select_for_update().filter(**own_marker).first() + if latest_summary_instance is None: + logger.warning(f"DREF summary run for DREF ({dref_id}) failed but was already superseded; leaving it as is.") + else: + latest_summary_instance.status = DrefSummary.SummaryStatus.FAILED + latest_summary_instance.save() + logger.warning(f"DREF summary generation failed for DREF ({dref_id})", exc_info=True) + return DrefSummaryGenerationResult.FAILED @shared_task diff --git a/dref/test_summary.py b/dref/test_summary.py new file mode 100644 index 000000000..8c725356b --- /dev/null +++ b/dref/test_summary.py @@ -0,0 +1,327 @@ +from datetime import date + +from django.test import TestCase, override_settings + +from dref.factories.dref import ( + DrefFactory, + DrefFinalReportFactory, + DrefOperationalUpdateFactory, + PlannedInterventionFactory, +) +from dref.models import Dref, DrefSummary +from dref.summary import SUMMARY_FIELDS, DrefSummaryGenerator +from dref.tasks import DrefSummaryGenerationResult, generate_dref_summary +from main.llm import ( + AzureOpenAiChat, + DummyDrefSummaryLLMClient, + get_dref_summary_llm_client, +) + + +class DrefSummaryLLMClientTest(TestCase): + def test_uses_dummy_client_during_tests_by_default(self): + # settings.TESTING is always true under the test runner, so the + # factory should return the dummy client even though + # USE_DUMMY_LLM_CLIENT itself defaults to False. + self.assertIsInstance(get_dref_summary_llm_client(), DummyDrefSummaryLLMClient) + + @override_settings( + TESTING=False, + USE_DUMMY_LLM_CLIENT=False, + AZURE_OPENAI_ENDPOINT="https://example.com", + AZURE_OPENAI_KEY="fake-key", + AZURE_OPENAI_DEPLOYMENT_NAME="fake-deployment", + ) + def test_uses_real_client_when_flag_off_and_not_testing(self): + self.assertIsInstance(get_dref_summary_llm_client(), AzureOpenAiChat) + + @override_settings(TESTING=False, USE_DUMMY_LLM_CLIENT=True) + def test_uses_dummy_client_when_flag_on(self): + self.assertIsInstance(get_dref_summary_llm_client(), DummyDrefSummaryLLMClient) + + +class DrefSummaryGeneratorTest(TestCase): + def test_generate_all_succeeds_end_to_end_for_dref(self): + dref = DrefFactory.create(title="Test Dref") + + results = DrefSummaryGenerator().generate_all(dref) + + self.assertEqual(set(results.keys()), set(SUMMARY_FIELDS)) + self.assertIn("DUMMY RESPONSE", results["situational_overview"]) + + def test_generate_all_succeeds_end_to_end_for_dref_operational_update(self): + ops_update = DrefOperationalUpdateFactory.create(dref=DrefFactory.create(), title="Test Ops Update") + + results = DrefSummaryGenerator().generate_all(ops_update) + + self.assertEqual(set(results.keys()), set(SUMMARY_FIELDS)) + self.assertIn("DUMMY RESPONSE", results["situational_overview"]) + + def test_generate_all_succeeds_end_to_end_for_dref_final_report(self): + planned_intervention = PlannedInterventionFactory.create(challenges="Some challenge", lessons_learnt="Some lesson") + final_report = DrefFinalReportFactory.create(title="Test Final Report", planned_interventions=[planned_intervention]) + + results = DrefSummaryGenerator().generate_all(final_report) + + self.assertEqual(set(results.keys()), set(SUMMARY_FIELDS)) + self.assertIn("DUMMY RESPONSE", results["situational_overview"]) + self.assertIn("DUMMY RESPONSE", results["challenges_identified"]) + self.assertIn("DUMMY RESPONSE", results["lessons_learned"]) + + def test_get_section_kwargs_for_dref(self): + dref = DrefFactory.create( + title="Test Dref", + event_description="Severe flooding", + type_of_dref=Dref.DrefType.RESPONSE, + event_scope="Affected 3 districts", + operation_objective="Provide shelter", + response_strategy="Cash and shelter support", + total_targeted_population=5000, + people_in_need=8000, + people_assisted="5000 people", + selection_criteria="Most vulnerable households", + ) + + kwargs = DrefSummaryGenerator.get_section_kwargs(dref) + + self.assertEqual(set(kwargs.keys()), {"situational_overview", "operational_strategy", "people_centered_approach"}) + self.assertEqual(kwargs["situational_overview"]["title"], "Test Dref") + self.assertEqual(kwargs["situational_overview"]["event_description"], "Severe flooding") + self.assertEqual(kwargs["situational_overview"]["event_scope"], "Affected 3 districts") + self.assertEqual(kwargs["operational_strategy"]["operation_objective"], "Provide shelter") + self.assertEqual(kwargs["operational_strategy"]["response_strategy"], "Cash and shelter support") + self.assertEqual(kwargs["operational_strategy"]["total_targeted_population"], 5000) + self.assertEqual(kwargs["operational_strategy"]["people_in_need"], 8000) + self.assertEqual(kwargs["people_centered_approach"]["people_assisted"], "5000 people") + self.assertEqual(kwargs["people_centered_approach"]["selection_criteria"], "Most vulnerable households") + self.assertNotIn("women", kwargs["people_centered_approach"]) + + def test_get_section_kwargs_for_dref_operational_update(self): + ops_update = DrefOperationalUpdateFactory.create( + dref=DrefFactory.create(), + title="Test Ops Update", + event_description="Flooding continues", + operation_objective="Extend shelter support", + response_strategy="Extended cash support", + total_dref_allocation=100000, + people_assisted="6000 people", + selection_criteria="Vulnerable households", + women=1000, + men=900, + girls=500, + boys=600, + new_operational_end_date=date(2025, 6, 1), + total_operation_timeframe=6, + ) + + kwargs = DrefSummaryGenerator.get_section_kwargs(ops_update) + + self.assertEqual(set(kwargs.keys()), {"situational_overview", "operational_strategy", "people_centered_approach"}) + self.assertEqual(kwargs["situational_overview"]["title"], "Test Ops Update") + self.assertEqual(kwargs["situational_overview"]["event_description"], "Flooding continues") + self.assertEqual(kwargs["operational_strategy"]["operation_objective"], "Extend shelter support") + self.assertEqual(kwargs["operational_strategy"]["total_dref_allocation"], 100000) + self.assertEqual(kwargs["operational_strategy"]["operation_end_date"], date(2025, 6, 1)) + self.assertEqual(kwargs["operational_strategy"]["operation_timeframe"], 6) + self.assertEqual(kwargs["people_centered_approach"]["people_assisted"], "6000 people") + self.assertEqual(kwargs["people_centered_approach"]["women"], 1000) + self.assertEqual(kwargs["people_centered_approach"]["men"], 900) + self.assertEqual(kwargs["people_centered_approach"]["girls"], 500) + self.assertEqual(kwargs["people_centered_approach"]["boys"], 600) + + def test_get_section_kwargs_for_dref_final_report(self): + planned_intervention = PlannedInterventionFactory.create( + challenges="Access constraints due to weather", + lessons_learnt="Earlier pre-positioning of stock helps", + ) + final_report = DrefFinalReportFactory.create( + title="Test Final Report", + event_description="Response concluded", + operation_objective="Deliver relief items", + response_strategy="Direct distribution", + total_dref_allocation=150000, + people_assisted="7000 people", + selection_criteria="Most affected households", + women=1200, + men=1100, + girls=600, + boys=700, + planned_interventions=[planned_intervention], + ) + + kwargs = DrefSummaryGenerator.get_section_kwargs(final_report) + + self.assertEqual( + set(kwargs.keys()), + { + "situational_overview", + "operational_strategy", + "people_centered_approach", + "challenges_identified", + "lessons_learned", + }, + ) + self.assertEqual(kwargs["situational_overview"]["title"], "Test Final Report") + self.assertEqual(kwargs["operational_strategy"]["total_dref_allocation"], 150000) + self.assertEqual(kwargs["people_centered_approach"]["women"], 1200) + + challenges = kwargs["challenges_identified"]["planned_interventions"] + self.assertEqual(len(challenges), 1) + self.assertEqual(challenges[0]["challenges"], "Access constraints due to weather") + + lessons = kwargs["lessons_learned"]["planned_interventions"] + self.assertEqual(len(lessons), 1) + self.assertEqual(lessons[0]["lessons_learnt"], "Earlier pre-positioning of stock helps") + + def test_get_section_kwargs_excludes_event_scope_for_imminent_dref(self): + imminent_dref = DrefFactory.create(type_of_dref=Dref.DrefType.IMMINENT, event_scope="scope text") + self.assertNotIn("event_scope", DrefSummaryGenerator.get_section_kwargs(imminent_dref)["situational_overview"]) + + response_dref = DrefFactory.create(type_of_dref=Dref.DrefType.RESPONSE, event_scope="scope text") + self.assertIn("event_scope", DrefSummaryGenerator.get_section_kwargs(response_dref)["situational_overview"]) + + def test_compute_source_hash_changes_with_content_and_is_deterministic(self): + dref = DrefFactory.create(title="Original Title", type_of_dref=Dref.DrefType.RESPONSE) + + hash_a = DrefSummaryGenerator.compute_source_hash(dref) + hash_b = DrefSummaryGenerator.compute_source_hash(dref) + self.assertEqual(hash_a, hash_b) + + dref.title = "Changed Title" + dref.save(update_fields=["title"]) + hash_c = DrefSummaryGenerator.compute_source_hash(dref) + self.assertNotEqual(hash_a, hash_c) + + def test_compute_source_hash_uses_provided_section_kwargs(self): + dref = DrefFactory.create(title="Test Dref", type_of_dref=Dref.DrefType.RESPONSE) + default_hash = DrefSummaryGenerator.compute_source_hash(dref) + + actual_kwargs = DrefSummaryGenerator.get_section_kwargs(dref) + self.assertEqual(default_hash, DrefSummaryGenerator.compute_source_hash(dref, section_kwargs=actual_kwargs)) + + custom_kwargs = {"situational_overview": {"title": "overridden"}} + self.assertNotEqual(default_hash, DrefSummaryGenerator.compute_source_hash(dref, section_kwargs=custom_kwargs)) + + def test_get_latest_approved_source_returns_none_when_dref_not_approved(self): + dref = DrefFactory.create(status=Dref.Status.FINALIZED) + self.assertIsNone(DrefSummaryGenerator.get_latest_approved_source(dref)) + + def test_get_latest_approved_source_follows_priority_order(self): + dref = DrefFactory.create(status=Dref.Status.APPROVED) + self.assertEqual(DrefSummaryGenerator.get_latest_approved_source(dref), (DrefSummary.SourceModel.DREF, dref)) + + # Unapproved updates don't count, so the Dref itself still wins. + DrefOperationalUpdateFactory.create(dref=dref, status=Dref.Status.FINALIZED, operational_update_number=1) + self.assertEqual(DrefSummaryGenerator.get_latest_approved_source(dref), (DrefSummary.SourceModel.DREF, dref)) + + # An approved Operational Update supersedes the Dref; the higher + # operational_update_number wins, regardless of insertion order. + DrefOperationalUpdateFactory.create(dref=dref, status=Dref.Status.APPROVED, operational_update_number=1) + ops_update_2 = DrefOperationalUpdateFactory.create(dref=dref, status=Dref.Status.APPROVED, operational_update_number=2) + self.assertEqual( + DrefSummaryGenerator.get_latest_approved_source(dref), + (DrefSummary.SourceModel.DREF_OPERATIONAL_UPDATE, ops_update_2), + ) + + # An approved Final Report supersedes everything else. + final_report = DrefFinalReportFactory.create(dref=dref, status=Dref.Status.APPROVED) + self.assertEqual( + DrefSummaryGenerator.get_latest_approved_source(dref), + (DrefSummary.SourceModel.DREF_FINAL_REPORT, final_report), + ) + + def test_returns_source_not_found_when_dref_missing_or_nothing_approved(self): + self.assertEqual(generate_dref_summary(dref_id=999999), DrefSummaryGenerationResult.SOURCE_NOT_FOUND) + + dref = DrefFactory.create(status=Dref.Status.FINALIZED) + self.assertEqual(generate_dref_summary(dref_id=dref.id), DrefSummaryGenerationResult.SOURCE_NOT_FOUND) + + def test_generates_from_dref_and_is_up_to_date_on_rerun(self): + dref = DrefFactory.create(status=Dref.Status.APPROVED) + + self.assertEqual(generate_dref_summary(dref_id=dref.id), DrefSummaryGenerationResult.SUCCESS) + summary = DrefSummary.objects.get(dref=dref) + self.assertEqual(summary.source, DrefSummary.SourceModel.DREF) + self.assertEqual(summary.source_id, dref.id) + + # Nothing changed since - re-running should not regenerate. + self.assertEqual(generate_dref_summary(dref_id=dref.id), DrefSummaryGenerationResult.UP_TO_DATE) + + def test_retrigger_with_overwrite_regenerates_even_when_up_to_date(self): + dref = DrefFactory.create(status=Dref.Status.APPROVED) + + self.assertEqual(generate_dref_summary(dref_id=dref.id), DrefSummaryGenerationResult.SUCCESS) + summary = DrefSummary.objects.get(dref=dref) + first_hash = summary.source_hash + first_updated_at = summary.updated_at + + # Nothing changed since, so a plain retrigger would be UP_TO_DATE ... + self.assertEqual(generate_dref_summary(dref_id=dref.id), DrefSummaryGenerationResult.UP_TO_DATE) + + # ... but overwrite=True must bypass that check and regenerate anyway. + self.assertEqual(generate_dref_summary(dref_id=dref.id, overwrite=True), DrefSummaryGenerationResult.SUCCESS) + + self.assertEqual(DrefSummary.objects.filter(dref=dref).count(), 1) + summary.refresh_from_db() + self.assertEqual(summary.status, DrefSummary.SummaryStatus.SUCCESS) + self.assertEqual(summary.source, DrefSummary.SourceModel.DREF) + self.assertEqual(summary.source_id, dref.id) + self.assertEqual(summary.source_hash, first_hash) + self.assertGreater(summary.updated_at, first_updated_at) + + def test_processing_in_flight_skips_same_source_but_regenerates_for_a_newer_one(self): + dref = DrefFactory.create(status=Dref.Status.APPROVED) + # Simulate a run already in flight for the Dref itself. + DrefSummary.objects.create( + dref=dref, + source=DrefSummary.SourceModel.DREF, + source_id=dref.id, + source_hash="in-flight-hash", + status=DrefSummary.SummaryStatus.PROCESSING, + ) + + # A duplicate trigger for the same in-flight source is dropped. + self.assertEqual(generate_dref_summary(dref_id=dref.id), DrefSummaryGenerationResult.ALREADY_IN_PROGRESS) + + # A newer approval lands while that run is still (supposedly) processing. + ops_update = DrefOperationalUpdateFactory.create(dref=dref, status=Dref.Status.APPROVED, operational_update_number=1) + + self.assertEqual(generate_dref_summary(dref_id=dref.id), DrefSummaryGenerationResult.SUCCESS) + summary = DrefSummary.objects.get(dref=dref) + self.assertEqual(summary.source, DrefSummary.SourceModel.DREF_OPERATIONAL_UPDATE) + self.assertEqual(summary.source_id, ops_update.id) + + def test_generate_all_succeeds_for_newer_ops_update_while_older_one_in_flight(self): + """Two Operational Updates on one Dref: a run in flight for the older + one must not block generation once a newer one is approved.""" + dref = DrefFactory.create(status=Dref.Status.APPROVED) + ops_update_1 = DrefOperationalUpdateFactory.create(dref=dref, status=Dref.Status.APPROVED, operational_update_number=1) + DrefSummary.objects.create( + dref=dref, + source=DrefSummary.SourceModel.DREF_OPERATIONAL_UPDATE, + source_id=ops_update_1.id, + source_hash="in-flight-hash", + status=DrefSummary.SummaryStatus.PROCESSING, + ) + + # A duplicate trigger for the same in-flight source (ops_update_1) is dropped. + self.assertEqual(generate_dref_summary(dref_id=dref.id), DrefSummaryGenerationResult.ALREADY_IN_PROGRESS) + + # A second, newer Operational Update is approved while that run is still (supposedly) processing. + ops_update_2 = DrefOperationalUpdateFactory.create(dref=dref, status=Dref.Status.APPROVED, operational_update_number=2) + + # The in-flight record is for a now-stale source (ops_update_1), so this + # trigger falls through and regenerates from ops_update_2 instead of + # being blocked as "already in progress". + self.assertEqual(generate_dref_summary(dref_id=dref.id), DrefSummaryGenerationResult.SUCCESS) + + # Exactly one DrefSummary row for the Dref - updated in place, not duplicated. + self.assertEqual(DrefSummary.objects.filter(dref=dref).count(), 1) + summary = DrefSummary.objects.get(dref=dref) + self.assertEqual(summary.source, DrefSummary.SourceModel.DREF_OPERATIONAL_UPDATE) + self.assertEqual(summary.source_id, ops_update_2.id) + self.assertEqual(summary.status, DrefSummary.SummaryStatus.SUCCESS) + self.assertNotEqual(summary.source_hash, "in-flight-hash") + + # Nothing changed since - re-running should not regenerate. + self.assertEqual(generate_dref_summary(dref_id=dref.id), DrefSummaryGenerationResult.UP_TO_DATE) diff --git a/dref/views.py b/dref/views.py index 480dd7c56..7dfe58e34 100644 --- a/dref/views.py +++ b/dref/views.py @@ -32,13 +32,7 @@ DrefOperationalUpdateFilter, DrefShareUserFilterSet, ) -from dref.models import ( - Dref, - DrefFile, - DrefFinalReport, - DrefOperationalUpdate, - DrefSummary, -) +from dref.models import Dref, DrefFile, DrefFinalReport, DrefOperationalUpdate from dref.permissions import ApproveDrefPermission from dref.serializers import ( AddDrefUserSerializer, @@ -143,11 +137,7 @@ def get_approved(self, request, pk=None, version=None): dref.status = Dref.Status.APPROVED dref.save(update_fields=["event", "status"]) - _dref_id = dref.id - transaction.on_commit( - lambda: generate_dref_summary.delay(source_model_name=DrefSummary.SourceModel.DREF, source_id=_dref_id) - ) - + transaction.on_commit(lambda: generate_dref_summary.delay(dref_id=dref.id)) return response.Response(DrefSerializer(dref, context={"request": request}).data) @extend_schema(request=None, responses=DrefSerializer) @@ -244,14 +234,7 @@ def get_approved(self, request, pk=None, version=None): operational_update.status = Dref.Status.APPROVED operational_update.date_of_approval = timezone.now().date() operational_update.save(update_fields=["status", "date_of_approval"]) - _ops_update_id = operational_update.pk - transaction.on_commit( - lambda: generate_dref_summary.delay( - source_model_name=DrefSummary.SourceModel.DREF_OPERATIONAL_UPDATE, - source_id=_ops_update_id, - overwrite=True, - ) - ) + transaction.on_commit(lambda: generate_dref_summary.delay(dref_id=operational_update.dref_id)) serializer = DrefOperationalUpdateSerializer(operational_update, context={"request": request}) return response.Response(serializer.data) @@ -320,14 +303,7 @@ def get_approved(self, request, pk=None, version=None): final_report.dref.is_active = False final_report.date_of_approval = timezone.now().date() final_report.dref.save(update_fields=["is_active", "date_of_approval"]) - _final_report_id = final_report.pk - transaction.on_commit( - lambda: generate_dref_summary.delay( - source_model_name=DrefSummary.SourceModel.DREF_FINAL_REPORT, - source_id=_final_report_id, - overwrite=True, - ) - ) + transaction.on_commit(lambda: generate_dref_summary.delay(dref_id=final_report.dref_id)) serializer = DrefFinalReportSerializer(final_report, context={"request": request}) return response.Response(serializer.data) diff --git a/main/llm.py b/main/llm.py new file mode 100644 index 000000000..87a33e4ff --- /dev/null +++ b/main/llm.py @@ -0,0 +1,183 @@ +import json +import logging +import re +from typing import Dict, List, Optional + +from django.conf import settings +from django.utils.functional import cached_property +from openai import AzureOpenAI + +logger = logging.getLogger(__name__) + +# A chat completion prompt: the ``[{"role": ..., "content": ...}, ...]`` list +# every real call site builds and passes in. +Message = Dict[str, str] +Messages = List[Message] + + +PLACEHOLDER_TEXT = ( + "This is placeholder text standing in for a real AI-generated summary. It is returned by " + "the dummy LLM client Lorem ipsum dolor sit amet, consectetur adipiscing elit. Sed do eiusmod tempor incididunt " + "ut labore et dolore magna aliqua. Ut enim ad minim veniam, quis nostrud exercitation ullamco " + "laboris nisi ut aliquip ex ea commodo consequat. Duis aute irure dolor in reprehenderit in " + "voluptate velit esse cillum dolore eu fugiat nulla pariatur." +) + + +class BaseLLMClient: + def get_response(self, messages: Messages) -> Optional[str]: + raise NotImplementedError + + @staticmethod + def _last_user_message(messages: Messages) -> str: + return next((message["content"] for message in reversed(messages) if message["role"] == "user"), "") + + def _fake_content(self) -> str: + """Placeholder text every dummy response uses in place of real LLM output.""" + return f"DUMMY RESPONSE, {PLACEHOLDER_TEXT}" + + +class AzureOpenAiChat(BaseLLMClient): + """The real Azure OpenAI-backed client, shared by every feature.""" + + def __init__(self, temperature: float = 0.5): + self.temperature = temperature + + if settings.TESTING: + # Tests never hit Azure, so a missing/placeholder config shouldn't fail construction. + return + + if not settings.AZURE_OPENAI_ENDPOINT or not settings.AZURE_OPENAI_KEY or not settings.AZURE_OPENAI_DEPLOYMENT_NAME: + raise Exception("Azure OpenAI configuration missing") + + @cached_property + def client(self): + return AzureOpenAI( + azure_endpoint=settings.AZURE_OPENAI_ENDPOINT, + api_key=settings.AZURE_OPENAI_KEY, + api_version="2023-05-15", + ) + + def get_response(self, messages: Messages) -> Optional[str]: + if settings.USE_DUMMY_LLM_CLIENT or settings.TESTING: + logger.info(f"{type(self).__name__}: dummy mode is on; returning a fake response instead of calling Azure OpenAI.") + return self._fake_content() + + try: + response = self.client.chat.completions.create( + model=settings.AZURE_OPENAI_DEPLOYMENT_NAME, + messages=messages, + temperature=self.temperature, + ) + return response.choices[0].message.content + except Exception as e: + logger.error(f"Error while generating LLM response: {e}", exc_info=True) + return None + + +class BaseDummyLLMClient(BaseLLMClient): + """Summaries generated for local development/testing, without hitting Azure OpenAI. Each subclass implements + ``_fake_response()`` to return a JSON object with the expected fields for that feature + """ + + def get_response(self, messages: Messages) -> Optional[str]: + return self._fake_response(messages) + + def _fake_response(self, messages: Messages) -> str: + raise NotImplementedError + + +class DummyDrefSummaryLLMClient(BaseDummyLLMClient): + """Fake response for DREF summary generation: a JSON object with all of + DREF's summary fields, each set to the same fake content, so + ``dref.summary.DrefSummaryGenerator`` parses/saves it unchanged. + """ + + def _fake_response(self, messages: Messages) -> str: + from dref.summary import SUMMARY_FIELDS + + logger.info("DummyDrefSummaryLLMClient: generating fake response for DREF summary fields") + fake_content = self._fake_content() + payload: Dict[str, str] = {field_name: fake_content for field_name in SUMMARY_FIELDS} + return json.dumps(payload) + + +class DummyOpsLearningLLMClient(BaseDummyLLMClient): + """Fake response for PER Ops Learning summaries.""" + + _EXCERPT_ID_RE = re.compile(r"^(\d+)\. In ", re.MULTILINE) + + def _prompt_excerpt_ids(self, user_content: str) -> List[int]: + return [int(excerpt_id) for excerpt_id in self._EXCERPT_ID_RE.findall(user_content)] + + def _fake_response(self, messages: Messages) -> str: + from per.models import OpsLearning + from per.ops_learning_summary import OpsLearningSummaryTask + + logger.info("DummyOpsLearningLLMClient: generating fake response for Ops Learning summary") + user_content = self._last_user_message(messages) + fake_content = self._fake_content() + excerpt_ids = self._prompt_excerpt_ids(user_content) + logger.info(f"DummyOpsLearningLLMClient: found {len(excerpt_ids)} excerpt id(s) in the prompt") + + all_excerpts_id = ", ".join(str(excerpt_id) for excerpt_id in excerpt_ids) + + payload: Dict[str, Dict[str, str]] + if OpsLearningSummaryTask.component_prompt in user_content: + logger.info("DummyOpsLearningLLMClient: detected the component prompt shape.") + excerpts = list( + OpsLearning.objects.filter(id__in=excerpt_ids, is_validated=True, per_component_validated__isnull=False) + .prefetch_related("per_component_validated") + .distinct() + ) + component = excerpts[0].per_component_validated.first() if excerpts else None + subtype = component.title if component else "Logistics" + excerpts_id = ", ".join(str(excerpt.id) for excerpt in excerpts) or all_excerpts_id + logger.info( + f"DummyOpsLearningLLMClient: {len(excerpts)} excerpt(s) matched a validated component; " + f"using subtype '{subtype}'." + ) + payload = {"0": {"type": "component", "subtype": subtype, "excerpts id": excerpts_id, "content": fake_content}} + + elif OpsLearningSummaryTask.sector_prompt in user_content: + logger.info("DummyOpsLearningLLMClient: detected the sector prompt shape.") + excerpts = list( + OpsLearning.objects.filter(id__in=excerpt_ids, is_validated=True, sector_validated__isnull=False) + .prefetch_related("sector_validated") + .distinct() + ) + sector = excerpts[0].sector_validated.first() if excerpts else None + subtype = sector.title if sector else "Shelter" + excerpts_id = ", ".join(str(excerpt.id) for excerpt in excerpts) or all_excerpts_id + logger.info( + f"DummyOpsLearningLLMClient: {len(excerpts)} excerpt(s) matched a validated sector; " + f"using subtype '{subtype}'." + ) + payload = {"0": {"type": "sector", "subtype": subtype, "excerpts id": excerpts_id, "content": fake_content}} + else: + logger.info( + f"DummyOpsLearningLLMClient: detected the primary prompt shape; " + f"returning 3 fake findings with excerpts id(s) [{all_excerpts_id}]." + ) + payload = { + str(i): { + "title": f"Dummy Finding {i + 1}", + "excerpts id": all_excerpts_id, + "content": fake_content, + "confidence level": "1/5", + } + for i in range(3) + } + return json.dumps(payload) + + +def get_dref_summary_llm_client() -> BaseLLMClient: + if settings.USE_DUMMY_LLM_CLIENT or settings.TESTING: + return DummyDrefSummaryLLMClient() + return AzureOpenAiChat(temperature=0.5) + + +def get_ops_learning_llm_client() -> BaseLLMClient: + if settings.USE_DUMMY_LLM_CLIENT or settings.TESTING: + return DummyOpsLearningLLMClient() + return AzureOpenAiChat(temperature=0.7) diff --git a/main/lock.py b/main/lock.py index 3d1fd2de1..9de5b7aaf 100644 --- a/main/lock.py +++ b/main/lock.py @@ -20,6 +20,7 @@ class RedisLockKey(models.TextChoices): OPERATION_LEARNING_SUMMARY = _BASE + "-operation-learning-summary-{0}" OPERATION_LEARNING_SUMMARY_EXPORT = _BASE + "-operation-learning-summary-export-{0}" MODEL_TRANSLATION = _BASE + "-{model_name}-translation-{id}" + DREF_SUMMARY = _BASE + "-dref-summary-{0}" @contextmanager diff --git a/main/settings.py b/main/settings.py index a500bef8c..b82853c20 100644 --- a/main/settings.py +++ b/main/settings.py @@ -146,6 +146,8 @@ AZURE_OPENAI_ENDPOINT=(str, None), AZURE_OPENAI_KEY=(str, None), AZURE_OPENAI_DEPLOYMENT_NAME=(str, None), + # Use a fake LLM client instead of calling Azure OpenAI + USE_DUMMY_LLM_CLIENT=(bool, False), # ReliefWeb appname RELIEF_WEB_APP_NAME=(str, None), # PowerBI @@ -868,6 +870,7 @@ def decode_base64(env_key, fallback_env_key): AZURE_OPENAI_ENDPOINT = env("AZURE_OPENAI_ENDPOINT") AZURE_OPENAI_KEY = env("AZURE_OPENAI_KEY") AZURE_OPENAI_DEPLOYMENT_NAME = env("AZURE_OPENAI_DEPLOYMENT_NAME") +USE_DUMMY_LLM_CLIENT = env("USE_DUMMY_LLM_CLIENT") OIDC_ENABLE = env("OIDC_ENABLE") OIDC_RSA_PRIVATE_KEY = None diff --git a/per/models.py b/per/models.py index 4cd7e7488..3b792c426 100644 --- a/per/models.py +++ b/per/models.py @@ -705,6 +705,9 @@ class OpsLearning(models.Model): modified_at = models.DateTimeField(verbose_name=_("modified_at"), auto_now=True) created_at = models.DateTimeField(verbose_name=_("created at"), auto_now_add=True) + # TYPING + id: int + class Meta: ordering = ("learning",) verbose_name = _("Operational Learning") diff --git a/per/ops_learning_summary.py b/per/ops_learning_summary.py index d2d3d6cd4..5a5d150ed 100644 --- a/per/ops_learning_summary.py +++ b/per/ops_learning_summary.py @@ -5,17 +5,15 @@ import pandas as pd import tiktoken -from django.conf import settings from django.db import transaction from django.db.models import F -from django.utils.functional import cached_property -from openai import AzureOpenAI from api.logger import logger from api.models import Country from api.utils import get_model_name from deployments.models import SectorTag from lang.tasks import translate_model_fields +from main.llm import get_ops_learning_llm_client from per.cache import OpslearningSummaryCacheHelper from per.models import ( FormComponent, @@ -29,24 +27,6 @@ ) -class AzureOpenAiChat: - - @cached_property - def client(self): - return AzureOpenAI( - azure_endpoint=settings.AZURE_OPENAI_ENDPOINT, api_key=settings.AZURE_OPENAI_KEY, api_version="2023-05-15" - ) - - def get_response(self, message): - try: - response = self.client.chat.completions.create( - model=settings.AZURE_OPENAI_DEPLOYMENT_NAME, messages=message, temperature=0.7 - ) - return response.choices[0].message.content - except Exception as e: - logger.error(f"Error while generating response: {e}", exc_info=True) - - class OpsLearningSummaryTask: PROMPT_DATA_LENGTH_LIMIT = 5000 @@ -815,9 +795,9 @@ def _summarize(prompt, type: OpsLearningPromptResponseCache.PromptType, system_m logger.warning("The length of the prompt might be too long.") return "{}" - # Using Azure OpenAI to summarize the prompt - client = AzureOpenAiChat() - response = client.get_response(message=messages) + # Using Azure OpenAI (or the dummy, per settings.USE_DUMMY_LLM_CLIENT) to summarize the prompt + client = get_ops_learning_llm_client() + response = client.get_response(messages) return response def _validate_format(summary, MAX_RETRIES=3):