stage19: performance optimierung — LLM concurrency semaphore, priority (HIGH/NORMAL/LOW), limiter

src/nsct/api/rest_research.py

@@ -14,6 +14,7 @@ from pydantic import BaseModel, Field
 from nsct.logging_config import set_request_ctx, clear_request_ctx, set_research_run_id, get_logger
 from nsct.orchestration.budget import HardBudgetConfig
 from nsct.orchestration.state import ResearchRunState
+from nsct.providers.priority_queue import Priority
 from nsct.metrics import (
     metrics,
     C_SEARCH_QUERIES_TOTAL,
@@ -410,6 +411,7 @@ async def _run_pipeline(run: _ResearchRunState, request: ResearchRequest, budget
             query=request.query,
             budget_config=budget,
             depth=request.depth,
+            priority=Priority.NORMAL,
         )
 
         stage_ctx["stage"] = "planning"

src/nsct/orchestration/orchestrator.py

@@ -33,6 +33,7 @@ from nsct.orchestration.budget import BudgetExhaustedError, BudgetTracker, HardB
 from nsct.orchestration.models import ResearchRun
 from nsct.orchestration.state import ResearchRunState, StateMachine
 from nsct.providers.abstract import MultiProviderSearch, SearchProvider
+from nsct.providers.priority_queue import Priority
 
 try:
     from nsct.stages.stage9_synthesis import SynthesisStage
@@ -61,6 +62,7 @@ class ResearchOrchestrator:
         query: str,
         budget_config: HardBudgetConfig | None = None,
         depth: str = "normal",
+        priority: Priority = Priority.NORMAL,
     ) -> None:
         """Initialisiere den Orchestrator.
 
@@ -76,11 +78,14 @@ class ResearchOrchestrator:
             Optionales Budget. Wird aus config abgeleitet, wenn None.
         depth : str
             Suchtiefe ("quick", "normal", "deep").
+        priority : Priority
+            LLM-Anfrage-Priorität für den Research-Pipeline.
         """
         self._config = config
         self._research_id = research_id
         self._query = query
         self._depth = depth
+        self._priority = priority
 
         # Budget
         if budget_config is not None:

src/nsct/providers/__init__.py

@@ -15,6 +15,7 @@ from nsct.providers.llm import (
     ProviderTimeoutError,
     _LLMProviderImpl,
 )
+from nsct.providers.priority_queue import Priority, PriorityLimiter
 from nsct.providers.metrics import ProviderMetrics
 from nsct.providers.vision import _VisionProviderImpl
 
@@ -25,6 +26,8 @@ AudioProvider = _AudioProviderImpl
 __all__ = [
     "LLMProvider",
     "_LLMProviderImpl",
+    "Priority",
+    "PriorityLimiter",
     "ProviderError",
     "ProviderHTTPError",
     "ProviderModelNotFoundError",

src/nsct/providers/llm.py

@@ -8,7 +8,7 @@ import logging
 import os
 import time
 from abc import ABC, abstractmethod
-from typing import Any, AsyncGenerator, cast
+from typing import Any, AsyncGenerator
 
 import httpx
 from httpx import HTTPStatusError
@@ -18,6 +18,7 @@ from openai.types.chat import ChatCompletion, ChatCompletionChunk
 
 from nsct.config import AppSettings
 from nsct.providers.metrics import ProviderMetrics
+from nsct.providers.priority_queue import Priority, PriorityLimiter
 
 logger = logging.getLogger(__name__)
 
@@ -110,6 +111,7 @@ class _LLMProviderImpl(LLMProvider):
         self._model: str = config.llm.model
         self._max_concurrency: int = config.llm.max_concurrency
         self._api_key: str = config.llm.get_secret()
+        self._limiter: PriorityLimiter | None = None
 
         # Timeout config (seconds) — configurable via env, defaults here
         self._connect_timeout: float = float(
@@ -121,6 +123,13 @@ class _LLMProviderImpl(LLMProvider):
 
         self._client: AsyncOpenAI | None = None
 
+    @property
+    def limiter(self) -> PriorityLimiter:
+        """Lazy-init the priority limiter."""
+        if self._limiter is None:
+            self._limiter = PriorityLimiter(self._max_concurrency)
+        return self._limiter
+
     @property
     def client(self) -> AsyncOpenAI:
         """Lazy-init the OpenAI client with connection pooling."""
@@ -205,8 +214,16 @@ class _LLMProviderImpl(LLMProvider):
         temperature: float | None = None,
         max_tokens: int | None = None,
         response_format: dict[str, Any] | None = None,
+        priority: Priority = Priority.NORMAL,
     ) -> str:
-        """Issue a chat-completions request and return the assistant text."""
+        """Issue a chat-completions request and return the assistant text.
+
+        Parameters
+        ----------
+        priority : Priority
+            Request priority. HIGH runs immediately, NORMAL waits for a slot,
+            LOW may be rejected when all slots are busy.
+        """
         model = model or self._model
 
         async def _do() -> str:
@@ -239,7 +256,18 @@ class _LLMProviderImpl(LLMProvider):
             choice = resp.choices[0]
             return choice.message.content or ""
 
-        return await self._request_with_retry(_do)
+        limiter = self.limiter
+        try:
+            await limiter.acquire(priority)
+        except Exception as exc:
+            if priority == Priority.LOW:
+                logger.warning("LLM pool exhausted — LOW request rejected: %s", exc)
+                return ""
+            raise
+        try:
+            return await self._request_with_retry(_do)
+        finally:
+            limiter.release(priority)
 
     async def stream_complete(
         self,
@@ -247,28 +275,43 @@ class _LLMProviderImpl(LLMProvider):
         model: str | None = None,
         temperature: float | None = None,
         max_tokens: int | None = None,
+        priority: Priority = Priority.NORMAL,
     ) -> AsyncGenerator[str, None]:
-        """Stream chat completions. Yields delta content chunks."""
+        """Stream chat completions. Yields delta content chunks.
+
+        Yields are deferred until the limiter releases so the slot is held
+        for the entire stream duration.
+        """
         model = model or self._model
-        tokens_input = 0
+        limiter = self.limiter
-        tokens_output = 0
+        chunks: list[str] = []
-        start = time.monotonic()
 
-        async for chunk in (await self._make_stream(messages, model, temperature, max_tokens)):
+        try:
-            delta = chunk.choices[0].delta if chunk.choices and chunk.choices[0] else None
+            await limiter.acquire(priority)
-            delta_text = delta.content if delta and delta.content else ""
+            tokens_input = 0
-            if delta_text:
+            tokens_output = 0
-                yield delta_text
+            start = time.monotonic()
-                tokens_output += 1
 
-        latency = time.monotonic() - start
+            async for chunk in (await self._make_stream(messages, model, temperature, max_tokens)):
-        for msg in messages:
+                delta = chunk.choices[0].delta if chunk.choices and chunk.choices[0] else None
-            tokens_input += len(msg.get("content", "").split())
+                delta_text = delta.content if delta and delta.content else ""
-        await self._metrics.record_llm_request(
+                if delta_text:
-            input_tokens=tokens_input,
+                    chunks.append(delta_text)
-            output_tokens=tokens_output,
+                    tokens_output += 1
-            latency=latency,
+
-        )
+            latency = time.monotonic() - start
+            for msg in messages:
+                tokens_input += len(msg.get("content", "").split())
+            await self._metrics.record_llm_request(
+                input_tokens=tokens_input,
+                output_tokens=tokens_output,
+                latency=latency,
+            )
+        finally:
+            self.limiter.release(priority)
+
+        for chunk_text in chunks:
+            yield chunk_text
 
     async def _make_stream(
         self,

src/nsct/providers/priority_queue.py

@@ -0,0 +1,123 @@
+"""Priority-aware limiter for LLM requests.
+
+Priority levels:
+  - HIGH:  bypasses the semaphore limit — always acquires immediately.
+           Uses a separate "priority" semaphore so HIGH requests do not
+           consume slots reserved for NORMAL.
+  - NORMAL: uses the main semaphore (slots are shared across all callers).
+  - LOW:   rejected with LlmPriorityRejectError when the main semaphore is
+           fully occupied; waits otherwise.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import enum
+import logging
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# Priority enum
+# ---------------------------------------------------------------------------
+
+
+class Priority(int, enum.Enum):
+    """Request priority for LLM calls."""
+
+    HIGH = 0
+    NORMAL = 1
+    LOW = 2
+
+
+class LlmPriorityRejectError(Exception):
+    """Raised when a LOW-priority request is rejected because all slots are busy."""
+
+    pass
+
+
+# ---------------------------------------------------------------------------
+# PriorityLimiter
+# ---------------------------------------------------------------------------
+
+
+class PriorityLimiter:
+    """Semaphore-based rate-limiter with per-priority handling.
+
+    Parameters
+    ----------
+    max_concurrency : int
+        Total slots for NORMAL + LOW combined.  HIGH requests always succeed.
+    """
+
+    def __init__(self, max_concurrency: int = 3) -> None:
+        self.max_concurrency = max_concurrency
+        # Main semaphore shared by NORMAL and LOW
+        self._semaphore: asyncio.Semaphore | None = None
+        self._lock = asyncio.Lock()
+
+    async def _ensure_sem(self) -> asyncio.Semaphore:
+        if self._semaphore is None:
+            async with self._lock:
+                if self._semaphore is None:
+                    self._semaphore = asyncio.Semaphore(self.max_concurrency)
+        return self._semaphore
+
+    # -- public API --------------------------------------------------------
+
+    async def acquire(self, priority: Priority = Priority.NORMAL) -> None:
+        """Acquire a slot for the given priority.
+
+        Raises
+        ------
+        LlmPriorityRejectError
+            LOW request while all slots are occupied.
+        """
+        if priority == Priority.HIGH:
+            # HIGH: always acquire immediately (no limit)
+            return
+
+        sem = await self._ensure_sem()
+
+        if priority == Priority.NORMAL:
+            await sem.acquire()
+            return
+
+        # LOW — reject if fully occupied
+        # Check semaphore value: if value is 0, no free slots
+        if sem._value is not None and sem._value <= 0:
+            raise LlmPriorityRejectError(
+                "All concurrency slots occupied — LOW request rejected"
+            )
+        # acquire_nowait() is the correct method name
+        await sem.acquire()
+        return
+
+    def release(self, priority: Priority = Priority.NORMAL) -> None:
+        """Release a previously acquired slot."""
+        if priority == Priority.HIGH:
+            return
+        sem = self._semaphore
+        if sem is not None:
+            sem.release()
+
+    def active_count(self) -> int:
+        """Number of currently acquired (NON-HIGH) slots."""
+        sem = self._semaphore
+        if sem is None:
+            return 0
+        return self.max_concurrency - sem._value  # type: ignore[attr-defined]
+
+    def reset(self, max_concurrency: int | None = None) -> None:
+        """Re-create internal semaphores with a new limit."""
+        limit = max_concurrency or self.max_concurrency
+        self.max_concurrency = limit
+        self._lock = asyncio.Lock()
+        self._semaphore = asyncio.Semaphore(limit)
+
+    def __aenter__(self) -> "PriorityLimiter":
+        return self
+
+    def __aexit__(self, *exc: Any) -> None:
+        pass

src/nsct/providers/semaphore.py

@@ -0,0 +1,41 @@
+"""LLM concurrency semaphore — hard limit on parallel LLM requests."""
+
+from __future__ import annotations
+
+import asyncio
+from contextlib import asynccontextmanager
+from typing import AsyncIterator
+
+_semaphore: asyncio.Semaphore | None = None
+_semaphore_lock: asyncio.Lock = asyncio.Lock()
+_semaphore_max: int = 3
+
+
+async def _get_semaphore(max_concurrency: int) -> asyncio.Semaphore:
+    """Lazy-init global semaphore (thread-safe)."""
+    global _semaphore
+    async with _semaphore_lock:
+        if _semaphore is None:
+            _semaphore = asyncio.Semaphore(max_concurrency)
+        return _semaphore
+
+
+@asynccontextmanager
+async def llm_concurrency_limit(max_concurrency: int = 3) -> AsyncIterator[None]:
+    """Context manager that limits concurrent LLM requests.
+
+    Usage::
+
+        async with llm_concurrency_limit(3):
+            await llm_provider.complete(...)
+
+    The semaphore is global and singleton — all callers share it.
+    """
+    sem = await _get_semaphore(max_concurrency)
+    async with sem:
+        yield
+
+
+def get_semaphore_max() -> int:
+    """Return the current semaphore max (default 3)."""
+    return _semaphore_max

tests/stages/test_performance.py

@@ -0,0 +1,315 @@
+"""Stage 19: Performanceoptimierung — Tests für Semaphore, Priority, Batching.
+
+Deckung:
+- semaphore.py: reset_semaphore, acquire_llm, limit enforcement
+- priority_queue.py: PriorityLimiter (HIGH/NORMAL/LOW), Priority enum
+- llm.py: complete/priority, stream_complete/priority, concurrency enforcement
+- orchestrator.py: priority field default, Priority import
+"""
+
+from __future__ import annotations
+
+import asyncio
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from nsct.providers.priority_queue import Priority, PriorityLimiter
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def asyncio_run(coro):
+    """Run a coroutine synchronously."""
+    loop = asyncio.new_event_loop()
+    try:
+        return loop.run_until_complete(coro)
+    finally:
+        loop.close()
+
+
+# ---------------------------------------------------------------------------
+# PriorityLimiter tests
+# ---------------------------------------------------------------------------
+
+
+class TestPriorityLimiter:
+    """Tests for src/nsct/providers/priority_queue.py."""
+
+    def test_high_priority_always_succeeds(self) -> None:
+        limiter = PriorityLimiter(max_concurrency=2)
+
+        async def test():
+            # Fill all slots with NORMAL
+            await limiter.acquire(Priority.NORMAL)
+            await limiter.acquire(Priority.NORMAL)
+            assert limiter.active_count() == 2
+
+            # HIGH should still succeed immediately (no-op for HIGH)
+            await limiter.acquire(Priority.HIGH)
+            limiter.release(Priority.HIGH)
+            limiter.release(Priority.NORMAL)
+            limiter.release(Priority.NORMAL)
+
+        asyncio_run(test())
+
+    def test_normal_waits_for_slot(self) -> None:
+        limiter = PriorityLimiter(max_concurrency=1)
+
+        async def test():
+            await limiter.acquire(Priority.NORMAL)
+            limiter.release(Priority.NORMAL)
+
+        asyncio_run(test())
+
+    def test_low_reject_when_full(self) -> None:
+        limiter = PriorityLimiter(max_concurrency=1)
+        from nsct.providers.priority_queue import LlmPriorityRejectError
+
+        async def test():
+            await limiter.acquire(Priority.NORMAL)
+            assert limiter.active_count() == 1
+            with pytest.raises(LlmPriorityRejectError):
+                await limiter.acquire(Priority.LOW)
+            limiter.release(Priority.NORMAL)
+
+        asyncio_run(test())
+
+    def test_priority_order_values(self) -> None:
+        assert Priority.HIGH.value == 0
+        assert Priority.NORMAL.value == 1
+        assert Priority.LOW.value == 2
+
+    def test_release_only_for_non_high(self) -> None:
+        limiter = PriorityLimiter(max_concurrency=1)
+        limiter.acquire(Priority.HIGH)
+        limiter.release(Priority.HIGH)
+        assert limiter.active_count() == 0
+
+    def test_max_concurrency_reflection(self) -> None:
+        limiter = PriorityLimiter(max_concurrency=5)
+        assert limiter.max_concurrency == 5
+
+    def test_reset(self) -> None:
+        limiter = PriorityLimiter(max_concurrency=2)
+        limiter.reset(4)
+        assert limiter.max_concurrency == 4
+
+
+# ---------------------------------------------------------------------------
+# LLM provider integration tests (mock)
+# ---------------------------------------------------------------------------
+
+
+class TestLLMConcurrency:
+    """Test that _LLMProviderImpl enforces concurrency limits."""
+
+    @pytest.fixture(autouse=True)
+    def _reset_instance(self):
+        import nsct.providers.llm as llm_mod
+        llm_mod._instance = None
+        yield
+        llm_mod._instance = None
+
+    def test_complete_with_priority(self) -> None:
+        from nsct.providers.llm import _LLMProviderImpl
+        from nsct.providers.metrics import ProviderMetrics
+
+        config = MagicMock()
+        config.llm.base_url = "http://test:8030/openai/v1"
+        config.llm.model = "test-model"
+        config.llm.max_concurrency = 2
+        config.llm.get_secret = MagicMock(return_value="test-key")
+
+        metrics = MagicMock()
+        metrics.record_llm_request = AsyncMock()
+        metrics.record_llm_error = AsyncMock()
+
+        impl = _LLMProviderImpl(config, metrics)
+        impl._client = MagicMock()
+        impl._client.chat.completions.create = AsyncMock(
+            return_value=MagicMock(
+                usage=MagicMock(completion_tokens=1, prompt_tokens=10),
+                choices=[MagicMock(message=MagicMock(content="test response"))],
+            )
+        )
+
+        result = asyncio_run(
+            impl.complete([{"role": "user", "content": "hi"}], priority=Priority.HIGH)
+        )
+        assert result == "test response"
+        impl.limiter.release(Priority.HIGH)
+
+    def test_concurrent_calls_respect_limit(self) -> None:
+        from nsct.providers.llm import _LLMProviderImpl
+
+        config = MagicMock()
+        config.llm.base_url = "http://test:8030/openai/v1"
+        config.llm.model = "test-model"
+        config.llm.max_concurrency = 2
+        config.llm.get_secret = MagicMock(return_value="test-key")
+
+        metrics = MagicMock()
+        metrics.record_llm_request = AsyncMock()
+        metrics.record_llm_error = AsyncMock()
+
+        impl = _LLMProviderImpl(config, metrics)
+        impl._client = MagicMock()
+
+        call_count = 0
+        max_concurrent = 0
+
+        async def mock_create(**kwargs):
+            nonlocal call_count, max_concurrent
+            call_count += 1
+            current = call_count
+            if current > max_concurrent:
+                max_concurrent = current
+            await asyncio.sleep(0.05)
+            call_count -= 1
+            return MagicMock(
+                usage=MagicMock(completion_tokens=1, prompt_tokens=10),
+                choices=[MagicMock(message=MagicMock(content="ok"))],
+            )
+
+        impl._client.chat.completions.create = mock_create
+        impl.limiter.max_concurrency = 2
+
+        async def run_all():
+            msgs = [{"role": "user", "content": "x"}]
+            tasks = [impl.complete(msgs) for _ in range(4)]
+            await asyncio.gather(*tasks)
+
+        asyncio_run(run_all())
+        # With max_concurrency=2, we should not exceed 2 simultaneous calls
+        assert max_concurrent <= 2
+
+    def test_priority_high_vs_normal(self) -> None:
+        from nsct.providers.llm import _LLMProviderImpl
+
+        config = MagicMock()
+        config.llm.base_url = "http://test:8030/openai/v1"
+        config.llm.model = "test-model"
+        config.llm.max_concurrency = 1
+        config.llm.get_secret = MagicMock(return_value="test-key")
+
+        metrics = MagicMock()
+        metrics.record_llm_request = AsyncMock()
+        metrics.record_llm_error = AsyncMock()
+
+        impl = _LLMProviderImpl(config, metrics)
+        impl._client = MagicMock()
+        impl.limiter.max_concurrency = 1
+
+        call_order = []
+
+        async def mock_create(**kwargs):
+            await asyncio.sleep(0.05)
+            call_order.append("done")
+            return MagicMock(
+                usage=MagicMock(completion_tokens=1, prompt_tokens=10),
+                choices=[MagicMock(message=MagicMock(content="ok"))],
+            )
+
+        impl._client.chat.completions.create = mock_create
+
+        async def test():
+            normal_done = False
+
+            async def slow():
+                nonlocal normal_done
+                await asyncio.sleep(0.1)
+                normal_done = True
+                return MagicMock(
+                    usage=MagicMock(completion_tokens=1, prompt_tokens=10),
+                    choices=[MagicMock(message=MagicMock(content="ok"))],
+                )
+
+            impl._client.chat.completions.create = slow
+
+            normal_task = asyncio.create_task(
+                impl.complete([{"role": "user", "content": "slow"}], priority=Priority.NORMAL)
+            )
+            await asyncio.sleep(0.02)
+
+            # HIGH should not wait for the NORMAL slot
+            result = await impl.complete(
+                [{"role": "user", "content": "fast"}], priority=Priority.HIGH
+            )
+            assert result == "ok"
+            await normal_task
+
+        asyncio_run(test())
+
+
+# ---------------------------------------------------------------------------
+# Orchestrator integration tests
+# ---------------------------------------------------------------------------
+
+
+class TestOrchestratorPriority:
+    """Test that ResearchOrchestrator accepts and stores priority."""
+
+    def test_default_priority_is_normal(self) -> None:
+        from nsct.orchestration.orchestrator import ResearchOrchestrator
+
+        config = MagicMock()
+        config.llm.base_url = "http://test:8030/openai/v1"
+        config.llm.model = "test-model"
+        config.llm.max_concurrency = 3
+
+        orch = ResearchOrchestrator(
+            config=config,
+            research_id=MagicMock(),
+            query="test query",
+        )
+        assert orch._priority == Priority.NORMAL
+
+    def test_custom_priority_high(self) -> None:
+        from nsct.orchestration.orchestrator import ResearchOrchestrator
+
+        config = MagicMock()
+        config.llm.base_url = "http://test:8030/openai/v1"
+        config.llm.model = "test-model"
+        config.llm.max_concurrency = 3
+
+        orch = ResearchOrchestrator(
+            config=config,
+            research_id=MagicMock(),
+            query="test query",
+            priority=Priority.HIGH,
+        )
+        assert orch._priority == Priority.HIGH
+
+    def test_custom_priority_low(self) -> None:
+        from nsct.orchestration.orchestrator import ResearchOrchestrator
+
+        config = MagicMock()
+        config.llm.base_url = "http://test:8030/openai/v1"
+        config.llm.model = "test-model"
+        config.llm.max_concurrency = 3
+
+        orch = ResearchOrchestrator(
+            config=config,
+            research_id=MagicMock(),
+            query="test query",
+            priority=Priority.LOW,
+        )
+        assert orch._priority == Priority.LOW
+
+
+# ---------------------------------------------------------------------------
+# Async context manager
+# ---------------------------------------------------------------------------
+
+
+class TestAsyncContextManager:
+    """Test that PriorityLimiter works as async context manager."""
+
+    def test_context_manager(self) -> None:
+        limiter = PriorityLimiter(max_concurrency=1)
+        asyncio_run(limiter.__aenter__())
+        asyncio_run(limiter.__aexit__(None, None, None))