Code / services/llm_provider_service/response_validator.py
services/llm_provider_service/response_validator.py
1
"""Response validation utilities for LLM Provider Service.
2
3
Performance optimized for sub-500ms response times - Phase 7 enhancement.
4
"""
5
6
from __future__ import annotations
7
8
import json
9
import re
10
import time
11
from typing import Any, Dict
12
from uuid import UUID
13
14
from pydantic import BaseModel, Field, ValidationError
15
16
from common_core import EssayComparisonWinner
17
from huleedu_service_libs.logging_utils import create_service_logger
18
from services.llm_provider_service.exceptions import (
19
raise_parsing_error,
20
raise_validation_error,
21
)
22
23
logger = create_service_logger("llm_provider_service.response_validator")
24
25
# Pre-compiled regex patterns for better performance
26
WINNER_PATTERN = re.compile(
27
rf"^({EssayComparisonWinner.ESSAY_A.value}|{EssayComparisonWinner.ESSAY_B.value})$"
28
)
29
WINNER_KEYWORDS = {"a", "essay a", "option a", "first", "left"}
30
WINNER_B_KEYWORDS = {"b", "essay b", "option b", "second", "right"}
31
32
33
class StandardizedLLMResponse(BaseModel):
34
"""Standardized response format for LLM comparison results."""
35
36
winner: str = Field(
37
description="Which essay is better",
38
pattern=rf"^({EssayComparisonWinner.ESSAY_A.value}|{EssayComparisonWinner.ESSAY_B.value})$",
39
)
40
justification: str | None = Field(
41
default=None, description="Optional explanation of the choice"
42
)
43
confidence: float = Field(description="Confidence score between 1.0 and 5.0", ge=1.0, le=5.0)
44
45
46
def validate_and_normalize_response(
47
raw_response: str | Dict[str, Any],
48
provider: str = "unknown",
49
correlation_id: UUID | None = None,
50
) -> StandardizedLLMResponse:
51
"""Validate and normalize LLM response to standardized format.
52
53
Performance optimized for sub-500ms response times.
54
55
Args:
56
raw_response: Raw response from LLM (JSON string or dict)
57
provider: Provider name for metrics tracking
58
correlation_id: Request correlation ID for error tracking
59
60
Returns:
61
Validated and normalized response
62
63
Raises:
64
HuleEduError: On validation or parsing failures
65
"""
66
start_time = time.perf_counter()
67
68
try:
69
# Fast path: check if already parsed dict
70
if isinstance(raw_response, dict):
71
parsed_data = raw_response
72
else:
73
try:
74
parsed_data = json.loads(raw_response)
75
except json.JSONDecodeError as e:
76
_record_validation_metrics(provider, "json_parse", start_time, False)
77
raise_parsing_error(
78
service="llm_provider_service",
79
operation="response_validation",
80
parse_target="json",
81
message=f"Invalid JSON format: {str(e)}",
82
correlation_id=correlation_id or UUID(int=0),
83
details={"provider": provider},
84
)
85
86
# Fast normalization with minimal string operations
87
normalized_data = _fast_normalize_fields(parsed_data)
88
89
# Validate against schema with pre-normalized data
90
validated_response = StandardizedLLMResponse.model_validate(normalized_data)
91
92
_record_validation_metrics(provider, "success", start_time, True)
93
return validated_response
94
95
except ValidationError as e:
96
_record_validation_metrics(provider, "validation_error", start_time, False)
97
error_msg = _format_validation_errors(e)
98
raise_validation_error(
99
service="llm_provider_service",
100
operation="response_validation",
101
field="response_fields",
102
message=f"Validation failed: {error_msg}",
103
correlation_id=correlation_id or UUID(int=0),
104
details={"provider": provider},
105
)
106
107
except Exception as e:
108
_record_validation_metrics(provider, "unexpected_error", start_time, False)
109
raise_validation_error(
110
service="llm_provider_service",
111
operation="response_validation",
112
field="unknown",
113
message=f"Unexpected validation error: {str(e)}",
114
correlation_id=correlation_id or UUID(int=0),
115
details={"provider": provider},
116
)
117
118
119
def _fast_normalize_fields(parsed_data: Dict[str, Any]) -> Dict[str, Any]:
120
"""Fast field normalization with contract enforcement.
121
122
LLM API Contract: Confidence should be 1-5 scale.
123
124
Contract Violation Handling:
125
- If LLM returns 0-1 scale: Convert to 1-5 (contract normalization)
126
- If LLM returns 1-5 scale: Clamp to range (contract enforcement)
127
128
This ensures downstream conversions work correctly:
129
- Provider: 1-5 → 0-1 (internal mathematical operations)
130
- Queue: 0-1 → 1-5 (event contract for domain language)
131
"""
132
winner = parsed_data.get("winner")
133
raw_justification = parsed_data.get("justification")
134
confidence = parsed_data.get("confidence")
135
136
# Fast winner normalization using pre-compiled patterns
137
if isinstance(winner, str) and not WINNER_PATTERN.match(winner):
138
winner_lower = winner.lower().strip()
139
if winner_lower in WINNER_B_KEYWORDS:
140
winner = EssayComparisonWinner.ESSAY_B.value
141
else:
142
winner = EssayComparisonWinner.ESSAY_A.value
143
144
# Fast confidence normalization with contract enforcement
145
if isinstance(confidence, (int, float)):
146
confidence = float(confidence)
147
# Handle contract violations: if LLM returns 0-1 scale instead of 1-5
148
if confidence <= 1.0:
149
# Convert 0-1 → 1-5 to match expected contract
150
confidence = confidence * 4.0 + 1.0
151
# Enforce 1-5 contract
152
confidence = min(5.0, max(1.0, confidence))
153
# Preserve provider output when present; never manufacture an explanation.
154
justification = None if raw_justification is None else str(raw_justification).strip() or None
155
156
return {"winner": winner, "justification": justification, "confidence": confidence}
157
158
159
def _format_validation_errors(e: ValidationError) -> str:
160
"""Efficiently format validation errors."""
161
if len(e.errors()) == 1:
162
error = e.errors()[0]
163
field = " -> ".join(str(loc) for loc in error["loc"])
164
return f"{field}: {error['msg']}"
165
166
return "; ".join(
167
f"{' -> '.join(str(loc) for loc in error['loc'])}: {error['msg']}" for error in e.errors()
168
)
169
170
171
def _record_validation_metrics(
172
provider: str, validation_type: str, start_time: float, success: bool
173
) -> None:
174
"""Record validation performance metrics."""
175
try:
176
from services.llm_provider_service.metrics import get_llm_metrics
177
178
metrics = get_llm_metrics()
179
180
duration = time.perf_counter() - start_time
181
validation_metric = metrics.get("llm_validation_duration_seconds")
182
if validation_metric:
183
validation_metric.labels(provider=provider, validation_type=validation_type).observe(
184
duration
185
)
186
187
# Log slow validations
188
if duration > 0.01: # 10ms threshold
189
logger.warning(
190
f"Slow validation detected: {duration:.4f}s for {provider} ({validation_type})"
191
)
192
except Exception:
193
# Don't let metrics recording break validation
194
pass
195
196
197
# NOTE: convert_to_internal_format function removed - no longer needed
198
# as internal models now use the same assessment domain language
199
200
201
def validate_and_normalize_response_fast(
202
raw_response: Dict[str, Any],
203
provider: str = "unknown",
204
correlation_id: UUID | None = None,
205
) -> StandardizedLLMResponse:
206
"""Ultra-fast validation for pre-parsed dict responses.
207
208
Optimized for scenarios where JSON is already parsed and we need minimal validation.
209
Skips some safety checks for maximum performance.
210
211
Args:
212
raw_response: Pre-parsed response dict
213
provider: Provider name for metrics tracking
214
correlation_id: Request correlation ID for error tracking
215
216
Returns:
217
Validated response
218
219
Raises:
220
HuleEduError: On validation failures
221
"""
222
start_time = time.perf_counter()
223
224
try:
225
# Fast normalization without extensive error checking
226
normalized_data = _ultra_fast_normalize_fields(raw_response)
227
228
# Direct model creation without extensive validation
229
validated_response = StandardizedLLMResponse.model_construct(**normalized_data)
230
231
_record_validation_metrics(provider, "fast_success", start_time, True)
232
return validated_response
233
234
except Exception as e:
235
_record_validation_metrics(provider, "fast_error", start_time, False)
236
raise_validation_error(
237
service="llm_provider_service",
238
operation="fast_response_validation",
239
field="response_fields",
240
message=f"Fast validation error: {str(e)}",
241
correlation_id=correlation_id or UUID(int=0),
242
details={"provider": provider},
243
)
244
245
246
def _ultra_fast_normalize_fields(parsed_data: Dict[str, Any]) -> Dict[str, Any]:
247
"""Ultra-fast field normalization with minimal safety checks."""
248
# Direct field access with minimal validation
249
winner = parsed_data["winner"]
250
raw_justification = parsed_data.get("justification")
251
confidence = float(parsed_data["confidence"])
252
253
# Minimal winner validation - just check first character
254
if winner and winner[0].upper() == "B":
255
winner = EssayComparisonWinner.ESSAY_B.value
256
else:
257
winner = EssayComparisonWinner.ESSAY_A.value
258
259
# Contract enforcement: handle 0-1 scale violations
260
if confidence <= 1.0:
261
confidence = confidence * 4.0 + 1.0
262
confidence = max(1.0, min(5.0, confidence))
263
264
justification = None if raw_justification is None else str(raw_justification).strip() or None
265
266
return {"winner": winner, "justification": justification, "confidence": confidence}
267
268
269
def batch_validate_responses(
270
responses: list[Dict[str, Any]],
271
provider: str = "unknown",
272
correlation_id: UUID | None = None,
273
) -> list[StandardizedLLMResponse]:
274
"""Batch validate multiple responses for improved performance.
275
276
Args:
277
responses: List of response dicts to validate
278
provider: Provider name for metrics tracking
279
correlation_id: Request correlation ID for error tracking
280
281
Returns:
282
List of validated responses
283
284
Raises:
285
HuleEduError: On batch validation failures
286
"""
287
start_time = time.perf_counter()
288
results = []
289
290
try:
291
for response in responses:
292
result = validate_and_normalize_response_fast(response, provider, correlation_id)
293
results.append(result)
294
295
_record_validation_metrics(provider, "batch_success", start_time, True)
296
return results
297
298
except Exception as e:
299
_record_validation_metrics(provider, "batch_error", start_time, False)
300
raise_validation_error(
301
service="llm_provider_service",
302
operation="batch_response_validation",
303
field="batch_responses",
304
message=f"Batch validation error: {str(e)}",
305
correlation_id=correlation_id or UUID(int=0),
306
details={"provider": provider, "processed_count": len(results)},
307
)
308
309
310
# JSON Schema for external validation (e.g., in API documentation)
311
STANDARDIZED_RESPONSE_SCHEMA = {
312
"type": "object",
313
"properties": {
314
"winner": {
315
"type": "string",
316
"enum": [EssayComparisonWinner.ESSAY_A.value, EssayComparisonWinner.ESSAY_B.value],
317
"description": (
318
f"Which essay is better: '{EssayComparisonWinner.ESSAY_A.value}' "
319
f"or '{EssayComparisonWinner.ESSAY_B.value}'"
320
),
321
},
322
"justification": {
323
"type": "string",
324
"description": "Optional brief explanation of why this essay was chosen",
325
},
326
"confidence": {
327
"type": "number",
328
"minimum": 1.0,
329
"maximum": 5.0,
330
"description": "Confidence score between 1.0 and 5.0",
331
},
332
},
333
"required": ["winner", "confidence"],
334
"additionalProperties": False,
335
}