Feedback and RLHF

+-------------------+     +-------------------+     +-------------------+
| Explicit Feedback |     | Implicit Signals  |     | System Metrics    |
| (ratings, comments|     | (clicks, dwell    |     | (latency, errors, |
|  corrections)     |     |  time, refinements)|    |  token usage)     |
+--------+----------+     +--------+----------+     +--------+----------+
         |                         |                          |
         v                         v                          v
+----------------------------------------------------------------+
|                    Feedback Collector                           |
|  Normalizes, validates, and routes feedback events             |
+----------------------------+-----------------------------------+
                             |
                             v
+----------------------------+-----------------------------------+
|                    Kafka Topic: feedback-events                 |
+----------------------------+-----------------------------------+
                             |
              +--------------+----------------+
              |              |                |
              v              v                v
    +---------+----+ +------+-------+ +------+--------+
    | Learning     | | Insights     | | Quality       |
    | Pipeline     | | Engine       | | Dashboard     |
    +---------+----+ +------+-------+ +------+--------+
              |              |                |
              v              v                v
    +----------------------------------------------------+
    |  Model Improvement: Prompt tuning, fine-tuning,    |
    |  example selection, routing optimization           |
    +----------------------------------------------------+

class FeedbackEvent(BaseModel):
    """Base feedback event model."""
    id: UUID = Field(default_factory=uuid4)
    type: FeedbackType
    source: FeedbackSource
    severity: FeedbackSeverity = FeedbackSeverity.INFO
    category: FeedbackCategory
    tenant_id: str
    user_id: str | None = None
    session_id: str | None = None
    message_id: str | None = None
    timestamp: datetime = Field(default_factory=datetime.utcnow)
    data: dict[str, Any] = Field(default_factory=dict)
    tags: list[str] = Field(default_factory=list)

class FeedbackType(str, Enum):
    LOG = "log"                    # System log event
    METRIC = "metric"              # Numeric metric
    INTERACTION = "interaction"    # User interaction
    QUERY = "query"                # Query-level feedback
    ERROR = "error"                # Error report
    PERFORMANCE = "performance"    # Performance measurement
    QUALITY = "quality"            # Quality assessment
    USER_EXPLICIT = "user_explicit"  # Explicit user feedback
    EXPLICIT = "explicit"          # Direct rating/comment
    IMPLICIT = "implicit"          # Behavioral signal

class FeedbackSource(str, Enum):
    # System sources
    APPLICATION_LOG = "application_log"
    SYSTEM_METRIC = "system_metric"
    TRACE = "trace"
 
    # User sources
    USER_RATING = "user_rating"
    USER_COMMENT = "user_comment"
    USER_CORRECTION = "user_correction"
 
    # Agent sources
    QUERY_RESULT = "query_result"
    MODEL_RESPONSE = "model_response"
    PIPELINE_OUTPUT = "pipeline_output"
 
    # External sources
    API_RESPONSE = "api_response"
    DATABASE_QUERY = "database_query"
    EXTERNAL_SERVICE = "external_service"

class FeedbackCategory(str, Enum):
    ACCURACY = "accuracy"              # Response correctness
    LATENCY = "latency"                # Response speed
    COST = "cost"                      # Token/compute cost
    ERROR = "error"                    # Error occurrence
    USER_SATISFACTION = "user_satisfaction"  # User happiness
    QUALITY = "quality"                # Overall quality
    SECURITY = "security"             # Security concern
    COMPLIANCE = "compliance"          # Compliance issue

# Thumbs up/down rating
{
    "type": "explicit",
    "source": "user_rating",
    "category": "accuracy",
    "data": {
        "rating": "positive",        # positive, negative
        "message_id": "msg-123",
        "query": "Show total sales",
        "response": "SELECT SUM(amount)...",
    }
}
 
# Text comment
{
    "type": "explicit",
    "source": "user_comment",
    "category": "quality",
    "data": {
        "comment": "The chart should show percentages, not raw numbers",
        "message_id": "msg-123",
    }
}
 
# SQL correction
{
    "type": "explicit",
    "source": "user_correction",
    "category": "accuracy",
    "data": {
        "original_sql": "SELECT SUM(amount) FROM orders",
        "corrected_sql": "SELECT SUM(amount) FROM orders WHERE status = 'completed'",
        "explanation": "Should only count completed orders",
    }
}

# Performance feedback
{
    "type": "performance",
    "source": "system_metric",
    "category": "latency",
    "data": {
        "endpoint": "/api/v1/text-to-sql",
        "latency_ms": 1250,
        "tokens_used": 3400,
        "llm_provider": "openai",
        "model": "gpt-4",
    }
}

class DecisionFeedback:
    """Tracks agent decision quality."""
 
    async def record_decision(
        self,
        session_id: str,
        tenant_id: str,
        decision_type: str,          # routing, sql_generation, visualization
        decision_data: dict[str, Any],
        outcome: str,                 # success, failure, correction
        confidence: float,
    ) -> None:
        """Record an agent decision and its outcome."""
        event = FeedbackEvent(
            type=FeedbackType.QUERY,
            source=FeedbackSource.MODEL_RESPONSE,
            category=FeedbackCategory.ACCURACY,
            tenant_id=tenant_id,
            session_id=session_id,
            data={
                "decision_type": decision_type,
                "decision_data": decision_data,
                "outcome": outcome,
                "confidence": confidence,
            },
        )
        await self._collector.collect(event)

class FeedbackPipeline:
    """Multi-stage feedback processing pipeline."""
 
    async def process(self, event: FeedbackEvent) -> None:
        """Process a feedback event through the pipeline."""
 
        # Stage 1: Validation
        validated = await self._validate(event)
        if not validated:
            return
 
        # Stage 2: Enrichment
        enriched = await self._enrich(event)
 
        # Stage 3: Classification
        classified = await self._classify(enriched)
 
        # Stage 4: Storage
        await self._store(classified)
 
        # Stage 5: Routing
        await self._route(classified)

# feedback/integration/kafka/config.py
FEEDBACK_TOPICS = {
    "feedback-events": "All feedback events",
    "feedback-ratings": "User ratings only",
    "feedback-corrections": "SQL corrections",
    "feedback-metrics": "System performance metrics",
}

class FeedbackDrivenPromptTuner:
    """Tunes prompts based on feedback signals."""
 
    async def analyze_and_tune(
        self,
        tenant_id: str,
        window_days: int = 7,
    ) -> list[PromptUpdate]:
        """Analyze recent feedback and suggest prompt updates."""
 
        # Collect negative feedback patterns
        negative = await self._get_negative_feedback(
            tenant_id, days=window_days
        )
 
        # Identify common failure patterns
        patterns = self._identify_patterns(negative)
 
        # Generate prompt improvements
        updates = []
        for pattern in patterns:
            improvement = await self._generate_improvement(pattern)
            updates.append(improvement)
 
        return updates

class FewShotSelector:
    """Selects few-shot examples based on feedback."""
 
    async def select_examples(
        self,
        query: str,
        tenant_id: str,
        k: int = 3,
    ) -> list[FewShotExample]:
        """Select the best few-shot examples for a query."""
 
        # Get positively-rated query pairs
        positive_pairs = await self._get_positive_pairs(tenant_id)
 
        # Find most similar to current query
        similar = self._find_similar(query, positive_pairs, k=k)
 
        return [
            FewShotExample(
                input_text=pair["question"],
                output_text=pair["sql"],
                explanation=pair.get("explanation"),
            )
            for pair in similar
        ]

class FeedbackInsights:
    """Generate insights from feedback data."""
 
    async def get_quality_report(
        self,
        tenant_id: str,
        period_days: int = 30,
    ) -> dict[str, Any]:
        return {
            "overall_satisfaction": await self._satisfaction_score(
                tenant_id, period_days
            ),
            "accuracy_rate": await self._accuracy_rate(
                tenant_id, period_days
            ),
            "top_failure_patterns": await self._failure_patterns(
                tenant_id, period_days
            ),
            "improvement_trend": await self._improvement_trend(
                tenant_id, period_days
            ),
            "most_corrected_queries": await self._most_corrected(
                tenant_id, period_days
            ),
        }

POST /api/v1/feedback                    # Submit feedback event
GET  /api/v1/feedback/insights/{tenant}  # Get feedback insights
GET  /api/v1/feedback/quality/{tenant}   # Get quality metrics
POST /api/v1/feedback/correction         # Submit SQL correction
GET  /api/v1/feedback/decisions/{session} # Get decision feedback