1from pydantic import BaseModel
2from enum import Enum
3from openai import OpenAI
4 
5client = OpenAI()
6 
7# Level 1: Confidence-tiered AI output
8# Transform raw model outputs into actionable confidence signals for users
Level 1: Confidence tiers convert raw model confidence into actionable UI signals
9 
10class ConfidenceTier(str, Enum):
11    HIGH = "high"        # Show recommendation prominently, default to accepting
12    MEDIUM = "medium"    # Show recommendation with review prompt
ConfidenceTier enum: 4 discrete levels each with a distinct UX treatment
13    LOW = "low"          # Show as "suggestion, needs review"
14    UNCERTAIN = "uncertain"  # Do not show as recommendation, show as question
15 
key_factors: the "why" that enables users to evaluate whether to follow the recommendation
16class AIRecommendation(BaseModel):
alternative: what to do if the recommendation is wrong — enables confident rejection without dead ends
17    recommendation: str
18    confidence_tier: ConfidenceTier
19    confidence_score: float          # 0.0 - 1.0
20    key_factors: list[str]           # Why: top 3 factors driving this recommendation
21    alternative: str | None          # What to do if this recommendation is wrong
22    review_flag: bool                # True if human review is recommended
23 
24def make_recommendation(context: str, domain: str) -> AIRecommendation:
25    """Generate a recommendation with calibrated confidence tier."""
26    response = client.beta.chat.completions.parse(
27        model="gpt-4o",
28        messages=[
29            {
30                "role": "system",
31                "content": f"""You are a {domain} recommendation system.
32Provide a recommendation and assess your confidence honestly.
33- HIGH confidence: clear answer supported by strong evidence
34- MEDIUM confidence: reasonable answer but meaningful uncertainty
35- LOW confidence: best guess but significant uncertainty
36- UNCERTAIN: insufficient information to recommend reliably
37Always provide the 2-3 key factors driving your recommendation."""
38            },
39            {"role": "user", "content": context}
40        ],
41        response_format=AIRecommendation,
42    )
UI rendering per tier: different badge colors, CTAs, and prominence levels — users learn the trust signal pattern
43    return response.choices[0].message.parsed
44 
45# Example UI rendering based on confidence tier
46def render_recommendation(rec: AIRecommendation) -> dict:
47    ui_config = {
48        ConfidenceTier.HIGH: {"badge": "✓ AI Recommendation", "badge_color": "green", "cta": "Accept"},
49        ConfidenceTier.MEDIUM: {"badge": "ⓘ AI Suggestion", "badge_color": "blue", "cta": "Review & Accept"},
50        ConfidenceTier.LOW: {"badge": "⚠ Needs Review", "badge_color": "yellow", "cta": "Review"},
51        ConfidenceTier.UNCERTAIN: {"badge": "? Unclear — Please Decide", "badge_color": "gray", "cta": "Decide"},
52    }
53    return {**ui_config[rec.confidence_tier], "recommendation": rec.recommendation,
54            "factors": rec.key_factors, "score": rec.confidence_score}

1from datetime import datetime
2from typing import Literal
3from pydantic import BaseModel
4import json
5 
6# Level 2: Structured feedback collection from user actions
7# User behavior (accept/reject/modify) is implicit feedback
8# Structured capture enables quality measurement and model improvement
9 
Level 2: Structured feedback — every user action is a labeled quality signal for the model
10class UserFeedback(BaseModel):
11    session_id: str
12    recommendation_id: str
user_modification: when users edit the AI output, they are implicitly labeling the correct output
13    action: Literal["accepted", "rejected", "modified", "ignored"]
time_to_decision: fast acceptance = high trust. Long deliberation = uncertainty. Skip = distrust.
14    user_modification: str | None  # If "modified" — what did the user change it to?
15    time_to_decision: float        # Seconds from showing recommendation to user action
16    confidence_tier_shown: str     # What confidence tier was displayed
17 
18class FeedbackPipeline:
19    def __init__(self, storage_backend):
20        self.storage = storage_backend
21        self.acceptance_rates = {}  # Track by confidence tier for calibration analysis
22 
23    def record_feedback(self, feedback: UserFeedback):
24        """Record user action as implicit quality signal."""
25        self.storage.write("user_feedback", feedback.dict())
26 
27        # Update acceptance rate tracking by confidence tier
28        tier = feedback.confidence_tier_shown
29        if tier not in self.acceptance_rates:
30            self.acceptance_rates[tier] = {"accepted": 0, "total": 0}
31 
32        self.acceptance_rates[tier]["total"] += 1
Calibration analysis: HIGH confidence recommendations accepted <70% indicates the tier is miscalibrated
33        if feedback.action == "accepted":
34            self.acceptance_rates[tier]["accepted"] += 1
35 
36    def compute_calibration(self) -> dict:
37        """
38        Check if confidence tiers match actual acceptance rates.
39        If HIGH confidence recommendations are only accepted 60% of the time,
40        the confidence tier is miscalibrated — users are overriding HIGH confidence.
41        """
42        calibration = {}
43        for tier, counts in self.acceptance_rates.items():
44            if counts["total"] > 0:
45                calibration[tier] = {
46                    "acceptance_rate": counts["accepted"] / counts["total"],
Corrections as training data: user modifications are the highest quality implicit labels you can collect
47                    "sample_size": counts["total"],
48                }
49        # Target: HIGH tier acceptance > 85%, LOW tier acceptance 50-70%
50        return calibration
51 
52    def get_correction_training_data(self) -> list[dict]:
53        """
54        User modifications = implicit corrections = training signal.
55        Users who change the AI's recommendation are teaching it what they wanted.
56        """
57        corrections = self.storage.query(
58            "user_feedback",
59            filters={"action": "modified"},
60        )
61        return [{"original": c["recommendation"], "corrected": c["user_modification"]}
62                for c in corrections if c["user_modification"]]

1from datetime import datetime, timedelta
2import pandas as pd
3 
4# Level 3: AI product metrics that measure real business impact
5# Technical metrics (accuracy, latency) ≠ product metrics (does it help users?)
6 
7class AIProductMetrics:
8    """
Level 3: Business impact metrics — connects AI quality to user outcomes, not just technical accuracy
9    Metrics framework that connects AI technical quality to business outcomes.
10    Answers: is the AI actually helping users accomplish their goals?
11    """
12 
13    def compute_human_ai_alignment(self, feedback_data: pd.DataFrame) -> dict:
14        """
15        Alignment: how often do users agree with the AI?
16        HIGH alignment on HIGH confidence = well-calibrated system.
Human-AI alignment: tracks how often users agree with the AI per confidence tier
17        LOW alignment on HIGH confidence = trust problem.
18        """
19        alignment_by_tier = feedback_data.groupby("confidence_tier_shown").apply(
20            lambda g: {
21                "agreement_rate": (g["action"] == "accepted").mean(),
22                "modification_rate": (g["action"] == "modified").mean(),
23                "rejection_rate": (g["action"] == "rejected").mean(),
24                "median_decision_time": g["time_to_decision"].median(),
25            }
26        )
27        return alignment_by_tier.to_dict()
28 
29    def compute_workflow_impact(
30        self,
Workflow impact: the fundamental product question — is the AI making users more successful?
31        task_completion_data: pd.DataFrame,
32        pre_ai_baseline: float
33    ) -> dict:
34        """
35        Workflow impact: does the AI change behavior in the desired direction?
36        Measure time savings, error rate reduction, or decision quality improvement.
37        """
38        ai_users = task_completion_data[task_completion_data["used_ai"] == True]
39        non_ai_users = task_completion_data[task_completion_data["used_ai"] == False]
40 
41        return {
time_savings_seconds: quantifies the efficiency value the AI provides to users
42            "task_completion_rate_with_ai": ai_users["completed"].mean(),
43            "task_completion_rate_without_ai": non_ai_users["completed"].mean(),
44            "time_savings_seconds": (
45                non_ai_users["task_time"].median() - ai_users["task_time"].median()
46            ),
AI reliance health: automation bias (accept without reading) and algorithm aversion (reject without considering) are both product failures
47            "ai_lift": ai_users["completed"].mean() - pre_ai_baseline,
48        }
49 
50    def compute_ai_reliance_health(self, feedback_data: pd.DataFrame) -> dict:
51        """
52        Healthy AI reliance: users accept good recommendations and reject bad ones.
53        Unhealthy pattern 1 (automation bias): accept everything, even wrong AI.
54        Unhealthy pattern 2 (algorithm aversion): reject everything, even right AI.
55        """
56        # Track override quality: when users override, were they right?
57        override_correctness = feedback_data[feedback_data["action"] == "rejected"].apply(
58            lambda row: self._evaluate_override_quality(row), axis=1
59        )
60        return {
61            "override_rate": (feedback_data["action"] != "accepted").mean(),
62            "override_quality": override_correctness.mean(),  # Were overrides correct decisions?
63            "automation_bias_risk": (feedback_data["time_to_decision"] < 2).mean(),  # Too fast = not reading
64        }