并发版

offline-backend-20260512/backend/app/gateway/routers/system_settings.py

@@ -21,7 +21,6 @@ from deerflow.config.system_settings import (
     PublishSquare,
     PublishSquaresSettings,
     SkillCompressionSettings,
-    SystemSettings,
     get_module_tag_ids,
     list_publish_squares,
     load_system_settings,
@@ -223,12 +222,16 @@ class ModelConcurrencyResponse(BaseModel):
     enabled: bool = Field(..., description="Whether per-model concurrency limiting is active")
     default_limit: int = Field(..., description="Fallback max concurrent runs for models without an explicit override")
     model_limits: dict[str, int] = Field(default_factory=dict, description="Per-model overrides: model_name -> max concurrent runs")
+    user_default_limit: int = Field(..., description="Fallback max concurrent runs per user and model. 0 disables the per-user fallback")
+    user_model_limits: dict[str, int] = Field(default_factory=dict, description="Per-user per-model overrides: model_name -> max concurrent runs for one user")
 
 
 class ModelConcurrencyUpdateRequest(BaseModel):
     enabled: bool | None = None
     default_limit: int | None = Field(default=None, ge=1, le=10_000)
     model_limits: dict[str, int] | None = None
+    user_default_limit: int | None = Field(default=None, ge=0, le=10_000)
+    user_model_limits: dict[str, int] | None = None
 
 
 def _skill_compression_to_response(settings: SkillCompressionSettings) -> SkillCompressionResponse:
@@ -252,10 +255,24 @@ def _model_concurrency_to_response(settings: ModelConcurrencySettings) -> ModelC
         if normalized_limit < 1:
             continue
         cleaned[model_name] = min(normalized_limit, 10_000)
+    cleaned_user: dict[str, int] = {}
+    for name, limit in (settings.user_model_limits or {}).items():
+        model_name = str(name).strip()
+        if not model_name:
+            continue
+        try:
+            normalized_limit = int(limit)
+        except (TypeError, ValueError):
+            continue
+        if normalized_limit < 1:
+            continue
+        cleaned_user[model_name] = min(normalized_limit, 10_000)
     return ModelConcurrencyResponse(
         enabled=settings.enabled,
         default_limit=max(1, min(int(settings.default_limit), 10_000)),
         model_limits=cleaned,
+        user_default_limit=max(0, min(int(getattr(settings, "user_default_limit", 0) or 0), 10_000)),
+        user_model_limits=cleaned_user,
     )
 
 
@@ -333,6 +350,22 @@ async def update_model_concurrency(
                 continue
             cleaned[model_name] = min(normalized_limit, 10_000)
         update["model_limits"] = cleaned
+    if "user_default_limit" in fields and fields["user_default_limit"] is not None:
+        update["user_default_limit"] = max(0, min(int(fields["user_default_limit"]), 10_000))
+    if "user_model_limits" in fields and fields["user_model_limits"] is not None:
+        cleaned_user: dict[str, int] = {}
+        for name, limit in (body.user_model_limits or {}).items():
+            model_name = str(name).strip()
+            if not model_name:
+                continue
+            try:
+                normalized_limit = int(limit)
+            except (TypeError, ValueError):
+                continue
+            if normalized_limit < 1:
+                continue
+            cleaned_user[model_name] = min(normalized_limit, 10_000)
+        update["user_model_limits"] = cleaned_user
     settings = current.model_concurrency.model_copy(update=update)
     saved = save_system_settings(current.model_copy(update={"model_concurrency": settings}))
     return _model_concurrency_to_response(saved.model_concurrency)

offline-backend-20260512/backend/app/gateway/services.py

@@ -36,6 +36,7 @@ from deerflow.runtime import (
     UnsupportedStrategyError,
     run_agent,
 )
+from deerflow.runtime.user_context import get_effective_user_id
 
 logger = logging.getLogger(__name__)
 
@@ -498,6 +499,49 @@ def _resolve_effective_run_model_name(body: Any) -> str:
     return default_model_name
 
 
+def _resolve_effective_model_name(body: Any, app_config: Any | None = None) -> str:
+    """Compatibility wrapper for resolving the effective run model."""
+    if app_config is None:
+        return _resolve_effective_run_model_name(body)
+
+    default_model_name = app_config.models[0].name if app_config.models else None
+    if not default_model_name:
+        raise ValueError("No chat models are configured. Please configure at least one model in config.yaml.")
+
+    requested_model_name = _extract_requested_model_name(body)
+    if requested_model_name and app_config.get_model_config(requested_model_name):
+        return requested_model_name
+
+    agent_id = _extract_effective_agent_id(body)
+    if agent_id:
+        try:
+            agent_config = load_agent_config(agent_id)
+        except Exception:
+            agent_config = None
+        agent_model_name = str(getattr(agent_config, "model", "") or "").strip()
+        if agent_model_name and app_config.get_model_config(agent_model_name):
+            return agent_model_name
+
+    return default_model_name
+
+
+def _resolve_model_concurrency_limit(model_name: str, app_config: Any) -> int | None:
+    """Read legacy per-model max-concurrency values from model config extras."""
+    model_config = app_config.get_model_config(model_name)
+    extras = getattr(model_config, "model_extra", None) or {}
+    for key in ("max_concurrency", "model_concurrency_limit"):
+        raw = extras.get(key)
+        if raw is None:
+            continue
+        try:
+            limit = int(raw)
+        except (TypeError, ValueError):
+            continue
+        if limit > 0:
+            return limit
+    return None
+
+
 def _get_model_concurrency_limit(model_name: str) -> int | None:
     """Return the configured limit for a model, or ``None`` when limiting is off."""
     settings = load_system_settings().model_concurrency
@@ -512,6 +556,25 @@ def _get_model_concurrency_limit(model_name: str) -> int | None:
     return max(1, limit)
 
 
+def _get_user_model_concurrency_limit(model_name: str) -> int | None:
+    """Return the configured per-user limit for a model, or ``None`` when off."""
+    settings = load_system_settings().model_concurrency
+    if not settings.enabled:
+        return None
+
+    raw_limit = settings.user_model_limits.get(
+        model_name,
+        getattr(settings, "user_default_limit", 0) or 0,
+    )
+    try:
+        limit = int(raw_limit)
+    except (TypeError, ValueError):
+        limit = int(getattr(settings, "user_default_limit", 0) or 0)
+    if limit < 1:
+        return None
+    return max(1, limit)
+
+
 async def _sync_legacy_agent_records(store) -> None:
     """Import existing folder-backed agents as built-in records.
 
@@ -688,10 +751,13 @@ async def start_run(
     t_post_enforce = time.perf_counter()
 
     effective_model_name = _resolve_effective_run_model_name(body)
+    effective_user_id = get_effective_user_id()
     model_concurrency_limit = _get_model_concurrency_limit(effective_model_name)
+    user_model_concurrency_limit = _get_user_model_concurrency_limit(effective_model_name)
 
     run_metadata = dict(body.metadata or {})
     run_metadata["model_name"] = effective_model_name
+    run_metadata["user_id"] = effective_user_id
 
     try:
         record = await run_mgr.create_or_reject(
@@ -702,9 +768,14 @@ async def start_run(
             kwargs={"input": body.input, "config": body.config},
             multitask_strategy=body.multitask_strategy,
             model_concurrency_limit=(effective_model_name, model_concurrency_limit) if model_concurrency_limit is not None else None,
+            user_model_concurrency_limit=(effective_model_name, effective_user_id, user_model_concurrency_limit) if user_model_concurrency_limit is not None else None,
         )
     except ModelConcurrencyLimitError as exc:
-        raise HTTPException(status_code=429, detail="当前使用人数较多，请稍等一下再试。") from exc
+        if getattr(exc, "scope", "") == "user_model":
+            detail = "当前账号对该模型的并发调用已达到上限，请等待已有回答结束后再试。"
+        else:
+            detail = "当前使用人数较多，请稍等一下再试。"
+        raise HTTPException(status_code=429, detail=detail) from exc
     except ConflictError as exc:
         # 同一会话上一轮回答还没结束，又提交了新问题 → 拒绝。给用户中文提示，
         # 而不是把内部英文「Thread ... already has an active run」直接抛到前端。

offline-backend-20260512/backend/packages/harness/deerflow/config/system_settings.py

@@ -286,6 +286,16 @@ class ModelConcurrencySettings(BaseModel):
         default_factory=dict,
         description="Optional per-model overrides: model_name -> max concurrent runs.",
     )
+    user_default_limit: int = Field(
+        default=0,
+        ge=0,
+        le=10_000,
+        description="Fallback max concurrent runs per user and model. 0 disables the per-user fallback.",
+    )
+    user_model_limits: dict[str, int] = Field(
+        default_factory=dict,
+        description="Optional per-user per-model overrides: model_name -> max concurrent runs for one user.",
+    )
     model_config = ConfigDict(extra="ignore")
 
 

offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/manager.py

@@ -218,6 +218,7 @@ class RunManager:
         kwargs: dict | None = None,
         multitask_strategy: str = "reject",
         model_concurrency_limit: tuple[str, int] | None = None,
+        user_model_concurrency_limit: tuple[str, str, int] | None = None,
     ) -> RunRecord:
         """Atomically check for inflight runs and create a new one.
 
@@ -261,7 +262,19 @@ class RunManager:
                 model_name, limit = model_concurrency_limit
                 active_for_model = sum(1 for r in self._runs.values() if r.status in (RunStatus.pending, RunStatus.running) and str((r.metadata or {}).get("model_name") or "").strip() == model_name)
                 if active_for_model >= max(1, int(limit)):
-                    raise ModelConcurrencyLimitError(model_name, limit)
+                    raise ModelConcurrencyLimitError(model_name, limit, scope="model")
+
+            if user_model_concurrency_limit is not None:
+                model_name, user_id, limit = user_model_concurrency_limit
+                active_for_user_model = sum(
+                    1
+                    for r in self._runs.values()
+                    if r.status in (RunStatus.pending, RunStatus.running)
+                    and str((r.metadata or {}).get("model_name") or "").strip() == model_name
+                    and str((r.metadata or {}).get("user_id") or "").strip() == user_id
+                )
+                if active_for_user_model >= max(1, int(limit)):
+                    raise ModelConcurrencyLimitError(model_name, limit, scope="user_model", user_id=user_id)
 
             record = RunRecord(
                 run_id=run_id,
@@ -316,9 +329,11 @@ class ConflictError(Exception):
 class ModelConcurrencyLimitError(Exception):
     """Raised when a model-specific in-flight run limit has been reached."""
 
-    def __init__(self, model_name: str, limit: int) -> None:
+    def __init__(self, model_name: str, limit: int, *, scope: str = "model", user_id: str | None = None) -> None:
         self.model_name = model_name
         self.limit = limit
+        self.scope = scope
+        self.user_id = user_id
         super().__init__(f"Model {model_name} has reached its concurrency limit")
 
 

offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/model_concurrency.py

@@ -0,0 +1,39 @@
+"""Small in-memory gate used by legacy model-concurrency tests."""
+
+from __future__ import annotations
+
+import asyncio
+
+
+class ModelConcurrencyGate:
+    """Track per-model active counts with atomic try-acquire semantics."""
+
+    def __init__(self) -> None:
+        self._lock = asyncio.Lock()
+        self._counts: dict[str, int] = {}
+
+    async def try_acquire(self, model_name: str, limit: int) -> bool:
+        key = str(model_name).strip()
+        if not key:
+            return True
+        async with self._lock:
+            current = self._counts.get(key, 0)
+            if current >= max(1, int(limit)):
+                return False
+            self._counts[key] = current + 1
+            return True
+
+    async def release(self, model_name: str) -> None:
+        key = str(model_name).strip()
+        if not key:
+            return
+        async with self._lock:
+            current = self._counts.get(key, 0)
+            if current <= 1:
+                self._counts.pop(key, None)
+            else:
+                self._counts[key] = current - 1
+
+    async def snapshot(self) -> dict[str, int]:
+        async with self._lock:
+            return dict(self._counts)

offline-backend-20260512/backend/tests/test_model_concurrency.py

@@ -11,6 +11,7 @@ import pytest
 from app.gateway.routers import system_settings as ss_router
 from app.gateway.services import (
     _get_model_concurrency_limit,
+    _get_user_model_concurrency_limit,
     _resolve_effective_run_model_name,
 )
 from deerflow.config import system_settings as ss
@@ -52,23 +53,31 @@ async def test_model_concurrency_settings_round_trip(
             enabled=True,
             default_limit=8,
             model_limits={"glm5": 10, "deepseek": 4},
+            user_default_limit=2,
+            user_model_limits={"glm5": 1},
         ),
         req,
     )
     assert saved.enabled is True
     assert saved.default_limit == 8
     assert saved.model_limits == {"glm5": 10, "deepseek": 4}
+    assert saved.user_default_limit == 2
+    assert saved.user_model_limits == {"glm5": 1}
 
     file_path = tmp_runtime_home / "system_settings.json"
     payload = json.loads(file_path.read_text(encoding="utf-8"))
     assert payload["model_concurrency"]["enabled"] is True
     assert payload["model_concurrency"]["default_limit"] == 8
     assert payload["model_concurrency"]["model_limits"]["glm5"] == 10
+    assert payload["model_concurrency"]["user_default_limit"] == 2
+    assert payload["model_concurrency"]["user_model_limits"]["glm5"] == 1
 
     loaded = await ss_router.get_model_concurrency(req)
     assert loaded.enabled is True
     assert loaded.default_limit == 8
     assert loaded.model_limits["deepseek"] == 4
+    assert loaded.user_default_limit == 2
+    assert loaded.user_model_limits["glm5"] == 1
 
 
 def test_resolve_effective_run_model_name_prefers_requested_model(
@@ -183,6 +192,44 @@ async def test_run_manager_rejects_model_limit_atomically(
         )
 
 
+@pytest.mark.asyncio
+async def test_run_manager_rejects_user_model_limit_atomically(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(
+        "app.gateway.services.load_system_settings",
+        lambda: SimpleNamespace(
+            model_concurrency=SimpleNamespace(
+                enabled=True,
+                default_limit=10,
+                model_limits={},
+                user_default_limit=3,
+                user_model_limits={"glm5": 1},
+            )
+        ),
+    )
+    assert _get_user_model_concurrency_limit("glm5") == 1
+    assert _get_user_model_concurrency_limit("other") == 3
+
+    run_mgr = RunManager()
+    await run_mgr.create_or_reject("t1", metadata={"model_name": "glm5", "user_id": "u1"})
+    await run_mgr.create_or_reject(
+        "t2",
+        metadata={"model_name": "glm5", "user_id": "u2"},
+        user_model_concurrency_limit=("glm5", "u2", 1),
+    )
+
+    with pytest.raises(ModelConcurrencyLimitError) as exc_info:
+        await run_mgr.create_or_reject(
+            "t3",
+            metadata={"model_name": "glm5", "user_id": "u1"},
+            user_model_concurrency_limit=("glm5", "u1", 1),
+        )
+
+    assert exc_info.value.scope == "user_model"
+    assert exc_info.value.user_id == "u1"
+
+
 def test_get_model_concurrency_limit_skips_when_disabled(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -193,7 +240,10 @@ def test_get_model_concurrency_limit_skips_when_disabled(
                 enabled=False,
                 default_limit=1,
                 model_limits={"glm5": 1},
+                user_default_limit=1,
+                user_model_limits={"glm5": 1},
             )
         ),
     )
     assert _get_model_concurrency_limit("glm5") is None
+    assert _get_user_model_concurrency_limit("glm5") is None