Method process_deficiency

train/eval.py:271–358

Process a single deficiency test sample: call API, classify deficiencies, and calculate metrics.

(self, item: Dict, max_retries: int = 3)

        self.compare_prompt = self._load_prompt_from_file("prompts/compare_question_prompt.txt")

    def process_deficiency(self, item: Dict, max_retries: int = 3) -> Dict:
        """Process a single deficiency test sample: call API, classify deficiencies, and calculate metrics."""
        try:
            # Convert image to base64
            image_path = os.path.join(self.image_root, item["image"]) if self.image_root else item["image"]
            if not os.path.exists(image_path):
                print(f"Debug - Deficiency image path: {image_path} (exists: {os.path.exists(image_path)})")
            base64_image_uri = image_to_base64_uri(image_path)
            if not base64_image_uri:
                return None

            messages = [
                {"role": "system", "content": self.system_prompt},
                {
                    "role": "user",
                    "content": [
                        {"type": "text", "text": self.deficiency_prompt},
                        {"type": "image_url", "image_url": {"url": base64_image_uri}}
                    ]
                }
            ]

            # Call API with retries
            generated_text = ""
            for attempt in range(max_retries):
                try:
                    completion = self.client.chat.completions.create(
                        model=self.test_model,
                        messages=messages,
                        temperature=1.0,
                        # top_p=0.95,
                        max_completion_tokens=1024*2,
                    )

                    generated_text = completion.choices[0].message.content or ""
                    break

                except Exception as e:
                    print(f"API call failed for slide {item.get('slide_id', 'unknown')} (attempt {attempt + 1}/{max_retries}): {e}")
                    if attempt < max_retries - 1:
                        time.sleep(2 ** attempt)  # Exponential backoff
                    else:
                        print(f"Failed after {max_retries} attempts for slide {item.get('slide_id', 'unknown')}")
                        return None

            # Predict major categories directly via boolean classification
            predicted_categories = set(classify_categories(generated_text))

            # Get ground truth specific deficiencies and map to categories
            ground_truth_deficiencies = [d["deficiency"] for d in item["deficiencies"]]
            ground_truth_categories = {
                DEFICIENCY_TO_CATEGORY_MAP[d] for d in ground_truth_deficiencies
                if d in DEFICIENCY_TO_CATEGORY_MAP
            }

            # Calculate metrics for CATEGORIES
            cat_true_positives = len(predicted_categories.intersection(ground_truth_categories))
            cat_false_positives = len(predicted_categories - ground_truth_categories)