| """
|
| Agent 2 - Visual Damage-Type Classifier
|
| ------------------------------------------
|
| Takes the flagged region crop from Agent 1 (plus its change_score and
|
| reliability signals) and identifies WHICH KIND of damage it is, using a
|
| Vision-Language Model (Gemini, same free-tier choice as Agent 3) - zero
|
| training, zero fine-tuning, same philosophy as the rest of Module 2.
|
|
|
| Classifies into exactly one of the 9 shared taxonomy categories:
|
| screen_display, structural_body, port_connector, missing_component,
|
| liquid_moisture, input_control, power_boot_failure, cosmetic_wear,
|
| other_unclassified
|
|
|
| Self-consistency: calls the VLM multiple times on the same crop and checks
|
| agreement - if the calls disagree on damage_type, that's flagged for human
|
| review rather than silently picking one guess. Same honest-about-uncertainty
|
| pattern used throughout Agent 1.
|
| """
|
|
|
| import os
|
| import json
|
| import re
|
| from collections import Counter
|
| from dataclasses import dataclass, field
|
|
|
| from google import genai
|
| from google.genai import types
|
|
|
|
|
| VLM_MODEL = "gemini-3.5-flash"
|
| SELF_CONSISTENCY_CALLS = 3
|
| AGREEMENT_THRESHOLD = 0.6
|
|
|
|
|
|
|
| DAMAGE_CATEGORIES = {
|
| "screen_display": "Cracked/scratched screen, dead pixels, flickering, no image/blank display, discoloration.",
|
| "structural_body": "Cracked/broken casing, dents, warping, broken hinge, missing structural piece.",
|
| "port_connector": "Damaged, bent, or loose port/connector (charging port, USB, HDMI, etc.).",
|
| "missing_component": "A part, cable, or accessory that should be present is visibly missing.",
|
| "liquid_moisture": "Visible liquid residue, staining, or corrosion consistent with a spill.",
|
| "input_control": "Broken/missing key, stuck button, damaged touchpad/touchscreen surface, broken control.",
|
| "power_boot_failure": "Visible signs consistent with power/battery failure (e.g. swelling, burn marks near battery) - "
|
| "NOTE: most power/boot failures have no visible symptom at all and should come from Agent 3, "
|
| "not this agent; only use this category if there IS a visible physical sign.",
|
| "cosmetic_wear": "Minor scuffs, scratches, or wear that does not affect function.",
|
| "other_unclassified": "Visible damage that doesn't clearly fit any category above, or the image is unclear/ambiguous.",
|
| }
|
|
|
| SEVERITY_LEVELS = ["Cosmetic", "Moderate", "Severe", "Critical"]
|
|
|
|
|
| @dataclass
|
| class DamageClassification:
|
| damage_type: str = None
|
| damage_level: str = None
|
| confidence: float = 0.0
|
| description: str = ""
|
| is_new_damage: bool = True
|
| raw_response: str = ""
|
| parse_ok: bool = True
|
| error: str = None
|
|
|
|
|
| @dataclass
|
| class ConsistencyResult:
|
| final_damage_type: str = None
|
| final_damage_level: str = None
|
| final_description: str = ""
|
| is_new_damage: bool = True
|
| agreement_ratio: float = 0.0
|
| requires_human_review: bool = True
|
| individual_calls: list = field(default_factory=list)
|
|
|
|
|
| def _build_prompt(item_type, change_score, alignment_warning=None, resolution_warning=None,
|
| has_full_context=False):
|
| """
|
| Build the classification prompt, including Agent 1's own diagnostics as
|
| context - NOT as ground truth, since Agent 1's confidence issues should
|
| make Agent 2 more cautious, not less.
|
|
|
| has_full_context: True when the full baseline and return photos are
|
| ALSO provided (not just the crop) - adds an explicit instruction to
|
| check whether the flagged mark is genuinely NEW (only visible in the
|
| return photo) or PRE-EXISTING (already visible in the baseline photo,
|
| e.g. an old scratch the item already had at pickup). Pre-existing
|
| damage should not be flagged as new damage from this session, even if
|
| Agent 1's pixel-level diff picked up a region near it.
|
| """
|
| category_list = "\n".join(f' - "{name}": {desc}' for name, desc in DAMAGE_CATEGORIES.items())
|
|
|
| context_notes = []
|
| if alignment_warning:
|
| context_notes.append(
|
| f"NOTE: the automated pre-screening flagged an alignment issue: \"{alignment_warning}\" "
|
| f"- treat the change_score below as less reliable than usual."
|
| )
|
| if resolution_warning:
|
| context_notes.append(f"NOTE: {resolution_warning}")
|
| context_block = ("\n".join(context_notes) + "\n") if context_notes else ""
|
|
|
| new_vs_old_instruction = ""
|
| if has_full_context:
|
| new_vs_old_instruction = (
|
| "\nYou have also been given the FULL baseline (pickup) and return photos, not just "
|
| "the cropped region - use them to check whether the damage shown in the crop is "
|
| "GENUINELY NEW (visible in the return photo but NOT in the baseline photo) or "
|
| "PRE-EXISTING (the same mark is already visible in the baseline photo - e.g. an old "
|
| "scratch or dent the item already had at pickup, which the automated crop detector "
|
| "may have flagged again due to lighting/angle differences, not because anything "
|
| "actually changed). If the damage is pre-existing rather than new, set "
|
| "\"is_new_damage\": false and explain this in your description - pre-existing damage "
|
| "should not be attributed to this rental session.\n"
|
| )
|
|
|
| return f"""You are inspecting a cropped photo region of a {item_type} that was automatically \
|
| flagged as visually changed between a pickup photo and a return photo.
|
|
|
| Automated pre-screening change_score: {change_score} (0-1 scale, higher = more visual difference \
|
| detected by pixel-level comparison - this is a rough signal, not a judgment of damage type or severity).
|
| {context_block}{new_vs_old_instruction}
|
| Classify the damage shown in this image into EXACTLY ONE of these categories:
|
| {category_list}
|
|
|
| Also rate severity as one of: {", ".join(SEVERITY_LEVELS)}.
|
| - Cosmetic: minor, does not affect function.
|
| - Moderate: visible damage, item still usable.
|
| - Severe: damage impairs normal use.
|
| - Critical: item non-functional or unsafe.
|
|
|
| If the image is unclear, ambiguous, or doesn't show a clear physical defect, use "other_unclassified" \
|
| and a low confidence score rather than guessing.
|
|
|
| Respond with ONLY a JSON object in this exact shape, nothing else - no preamble, no markdown fences:
|
| {{"damage_type": "screen_display", "damage_level": "Severe", "confidence": 0.85, "is_new_damage": true, "description": "A short, human-readable description of what you see, for an admin dashboard."}}
|
| """
|
|
|
|
|
| def _read_image_part(image_path):
|
| with open(image_path, "rb") as f:
|
| data = f.read()
|
| mime_type = "image/png" if image_path.lower().endswith(".png") else "image/jpeg"
|
| return types.Part.from_bytes(data=data, mime_type=mime_type)
|
|
|
|
|
| def _parse_json_response(text):
|
| cleaned = text.strip()
|
| if cleaned.startswith("```"):
|
| cleaned = cleaned.strip("`")
|
| if cleaned.lower().startswith("json"):
|
| cleaned = cleaned[4:]
|
| return json.loads(cleaned.strip())
|
|
|
|
|
| def classify_damage_single_call(crop_path, item_type, change_score, client=None,
|
| baseline_path=None, return_path=None,
|
| alignment_warning=None, resolution_warning=None) -> DamageClassification:
|
| """
|
| One VLM call classifying a single crop. Optionally includes the full
|
| baseline/return photos for context - both for cases like a crack near
|
| a hinge reading differently than one mid-screen, AND (the main reason
|
| this is now used by default from Agent 2's real entry points) so the
|
| model can distinguish genuinely NEW damage from PRE-EXISTING damage the
|
| item already had at pickup.
|
| """
|
| if client is None:
|
| client = genai.Client(api_key=os.environ.get("GOOGLE_API_KEY"))
|
|
|
| has_full_context = baseline_path is not None and return_path is not None
|
| prompt = _build_prompt(item_type, change_score, alignment_warning, resolution_warning,
|
| has_full_context=has_full_context)
|
|
|
| contents = [_read_image_part(crop_path)]
|
| if baseline_path:
|
| contents.append("(This is the BASELINE/pickup photo for context:)")
|
| contents.append(_read_image_part(baseline_path))
|
| if return_path:
|
| contents.append("(This is the RETURN photo for context:)")
|
| contents.append(_read_image_part(return_path))
|
| contents.append(prompt)
|
|
|
| raw_text = ""
|
| try:
|
| response = client.models.generate_content(
|
| model=VLM_MODEL,
|
| contents=contents,
|
| config=types.GenerateContentConfig(temperature=0.2),
|
|
|
|
|
| )
|
| raw_text = response.text
|
| parsed = _parse_json_response(raw_text)
|
|
|
| damage_type = parsed.get("damage_type")
|
| if damage_type not in DAMAGE_CATEGORIES:
|
| damage_type = "other_unclassified"
|
|
|
| damage_level = parsed.get("damage_level")
|
| if damage_level not in SEVERITY_LEVELS:
|
| damage_level = "Moderate"
|
|
|
| return DamageClassification(
|
| damage_type=damage_type,
|
| damage_level=damage_level,
|
| confidence=float(parsed.get("confidence", 0.5)),
|
| description=parsed.get("description", ""),
|
| is_new_damage=bool(parsed.get("is_new_damage", True)),
|
| raw_response=raw_text,
|
| parse_ok=True,
|
| )
|
|
|
| except (json.JSONDecodeError, KeyError, ValueError) as e:
|
| return DamageClassification(raw_response=raw_text, parse_ok=False,
|
| error=f"Failed to parse VLM response: {e}")
|
| except Exception as e:
|
| return DamageClassification(parse_ok=False, error=f"VLM call failed: {e}")
|
|
|
|
|
| def classify_damage_with_consistency(crop_path, item_type, change_score,
|
| baseline_path=None, return_path=None,
|
| alignment_warning=None, resolution_warning=None,
|
| n_calls=SELF_CONSISTENCY_CALLS,
|
| api_key=None) -> ConsistencyResult:
|
| """
|
| Main entry point for Agent 2. Calls the VLM classifier N times on the
|
| same crop and checks agreement, instead of trusting a single call.
|
|
|
| If the calls agree strongly on damage_type: trust it, requires_human_review=False.
|
| If they disagree: still return the most common answer as a best guess,
|
| but flag requires_human_review=True so a human/Agent 4 double-checks
|
| rather than the system silently acting on an uncertain classification.
|
| """
|
| client = genai.Client(api_key=api_key or os.environ.get("GOOGLE_API_KEY"))
|
|
|
| calls = []
|
| for _ in range(n_calls):
|
| result = classify_damage_single_call(
|
| crop_path, item_type, change_score, client=client,
|
| baseline_path=baseline_path, return_path=return_path,
|
| alignment_warning=alignment_warning, resolution_warning=resolution_warning,
|
| )
|
| calls.append(result)
|
|
|
| valid_calls = [c for c in calls if c.parse_ok]
|
| if not valid_calls:
|
| return ConsistencyResult(
|
| final_damage_type="other_unclassified",
|
| final_damage_level="Moderate",
|
| final_description="Classification failed - all VLM calls errored.",
|
| is_new_damage=True,
|
| agreement_ratio=0.0,
|
| requires_human_review=True,
|
| individual_calls=[c.__dict__ for c in calls],
|
| )
|
|
|
| type_counts = Counter(c.damage_type for c in valid_calls)
|
| most_common_type, count = type_counts.most_common(1)[0]
|
| agreement_ratio = count / len(valid_calls)
|
|
|
|
|
|
|
| agreeing_calls = [c for c in valid_calls if c.damage_type == most_common_type]
|
| best_call = max(agreeing_calls, key=lambda c: c.confidence)
|
|
|
|
|
|
|
|
|
|
|
| new_damage_votes = [c.is_new_damage for c in agreeing_calls]
|
| is_new_damage = sum(new_damage_votes) >= len(new_damage_votes) / 2
|
|
|
| return ConsistencyResult(
|
| final_damage_type=most_common_type,
|
| final_damage_level=best_call.damage_level,
|
| final_description=best_call.description,
|
| is_new_damage=is_new_damage,
|
| agreement_ratio=round(agreement_ratio, 2),
|
| requires_human_review=(agreement_ratio < AGREEMENT_THRESHOLD),
|
| individual_calls=[
|
| {"damage_type": c.damage_type, "damage_level": c.damage_level,
|
| "confidence": c.confidence, "is_new_damage": c.is_new_damage}
|
| for c in valid_calls
|
| ],
|
| )
|
|
|
