gunn.adapters.llm package

Submodules

gunn.adapters.llm.dummy module

Dummy LLM adapter for cancellation testing.

This module provides a mock LLM adapter that simulates token generation with configurable timing and proper cancellation support for testing the 100ms cancellation SLO.

class gunn.adapters.llm.dummy.LLMAdapter(*args, **kwargs)[ソース]

ベースクラス: Protocol

Protocol for LLM adapters with streaming support.

async generate_stream(prompt, cancel_token, max_tokens=100)[ソース]

Generate streaming tokens with cancellation support.

戻り値の型:

AsyncGenerator[str, None]

パラメータ:
Args:

prompt: Input prompt for generation cancel_token: Token for cancellation signaling max_tokens: Maximum number of tokens to generate

Yields:

Generated tokens as strings

Raises:

asyncio.CancelledError: If generation is cancelled

class gunn.adapters.llm.dummy.DummyLLMAdapter(token_interval_ms=25.0, generation_time_ms=1000.0, tokens_per_second=40.0)[ソース]

ベースクラス: object

Mock LLM adapter for testing cancellation behavior.

This adapter simulates token generation with configurable timing to test the 100ms cancellation SLO requirement. It yields control every 20-30ms to ensure responsive cancellation.

Requirements addressed: - 6.2: Monitor for cancellation signals at token boundaries - 6.3: Immediately halt token generation within 100ms - 11.2: Cancel-to-halt latency ≤ 100ms at token boundaries

パラメータ:

  • token_interval_ms (float)

  • generation_time_ms (float)

  • tokens_per_second (float)

async generate_stream(prompt, cancel_token, max_tokens=100)[ソース]

Generate streaming tokens with cancellation support.

Yields control every token_interval_ms to check for cancellation. Ensures cancellation response within 100ms SLO.

戻り値の型:

AsyncGenerator[str, None]

パラメータ:
Args:

prompt: Input prompt for generation cancel_token: Token for cancellation signaling max_tokens: Maximum number of tokens to generate

Yields:

Generated tokens as strings

Raises:

asyncio.CancelledError: If generation is cancelled via cancel_token

async generate_with_timing(prompt, cancel_token, max_tokens=100)[ソース]

Generate tokens and return timing information for testing.

戻り値の型:

tuple[list[str], float, float | None]

パラメータ:
Args:

prompt: Input prompt for generation cancel_token: Token for cancellation signaling max_tokens: Maximum number of tokens to generate

Returns:

Tuple of (tokens, generation_time_ms, cancellation_time_ms) cancellation_time_ms is None if not cancelled

Raises:

asyncio.CancelledError: If generation is cancelled

class gunn.adapters.llm.dummy.CancellationTestHelper[ソース]

ベースクラス: object

Helper class for testing cancellation timing accuracy.

async static test_cancellation_timing(adapter, cancel_token, cancel_after_ms, tolerance_ms=5.0)[ソース]

Test cancellation timing accuracy.

戻り値の型:

tuple[bool, float, float]

パラメータ:
Args:

adapter: LLM adapter to test cancel_token: Token to cancel cancel_after_ms: Time to wait before cancelling tolerance_ms: Acceptable timing tolerance

Returns:

Tuple of (within_tolerance, actual_cancel_time_ms, expected_cancel_time_ms)

async static measure_token_yield_interval(adapter, cancel_token, measurement_duration_ms=200.0)[ソース]

Measure actual token yield intervals.

戻り値の型:

list[float]

パラメータ:
Args:

adapter: LLM adapter to test cancel_token: Token for generation measurement_duration_ms: How long to measure

Returns:

List of intervals between token yields in milliseconds

gunn.adapters.llm.streaming module

Module contents

LLM streaming integration adapter.

This package provides adapters for integrating with LLM services, including production streaming adapters and a dummy adapter for testing.

class gunn.adapters.llm.CancellationTestHelper[ソース]

ベースクラス: object

Helper class for testing cancellation timing accuracy.

async static measure_token_yield_interval(adapter, cancel_token, measurement_duration_ms=200.0)[ソース]

Measure actual token yield intervals.

戻り値の型:

list[float]

パラメータ:
Args:

adapter: LLM adapter to test cancel_token: Token for generation measurement_duration_ms: How long to measure

Returns:

List of intervals between token yields in milliseconds

async static test_cancellation_timing(adapter, cancel_token, cancel_after_ms, tolerance_ms=5.0)[ソース]

Test cancellation timing accuracy.

戻り値の型:

tuple[bool, float, float]

パラメータ:
Args:

adapter: LLM adapter to test cancel_token: Token to cancel cancel_after_ms: Time to wait before cancelling tolerance_ms: Acceptable timing tolerance

Returns:

Tuple of (within_tolerance, actual_cancel_time_ms, expected_cancel_time_ms)

class gunn.adapters.llm.DummyLLMAdapter(token_interval_ms=25.0, generation_time_ms=1000.0, tokens_per_second=40.0)[ソース]

ベースクラス: object

Mock LLM adapter for testing cancellation behavior.

This adapter simulates token generation with configurable timing to test the 100ms cancellation SLO requirement. It yields control every 20-30ms to ensure responsive cancellation.

Requirements addressed: - 6.2: Monitor for cancellation signals at token boundaries - 6.3: Immediately halt token generation within 100ms - 11.2: Cancel-to-halt latency ≤ 100ms at token boundaries

パラメータ:

  • token_interval_ms (float)

  • generation_time_ms (float)

  • tokens_per_second (float)

async generate_stream(prompt, cancel_token, max_tokens=100)[ソース]

Generate streaming tokens with cancellation support.

Yields control every token_interval_ms to check for cancellation. Ensures cancellation response within 100ms SLO.

戻り値の型:

AsyncGenerator[str, None]

パラメータ:
Args:

prompt: Input prompt for generation cancel_token: Token for cancellation signaling max_tokens: Maximum number of tokens to generate

Yields:

Generated tokens as strings

Raises:

asyncio.CancelledError: If generation is cancelled via cancel_token

async generate_with_timing(prompt, cancel_token, max_tokens=100)[ソース]

Generate tokens and return timing information for testing.

戻り値の型:

tuple[list[str], float, float | None]

パラメータ:
Args:

prompt: Input prompt for generation cancel_token: Token for cancellation signaling max_tokens: Maximum number of tokens to generate

Returns:

Tuple of (tokens, generation_time_ms, cancellation_time_ms) cancellation_time_ms is None if not cancelled

Raises:

asyncio.CancelledError: If generation is cancelled

class gunn.adapters.llm.GenerationRequest(**data)[ソース]

ベースクラス: BaseModel

Request for LLM generation.

パラメータ:
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

prompt: str
max_tokens: int | None
temperature: float | None
stop_sequences: list[str] | None
metadata: dict[str, Any] | None
class gunn.adapters.llm.GenerationResponse(**data)[ソース]

ベースクラス: BaseModel

Response from LLM generation.

パラメータ:

  • content (str)

  • token_count (int)

  • generation_time_ms (float)

  • cancelled (bool)

  • cancellation_time_ms (float | None)

  • error (str | None)

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

content: str
token_count: int
generation_time_ms: float
cancelled: bool
cancellation_time_ms: float | None
error: str | None
class gunn.adapters.llm.LLMAdapter(*args, **kwargs)[ソース]

ベースクラス: Protocol

Protocol for LLM adapters with streaming support.

async generate_stream(prompt, cancel_token, max_tokens=100)[ソース]

Generate streaming tokens with cancellation support.

戻り値の型:

AsyncGenerator[str, None]

パラメータ:
Args:

prompt: Input prompt for generation cancel_token: Token for cancellation signaling max_tokens: Maximum number of tokens to generate

Yields:

Generated tokens as strings

Raises:

asyncio.CancelledError: If generation is cancelled