diff --git a/README.md b/README.md index 20c7e1560..3db8fc8d8 100644 --- a/README.md +++ b/README.md @@ -110,6 +110,10 @@ That prompt is intended for coding agents. It tells the agent to clone the repo ### Configuration +Optional per-model [`request_admission`](backend/docs/CONFIGURATION.md#model-request-admission) +paces requests to help stay within provider request-per-minute limits. +It is disabled by default; see the linked guide to enable it. + 1. **Clone the DeerFlow repository** ```bash diff --git a/backend/docs/CONFIGURATION.md b/backend/docs/CONFIGURATION.md index 3ef517b1d..a5ca7e401 100644 --- a/backend/docs/CONFIGURATION.md +++ b/backend/docs/CONFIGURATION.md @@ -2,6 +2,51 @@ This guide explains how to configure DeerFlow for your environment. +## Model request admission + +For request-per-minute limits, opt into pacing on each relevant `models[]` +entry. For example, add this alongside its `name`, `use`, and `model` fields: + +```yaml +request_admission: + requests_per_minute: 60 + group: shared-provider-account + max_wait_seconds: 300 + max_queue_size: 256 +``` + +Calls wait in a bounded FIFO before dispatch. At 60 RPM, admissions are spaced +at least one second apart, even after idle periods. The first call can proceed +immediately. Async waiting is cancellable; a cancelled or expired waiter spends +no admission. Queue overflow and wait expiry fail locally before dispatch. +The deadline covers admission waiting only, not the provider's response time. + +An explicit `group` shares the budget across model profiles using the same +provider quota. Omit it for a separate budget per configured model name. +All profiles in a group must have identical settings. The limiter is shared by +factory-created model instances across threads and event loops, including +lead agents, subagents and auxiliary models using the standard LangChain +BaseChatModel invoke/stream hooks. Restart the Gateway after changing, +disabling or regrouping active policies; conflicting settings fail model +construction instead of resetting a live budget. + +When enabled, exposed SDK `max_retries` settings are set to zero: SDK retries +would bypass the admission hook. Agent middleware retries still work and each +new attempt is paced. Calls outside that middleware no longer get SDK retries. +Custom providers that bypass BaseChatModel admission hooks or perform hidden +retries need their own integration. A caller-supplied `rate_limiter` cannot be +combined with `request_admission`. + +This is **process-local RPM pacing**, not TPM accounting or a distributed quota +service. Divide the provider allowance among Gateway workers/replicas and allow +headroom for other applications. Provider token limits, external consumption, +billing failures and permanent errors can still fail a task. Existing +`llm_call.max_concurrent_calls` remains independent: when enabled, its slot is +held while the underlying model waits for admission, so use shared groups and +concurrency settings deliberately. Waiting contributes to model-call latency +and remains subject to the enclosing run's timeout. This option is off by +default and does not promise unlimited retries or eventual task completion. + ## Config Versioning `config.example.yaml` contains a `config_version` field that tracks schema changes. When the example version is higher than your local `config.yaml`, the application emits a startup warning: diff --git a/backend/packages/harness/deerflow/agents/middlewares/llm_error_handling_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/llm_error_handling_middleware.py index 1cd56d918..7df38847b 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/llm_error_handling_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/llm_error_handling_middleware.py @@ -23,6 +23,7 @@ from langchain_core.messages import AIMessage from langgraph.errors import GraphBubbleUp from deerflow.config.app_config import AppConfig +from deerflow.models.request_admission import AdmissionError from deerflow.utils.custom_events import aemit_custom_event, emit_custom_event logger = logging.getLogger(__name__) @@ -504,6 +505,8 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): self._circuit_probe_token = None def _classify_error(self, exc: BaseException) -> tuple[bool, str]: + if isinstance(exc, AdmissionError): + return False, "admission" detail = _extract_error_detail(exc) lowered = detail.lower() error_code = _extract_error_code(exc) diff --git a/backend/packages/harness/deerflow/config/AGENTS.md b/backend/packages/harness/deerflow/config/AGENTS.md index a72bc07fe..8b7958e03 100644 --- a/backend/packages/harness/deerflow/config/AGENTS.md +++ b/backend/packages/harness/deerflow/config/AGENTS.md @@ -47,6 +47,14 @@ Configuration priority: Config values starting with `$` are resolved as environment variables (e.g., `$OPENAI_API_KEY`). `ModelConfig` also declares `use_responses_api` and `output_version` so OpenAI `/v1/responses` can be enabled explicitly while still using `langchain_openai:ChatOpenAI`. +`ModelConfig.request_admission` is optional and is not a provider parameter. +Its positive RPM, finite wait deadline, queue bound and optional quota-group name +configure process-local model pacing. Models sharing an explicit group must use +identical policies. Restart after changing, disabling or regrouping an active +policy; conflicting policies fail construction rather than silently resetting +an active budget. This nested model option is enforced by its limiter registry, +not by the top-level infrastructure reload-boundary registry. + **Extensions Configuration** (`extensions_config.json`): MCP servers and skills are configured together in `extensions_config.json` in project root: diff --git a/backend/packages/harness/deerflow/config/model_config.py b/backend/packages/harness/deerflow/config/model_config.py index 36ba5ed90..590e80975 100644 --- a/backend/packages/harness/deerflow/config/model_config.py +++ b/backend/packages/harness/deerflow/config/model_config.py @@ -1,10 +1,21 @@ from pydantic import BaseModel, ConfigDict, Field +class RequestAdmissionConfig(BaseModel): + """Optional per-process pacing of model requests, shared by quota group.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + requests_per_minute: int = Field(gt=0, strict=True) + group: str | None = Field(default=None, min_length=1, max_length=100, pattern=r"^[A-Za-z0-9_.-]+$") + max_wait_seconds: float = Field(default=300, gt=0, allow_inf_nan=False) + max_queue_size: int = Field(default=256, gt=0, strict=True) + + class ModelConfig(BaseModel): """Config section for a model""" name: str = Field(..., description="Unique name for the model") + request_admission: RequestAdmissionConfig | None = Field(default=None, description="Opt-in process-local RPM pacing. Changing an active group's policy requires a process restart.") display_name: str | None = Field(..., default_factory=lambda: None, description="Display name for the model") description: str | None = Field(..., default_factory=lambda: None, description="Description for the model") use: str = Field( diff --git a/backend/packages/harness/deerflow/models/AGENTS.md b/backend/packages/harness/deerflow/models/AGENTS.md index ca9707ed0..81398eb7d 100644 --- a/backend/packages/harness/deerflow/models/AGENTS.md +++ b/backend/packages/harness/deerflow/models/AGENTS.md @@ -1,5 +1,10 @@ ### Model Factory (`packages/harness/deerflow/models/factory.py`) +Request-admission waits follow the next scheduled admission and configured +interval, capped at 50 ms; the cap must not become a minimum poll interval that +limits high-RPM throughput. Local `AdmissionError` is structurally non-retriable +in LLM error handling regardless of its message text. + - `create_chat_model(name, thinking_enabled)` instantiates LLM from config via reflection - Supports `thinking_enabled` flag with per-model `when_thinking_enabled` overrides - Supports vLLM-style thinking toggles via `when_thinking_enabled.extra_body.chat_template_kwargs.enable_thinking` for Qwen reasoning models, while normalizing legacy `thinking` configs for backward compatibility @@ -7,6 +12,7 @@ - Supports `supports_vision` flag for image understanding models - Config values starting with `$` resolved as environment variables - Missing provider modules surface actionable install hints from reflection resolvers (for example `uv add langchain-google-genai`) +- Optional `models[].request_admission` attaches a process-shared `BaseRateLimiter` at the model factory. Identical explicit groups share one FIFO across model instances, threads and event loops; implicit groups use the configured model name. Policies are immutable once registered and conflicting settings fail construction. A monotonic minimum interval spaces requests without idle-time burst credit; bounded waiters poll without occupying executor threads and unregister in `finally`. The factory strips the policy from provider kwargs and sets exposed SDK `max_retries=0` so middleware retries re-enter admission. This limits model invocations, not tokens or a distributed provider account; custom providers bypassing BaseChatModel hooks are outside the contract. Tests: `test_model_request_admission.py`. ### Claude Code Credentials (`packages/harness/deerflow/models/credential_loader.py`) diff --git a/backend/packages/harness/deerflow/models/factory.py b/backend/packages/harness/deerflow/models/factory.py index f965135d6..1bc837ab7 100644 --- a/backend/packages/harness/deerflow/models/factory.py +++ b/backend/packages/harness/deerflow/models/factory.py @@ -226,6 +226,7 @@ def create_chat_model(name: str | None = None, thinking_enabled: bool = False, * # display) — must never reach the provider client, which would # forward unknown kwargs into the completion request payload. "pricing", + "request_admission", }, ) # Layer per-caller sampling overrides (e.g. a custom agent's temperature / @@ -325,6 +326,18 @@ def create_chat_model(name: str | None = None, thinking_enabled: bool = False, * # explicit profile from a caller or model_overrides is never clobbered. translate_context_window = bool(model_config.context_window) and "profile" not in kwargs and "profile" not in model_settings_from_config + if model_config.request_admission is not None: + from deerflow.models.request_admission import get_request_admission + + if "rate_limiter" in kwargs or "rate_limiter" in model_settings_from_config: + raise ValueError("request_admission cannot be combined with a custom rate_limiter") + model_settings_from_config["rate_limiter"] = get_request_admission(name, model_config.request_admission) + # SDK-internal retries do not re-enter BaseChatModel's admission hook. + # Keep retries at the middleware layer where each attempt is paced. + if "max_retries" in model_class.model_fields: + kwargs.pop("max_retries", None) + model_settings_from_config["max_retries"] = 0 + _warn_unknown_model_settings(model_class, name, model_settings_from_config) model_instance = model_class(**kwargs, **model_settings_from_config) diff --git a/backend/packages/harness/deerflow/models/request_admission.py b/backend/packages/harness/deerflow/models/request_admission.py new file mode 100644 index 000000000..ebe08e1c9 --- /dev/null +++ b/backend/packages/harness/deerflow/models/request_admission.py @@ -0,0 +1,113 @@ +"""Bounded FIFO admission shared by synchronous calls and independent loops. + +No executor worker or timer task is retained while waiting. Short polling sleeps +allow cancellation and deadlines without storing references to caller loops. +Requests are evenly spaced; idle time never accumulates a burst allowance. +""" + +import asyncio +import threading +import time +from collections import deque +from time import monotonic + +from langchain_core.rate_limiters import BaseRateLimiter + +from deerflow.config.model_config import RequestAdmissionConfig + + +class AdmissionError(RuntimeError): + """Local admission failed before making an upstream request.""" + + +class RequestAdmission(BaseRateLimiter): + def __init__(self, config: RequestAdmissionConfig): + self.config = config + self._interval = 60 / config.requests_per_minute + self._next = 0.0 + self._lock = threading.Lock() + self._waiters: deque[object] = deque() + + def _try(self, ticket: object | None) -> bool: + with self._lock: + now = monotonic() + if self._waiters and self._waiters[0] is not ticket: + return False + if now < self._next: + return False + self._next = now + self._interval + return True + + def _enqueue(self) -> object: + ticket = object() + with self._lock: + if len(self._waiters) >= self.config.max_queue_size: + raise AdmissionError("LLM admission queue is full; reduce workload or increase queue capacity.") + self._waiters.append(ticket) + return ticket + + def _remove(self, ticket: object) -> None: + with self._lock: + self._waiters.remove(ticket) + + def _delay(self, deadline: float) -> float: + now = monotonic() + remaining = deadline - now + if remaining <= 0: + raise AdmissionError("LLM admission timed out before dispatch; increase max_wait_seconds or reduce workload.") + with self._lock: + until_next = self._next - now + # Track short admission intervals rather than imposing a 20/s ceiling. + # Non-head waiters still yield when the schedule is already due. + return min(0.05, self._interval, until_next if until_next > 0 else self._interval, remaining) + + def acquire(self, *, blocking: bool = True) -> bool: + if self._try(None): + return True + if not blocking: + return False + deadline = monotonic() + self.config.max_wait_seconds + ticket = self._enqueue() + try: + while True: + delay = self._delay(deadline) + if self._try(ticket): + return True + time.sleep(delay) + finally: + self._remove(ticket) + + async def aacquire(self, *, blocking: bool = True) -> bool: + if self._try(None): + return True + if not blocking: + return False + deadline = monotonic() + self.config.max_wait_seconds + ticket = self._enqueue() + try: + while True: + delay = self._delay(deadline) + if self._try(ticket): + return True + await asyncio.sleep(delay) + finally: + self._remove(ticket) + + +_registry_lock = threading.Lock() +_registry: dict[tuple[str, str], RequestAdmission] = {} + + +def get_request_admission(model_name: str, config: RequestAdmissionConfig) -> RequestAdmission: + # Explicit group names never collide with implicit model names. Operator + # configuration determines the finite set; no user IDs or secrets are keys. + key = ("group", config.group) if config.group else ("model", model_name) + with _registry_lock: + existing = _registry.get(key) + if existing is not None: + if existing.config != config: + raise ValueError("Model request admission policy changed or conflicts within a shared group; align settings and restart the Gateway.") + return existing + limiter = RequestAdmission(config) + _registry[key] = limiter + return limiter diff --git a/backend/tests/test_model_request_admission.py b/backend/tests/test_model_request_admission.py new file mode 100644 index 000000000..3632361a4 --- /dev/null +++ b/backend/tests/test_model_request_admission.py @@ -0,0 +1,228 @@ +"""Offline request pacing, queue lifecycle, and model-factory integration.""" + +import asyncio +import threading +from concurrent.futures import ThreadPoolExecutor + +import pytest +from pydantic import ValidationError + +from deerflow.config.model_config import RequestAdmissionConfig +from deerflow.models import request_admission as admission + + +@pytest.fixture +def clock(monkeypatch): + now = [0.0] + monkeypatch.setattr(admission, "monotonic", lambda: now[0]) + return now + + +def test_pacing_has_no_catch_up_burst(clock): + limiter = admission.RequestAdmission(RequestAdmissionConfig(requests_per_minute=60)) + assert limiter.acquire(blocking=False) + assert not limiter.acquire(blocking=False) + clock[0] = 0.999 + assert not limiter.acquire(blocking=False) + clock[0] = 1 + assert limiter.acquire(blocking=False) + clock[0] = 100 + assert limiter.acquire(blocking=False) + assert not limiter.acquire(blocking=False) + + +def test_high_rpm_wait_tracks_next_admission(clock): + limiter = admission.RequestAdmission(RequestAdmissionConfig(requests_per_minute=6000)) + limiter.acquire() + assert limiter._delay(300) == pytest.approx(0.01) + clock[0] = 0.009 + assert limiter._delay(300) == pytest.approx(0.001) + clock[0] = 0.02 + # A non-head waiter must yield, rather than spin on an overdue schedule. + assert 0 < limiter._delay(300) <= 0.01 + + +@pytest.mark.asyncio +async def test_fifo_cancellation_and_queue_capacity(clock): + limiter = admission.RequestAdmission(RequestAdmissionConfig(requests_per_minute=60, max_queue_size=2)) + await limiter.aacquire() + first = asyncio.create_task(limiter.aacquire()) + second = asyncio.create_task(limiter.aacquire()) + await asyncio.sleep(0) + try: + with pytest.raises(admission.AdmissionError, match="queue is full"): + await limiter.aacquire() + assert not limiter.acquire(blocking=False) + first.cancel() + with pytest.raises(asyncio.CancelledError): + await first + clock[0] = 1 + assert await asyncio.wait_for(second, 1) + assert not limiter._waiters + finally: + for task in (first, second): + task.cancel() + await asyncio.gather(first, second, return_exceptions=True) + + +@pytest.mark.asyncio +async def test_wait_deadline_does_not_spend_a_permit(clock): + limiter = admission.RequestAdmission(RequestAdmissionConfig(requests_per_minute=1, max_wait_seconds=1)) + await limiter.aacquire() + waiter = asyncio.create_task(limiter.aacquire()) + await asyncio.sleep(0) + clock[0] = 2 + with pytest.raises(admission.AdmissionError, match="timed out"): + await asyncio.wait_for(waiter, 1) + assert not limiter._waiters + clock[0] = 60 + assert limiter.acquire(blocking=False) + + +def test_sync_and_foreign_loops_share_one_budget(clock): + limiter = admission.RequestAdmission(RequestAdmissionConfig(requests_per_minute=1)) + barrier = threading.Barrier(8) + + def attempt(index): + barrier.wait(timeout=5) + return limiter.acquire(blocking=False) if index % 2 else asyncio.run(limiter.aacquire(blocking=False)) + + with ThreadPoolExecutor(max_workers=8) as executor: + assert sum(executor.map(attempt, range(8))) == 1 + + +@pytest.mark.parametrize("values", [{"requests_per_minute": 0}, {"requests_per_minute": True}, {"requests_per_minute": 1, "max_wait_seconds": float("inf")}, {"requests_per_minute": 1, "max_queue_size": 0}]) +def test_invalid_configuration(values): + with pytest.raises(ValidationError): + RequestAdmissionConfig(**values) + + +def test_sync_timeout_removes_waiter(clock, monkeypatch): + limiter = admission.RequestAdmission(RequestAdmissionConfig(requests_per_minute=1, max_wait_seconds=1)) + limiter.acquire() + monkeypatch.setattr(admission.time, "sleep", lambda _: clock.__setitem__(0, 2)) + with pytest.raises(admission.AdmissionError, match="timed out"): + limiter.acquire() + assert not limiter._waiters + + +@pytest.mark.asyncio +async def test_fifo_prevents_newcomers_overtaking(clock): + limiter = admission.RequestAdmission(RequestAdmissionConfig(requests_per_minute=60)) + limiter.acquire() + order = [] + + async def wait(index): + await limiter.aacquire() + order.append(index) + + first = asyncio.create_task(wait(1)) + second = asyncio.create_task(wait(2)) + await asyncio.sleep(0) + try: + clock[0] = 1 + assert not limiter.acquire(blocking=False) + await asyncio.wait_for(first, 1) + assert order == [1] + clock[0] = 2 + await asyncio.wait_for(second, 1) + assert order == [1, 2] + finally: + first.cancel() + second.cancel() + await asyncio.gather(first, second, return_exceptions=True) + + +@pytest.fixture +def registry(monkeypatch): + monkeypatch.setattr(admission, "_registry", {}) + + +def test_group_sharing_isolation_and_conflict(registry, clock): + config = RequestAdmissionConfig(requests_per_minute=60, group="shared") + a = admission.get_request_admission("a", config) + assert admission.get_request_admission("b", config) is a + assert a.acquire(blocking=False) + assert not admission.get_request_admission("b", config).acquire(blocking=False) + # An implicit model named 'shared' must not collide with that group. + b = admission.get_request_admission("shared", RequestAdmissionConfig(requests_per_minute=60)) + assert b.acquire(blocking=False) + with pytest.raises(ValueError, match="restart"): + admission.get_request_admission("a", config.model_copy(update={"requests_per_minute": 30})) + + +def make_model(monkeypatch, *, name="a", policy=None, provider=False): + from langchain_core.language_models.fake_chat_models import FakeListChatModel + + from deerflow.config.app_config import AppConfig + from deerflow.config.model_config import ModelConfig + from deerflow.config.sandbox_config import SandboxConfig + from deerflow.models import factory + + model = ModelConfig(name=name, use="langchain_openai:ChatOpenAI", model="test", api_key="offline-test-key", request_admission=policy) + config = AppConfig(models=[model], sandbox=SandboxConfig(use="deerflow.sandbox.local:LocalSandboxProvider")) + if not provider: + monkeypatch.setattr(factory, "resolve_class", lambda *args: FakeListChatModel) + return factory.create_chat_model(name, app_config=config, attach_tracing=False, **({} if provider else {"responses": ["ok"]})) + + +@pytest.mark.asyncio +async def test_factory_invoke_and_stream_share_budget(monkeypatch, registry, clock): + policy = RequestAdmissionConfig(requests_per_minute=60, group="account") + a = make_model(monkeypatch, policy=policy) + b = make_model(monkeypatch, name="b", policy=policy) + assert a.rate_limiter is b.rate_limiter + assert a.invoke("hello").content == "ok" + pending = asyncio.create_task(b.ainvoke("hello")) + # Drive the normal BaseChatModel pipeline up to admission, not just a mock. + for _ in range(100): + if a.rate_limiter._waiters: + break + await asyncio.sleep(0) + try: + assert a.rate_limiter._waiters + assert not pending.done() + clock[0] = 1 + assert (await asyncio.wait_for(pending, 1)).content == "ok" + clock[0] = 2 + assert "".join(chunk.content for chunk in a.stream("hello")) == "ok" + assert not a.rate_limiter.acquire(blocking=False) + clock[0] = 3 + assert "".join([chunk.content async for chunk in b.astream("hello")]) == "ok" + assert not a.rate_limiter.acquire(blocking=False) + finally: + pending.cancel() + await asyncio.gather(pending, return_exceptions=True) + + +def test_real_openai_factory_does_not_forward_policy_or_retry_inside_sdk(monkeypatch, registry): + model = make_model(monkeypatch, policy=RequestAdmissionConfig(requests_per_minute=30), provider=True) + assert isinstance(model.rate_limiter, admission.RequestAdmission) + assert model.max_retries == 0 + assert "request_admission" not in model.model_kwargs + assert "request_admission" not in model._default_params + + +def test_disabled_factory_preserves_default(monkeypatch, registry): + model = make_model(monkeypatch) + assert model.rate_limiter is None + assert not admission._registry + + +def test_admission_failures_are_not_retried_as_provider_errors(monkeypatch): + from deerflow.agents.middlewares import llm_error_handling_middleware as errors + from deerflow.config.app_config import AppConfig + from deerflow.config.sandbox_config import SandboxConfig + + monkeypatch.setattr(errors, "_PROCESS_LIMITER", None) + monkeypatch.setattr(errors, "_CAP_RESOLVED", False) + middleware = errors.LLMErrorHandlingMiddleware(app_config=AppConfig(sandbox=SandboxConfig(use="test"))) + for message in ( + "LLM admission queue is full; reduce workload or increase queue capacity.", + "LLM admission timed out before dispatch; increase max_wait_seconds or reduce workload.", + "rate limit exceeded locally", + "provider quota", + "server busy", + ): + retry, _ = middleware._classify_error(admission.AdmissionError(message)) + assert retry is False diff --git a/config.example.yaml b/config.example.yaml index 5091424df..f36c3f3c3 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -20,7 +20,17 @@ # ============================================================================ # Bump this number when the config schema changes. # Run `make config-upgrade` to merge new fields into your local config.yaml. -config_version: 43 +config_version: 44 + +# Optional per-model request pacing (inside a models[] entry): +# request_admission: +# requests_per_minute: 60 +# group: shared-provider-account # optional; defaults to the model config name +# max_wait_seconds: 300 +# max_queue_size: 256 +# Off unless configured. Shared groups require identical settings; restart after +# changes. Process-local RPM only, not TPM or a cluster-wide quota. See the +# "Model request admission" section in backend/docs/CONFIGURATION.md. # ============================================================================ # Logging diff --git a/deploy/helm/deer-flow/README.md b/deploy/helm/deer-flow/README.md index a63ecf608..edb3cba8b 100644 --- a/deploy/helm/deer-flow/README.md +++ b/deploy/helm/deer-flow/README.md @@ -131,7 +131,7 @@ they resolve from the `secrets` map): ```yaml config: | - config_version: 43 + config_version: 44 models: - name: gpt-4 use: langchain_openai:ChatOpenAI diff --git a/deploy/helm/deer-flow/values.yaml b/deploy/helm/deer-flow/values.yaml index 044f9206b..10439fda9 100644 --- a/deploy/helm/deer-flow/values.yaml +++ b/deploy/helm/deer-flow/values.yaml @@ -249,7 +249,7 @@ ingress: # -- DeerFlow config.yaml content. Secrets MUST stay as $VAR references — never # inline literal secret values here. The default enables provisioner sandbox. config: | - config_version: 43 + config_version: 44 log_level: info recursion_limit: 100 max_recursion_limit: 1000