qfa.services.llm_call_executor#

Shared LLM-call scaffolding for the application services.

Every use case in qfa.services wraps its LLM calls in the same four concerns: redact PII before the call (and restore it afterwards), derive a per-call timeout from the request deadline, guard the token budget, and bound concurrent calls with a semaphore. LLMCallExecutor owns those four and nothing else, so a use-case service can be read without them.

Per ADR-017 this is a plain concrete class, deliberately not a Protocol and not a base class:

  • It is not a driven port. Ports in this codebase invert dependencies on infrastructure; this object wraps no external system, it orchestrates calls to an LLMPort that already exists. It is therefore not declared in qfa.domain.ports.

  • Services receive it as a constructor dependency and delegate to it. Nothing inherits from it — behaviour reuse in this codebase is always composition, and class X(Y): stays readable as “X implements port Y”.

Tests construct the real executor over the existing FakeLLMPort / FakeAnonymizer doubles; there is no fake executor.

Classes

LLMCallExecutor(llm, anonymizer, settings, ...)

Deadline-, budget- and concurrency-aware wrapper around an LLM port.

SlotTiming([queued_seconds, call_seconds])

Split timing for one semaphore-bounded hierarchical LLM call.
class qfa.services.llm_call_executor.SlotTiming(queued_seconds: float = 0.0, call_seconds: float = 0.0)[source]#

Bases: object

Split timing for one semaphore-bounded hierarchical LLM call.

queued_seconds is the time spent waiting to acquire the concurrency semaphore; call_seconds is the LLM completion itself, measured only after the slot was acquired. Keeping them apart makes the per-chunk debug lines honest: a single combined duration folds queue-wait into “call time”, so a 5s call that waited 100s in the queue looked like a 110s call (and like a near-timeout when it was nothing of the sort).

queued_seconds: float = 0.0#
call_seconds: float = 0.0#
class qfa.services.llm_call_executor.LLMCallExecutor(llm: LLMPort, anonymizer: AnonymizationPort, settings: OrchestratorSettings, llm_timeout_seconds: float, max_total_tokens: int)[source]#

Bases: object

Deadline-, budget- and concurrency-aware wrapper around an LLM port.

Parameters:

  • llm (LLMPort) – The LLM provider adapter every call runs on by default. Call sites that must use a different connection — the LLM-as-judge calls, which may be configured onto their own model — pass it explicitly to bounded_complete().

  • anonymizer (AnonymizationPort) – The anonymisation adapter used to redact PII before LLM calls.

  • settings (OrchestratorSettings) – Cross-cutting configuration; this object reads chars_per_token for its token estimate.

  • llm_timeout_seconds (float) – Maximum time in seconds for a single LLM call, before the deadline is taken into account.

  • max_total_tokens (int) – Maximum estimated total tokens for a single request.

anonymize_records(records: tuple[FeedbackRecordModel, ...], anonymize: bool) → tuple[tuple[FeedbackRecordModel, ...], dict[str, str]][source]#

Anonymise each record’s text, returning new records + merged mapping.

Metadata is left untouched (codes/dates are not PII and feed the deterministic trend table). When anonymize is False, records are returned unchanged with an empty mapping.

anonymize_text(text: str) → tuple[str, dict[str, str]][source]#

Redact PII from one assembled message, returning text + mapping.

The single-record use cases anonymise the assembled user message rather than each record’s content, because the envelope they send is built before the call; anonymize_records() is the batch equivalent for the paths that chunk records first. Pair this with deanonymize_json() to restore the redacted values in the model’s response.

async bounded_complete(semaphore: Semaphore, *, llm: LLMPort | None = None, system_message: str, user_message: str, tenant_id: str, response_model: type[T_Response], deadline: datetime, timing: SlotTiming | None = None) → LLMResponse[source]#

Run one LLM completion, bounded by semaphore and the deadline.

This is complete() run under semaphore, with the queue-wait measured around the acquire.

semaphore caps how many completions run at once across the whole hierarchical pipeline (map, leaf judge, reduce), so concurrency stays within max_concurrent_chunks across every phase. The deadline/timeout is computed after acquiring a slot, so a completion that queued behind others still honours the remaining budget (and raises AnalysisTimeoutError if the deadline passed while it waited).

llm overrides the connection for this one call. It exists because this helper serves both map/reduce and the leaf judges, which may run on different connections (see judge_llm); None uses the executor’s own client. Every current call site passes it explicitly so the connection each call uses is visible where the call is written. Note the semaphore is shared regardless: the bound is on total in-flight calls, not per connection.

When timing is supplied it is populated with the queue-wait and the post-acquire call duration as two separate fields, so callers can log them apart rather than reporting one combined number that hides how long the call sat waiting for a slot.

check_deadline_and_get_timeout(deadline: datetime) → float[source]#

Raise if the deadline has passed or too little time remains.

Return a timeout (seconds) bounded by the deadline and the configured per-call limit.

check_token_limit(system_message: str, user_message: str) → None[source]#

Estimate total tokens and raise if over the limit.

Parameters:

  • system_message (str) – The assembled system message.

  • user_message (str) – The assembled user message containing the feedback records.

Raises:

FeedbackTooLargeError – When estimated tokens exceed the configured limit.

async complete(*, llm: LLMPort | None = None, system_message: str, user_message: str, tenant_id: str, response_model: type[T_Response], deadline: datetime) → LLMResponse[source]#

Run one LLM completion bounded by the deadline.

The per-call timeout is derived from deadline immediately before the call, so a completion issued late in a request still honours the remaining budget (and raises AnalysisTimeoutError rather than issuing a call that cannot finish in time).

Use this for the single-call use cases. The hierarchical pipeline uses bounded_complete(), which adds the concurrency semaphore and the queue-wait timing on top of this method.

llm overrides the connection for this one call — see bounded_complete() for why that override exists; None uses the executor’s own client.

deanonymize_json(payload: str, mapping: dict[str, str]) → str[source]#

Restore redacted values inside a serialized JSON payload.

The counterpart to anonymize_text() for structured responses: the mapping’s values are escaped the way json.dumps would before substitution, so PII containing a quote or a newline cannot corrupt the JSON the caller is about to re-validate.