Code / services/cj_assessment_service/implementations/llm_interaction_impl.py
services/cj_assessment_service/implementations/llm_interaction_impl.py
1
"""LLM comparison interaction orchestration for CJ Assessment.
2
3
Purpose:
4
Send CJ comparison tasks to the configured LLM Provider boundary while
5
preserving callback correlation metadata, prompt-block evidence, usage, and
6
provider execution constraints.
7
8
Relationships:
9
- Called by CJ comparison workflow components after pair generation.
10
- Publishes provider requests through `LLMProviderProtocol` and receives
11
results asynchronously through callback persistence.
12
"""
13
14
from __future__ import annotations
15
16
import asyncio
17
from hashlib import sha256
18
from uuid import UUID
19
20
from pydantic import JsonValue
21
22
from common_core import Environment, LLMProviderType
23
from common_core.error_enums import ErrorCode
24
from huleedu_service_libs.error_handling.error_detail_factory import (
25
create_error_detail_with_context,
26
)
27
from huleedu_service_libs.logging_utils import create_service_logger
28
from services.cj_assessment_service.config import Settings
29
from services.cj_assessment_service.metrics import get_business_metrics
30
from services.cj_assessment_service.models_api import (
31
CJLLMComparisonMetadata,
32
ComparisonAdmissionResult,
33
ComparisonTask,
34
ComparisonTrackingKey,
35
)
36
from services.cj_assessment_service.protocols import (
37
LLMInteractionProtocol,
38
LLMProviderProtocol,
39
)
40
41
logger = create_service_logger("cj_assessment_service.llm_interaction_impl")
42
43
44
class LLMInteractionImpl(LLMInteractionProtocol):
45
"""Implementation of LLMInteractionProtocol for orchestrating LLM comparisons."""
46
47
def __init__(
48
self,
49
providers: dict[LLMProviderType, LLMProviderProtocol],
50
settings: Settings,
51
) -> None:
52
"""Initialize LLM interaction orchestrator.
53
54
Args:
55
providers: Dictionary of available LLM providers.
56
settings: Application settings.
57
"""
58
self.providers = providers
59
self.settings = settings
60
61
def _get_provider_for_model(self) -> LLMProviderProtocol:
62
"""Get the appropriate provider for the configured model."""
63
# Use the default provider from structured configuration
64
provider_key = self.settings.DEFAULT_LLM_PROVIDER
65
66
if provider_key not in self.providers:
67
available_providers = list(self.providers.keys())
68
logger.warning(
69
f"Default provider '{provider_key}' not available. Available providers: "
70
f"{available_providers}. Using first available provider.",
71
)
72
provider_key = available_providers[0] if available_providers else LLMProviderType.OPENAI
73
74
logger.debug(
75
f"Using provider '{provider_key}' with model '{self.settings.DEFAULT_LLM_MODEL}'",
76
)
77
return self.providers[provider_key]
78
79
async def perform_comparisons(
80
self,
81
tasks: list[ComparisonTask],
82
correlation_id: UUID,
83
tracking_map: dict[ComparisonTrackingKey, UUID] | None = None,
84
bos_batch_id: str | None = None,
85
model_override: str | None = None,
86
temperature_override: float | None = None,
87
max_tokens_override: int | None = None,
88
system_prompt_override: str | None = None,
89
provider_override: str | LLMProviderType | None = None,
90
metadata_context: dict[str, JsonValue] | None = None,
91
) -> list[ComparisonAdmissionResult]:
92
"""Perform multiple comparison tasks using configured LLM providers.
93
94
Args:
95
tasks: List of comparison tasks to process.
96
correlation_id: Request correlation ID for tracing
97
tracking_map: Optional map of (essay_a_id, essay_b_id) to unique correlation IDs
98
bos_batch_id: Optional BOS batch identifier propagated for ENG5 correlation
99
model_override: Optional model name override
100
temperature_override: Optional temperature override (0.0-2.0)
101
max_tokens_override: Optional max tokens override
102
system_prompt_override: Optional system prompt override per batch
103
provider_override: Optional provider name override forwarded to LPS
104
metadata_context: Optional metadata fields merged into CJLLMComparisonMetadata
105
106
Returns:
107
List of comparison results corresponding to the input tasks.
108
"""
109
if not tasks:
110
logger.warning("No comparison tasks provided")
111
return []
112
113
# Get the provider for the current model configuration
114
provider = self._get_provider_for_model()
115
116
# Get business metrics for recording LLM operations
117
business_metrics = get_business_metrics()
118
llm_api_calls_metric = business_metrics.get("llm_api_calls")
119
120
def _serialize_prompt_blocks(
121
task: ComparisonTask,
122
task_correlation_id: UUID,
123
) -> list[dict[str, JsonValue]] | None:
124
"""Serialize prompt blocks or fail before provider submission."""
125
if task.prompt_blocks is None:
126
return None
127
128
try:
129
return task.prompt_blocks.to_api_dict_list()
130
except (TypeError, ValueError) as serialization_error:
131
logger.warning(
132
"Failed to serialize prompt_blocks; aborting provider submission",
133
exc_info=True,
134
extra={
135
"essay_a": task.essay_a.id,
136
"essay_b": task.essay_b.id,
137
"correlation_id": str(task_correlation_id),
138
},
139
)
140
raise serialization_error
141
142
def _with_prompt_block_evidence(
143
request_metadata: dict[str, JsonValue],
144
prompt_blocks_payload: list[dict[str, JsonValue]] | None,
145
rendered_prompt_text: str,
146
) -> dict[str, JsonValue]:
147
"""Add callback-safe prompt-block evidence to request metadata."""
148
if not prompt_blocks_payload:
149
return request_metadata
150
151
metadata = dict(request_metadata)
152
cacheable_hashes: list[JsonValue] = []
153
block_order: list[JsonValue] = []
154
block_hashes: list[JsonValue] = []
155
for block in prompt_blocks_payload:
156
target = block.get("target")
157
if not isinstance(target, str):
158
raise ValueError("prompt block target evidence must be a string")
159
content_hash = block.get("content_hash")
160
if not isinstance(content_hash, str):
161
raise ValueError("prompt block content hash evidence must be a string")
162
block_order.append(f"block_{len(block_order)}_{target}")
163
block_hashes.append(content_hash)
164
if block.get("cacheable") is True:
165
cacheable_hashes.append(content_hash)
166
metadata["prompt_blocks_present"] = True
167
metadata["prompt_block_count"] = len(prompt_blocks_payload)
168
metadata["cacheable_prompt_block_count"] = len(cacheable_hashes)
169
metadata["prompt_block_order"] = block_order
170
metadata["prompt_block_sha256s"] = block_hashes
171
metadata["rendered_prompt_sha256"] = sha256(
172
rendered_prompt_text.encode("utf-8")
173
).hexdigest()
174
metadata["rendered_prompt_text"] = rendered_prompt_text
175
metadata["rendered_request_shape"] = "prompt_blocks"
176
if cacheable_hashes:
177
metadata["static_prompt_block_sha256"] = cacheable_hashes[0]
178
metadata["cacheable_prompt_block_sha256s"] = list(cacheable_hashes)
179
return metadata
180
181
def task_correlation_id(task: ComparisonTask) -> UUID:
182
if tracking_map is None:
183
return correlation_id
184
explicit = task.explicit_comparison
185
key: ComparisonTrackingKey = (
186
explicit.comparison_id
187
if explicit is not None
188
else (task.essay_a.id, task.essay_b.id)
189
)
190
return tracking_map.get(key, correlation_id)
191
192
async def process_task(task: ComparisonTask) -> ComparisonAdmissionResult:
193
"""Request provider admission for a single comparison task."""
194
# Get the unique correlation ID for this specific comparison
195
# If tracking_map is provided, use the unique ID,
196
# otherwise use the batch correlation_id
197
request_correlation_id = task_correlation_id(task)
198
if not tracking_map:
199
logger.warning(
200
f"No tracking_map provided, using batch correlation ID {correlation_id}",
201
extra={"essay_a": task.essay_a.id, "essay_b": task.essay_b.id},
202
)
203
204
# Make direct LLM API request - no caching
205
metadata_adapter = CJLLMComparisonMetadata.from_comparison_task(
206
task,
207
bos_batch_id=bos_batch_id,
208
)
209
if metadata_context:
210
metadata_adapter = metadata_adapter.with_additional_context(**metadata_context)
211
request_metadata = metadata_adapter.to_request_metadata()
212
213
prompt_blocks_payload = _serialize_prompt_blocks(
214
task=task,
215
task_correlation_id=request_correlation_id,
216
)
217
request_metadata = _with_prompt_block_evidence(
218
request_metadata=request_metadata,
219
prompt_blocks_payload=prompt_blocks_payload,
220
rendered_prompt_text=task.prompt,
221
)
222
223
try:
224
admission = await provider.generate_comparison(
225
user_prompt=task.prompt,
226
correlation_id=request_correlation_id,
227
prompt_blocks=prompt_blocks_payload,
228
system_prompt_override=system_prompt_override,
229
model_override=model_override,
230
temperature_override=temperature_override,
231
max_tokens_override=max_tokens_override,
232
provider_override=provider_override,
233
request_metadata=request_metadata,
234
)
235
236
# Record queued LLM API call metric
237
if llm_api_calls_metric:
238
llm_api_calls_metric.labels(
239
provider=self.settings.DEFAULT_LLM_PROVIDER.value,
240
model=model_override or self.settings.DEFAULT_LLM_MODEL,
241
status="queued",
242
).inc()
243
244
return ComparisonAdmissionResult(
245
task=task,
246
request_correlation_id=request_correlation_id,
247
admission=admission,
248
)
249
250
except Exception as e:
251
# Record failed LLM API call metric
252
if llm_api_calls_metric:
253
llm_api_calls_metric.labels(
254
provider=self.settings.DEFAULT_LLM_PROVIDER.value,
255
model=model_override or self.settings.DEFAULT_LLM_MODEL,
256
status="error",
257
).inc()
258
259
# Note: exc_info omitted to avoid Rich+Mock interaction in tests
260
logger.error(
261
f"Unexpected error processing task {task.essay_a.id}-{task.essay_b.id}: {e}",
262
)
263
return ComparisonAdmissionResult(
264
task=task,
265
request_correlation_id=request_correlation_id,
266
error_detail=create_error_detail_with_context(
267
error_code=ErrorCode.PROCESSING_ERROR,
268
message=f"Unexpected error processing task: {e!s}",
269
service="cj_assessment_service",
270
operation="process_comparison_task",
271
correlation_id=correlation_id,
272
details={
273
"essay_a_id": task.essay_a.id,
274
"essay_b_id": task.essay_b.id,
275
},
276
),
277
)
278
279
# Process all tasks concurrently
280
try:
281
results = await asyncio.gather(
282
*[process_task(task) for task in tasks],
283
return_exceptions=True,
284
)
285
286
# Handle any exceptions and async results from gather
287
final_results: list[ComparisonAdmissionResult] = []
288
for i, result in enumerate(results):
289
if isinstance(result, Exception):
290
if self.settings.ENVIRONMENT is not Environment.PRODUCTION and isinstance(
291
result, (TypeError, ValueError)
292
):
293
raise result
294
295
# Note: exc_info omitted here as exceptions should have been logged with full
296
# traceback in the inner except block of process_task(). Logging here with
297
# exc_info=result can trigger Rich's introspection of Mock objects in test
298
# environments, causing TypeError when Rich tries to render local variables.
299
logger.error(
300
f"Task with essays {tasks[i].essay_a.id}-{tasks[i].essay_b.id} "
301
f"raised an exception: {result}",
302
)
303
final_results.append(
304
ComparisonAdmissionResult(
305
task=tasks[i],
306
request_correlation_id=task_correlation_id(tasks[i]),
307
error_detail=create_error_detail_with_context(
308
error_code=ErrorCode.PROCESSING_ERROR,
309
message=f"Task execution failed: {result!s}",
310
service="cj_assessment_service",
311
operation="process_comparisons_batch",
312
correlation_id=correlation_id,
313
details={
314
"essay_a_id": tasks[i].essay_a.id,
315
"essay_b_id": tasks[i].essay_b.id,
316
},
317
),
318
),
319
)
320
elif isinstance(result, ComparisonAdmissionResult):
321
final_results.append(result)
322
323
return final_results
324
325
except Exception as e:
326
if self.settings.ENVIRONMENT is not Environment.PRODUCTION and isinstance(
327
e, (TypeError, ValueError)
328
):
329
raise
330
331
logger.error(
332
f"Critical error during comparison processing: {e}",
333
exc_info=True,
334
)
335
# Return error results for all tasks
336
return [
337
ComparisonAdmissionResult(
338
task=task,
339
request_correlation_id=task_correlation_id(task),
340
error_detail=create_error_detail_with_context(
341
error_code=ErrorCode.PROCESSING_ERROR,
342
message=f"Critical processing error: {e!s}",
343
service="cj_assessment_service",
344
operation="process_comparisons_critical_error",
345
correlation_id=correlation_id,
346
details={"essay_a_id": task.essay_a.id, "essay_b_id": task.essay_b.id},
347
),
348
)
349
for task in tasks
350
]