775 lines
36 KiB
Python
775 lines
36 KiB
Python
"""Phase 3: Evaluation harness for AI Insight quality.
|
|
Tests use stubbed LLM to assert prompt/contract behavior without live Ollama.
|
|
Golden fixtures: one per topic + edge cases.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
from unittest.mock import patch, MagicMock
|
|
from pathlib import Path
|
|
|
|
import django
|
|
from django.test import TestCase, override_settings
|
|
from django.utils import timezone
|
|
|
|
# Setup Django
|
|
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings")
|
|
django.setup()
|
|
|
|
from apps.operations.services import insight_service
|
|
from apps.operations.services import cp707_knowledge as cp707
|
|
from apps.operations.services import root_cause
|
|
|
|
|
|
# ─── Golden fixtures ──────────────────────────────────────────────────────
|
|
FIXTURE_DIR = Path(__file__).parent.parent.parent.parent / "fixtures" / "insight"
|
|
|
|
|
|
def load_fixture(name: str) -> dict:
|
|
with open(FIXTURE_DIR / name, encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
|
|
GOLDEN_FIXTURES = {
|
|
"hitung_ayam": load_fixture("gold-cycle-h28-kandang-01.json"),
|
|
"end_cycle": load_fixture("gold-cycle-end-kandang-01.json"),
|
|
"fcr": load_fixture("kandang-01.json"),
|
|
"berat_ayam": load_fixture("kandang-02.json"),
|
|
"eef": load_fixture("kandang-03.json"),
|
|
"hitung_karung": load_fixture("kandang-04.json"),
|
|
"iot_panel": load_fixture("kandang-05.json"),
|
|
# Edge cases
|
|
"empty_context": load_fixture("empty-context-kandang-03.json"),
|
|
"half_good_h28": load_fixture("half-good-h28.json"),
|
|
"half_bad_h28": load_fixture("half-bad-h28.json"),
|
|
}
|
|
|
|
|
|
# ─── Stubbed LLM responses ────────────────────────────────────────────────
|
|
STUB_RESPONSES = {
|
|
"hitung_ayam": {
|
|
"kesimpulan": "Mortalitas kumulatif 9.91% adalah SANGAT TINGGI melebihi ambang CP 707 (>7%).",
|
|
"insight": "Mortalitas kumulatif 9.91% jauh melebihi standar CP 707 (5-7% TINGGI, >7% SANGAT TINGGI). Ini menunjukkan adanya tekanan penyakit atau manajemen yang perlu dievaluasi. Segera lakukan nekropsi pada ayam mati untuk mengidentifikasi penyebab, perketat biosekuriti, dan pantau mortalitas harian hingga tren menurun."
|
|
},
|
|
"fcr": {
|
|
"kesimpulan": "FCR 1.40 pada hari ke-28 di atas standar CP 707 (1.315).",
|
|
"insight": "FCR 1.40 melebihi standar CP 707 pada hari ke-28 (1.315) sebesar 6.5%. Hal ini menandakan efisiensi pakan menurun. Periksa kualitas pakan, pastikan akses feeder memadai, dan evaluasi program pakan. Tindakan: cek sisa pakan di feeder, sesuaikan jadwal pemberian, dan bandingkan FCR mingguan dengan standar."
|
|
},
|
|
"berat_ayam": {
|
|
"kesimpulan": "Bobot rata-rata 1550g pada hari ke-28 sesuai standar CP 707 (1543g).",
|
|
"insight": "Bobot rata-rata 1550g mendekati standar CP 707 hari ke-28 (1543g). Pertumbuhan berada dalam rentang normal. Lanjutkan manajemen yang saat ini berjalan baik. Pantau ADG harian agar tetap stabil."
|
|
},
|
|
"eef": {
|
|
"kesimpulan": "EEF/IP 280 berada di kisaran DI BAWAH RATA-RATA (skala manajer, di luar buku CP 707).",
|
|
"insight": "EEF/IP 280 tergolong DI BAWAH RATA-RATA pada skala manajer (200-300). FCR dan mortalitas adalah penyebab utama indeks rendah. Perbaiki efisiensi pakan dengan mengevaluasi program nutrisi dan akses feeder. Pantau EEF harian untuk memastikan tren naik."
|
|
},
|
|
"hitung_karung": {
|
|
"kesimpulan": "Saldo karung 15 karung (masuk 100 - tuang 85 - sisa 0).",
|
|
"insight": "Saldo karung 15 karung menunjukkan adanya selisih antara pakan masuk dan yang dituang. Periksa pencatatan pakan masuk, pastikan tidak ada kebocoran di gudang, dan rekonsiliasi stok fisik dengan buku. Tindakan: audit stok mingguan dan perbaiki prosedur pencatatan."
|
|
},
|
|
"iot_panel": {
|
|
"kesimpulan": "Suhu rata-rata 28.5°C pada hari ke-28 di atas TET standar CP 707.",
|
|
"insight": "Suhu kandang 28.5°C melebihi Target Efektif Temperatur CP 707 untuk hari ke-28. Hal ini dapat menurunkan nafsu makan. Turunkan suhu kandang dengan menambah ventilasi atau cooling pad. Pantau suhu setiap jam dan kelembapan agar tetap 50-70%."
|
|
},
|
|
"end_cycle": {
|
|
"kesimpulan": "Siklus berakhir dengan FCR 1.72 dan EEF 280 — keduanya di bawah standar optimal.",
|
|
"masalah": [
|
|
"FCR akhir 1.72 melampaui standar CP 707 hari ke-48",
|
|
"EEF/IP 280 tergolong DI BAWAH RATA-RATA (skala manajer)"
|
|
],
|
|
"akar_penyebab": [
|
|
{"id": "pakan_akses", "hipotesis": "Pakan / akses pakan", "bukti": ["FCR akhir 1.72 jauh di atas standar"], "confidence": "high", "fase": "growth"},
|
|
{"id": "fcr_tinggi", "hipotesis": "Efisiensi pakan menurun", "bukti": ["FCR minggu 6-7 rata-rata 1.68-1.72"], "confidence": "med", "fase": "growth"}
|
|
],
|
|
"perbaikan_siklus_berikutnya": [
|
|
{"id": "pakan_akses", "fase": "growth", "aksi": "Pastikan ketersediaan dan akses pakan (feeder space, jadwal isi, kualitas fisik pakan) di fase growth.", "metrik_pantau": "FCR harian vs CP 707; bobot rata-rata vs target"},
|
|
{"id": "fcr_tinggi", "fase": "growth", "aksi": "Review program pakan dan cegah waste; bandingkan FCR mingguan dengan standar CP 707.", "metrik_pantau": "FCR akhir minggu; konsumsi karung per 1000 ekor"}
|
|
],
|
|
"insight": "FCR akhir 1.72 dan EEF 280 menandakan efisiensi pakan rendah sepanjang siklus. Penyebab utama adalah akses pakan tidak optimal di fase growth (FCR minggu 3-7 terus naik). Perbaikan siklus berikutnya: pastikan feeder space memadai, evaluasi program nutrisi, dan monitoring FCR mingguan ketat."
|
|
},
|
|
}
|
|
|
|
|
|
def make_stub_llm_response(fixture_key: str):
|
|
"""Return a mock call_ollama that returns a fixed response for the fixture."""
|
|
resp = STUB_RESPONSES.get(fixture_key, STUB_RESPONSES["fcr"])
|
|
return lambda *a, **kw: (json.dumps(resp, ensure_ascii=False), {"prompt_eval_count": 1000, "eval_count": 200})
|
|
|
|
|
|
# ─── Test assertions ──────────────────────────────────────────────────────
|
|
|
|
FORBIDDEN_INSTRUCTION_MARKERS = [
|
|
"BALAS HANYA",
|
|
"STATUS AKTUAL",
|
|
"FAKTA GRADED",
|
|
"PANJANG WAJIB",
|
|
"ANALISIS AKAR MASALAH",
|
|
"ROOT CAUSE",
|
|
"KONTEKS:",
|
|
"NAMA WAJIB:",
|
|
]
|
|
|
|
FORBIDDEN_FOREIGN_TERMS = {
|
|
"fcr": ["mortalitas", "mati", "kematian", "panen", "tanaman", "pertanian"],
|
|
"berat_ayam": ["mortalitas", "mati", "kematian", "tanaman", "pertanian"],
|
|
"hitung_ayam": ["fcr", "tanaman", "pertanian"],
|
|
"eef": ["tanaman", "pertanian"],
|
|
"hitung_karung": ["mortalitas", "suhu", "amonia", "tanaman", "pertanian"],
|
|
"iot_panel": ["mortalitas", "kematian", "karung", "tanaman", "pertanian"],
|
|
}
|
|
|
|
|
|
def assert_no_instruction_echo(text: str, field: str):
|
|
"""Assert LLM output doesn't copy prompt instruction markers."""
|
|
for marker in FORBIDDEN_INSTRUCTION_MARKERS:
|
|
assert marker.lower() not in text.lower(), f"{field} contains instruction marker: {marker}"
|
|
|
|
|
|
def assert_no_foreign_terms(text: str, topic: str, field: str):
|
|
"""Assert LLM output doesn't mention metrics from other topics."""
|
|
forbidden = FORBIDDEN_FOREIGN_TERMS.get(topic, [])
|
|
for term in forbidden:
|
|
assert term.lower() not in text.lower(), f"{field} mentions foreign term '{term}' for topic {topic}"
|
|
|
|
|
|
def assert_min_length(text: str, field: str, min_chars: int = 50):
|
|
"""Assert narrative field has meaningful length."""
|
|
assert len(text.strip()) >= min_chars, f"{field} too short ({len(text.strip())} chars, min {min_chars})"
|
|
|
|
|
|
def assert_status_consistency(graded: dict, kesimpulan: str, insight: str):
|
|
"""Assert critical/warning status is reflected in narrative."""
|
|
analyses = (graded or {}).get("analyses") or {}
|
|
for key, block in analyses.items():
|
|
if not isinstance(block, dict):
|
|
continue
|
|
status = str(block.get("status") or "").lower()
|
|
if status in ("warning", "critical"):
|
|
# The narrative should mention the metric or the issue
|
|
metric_mentioned = key.lower() in kesimpulan.lower() or key.lower() in insight.lower()
|
|
# Also accept severity labels
|
|
severity = str(block.get("severity_label") or "").lower()
|
|
severity_mentioned = severity in kesimpulan.lower() or severity in insight.lower()
|
|
assert metric_mentioned or severity_mentioned, (
|
|
f"Status {status} for {key} not reflected in narrative. "
|
|
f"kesimpulan={kesimpulan[:80]}... insight={insight[:80]}..."
|
|
)
|
|
|
|
|
|
def assert_kesimpulan_verdict_only(kesimpulan: str):
|
|
"""Assert kesimpulan contains only verdict/assessment, no actions."""
|
|
action_words = ["lakukan", "periksa", "sebaiknya", "konsultasikan", "harus", "perlu", "diperlukan", "tindakan"]
|
|
for word in action_words:
|
|
assert word.lower() not in kesimpulan.lower(), f"kesimpulan contains action word '{word}': {kesimpulan}"
|
|
|
|
|
|
def assert_insight_has_actions(insight: str):
|
|
"""Assert insight contains at least one action word."""
|
|
action_words = ["lakukan", "periksa", "sebaiknya", "konsultasikan", "tindakan", "perbaiki", "evaluasi", "pantau"]
|
|
has_action = any(word.lower() in insight.lower() for word in action_words)
|
|
assert has_action, f"insight lacks action words: {insight}"
|
|
|
|
|
|
# ─── Test classes ────────────────────────────────────────────────────────
|
|
|
|
class InsightQualityTestBase(TestCase):
|
|
"""Base class with common setup for quality tests."""
|
|
|
|
kandang_name = "Test Kandang"
|
|
|
|
def setUp(self):
|
|
from apps.farms.models import Kandang, Cycle, Site
|
|
from apps.accounts.models import User
|
|
|
|
# Create minimal test data
|
|
user = User.objects.create_user(user_name="testuser", password="testpass")
|
|
site = Site.objects.create(site_name="Test Site", user=user)
|
|
kandang = Kandang.objects.create(kandang_name=self.kandang_name, site=site)
|
|
cycle = Cycle.objects.create(
|
|
kandang=kandang,
|
|
doc_in_count=25000,
|
|
doc_in_weight=42000, # grams
|
|
start_date=timezone.now().date()
|
|
)
|
|
|
|
self.kandang_id = kandang.pk
|
|
self.cycle_id = cycle.pk
|
|
self.kandang_name = kandang.kandang_name
|
|
|
|
def _run_generate_insight(self, topic: str, context: dict, report_type: str = "page", report_period: str = "current"):
|
|
"""Run generate_insight with stubbed LLM."""
|
|
with patch("apps.operations.services.insight_service.call_ollama", make_stub_llm_response(topic)):
|
|
return insight_service.generate_insight(
|
|
cycle_id=self.cycle_id,
|
|
kandang_id=self.kandang_id,
|
|
topic=topic,
|
|
context=context,
|
|
report_type=report_type,
|
|
report_period=report_period,
|
|
force_refresh=True,
|
|
)
|
|
|
|
def assert_no_quality_issues(self, topic: str, result: dict, ctx: dict):
|
|
"""Run all quality assertions on a result."""
|
|
kesimpulan = result.get("summary") or result.get("kesimpulan") or ""
|
|
insight = result.get("insight") or ""
|
|
|
|
# Length
|
|
assert_min_length(kesimpulan, "kesimpulan", min_chars=30)
|
|
assert_min_length(insight, "insight", min_chars=80)
|
|
|
|
# No instruction echo
|
|
assert_no_instruction_echo(kesimpulan, "kesimpulan")
|
|
assert_no_instruction_echo(insight, "insight")
|
|
|
|
# No foreign terms
|
|
assert_no_foreign_terms(kesimpulan, topic, "kesimpulan")
|
|
assert_no_foreign_terms(insight, topic, "insight")
|
|
|
|
# Field roles
|
|
assert_kesimpulan_verdict_only(kesimpulan)
|
|
assert_insight_has_actions(insight)
|
|
|
|
# Status consistency
|
|
graded = insight_service.grade_context(ctx, topic)
|
|
assert_status_consistency(graded, kesimpulan, insight)
|
|
|
|
# End-cycle structure
|
|
if topic == "end_cycle" or result.get("structured_end_cycle"):
|
|
self.assert_end_cycle_structure(result)
|
|
|
|
|
|
class TopicCoverageTests(InsightQualityTestBase):
|
|
"""Test each topic produces valid, on-topic insights."""
|
|
|
|
def test_hitung_ayam_topic(self):
|
|
ctx = GOLDEN_FIXTURES["hitung_ayam"]
|
|
result = self._run_generate_insight("hitung_ayam", ctx)
|
|
self.assertTrue(result["success"])
|
|
self.assert_no_quality_issues("hitung_ayam", result, ctx)
|
|
|
|
def test_fcr_topic(self):
|
|
ctx = GOLDEN_FIXTURES["fcr"]
|
|
result = self._run_generate_insight("fcr", ctx)
|
|
self.assertTrue(result["success"])
|
|
self.assert_no_quality_issues("fcr", result, ctx)
|
|
|
|
def test_berat_ayam_topic(self):
|
|
ctx = GOLDEN_FIXTURES["berat_ayam"]
|
|
result = self._run_generate_insight("berat_ayam", ctx)
|
|
self.assertTrue(result["success"])
|
|
self.assert_no_quality_issues("berat_ayam", result, ctx)
|
|
|
|
def test_eef_topic(self):
|
|
ctx = GOLDEN_FIXTURES["eef"]
|
|
result = self._run_generate_insight("eef", ctx)
|
|
self.assertTrue(result["success"])
|
|
self.assert_no_quality_issues("eef", result, ctx)
|
|
|
|
def test_hitung_karung_topic(self):
|
|
ctx = GOLDEN_FIXTURES["hitung_karung"]
|
|
result = self._run_generate_insight("hitung_karung", ctx)
|
|
self.assertTrue(result["success"])
|
|
self.assert_no_quality_issues("hitung_karung", result, ctx)
|
|
|
|
def test_iot_panel_topic(self):
|
|
ctx = GOLDEN_FIXTURES["iot_panel"]
|
|
result = self._run_generate_insight("iot_panel", ctx)
|
|
self.assertTrue(result["success"])
|
|
self.assert_no_quality_issues("iot_panel", result, ctx)
|
|
|
|
def test_end_cycle_topic(self):
|
|
ctx = GOLDEN_FIXTURES["end_cycle"]
|
|
result = self._run_generate_insight("dashboard", ctx, report_type="end_cycle")
|
|
self.assertTrue(result["success"])
|
|
self.assert_no_quality_issues("end_cycle", result, ctx)
|
|
|
|
def assert_no_quality_issues(self, topic: str, result: dict, ctx: dict):
|
|
"""Run all quality assertions on a result."""
|
|
kesimpulan = result.get("summary") or result.get("kesimpulan") or ""
|
|
insight = result.get("insight") or ""
|
|
|
|
# Length
|
|
assert_min_length(kesimpulan, "kesimpulan", min_chars=30)
|
|
assert_min_length(insight, "insight", min_chars=80)
|
|
|
|
# No instruction echo
|
|
assert_no_instruction_echo(kesimpulan, "kesimpulan")
|
|
assert_no_instruction_echo(insight, "insight")
|
|
|
|
# No foreign terms
|
|
assert_no_foreign_terms(kesimpulan, topic, "kesimpulan")
|
|
assert_no_foreign_terms(insight, topic, "insight")
|
|
|
|
# Field roles
|
|
assert_kesimpulan_verdict_only(kesimpulan)
|
|
assert_insight_has_actions(insight)
|
|
|
|
# Status consistency
|
|
graded = insight_service.grade_context(ctx, topic)
|
|
assert_status_consistency(graded, kesimpulan, insight)
|
|
|
|
# End-cycle structure
|
|
if topic == "end_cycle" or result.get("structured_end_cycle"):
|
|
self.assert_end_cycle_structure(result)
|
|
|
|
def assert_end_cycle_structure(self, result: dict):
|
|
"""Validate end-cycle JSON structure."""
|
|
structured = result.get("structured_end_cycle")
|
|
self.assertIsNotNone(structured)
|
|
self.assertIn("masalah", structured)
|
|
self.assertIn("akar_penyebab", structured)
|
|
self.assertIn("perbaikan_siklus_berikutnya", structured)
|
|
self.assertIsInstance(structured["masalah"], list)
|
|
self.assertIsInstance(structured["akar_penyebab"], list)
|
|
self.assertIsInstance(structured["perbaikan_siklus_berikutnya"], list)
|
|
|
|
# Each akar_penyebab must have required fields
|
|
for akar in structured["akar_penyebab"]:
|
|
self.assertIn("id", akar)
|
|
self.assertIn("hipotesis", akar)
|
|
self.assertIn("bukti", akar)
|
|
self.assertIn("confidence", akar)
|
|
self.assertIn("fase", akar)
|
|
|
|
# Each perbaikan must have required fields
|
|
for fix in structured["perbaikan_siklus_berikutnya"]:
|
|
self.assertIn("id", fix)
|
|
self.assertIn("fase", fix)
|
|
self.assertIn("aksi", fix)
|
|
self.assertIn("metrik_pantau", fix)
|
|
|
|
|
|
class EdgeCaseTests(InsightQualityTestBase):
|
|
"""Test edge cases: empty context, placeholder bait, etc."""
|
|
|
|
def test_empty_context_fails_loud(self):
|
|
"""Empty context should raise RuntimeError (no silent fallback)."""
|
|
ctx = GOLDEN_FIXTURES["empty_context"]
|
|
# With empty context, the LLM might still return something
|
|
# The test verifies that degenerate output is rejected
|
|
with patch("apps.operations.services.insight_service.call_ollama", make_stub_llm_response("fcr")):
|
|
result = insight_service.generate_insight(
|
|
cycle_id=self.cycle_id,
|
|
kandang_id=self.kandang_id,
|
|
topic="fcr",
|
|
context=ctx,
|
|
force_refresh=True,
|
|
)
|
|
# Should either fail or produce valid output with unknown status
|
|
if result.get("success"):
|
|
# If it succeeds, verify the graded status shows unknown
|
|
graded = insight_service.grade_context(ctx, "fcr")
|
|
fcr_block = graded.get("analyses", {}).get("fcr", {})
|
|
self.assertEqual(fcr_block.get("status"), "unknown")
|
|
else:
|
|
# If it fails, it should be a RuntimeError
|
|
pass
|
|
|
|
def test_placeholder_bait_rejected(self):
|
|
"""Output with '...' placeholder should be rejected on retry then fail."""
|
|
ctx = GOLDEN_FIXTURES["fcr"].copy()
|
|
ctx["fcr_terakhir"] = None # Force missing data
|
|
|
|
def placeholder_response(*a, **kw):
|
|
return ('{"kesimpulan": "...", "insight": "..."}', {"prompt_eval_count": 100, "eval_count": 10})
|
|
|
|
with patch("apps.operations.services.insight_service.call_ollama", placeholder_response):
|
|
with self.assertRaises(RuntimeError) as cm:
|
|
insight_service.generate_insight(
|
|
cycle_id=self.cycle_id,
|
|
kandang_id=self.kandang_id,
|
|
topic="fcr",
|
|
context=ctx,
|
|
force_refresh=True,
|
|
)
|
|
self.assertIn("degenerate", str(cm.exception).lower())
|
|
|
|
def test_half_good_h28_produces_valid(self):
|
|
"""Half-good fixture should produce valid output."""
|
|
ctx = GOLDEN_FIXTURES["half_good_h28"]
|
|
result = self._run_generate_insight("fcr", ctx)
|
|
self.assertTrue(result["success"])
|
|
self.assert_no_quality_issues("fcr", result, ctx)
|
|
|
|
def test_half_bad_h28_grades_critical(self):
|
|
"""Half-bad fixture should show critical in graded and narrative."""
|
|
ctx = GOLDEN_FIXTURES["half_bad_h28"]
|
|
graded = insight_service.grade_context(ctx, "fcr")
|
|
# Check that grading produces critical status
|
|
fcr_block = graded.get("analyses", {}).get("fcr", {})
|
|
self.assertIn(fcr_block.get("status"), ("warning", "critical"))
|
|
|
|
result = self._run_generate_insight("fcr", ctx)
|
|
self.assertTrue(result["success"])
|
|
assert_status_consistency(graded, result.get("summary", ""), result.get("insight", ""))
|
|
|
|
|
|
class GradingScopeTests(TestCase):
|
|
"""Test TOPIC_METRICS grading scope is enforced."""
|
|
|
|
def test_fcr_topic_only_grades_fcr(self):
|
|
ctx = GOLDEN_FIXTURES["fcr"]
|
|
graded = insight_service.grade_context(ctx, "fcr")
|
|
analyses = graded.get("analyses", {})
|
|
self.assertIn("fcr", analyses)
|
|
self.assertNotIn("bw", analyses)
|
|
self.assertNotIn("mortality", analyses)
|
|
|
|
def test_berat_ayam_topic_grades_bw_and_adg(self):
|
|
ctx = GOLDEN_FIXTURES["berat_ayam"]
|
|
graded = insight_service.grade_context(ctx, "berat_ayam")
|
|
analyses = graded.get("analyses", {})
|
|
self.assertIn("bw", analyses)
|
|
self.assertIn("adg", analyses)
|
|
self.assertNotIn("fcr", analyses)
|
|
self.assertNotIn("mortality", analyses)
|
|
|
|
def test_hitung_ayam_topic_grades_mortality_only(self):
|
|
ctx = GOLDEN_FIXTURES["hitung_ayam"]
|
|
graded = insight_service.grade_context(ctx, "hitung_ayam")
|
|
analyses = graded.get("analyses", {})
|
|
self.assertIn("mortality", analyses)
|
|
# daily_mortality_doc only graded when mortalitas_hari_ini_ekor is provided
|
|
# Fixture doesn't have daily deaths data, so it won't be present
|
|
self.assertNotIn("bw", analyses)
|
|
self.assertNotIn("fcr", analyses)
|
|
|
|
def test_eef_topic_grades_eef_and_fcr(self):
|
|
ctx = GOLDEN_FIXTURES["eef"]
|
|
graded = insight_service.grade_context(ctx, "eef")
|
|
analyses = graded.get("analyses", {})
|
|
self.assertIn("eef", analyses)
|
|
self.assertIn("fcr", analyses)
|
|
self.assertNotIn("mortality", analyses)
|
|
|
|
def test_unknown_only_for_core_metrics(self):
|
|
"""Unknown status only emitted for topic's core metrics."""
|
|
ctx = {"hari_ke": 28} # Only day_age
|
|
graded = insight_service.grade_context(ctx, "fcr")
|
|
analyses = graded.get("analyses", {})
|
|
fcr_block = analyses.get("fcr", {})
|
|
self.assertEqual(fcr_block.get("status"), "unknown")
|
|
# No bw/mortality unknown blocks
|
|
self.assertNotIn("bw", analyses)
|
|
self.assertNotIn("mortality", analyses)
|
|
|
|
|
|
class PromptContractTests(TestCase):
|
|
"""Test prompt template contract compliance."""
|
|
|
|
def test_build_user_prompt_has_ordered_sections(self):
|
|
"""Prompt should have header → data → status → root cause → contract → schema."""
|
|
topic = "fcr"
|
|
context = GOLDEN_FIXTURES["fcr"]
|
|
graded = insight_service.grade_context(context, topic)
|
|
prompt = insight_service.build_insight_user_prompt(
|
|
topic=topic,
|
|
kandang_name="Kandang 01",
|
|
report_type="page",
|
|
report_period="current",
|
|
context=context,
|
|
graded=graded,
|
|
is_end_cycle=False,
|
|
)
|
|
|
|
# Check order
|
|
sections = [
|
|
"Buat insight topik",
|
|
"[Data Halaman (JSON)]",
|
|
"STATUS AKTUAL",
|
|
"FAKTA GRADED",
|
|
"PANJANG WAJIB",
|
|
"BALAS HANYA JSON",
|
|
]
|
|
positions = [prompt.find(s) for s in sections]
|
|
for i, (pos, sec) in enumerate(zip(positions, sections)):
|
|
self.assertGreaterEqual(pos, 0, f"Missing section: {sec}")
|
|
if i > 0:
|
|
self.assertGreater(pos, positions[i-1], f"Section order wrong: {sec} before {sections[i-1]}")
|
|
|
|
def test_end_cycle_prompt_includes_root_cause(self):
|
|
"""End-cycle prompt should include ANALISIS AKAR MASALAH section."""
|
|
topic = "dashboard"
|
|
context = GOLDEN_FIXTURES["end_cycle"]
|
|
graded = insight_service.grade_context(context, topic)
|
|
hypotheses = root_cause.build_root_cause_hypotheses(graded, context)
|
|
prompt = insight_service.build_insight_user_prompt(
|
|
topic=topic,
|
|
kandang_name="Kandang 01",
|
|
report_type="end_cycle",
|
|
report_period="end",
|
|
context=context,
|
|
graded=graded,
|
|
is_end_cycle=True,
|
|
hypotheses=hypotheses,
|
|
)
|
|
self.assertIn("[ANALISIS AKAR MASALAH]", prompt)
|
|
self.assertIn("brooding", prompt.lower()) # hypothesis label
|
|
|
|
def test_field_role_contract_in_prompt(self):
|
|
"""Prompt should contain field role contract (kesimpulan=verdict only, insight=causes+actions)."""
|
|
prompt = insight_service.build_insight_user_prompt(
|
|
topic="fcr",
|
|
kandang_name="Kandang 01",
|
|
report_type="page",
|
|
report_period="current",
|
|
context={},
|
|
graded={},
|
|
is_end_cycle=False,
|
|
)
|
|
self.assertIn("PENILAIAN DATA SAJA (tanpa sebab/aksi)", prompt)
|
|
self.assertIn("DILARANG menulis kata aksi", prompt)
|
|
self.assertIn("insight HARUS mengandung minimal satu kata aksi", prompt)
|
|
|
|
def test_no_echo_prone_status_sentence(self):
|
|
"""Prompt should have FAKTA GRADED as list, not echo-prone sentence."""
|
|
prompt = insight_service.build_insight_user_prompt(
|
|
topic="fcr",
|
|
kandang_name="Kandang 01",
|
|
report_type="page",
|
|
report_period="current",
|
|
context={"fcr_terakhir": 1.4},
|
|
graded={"analyses": {"fcr": {"status": "warning", "message": "FCR tinggi"}}},
|
|
is_end_cycle=False,
|
|
)
|
|
# FAKTA GRADED should be list format
|
|
self.assertIn("- fcr = warning:", prompt)
|
|
# STATUS AKTUAL line is present but that's OK - it's a single line, not the echo-prone pattern
|
|
# The echo risk is from "STATUS AKTUAL: x. Kesimpulan harus..." which we avoid
|
|
|
|
|
|
class GraderAccuracyTests(TestCase):
|
|
"""Test new grader functions produce expected outputs."""
|
|
|
|
def test_analyze_eef_with_book_standard(self):
|
|
"""EEF within book range (day 28) uses book standard."""
|
|
result = cp707.analyze_eef(eef_terakhir=320, day_age=28, eef_seri=[{"hari": 26, "eef": 310}, {"hari": 27, "eef": 315}, {"hari": 28, "eef": 320}])
|
|
self.assertEqual(result["source"], "book_standard")
|
|
self.assertEqual(result["standard"], 374) # IP standard day 28 (from PERFORMANCE_STANDARD_DAILY)
|
|
self.assertIn("status", result)
|
|
# 320 vs 374 = -14.4% → warning (between -15% and -5%)
|
|
self.assertEqual(result["status"], "warning")
|
|
|
|
def test_analyze_eef_outside_book_uses_manager_scale(self):
|
|
"""EEF outside book range (day 40) uses manager absolute scale."""
|
|
result = cp707.analyze_eef(eef_terakhir=380, day_age=40, eef_seri=[{"hari": 38, "eef": 370}, {"hari": 39, "eef": 375}, {"hari": 40, "eef": 380}])
|
|
self.assertEqual(result["source"], "manager_absolute_scale")
|
|
self.assertEqual(result["status"], "ok")
|
|
self.assertIn("BAIK (skala manajer", result["message"]) # 380 is in 350-399 range = BAIK
|
|
|
|
def test_analyze_eef_invalid_above_500(self):
|
|
"""EEF >500 is invalid - only when book standard unavailable (day >37)."""
|
|
# Day 40 has no book standard, so uses manager absolute scale
|
|
result = cp707.analyze_eef(eef_terakhir=550, day_age=40)
|
|
self.assertEqual(result["status"], "invalid")
|
|
self.assertIn("tidak realistis", result["message"])
|
|
|
|
def test_analyze_eef_sharp_drop_escalates(self):
|
|
"""Sharp drop (>20 pts in 3 days) escalates verdict."""
|
|
# Day 28 has book standard IP=374
|
|
# EEF 320 vs 374 = -14.4% → warning
|
|
# But sharp drop (345→320 = -25 in 3 days) escalates warning → critical
|
|
eef_seri = [{"hari": 26, "eef": 345}, {"hari": 27, "eef": 335}, {"hari": 28, "eef": 320}]
|
|
result = cp707.analyze_eef(eef_terakhir=320, day_age=28, eef_seri=eef_seri)
|
|
# Base: 320 vs 374 = -14.4% → warning
|
|
# Sharp drop escalates warning → critical
|
|
self.assertEqual(result["status"], "critical")
|
|
self.assertIn("eskalasi", result["message"].lower())
|
|
|
|
def test_analyze_adg_uses_book_standard(self):
|
|
"""ADG uses book daily standard."""
|
|
result = cp707.analyze_adg(actual_adg=85, day_age=28)
|
|
self.assertEqual(result["standard"], 88) # ADG standard day 28
|
|
# 85 vs 88 = -3.4% → between -5% and +10% = ok (normal range)
|
|
self.assertEqual(result["status"], "ok")
|
|
self.assertIn("sesuai standar", result["message"])
|
|
|
|
def test_analyze_daily_mortality_doc_thresholds(self):
|
|
"""Daily mortality % DOC thresholds: >0.05% warning, >=0.1% critical."""
|
|
# 50 deaths / 25000 DOC = 0.2% → critical
|
|
result = cp707.analyze_daily_mortality_doc(50, 25000)
|
|
self.assertEqual(result["status"], "critical")
|
|
self.assertEqual(result["actual_pct"], 0.2)
|
|
|
|
# 10 deaths / 25000 DOC = 0.04% → ok
|
|
result = cp707.analyze_daily_mortality_doc(10, 25000)
|
|
self.assertEqual(result["status"], "ok")
|
|
|
|
# 15 deaths / 25000 DOC = 0.06% → warning
|
|
result = cp707.analyze_daily_mortality_doc(15, 25000)
|
|
self.assertEqual(result["status"], "warning")
|
|
|
|
def test_analyze_feed_balance(self):
|
|
"""Feed balance computes correctly."""
|
|
result = cp707.analyze_feed_balance(karung_masuk=100, karung_tuang=85, karung_sisa=10, target_harian=5)
|
|
self.assertEqual(result["balance_karung"], 5)
|
|
self.assertEqual(result["status"], "ok")
|
|
|
|
def test_grade_context_emits_unknown_for_core_only(self):
|
|
"""Missing core metric gets unknown; foreign metrics not emitted."""
|
|
ctx = {"hari_ke": 28}
|
|
graded = insight_service.grade_context(ctx, "fcr")
|
|
analyses = graded.get("analyses", {})
|
|
self.assertEqual(analyses.get("fcr", {}).get("status"), "unknown")
|
|
self.assertNotIn("bw", analyses)
|
|
self.assertNotIn("mortality", analyses)
|
|
|
|
|
|
class RootCauseTests(TestCase):
|
|
"""Test root cause hypotheses are grounded in graded facts."""
|
|
|
|
def test_root_cause_only_from_graded_facts(self):
|
|
"""Hypotheses only reference metrics that have warning/critical status."""
|
|
graded = {
|
|
"analyses": {
|
|
"fcr": {"status": "critical", "message": "FCR 1.72 vs 1.55"},
|
|
"mortality": {"status": "ok", "message": "Mortalitas 3%"},
|
|
}
|
|
}
|
|
context = {"weekly_summaries": [{"minggu_ke": 3, "fcr_akhir_rasio": 1.5}, {"minggu_ke": 4, "fcr_akhir_rasio": 1.6}]}
|
|
hypotheses = root_cause.build_root_cause_hypotheses(graded, context)
|
|
|
|
# Only FCR-related hypotheses (mortality is ok)
|
|
for h in hypotheses:
|
|
self.assertIn(h["id"], ("pakan_akses", "fcr_tinggi"))
|
|
# bukti should reference FCR
|
|
bukti_text = " ".join(h.get("bukti", [])).lower()
|
|
self.assertIn("fcr", bukti_text)
|
|
|
|
def test_sanitize_end_cycle_drops_invented_causes(self):
|
|
"""LLM-invented causes not in candidates are dropped."""
|
|
hypotheses = [
|
|
{"id": "pakan_akses", "hipotesis": "Pakan / akses pakan", "bukti": ["FCR tinggi"], "confidence": "high", "fase": "growth"},
|
|
]
|
|
actions = [{"id": "pakan_akses", "fase": "growth", "aksi": "Cek feeder", "metrik_pantau": "FCR"}]
|
|
masalah = ["FCR kritis"]
|
|
|
|
# LLM returns invented cause
|
|
llm_output = {
|
|
"kesimpulan": "FCR kritis",
|
|
"masalah": ["FCR kritis"],
|
|
"akar_penyebab": [
|
|
{"id": "pakan_akses", "hipotesis": "Pakan / akses pakan", "bukti": ["FCR tinggi"], "confidence": "high", "fase": "growth"},
|
|
{"id": "invented", "hipotesis": "Cuaca buruk", "bukti": ["Hujan"], "confidence": "low", "fase": "finisher"}, # NOT in candidates
|
|
],
|
|
"perbaikan_siklus_berikutnya": [
|
|
{"id": "pakan_akses", "fase": "growth", "aksi": "Cek feeder", "metrik_pantau": "FCR"},
|
|
{"id": "invented_fix", "fase": "finisher", "aksi": "Tutup ventilasi", "metrik_pantau": "Suhu"}, # NOT in actions
|
|
],
|
|
"insight": "FCR tinggi karena pakan dan cuaca",
|
|
}
|
|
|
|
sanitized = root_cause.sanitize_end_cycle_payload(llm_output, hypotheses=hypotheses, actions=actions, masalah=masalah)
|
|
|
|
# Invented cause dropped
|
|
self.assertEqual(len(sanitized["akar_penyebab"]), 1)
|
|
self.assertEqual(sanitized["akar_penyebab"][0]["id"], "pakan_akses")
|
|
|
|
# Invented fix dropped
|
|
self.assertEqual(len(sanitized["perbaikan_siklus_berikutnya"]), 1)
|
|
self.assertEqual(sanitized["perbaikan_siklus_berikutnya"][0]["id"], "pakan_akses")
|
|
|
|
|
|
class DailyAkarMasalahTests(TestCase):
|
|
"""Test daily AI insight 3-section format (kesimpulan, akar_masalah, insight)."""
|
|
|
|
def test_parse_llm_json_with_akar_masalah(self):
|
|
payload = json.dumps({
|
|
"kesimpulan": "FCR 1.40 pada hari ke-28 berada di atas standar CP 707 (1.315).",
|
|
"akar_masalah": "Fluktuasi suhu kandang malam hari dan akses pakan tidak merata.",
|
|
"insight": "Lakukan kalibrasi pemanas dan ratakan distribusi pakan di sepanjang brooding area."
|
|
})
|
|
parsed = insight_service.parse_llm_json(payload)
|
|
self.assertIsNotNone(parsed)
|
|
self.assertEqual(parsed["kesimpulan"], "FCR 1.40 pada hari ke-28 berada di atas standar CP 707 (1.315).")
|
|
self.assertEqual(parsed["akar_masalah"], "Fluktuasi suhu kandang malam hari dan akses pakan tidak merata.")
|
|
self.assertIn("kalibrasi pemanas", parsed["insight"])
|
|
|
|
def test_parse_llm_json_variants(self):
|
|
payload = json.dumps({
|
|
"summary": "Kondisi pakan normal.",
|
|
"root_cause": "Kondisi operasional optimal, tidak terdeteksi indikasi deviasi.",
|
|
"rekomendasi": "Pertahankan SOP saat ini."
|
|
})
|
|
parsed = insight_service.parse_llm_json(payload)
|
|
self.assertIsNotNone(parsed)
|
|
self.assertEqual(parsed["kesimpulan"], "Kondisi pakan normal.")
|
|
self.assertEqual(parsed["akar_masalah"], "Kondisi operasional optimal, tidak terdeteksi indikasi deviasi.")
|
|
self.assertEqual(parsed["insight"], "Pertahankan SOP saat ini.")
|
|
|
|
def test_parse_llm_json_legacy_fallback(self):
|
|
payload = json.dumps({
|
|
"kesimpulan": "Data normal.",
|
|
"insight": "Lanjutkan pemantauan harian."
|
|
})
|
|
parsed = insight_service.parse_llm_json(payload)
|
|
self.assertIsNotNone(parsed)
|
|
self.assertEqual(parsed["kesimpulan"], "Data normal.")
|
|
self.assertEqual(parsed["akar_masalah"], "")
|
|
self.assertEqual(parsed["insight"], "Lanjutkan pemantauan harian.")
|
|
|
|
def test_embedded_daily_serialization(self):
|
|
clean_insight = "Lakukan penyesuaian tirai kandang dan ventilasi minimal."
|
|
akar_masalah = "Suhu brooding terlalu dingin di malam hari (26C vs 32C)."
|
|
daily_json = json.dumps({"akar_masalah": akar_masalah, "insight": clean_insight}, ensure_ascii=False)
|
|
stored_text = f"{clean_insight}\n\n{insight_service.DAILY_JSON_MARKER}\n{daily_json}"
|
|
|
|
mock_row = MagicMock()
|
|
mock_row.pk = 1
|
|
mock_row.cycle_id = 10
|
|
mock_row.kandang_id = 2
|
|
mock_row.topic = "fcr"
|
|
mock_row.report_type = "page"
|
|
mock_row.report_period = "current"
|
|
mock_row.summary = "FCR tinggi pada hari ke-7."
|
|
mock_row.insight_text = stored_text
|
|
mock_row.alert = "warning"
|
|
mock_row.citations = "[]"
|
|
mock_row.date = None
|
|
mock_row.created_at = None
|
|
mock_row.updated_at = None
|
|
|
|
serialized = insight_service.serialize_insight(mock_row)
|
|
self.assertTrue(serialized["success"])
|
|
self.assertEqual(serialized["summary"], "FCR tinggi pada hari ke-7.")
|
|
self.assertEqual(serialized["akar_masalah"], akar_masalah)
|
|
self.assertEqual(serialized["insight"], clean_insight)
|
|
self.assertIn(akar_masalah, serialized["insight_text"])
|
|
self.assertIn(clean_insight, serialized["insight_text"])
|
|
|
|
def test_daily_prompt_structure(self):
|
|
prompt = insight_service.build_insight_user_prompt(
|
|
topic="fcr",
|
|
kandang_name="Kandang 01",
|
|
report_type="page",
|
|
report_period="current",
|
|
context={"fcr_terakhir": 1.4},
|
|
graded={"analyses": {"fcr": {"status": "warning", "message": "FCR tinggi"}}},
|
|
is_end_cycle=False,
|
|
)
|
|
self.assertIn("akar_masalah", prompt)
|
|
self.assertIn("ANALISIS AKAR PENYEBAB", prompt)
|
|
self.assertIn('"akar_masalah":"<analisis faktor pemicu', prompt)
|
|
|
|
|
|
class TraceLintTests(TestCase):
|
|
"""Test rag_trace.jsonl linting logic."""
|
|
|
|
def test_prompt_version_v3(self):
|
|
"""After rework, prompt_version should be v3.0."""
|
|
# This is a placeholder - actual trace log check would read the file
|
|
# The generate_insight function currently logs v2.1
|
|
# After full rework it should be v3.0
|
|
self.assertEqual(insight_service.generate_insight.__module__, "apps.operations.services.insight_service")
|
|
|
|
def test_completion_tokens_threshold(self):
|
|
"""Completion tokens < 60 indicates degenerate output."""
|
|
# Stub would return 200 tokens - well above threshold
|
|
self.assertGreater(200, 60)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import unittest
|
|
unittest.main() |