Code / services/cj_assessment_service/implementations/llm_provider_service_client.py
services/cj_assessment_service/implementations/llm_provider_service_client.py
1
"""HTTP client for CJ-owned requests to the LLM Provider Service.
2
3
Purpose:
4
Convert comparative-judgement LLM requests into the centralized LLM
5
Provider Service HTTP contract while preserving the distinction between
6
explicit caller overrides and provider-owned defaults.
7
8
Relationships:
9
- Called by CJ comparison orchestration when essay pairs need LLM judgement.
10
- Sends queued comparison requests to the LLM Provider Service `/comparison`
11
endpoint and relies on Kafka callbacks for results.
12
- Carries ENG runner metadata such as reasoning effort and output verbosity
13
into provider-service overrides.
14
"""
15
16
from __future__ import annotations
17
18
import json
19
import typing
20
from uuid import UUID
21
22
import aiohttp
23
from pydantic import JsonValue
24
25
from common_core import LLMConfigOverridesHTTP as ProviderLLMConfigOverrides
26
from common_core import LLMProviderType
27
from common_core.events.cj_assessment_events import LLMConfigOverrides as CJLLMConfigOverrides
28
from huleedu_service_libs.error_handling import (
29
HuleEduError,
30
raise_authentication_error,
31
raise_external_service_error,
32
raise_invalid_request,
33
raise_invalid_response,
34
raise_parsing_error,
35
raise_rate_limit_error,
36
raise_resource_not_found,
37
raise_service_unavailable,
38
raise_timeout_error,
39
)
40
from huleedu_service_libs.logging_utils import create_service_logger
41
from services.cj_assessment_service.config import Settings
42
from services.cj_assessment_service.models_api import LLMProviderAdmission
43
from services.cj_assessment_service.protocols import LLMProviderProtocol, RetryManagerProtocol
44
45
logger = create_service_logger("cj_assessment_service.llm_provider_service_client")
46
47
type OpenAIAllowanceModeValue = typing.Literal["free_allowance_only", "allow_paid_overage"]
48
49
50
def _build_llm_config_override_payload(
51
*,
52
overrides: CJLLMConfigOverrides | ProviderLLMConfigOverrides | None = None,
53
provider_override: str | LLMProviderType | None = None,
54
model_override: str | None = None,
55
endpoint_profile: str | None = None,
56
temperature_override: float | None = None,
57
system_prompt_override: str | None = None,
58
max_tokens_override: int | None = None,
59
reasoning_effort: str | None = None,
60
output_verbosity: str | None = None,
61
allowance_mode: OpenAIAllowanceModeValue = "free_allowance_only",
62
) -> dict[str, JsonValue] | None:
63
"""Convert CJ overrides into the provider service payload."""
64
if overrides is not None:
65
provider_override = provider_override or overrides.provider_override
66
model_override = model_override or overrides.model_override
67
endpoint_profile = endpoint_profile or overrides.endpoint_profile
68
temp_attr = overrides.temperature_override
69
if temperature_override is None and temp_attr is not None:
70
temperature_override = temp_attr
71
max_attr = overrides.max_tokens_override
72
if max_tokens_override is None and max_attr is not None:
73
max_tokens_override = max_attr
74
prompt_attr = overrides.system_prompt_override
75
if system_prompt_override is None and prompt_attr is not None:
76
system_prompt_override = prompt_attr
77
reasoning_attr = overrides.reasoning_effort
78
if reasoning_effort is None and reasoning_attr is not None:
79
reasoning_effort = reasoning_attr
80
verbosity_attr = overrides.output_verbosity
81
if output_verbosity is None and verbosity_attr is not None:
82
output_verbosity = verbosity_attr
83
allowance_mode = _allowance_mode_value(overrides.allowance_mode)
84
provider_enum: LLMProviderType | None = None
85
if isinstance(provider_override, LLMProviderType):
86
provider_enum = provider_override
87
elif isinstance(provider_override, str):
88
normalized = provider_override.strip().lower()
89
try:
90
provider_enum = LLMProviderType(normalized)
91
except ValueError:
92
logger.warning(
93
"Unknown provider_override supplied; default provider will be used",
94
extra={"provider_override": provider_override},
95
)
96
97
overrides_model = ProviderLLMConfigOverrides(
98
provider_override=provider_enum,
99
model_override=model_override,
100
endpoint_profile=endpoint_profile,
101
temperature_override=temperature_override,
102
system_prompt_override=system_prompt_override,
103
max_tokens_override=max_tokens_override,
104
reasoning_effort=reasoning_effort,
105
output_verbosity=output_verbosity,
106
allowance_mode=allowance_mode,
107
)
108
payload = overrides_model.model_dump(exclude_none=True, exclude_defaults=True)
109
return payload or None
110
111
112
def _allowance_mode_value(value: str | None) -> OpenAIAllowanceModeValue:
113
if value is None or value == "free_allowance_only":
114
return "free_allowance_only"
115
if value == "allow_paid_overage":
116
return "allow_paid_overage"
117
raise ValueError(
118
"allowance_mode must be free_allowance_only or explicitly human-authorized "
119
"allow_paid_overage"
120
)
121
122
123
def _optional_metadata_text(
124
metadata: dict[str, JsonValue],
125
key: str,
126
) -> str | None:
127
value = metadata.get(key)
128
if value is None:
129
return None
130
if isinstance(value, str) and value.strip():
131
return value
132
raise ValueError(f"request metadata {key} must be a non-blank string")
133
134
135
class LLMProviderServiceClient(LLMProviderProtocol):
136
"""HTTP client for centralized LLM Provider Service."""
137
138
def __init__(
139
self,
140
session: aiohttp.ClientSession,
141
settings: Settings,
142
retry_manager: RetryManagerProtocol,
143
) -> None:
144
"""Initialize LLM Provider Service client.
145
146
Args:
147
session: HTTP session for making requests
148
settings: Application settings containing service URL
149
retry_manager: Retry manager for handling transient failures
150
"""
151
self.session = session
152
self.settings = settings
153
self.retry_manager = retry_manager
154
self.base_url = settings.LLM_PROVIDER_SERVICE_URL.rstrip("/")
155
156
async def generate_comparison(
157
self,
158
user_prompt: str,
159
correlation_id: UUID,
160
prompt_blocks: list[dict[str, JsonValue]] | None = None,
161
system_prompt_override: str | None = None,
162
model_override: str | None = None,
163
temperature_override: float | None = None,
164
max_tokens_override: int | None = None,
165
provider_override: str | None = None,
166
request_metadata: dict[str, JsonValue] | None = None,
167
) -> LLMProviderAdmission:
168
"""Generate comparison via LLM Provider Service.
169
170
Accept only a queued (202) response carrying a valid provider queue ID.
171
172
Args:
173
user_prompt: The user prompt containing essay comparison request
174
correlation_id: Correlation ID for request tracing
175
system_prompt_override: Optional system prompt override
176
model_override: Optional model name override
177
temperature_override: Optional temperature override (0.0-2.0)
178
max_tokens_override: Optional max tokens override
179
provider_override: Optional provider name override
180
181
Returns:
182
Typed provider admission. Results still arrive through callbacks.
183
"""
184
# Build request body for LLM Provider Service
185
# Note: user_prompt now contains the complete prompt with essays embedded
186
metadata = request_metadata or {}
187
overrides_payload = _build_llm_config_override_payload(
188
provider_override=provider_override or self.settings.DEFAULT_LLM_PROVIDER.value,
189
model_override=model_override or self.settings.DEFAULT_LLM_MODEL,
190
endpoint_profile=_optional_metadata_text(metadata, "endpoint_profile"),
191
temperature_override=temperature_override,
192
system_prompt_override=system_prompt_override,
193
max_tokens_override=max_tokens_override,
194
reasoning_effort=_optional_metadata_text(metadata, "reasoning_effort"),
195
output_verbosity=_optional_metadata_text(metadata, "output_verbosity"),
196
allowance_mode=_allowance_mode_value(
197
_optional_metadata_text(metadata, "allowance_mode")
198
),
199
)
200
201
request_body: dict[str, JsonValue] = {
202
"user_prompt": user_prompt,
203
"metadata": metadata,
204
"correlation_id": str(correlation_id),
205
"callback_topic": self.settings.LLM_PROVIDER_CALLBACK_TOPIC, # Required field
206
}
207
if prompt_blocks:
208
prompt_block_payload: list[JsonValue] = [dict(block) for block in prompt_blocks]
209
request_body["prompt_blocks"] = prompt_block_payload
210
if overrides_payload:
211
request_body["llm_config_overrides"] = overrides_payload
212
213
# Make initial HTTP request with retry logic
214
url = f"{self.base_url}/comparison"
215
216
async def make_request() -> LLMProviderAdmission:
217
try:
218
async with self.session.post(
219
url,
220
json=request_body,
221
headers={"Content-Type": "application/json"},
222
timeout=aiohttp.ClientTimeout(total=60), # 60 second timeout
223
) as response:
224
response_text = await response.text()
225
226
if response.status == 202:
227
# ALL LLM requests are async - this is the ONLY valid success response
228
# LLM processing takes 10-30+ seconds, blocking HTTP is an anti-pattern
229
return await self._handle_queued_response(
230
response_text,
231
correlation_id,
232
)
233
234
elif response.status == 200:
235
# This should never happen in production - LLM Provider should always queue
236
logger.error(
237
"LLM Provider returned 200 (sync) - "
238
"this is an architectural violation. "
239
"All LLM calls must be async (202) with Kafka callbacks.",
240
extra={
241
"correlation_id": str(correlation_id),
242
"provider": provider_override
243
or self.settings.DEFAULT_LLM_PROVIDER.value,
244
},
245
)
246
# Treat as an error - we don't support synchronous LLM calls
247
raise_invalid_response(
248
service="cj_assessment_service",
249
operation="generate_comparison",
250
message="LLM Provider returned synchronous response (200). "
251
"Only async (202) responses are supported.",
252
correlation_id=correlation_id,
253
response_status=200,
254
)
255
256
else:
257
# Check if error is retryable
258
try:
259
error_data = json.loads(response_text)
260
is_retryable = response.status in [
261
429,
262
500,
263
502,
264
503,
265
504,
266
] or error_data.get("is_retryable", False)
267
268
if is_retryable:
269
# Log before raising for better debugging
270
logger.warning(
271
"Retryable error from LLM Provider Service",
272
extra={
273
"status_code": response.status,
274
"error": error_data.get("error", "Unknown"),
275
"correlation_id": str(correlation_id),
276
"provider": provider_override
277
or self.settings.DEFAULT_LLM_PROVIDER.value,
278
},
279
)
280
# Create proper exception with details
281
raise aiohttp.ClientResponseError(
282
request_info=response.request_info,
283
history=response.history,
284
status=response.status,
285
message=error_data.get("error", str(response.status)),
286
headers=response.headers,
287
)
288
except json.JSONDecodeError:
289
# If we can't parse JSON, use status code to decide
290
if response.status in [429, 500, 502, 503, 504]:
291
logger.warning(
292
"Retryable HTTP error (no JSON body)",
293
extra={
294
"status_code": response.status,
295
"correlation_id": str(correlation_id),
296
},
297
)
298
response.raise_for_status()
299
300
# Non-retryable error - handle and raise
301
await self._handle_error_response(
302
response.status, response_text, correlation_id
303
)
304
# This should never be reached as _handle_error_response raises
305
raise AssertionError("_handle_error_response should have raised")
306
307
except aiohttp.ClientError:
308
# Re-raise to let retry manager handle it
309
raise
310
except HuleEduError:
311
# Re-raise HuleEduError as-is (preserve error code semantics)
312
raise
313
except Exception as e:
314
# Only convert unexpected raw exceptions to structured errors
315
raise_external_service_error(
316
service="cj_assessment_service",
317
operation="generate_comparison.make_request",
318
external_service="llm_provider_service",
319
message=f"Unexpected error calling LLM Provider Service: {str(e)}",
320
correlation_id=correlation_id,
321
exception_type=type(e).__name__,
322
)
323
324
# Use retry manager directly - circuit breaker should be integrated into the HTTP client
325
# or applied at the transport layer for async compatibility
326
return await self.retry_manager.with_retry(make_request)
327
328
# REMOVED: _handle_immediate_response method
329
# This method was removed as part of the architectural simplification.
330
# ALL LLM calls are async (202) with Kafka callbacks.
331
# Synchronous responses (200) are an anti-pattern for LLM integration.
332
# LLMs take 10-30+ seconds - blocking HTTP for this duration is wasteful.
333
334
async def _handle_queued_response(
335
self,
336
response_text: str,
337
correlation_id: UUID,
338
) -> LLMProviderAdmission:
339
"""Handle queued (202) response from LLM Provider Service.
340
341
No polling is performed - the service will send results via callback.
342
343
Args:
344
response_text: Raw response text from the HTTP response
345
correlation_id: Correlation ID for request tracing
346
347
Returns:
348
Typed queue admission; results will be delivered via callback.
349
"""
350
try:
351
queue_response = json.loads(response_text)
352
queue_id = queue_response.get("queue_id")
353
354
if not queue_id:
355
raise_invalid_response(
356
service="cj_assessment_service",
357
operation="_handle_queued_response",
358
expected="queue_id in response",
359
actual="missing queue_id",
360
message="Queue response missing queue_id",
361
correlation_id=correlation_id,
362
response_preview=response_text[:200],
363
)
364
365
try:
366
return LLMProviderAdmission(queue_id=queue_id)
367
except ValueError:
368
raise_invalid_response(
369
service="cj_assessment_service",
370
operation="_handle_queued_response",
371
expected="UUID queue_id in response",
372
actual="invalid queue_id",
373
message="Queue response contains an invalid queue_id",
374
correlation_id=correlation_id,
375
response_preview=response_text[:200],
376
)
377
378
except json.JSONDecodeError as e:
379
raise_parsing_error(
380
service="cj_assessment_service",
381
operation="_handle_queued_response",
382
parse_target="queued_response_json",
383
message=f"Failed to parse queued response JSON: {str(e)}",
384
correlation_id=correlation_id,
385
response_preview=response_text[:200],
386
)
387
388
async def _handle_error_response(
389
self, status_code: int, response_text: str, correlation_id: UUID
390
) -> None:
391
"""Handle error responses from LLM Provider Service.
392
393
Args:
394
status_code: HTTP status code
395
response_text: Raw response text from the HTTP response
396
correlation_id: Correlation ID for request tracing
397
398
Raises:
399
HuleEduError: Always raises appropriate error based on status code
400
"""
401
try:
402
error_data = json.loads(response_text)
403
error_msg = error_data.get("error", f"HTTP {status_code}")
404
details_text = error_data.get("details", "")
405
is_retryable = error_data.get("is_retryable", status_code in [429, 500, 502, 503, 504])
406
407
details = {
408
"status_code": status_code,
409
"is_retryable": is_retryable,
410
"details_text": details_text,
411
"response_preview": response_text[:200],
412
}
413
except json.JSONDecodeError:
414
error_msg = f"HTTP {status_code}"
415
details = {
416
"status_code": status_code,
417
"is_retryable": status_code in [429, 500, 502, 503, 504],
418
"response_preview": response_text[:200],
419
}
420
421
# Map status codes directly to appropriate factory functions
422
if status_code == 400:
423
raise_invalid_request(
424
service="cj_assessment_service",
425
operation="_handle_error_response",
426
message=f"LLM Provider Service error: {error_msg}",
427
correlation_id=correlation_id,
428
**details,
429
)
430
elif status_code in [401, 403]:
431
raise_authentication_error(
432
service="cj_assessment_service",
433
operation="_handle_error_response",
434
message=f"LLM Provider Service error: {error_msg}",
435
correlation_id=correlation_id,
436
**details,
437
)
438
elif status_code == 404:
439
raise_resource_not_found(
440
service="cj_assessment_service",
441
operation="_handle_error_response",
442
resource_type="LLM Provider endpoint",
443
resource_id="N/A",
444
message=f"LLM Provider Service error: {error_msg}",
445
correlation_id=correlation_id,
446
**details,
447
)
448
elif status_code in [408, 504]:
449
raise_timeout_error(
450
service="cj_assessment_service",
451
operation="_handle_error_response",
452
timeout_seconds=60,
453
message=f"LLM Provider Service error: {error_msg}",
454
correlation_id=correlation_id,
455
**details,
456
)
457
elif status_code == 429:
458
raise_rate_limit_error(
459
service="cj_assessment_service",
460
operation="_handle_error_response",
461
message=f"LLM Provider Service error: {error_msg}",
462
correlation_id=correlation_id,
463
**details,
464
)
465
elif status_code in [502, 503]:
466
raise_service_unavailable(
467
service="cj_assessment_service",
468
operation="_handle_error_response",
469
unavailable_service="llm_provider_service",
470
message=f"LLM Provider Service error: {error_msg}",
471
correlation_id=correlation_id,
472
**details,
473
)
474
else:
475
# Default to external service error
476
raise_external_service_error(
477
service="cj_assessment_service",
478
operation="_handle_error_response",
479
external_service="llm_provider_service",
480
message=f"LLM Provider Service error: {error_msg}",
481
correlation_id=correlation_id,
482
**details,
483
)