Code / services/llm_provider_service/protocols.py
services/llm_provider_service/protocols.py
1
"""Provider, orchestration, queue, and comparison contracts for LLM Provider."""
2
3
from collections.abc import Mapping
4
from typing import Awaitable, Callable, Optional, Protocol, TypeVar, runtime_checkable
5
from uuid import UUID
6
7
from pydantic import BaseModel
8
9
from common_core import LLMProviderType, QueueStatus
10
from common_core.events.envelope import EventEnvelope
11
from services.llm_provider_service.comparison_bundle_models import (
12
ProviderComparisonBundleResponse,
13
)
14
from services.llm_provider_service.internal_models import (
15
BatchComparisonItem,
16
LLMOrchestratorResponse,
17
LLMProviderResponse,
18
LLMQueuedResult,
19
)
20
from services.llm_provider_service.queue_models import (
21
QueuedRequest,
22
QueueMaintenanceResult,
23
QueueStats,
24
)
25
26
PromptBlock = dict[str, object]
27
RequestMetadata = Mapping[str, object]
28
ProviderOverrides = dict[str, object]
29
EventPayload = TypeVar("EventPayload", bound=BaseModel)
30
31
32
class LLMProviderProtocol(Protocol):
33
"""Protocol for individual LLM provider implementations."""
34
35
async def generate_comparison(
36
self,
37
user_prompt: str,
38
correlation_id: UUID,
39
prompt_blocks: list[PromptBlock] | None = None,
40
system_prompt_override: str | None = None,
41
model_override: str | None = None,
42
endpoint_profile: str | None = None,
43
temperature_override: float | None = None,
44
max_tokens_override: int | None = None,
45
reasoning_effort: str | None = None,
46
output_verbosity: str | None = None,
47
allowance_mode: str | None = None,
48
) -> LLMProviderResponse:
49
"""Generate LLM comparison response.
50
51
Args:
52
user_prompt: Complete comparison prompt with essays embedded
53
correlation_id: Request correlation ID for tracing
54
system_prompt_override: Optional system prompt override
55
model_override: Optional model override
56
endpoint_profile: Optional exact provider endpoint profile
57
temperature_override: Optional temperature override
58
max_tokens_override: Optional max tokens override
59
reasoning_effort: Optional reasoning effort hint for GPT-5 family models
60
output_verbosity: Optional output verbosity hint for GPT-5 family models
61
62
Returns:
63
The LLM provider response containing comparison result
64
65
Raises:
66
HuleEduError: On any failure to generate comparison
67
"""
68
...
69
70
71
@runtime_checkable
72
class ComparisonBundleProviderProtocol(Protocol):
73
"""Provider capability for one synchronous call containing many CJ items."""
74
75
async def generate_comparison_bundle(
76
self,
77
items: tuple[BatchComparisonItem, ...],
78
correlation_id: UUID,
79
system_prompt_override: str | None = None,
80
model_override: str | None = None,
81
endpoint_profile: str | None = None,
82
temperature_override: float | None = None,
83
max_tokens_override: int | None = None,
84
reasoning_effort: str | None = None,
85
output_verbosity: str | None = None,
86
allowance_mode: str | None = None,
87
) -> ProviderComparisonBundleResponse:
88
"""Generate all identified CJ decisions through one provider request."""
89
...
90
91
92
class LLMOrchestratorProtocol(Protocol):
93
"""Protocol for LLM orchestration service."""
94
95
async def perform_comparison(
96
self,
97
provider: LLMProviderType,
98
user_prompt: str,
99
correlation_id: UUID,
100
prompt_blocks: list[PromptBlock] | None = None,
101
request_metadata: RequestMetadata | None = None,
102
callback_topic: str | None = None,
103
**overrides: object,
104
) -> LLMQueuedResult:
105
"""Perform LLM comparison with provider selection.
106
107
Args:
108
provider: LLM provider to use
109
user_prompt: Complete comparison prompt with essays embedded
110
correlation_id: Request correlation ID
111
request_metadata: Arbitrary metadata to echo back in callbacks
112
**overrides: Additional parameter overrides
113
114
Returns:
115
LLMQueuedResult - ALL requests are queued for async processing with callback delivery
116
117
Raises:
118
HuleEduError: On any failure to perform comparison
119
"""
120
...
121
122
async def process_queued_request(
123
self,
124
provider: LLMProviderType,
125
user_prompt: str,
126
correlation_id: UUID,
127
prompt_blocks: list[PromptBlock] | None = None,
128
**overrides: object,
129
) -> LLMOrchestratorResponse:
130
"""Process a queued request directly (internal use only).
131
132
This method bypasses queueing and directly calls providers.
133
ONLY used by QueueProcessor for processing already-queued requests.
134
135
Args:
136
provider: LLM provider to use
137
user_prompt: Complete comparison prompt with essays embedded
138
correlation_id: Request correlation ID
139
**overrides: Additional parameter overrides
140
141
Returns:
142
LLMOrchestratorResponse with immediate result
143
144
Raises:
145
HuleEduError: On any failure to process request
146
"""
147
...
148
149
async def test_provider(self, provider: LLMProviderType, correlation_id: UUID) -> bool:
150
"""Test provider connectivity and availability.
151
152
Args:
153
provider: Provider to test
154
correlation_id: Request correlation ID for tracing
155
156
Returns:
157
True if provider is available and functional
158
159
Raises:
160
HuleEduError: On provider test failure with details
161
"""
162
...
163
164
165
class LLMEventPublisherProtocol(Protocol):
166
"""Protocol for LLM usage event publishing."""
167
168
async def publish_llm_request_started(
169
self,
170
provider: str,
171
correlation_id: UUID,
172
metadata: RequestMetadata,
173
) -> None:
174
"""Publish LLM request started event.
175
176
Args:
177
provider: LLM provider name
178
correlation_id: Request correlation ID
179
metadata: Additional event metadata
180
"""
181
...
182
183
async def publish_llm_request_completed(
184
self,
185
provider: str,
186
correlation_id: UUID,
187
success: bool,
188
response_time_ms: int,
189
metadata: RequestMetadata,
190
) -> None:
191
"""Publish LLM request completed event.
192
193
Args:
194
provider: LLM provider name
195
correlation_id: Request correlation ID
196
success: Whether request was successful
197
response_time_ms: Response time in milliseconds
198
metadata: Additional event metadata
199
"""
200
...
201
202
async def publish_llm_provider_failure(
203
self,
204
provider: str,
205
failure_type: str,
206
correlation_id: UUID,
207
error_details: str,
208
circuit_breaker_opened: bool = False,
209
) -> None:
210
"""Publish LLM provider failure event.
211
212
Args:
213
provider: LLM provider name
214
failure_type: Type of failure
215
correlation_id: Request correlation ID
216
error_details: Error details
217
circuit_breaker_opened: Whether circuit breaker opened
218
"""
219
...
220
221
async def publish_to_topic(
222
self,
223
topic: str,
224
envelope: EventEnvelope[EventPayload],
225
key: Optional[str] = None,
226
) -> None:
227
"""Publish event to specific topic.
228
229
Args:
230
topic: Kafka topic name
231
envelope: Event envelope to publish
232
key: Optional message key for partitioning
233
"""
234
...
235
236
237
class LLMRetryManagerProtocol(Protocol):
238
"""Protocol for LLM request retry management."""
239
240
async def with_retry(
241
self,
242
operation: Callable[..., Awaitable[object]],
243
operation_name: str,
244
**kwargs: object,
245
) -> object:
246
"""Execute operation with retry logic.
247
248
Args:
249
operation: Async operation to execute
250
operation_name: Name of operation for logging
251
**kwargs: Arguments to pass to operation
252
253
Returns:
254
Result from operation
255
"""
256
...
257
258
259
class QueueManagerProtocol(Protocol):
260
"""Protocol for request queue management."""
261
262
async def enqueue(self, request: QueuedRequest) -> bool:
263
"""Enqueue a request for later processing.
264
265
Args:
266
request: The request to queue
267
268
Returns:
269
True if enqueued successfully, False if queue is full
270
"""
271
...
272
273
async def dequeue(self, provider: LLMProviderType) -> Optional[QueuedRequest]:
274
"""Get the next request for one provider lane.
275
276
Returns:
277
Next request or None if queue is empty
278
"""
279
...
280
281
async def reserve_compatible(
282
self,
283
primary_request: QueuedRequest,
284
provider: LLMProviderType,
285
maximum_items: int,
286
) -> list[QueuedRequest]:
287
"""Claim compatible queued work for an already-claimed bundle primary."""
288
...
289
290
async def remove(self, queue_id: UUID) -> bool:
291
"""Remove a request from the queue by ID.
292
293
Args:
294
queue_id: The queue ID to remove
295
296
Returns:
297
True if removed successfully, False if not found
298
"""
299
...
300
301
async def complete(self, queue_id: UUID, lease_token: str | None) -> bool:
302
"""Race-safe completion of a processing request.
303
304
Returns True only if the caller still held the processing lease
305
(commit token). Provider result handlers call this only after Kafka
306
accepts the corresponding callback.
307
308
Args:
309
queue_id: The queue ID to complete
310
lease_token: Opaque token from the current processing claim
311
312
Returns:
313
True if the lease was held and the item was removed
314
"""
315
...
316
317
async def renew_lease(self, request: QueuedRequest) -> bool:
318
"""Extend the processing lease held by the current request claim."""
319
...
320
321
async def renew_leases(self, requests: list[QueuedRequest]) -> bool:
322
"""Atomically extend every lease held by one provider-call bundle."""
323
...
324
325
async def release(self, request: QueuedRequest) -> bool:
326
"""Return a processing request to its provider queue while its claim is held."""
327
...
328
329
async def get_status(self, queue_id: UUID) -> Optional[QueuedRequest]:
330
"""Get status of a queued request.
331
332
Args:
333
queue_id: The queue ID to check
334
335
Returns:
336
Request details or None if not found
337
"""
338
...
339
340
async def update_status(
341
self,
342
queue_id: UUID,
343
status: QueueStatus,
344
message: Optional[str] = None,
345
result_location: Optional[str] = None,
346
) -> bool:
347
"""Update status of a queued request.
348
349
Args:
350
queue_id: The queue ID to update
351
status: New status
352
message: Optional error or status message
353
result_location: Optional result location
354
355
Returns:
356
True if updated successfully
357
358
Raises:
359
HuleEduError: On failure to update status
360
"""
361
...
362
363
async def get_queue_stats(self) -> QueueStats:
364
"""Get current queue statistics.
365
366
Returns:
367
Queue statistics
368
"""
369
...
370
371
async def cleanup_expired(self) -> QueueMaintenanceResult:
372
"""Remove expired requests from queue and reclaim stale leases.
373
374
Returns:
375
Queue maintenance result including terminal stale failures
376
"""
377
...
378
379
380
class ComparisonProcessorProtocol(Protocol):
381
"""Protocol for processing LLM comparisons - domain logic only."""
382
383
async def process_comparison(
384
self,
385
provider: LLMProviderType,
386
user_prompt: str,
387
correlation_id: UUID,
388
prompt_blocks: list[PromptBlock] | None = None,
389
**overrides: object,
390
) -> "LLMOrchestratorResponse":
391
"""Process LLM comparison without infrastructure concerns.
392
393
This is pure domain logic - no queuing, no callbacks.
394
Used by QueueProcessor for actual LLM invocation.
395
396
Args:
397
provider: LLM provider to use
398
user_prompt: Complete comparison prompt with essays embedded
399
correlation_id: Request correlation ID
400
**overrides: Additional parameter overrides
401
402
Returns:
403
LLMOrchestratorResponse with immediate result
404
405
Raises:
406
HuleEduError: On any failure to process comparison
407
"""
408
...
409
410
async def process_comparison_bundle(
411
self,
412
items: tuple[BatchComparisonItem, ...],
413
) -> list["LLMOrchestratorResponse"]:
414
"""Process one homogeneous CJ bundle through one provider call."""
415
...