diff --git a/frontend-web/src/components/page-sidebar.tsx b/frontend-web/src/components/page-sidebar.tsx index dcd6d382..40ab6267 100644 --- a/frontend-web/src/components/page-sidebar.tsx +++ b/frontend-web/src/components/page-sidebar.tsx @@ -7,6 +7,7 @@ import { ChevronsUpDown, Folder, FolderOpen, + Gauge, ListTree, MousePointerClick, Settings, @@ -231,6 +232,14 @@ function SecondaryNavSidebar() { 白名单管理 )} + {isAdmin && ( + navigate("/page/strategy/settings/model-concurrency")} + > + + 模型并发限制 + + )} {canManageSensitiveWords() && ( navigate("/page/strategy/admin/sensitive-words")} diff --git a/frontend-web/src/core/system-settings/api.ts b/frontend-web/src/core/system-settings/api.ts index 6ca628c2..778b273a 100644 --- a/frontend-web/src/core/system-settings/api.ts +++ b/frontend-web/src/core/system-settings/api.ts @@ -342,6 +342,18 @@ export interface SkillCompressionUpdate { keep_index_in_prompt?: boolean; } +export interface ModelConcurrencySettings { + enabled: boolean; + default_limit: number; + model_limits: Record; +} + +export interface ModelConcurrencyUpdate { + enabled?: boolean; + default_limit?: number; + model_limits?: Record; +} + export async function loadSkillCompressionSettings(): Promise { const response = await apiFetch( `${getBackendBaseURL()}/api/system-settings/skill-compression`, @@ -361,6 +373,53 @@ export async function loadSkillCompressionSettings(): Promise { + const response = await apiFetch( + `${getBackendBaseURL()}/api/system-settings/model-concurrency`, + ); + if (!response.ok) { + throw new Error( + `Failed to load model concurrency settings: ${response.status}`, + ); + } + const data = (await response.json()) as ModelConcurrencySettings; + writeYLog({ + optType: OptType.view, + optDetail: "CM助手-全量-系统设置-模型并发限制查看", + url: "/api/system-settings/model-concurrency", + returnParam: data, + }); + return data; +} + +export async function updateModelConcurrencySettings( + body: ModelConcurrencyUpdate, +): Promise { + const response = await apiFetch( + `${getBackendBaseURL()}/api/system-settings/model-concurrency`, + { + method: "PUT", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(body), + }, + ); + if (!response.ok) { + const detail = await response + .json() + .catch(() => ({ detail: "Failed to update model concurrency settings" })); + throw new Error(detail.detail ?? "Failed to update model concurrency settings"); + } + const data = (await response.json()) as ModelConcurrencySettings; + writeYLog({ + optType: OptType.edit, + optDetail: "CM助手-全量-系统设置-模型并发限制编辑", + url: "/api/system-settings/model-concurrency", + reqParam: body, + returnParam: data, + }); + return data; +} + export async function updateSkillCompressionSettings( body: SkillCompressionUpdate, ): Promise { diff --git a/frontend-web/src/core/system-settings/hooks.ts b/frontend-web/src/core/system-settings/hooks.ts index 0060b22a..5ae171df 100644 --- a/frontend-web/src/core/system-settings/hooks.ts +++ b/frontend-web/src/core/system-settings/hooks.ts @@ -7,6 +7,7 @@ import { loadAppearanceSettings, loadCitationDisplaySettings, loadLeaderboardSettings, + loadModelConcurrencySettings, loadPrivateSquareStatus, loadPromptPrefixSettings, loadPublishSquares, @@ -15,6 +16,7 @@ import { updateAppearanceSettings, updateCitationDisplaySettings, updateLeaderboardSettings, + updateModelConcurrencySettings, updatePrivateSquareSettings, updatePromptPrefixSettings, updatePublishSquares, @@ -22,6 +24,7 @@ import { type AppearanceUpdate, type CitationDisplayUpdate, type LeaderboardSettingsUpdate, + type ModelConcurrencyUpdate, type PrivateSquareUpdate, type PromptPrefixUpdate, type PublishSquaresUpdate, @@ -37,6 +40,7 @@ const APPEARANCE_KEY = ["systemSettings", "appearance"] as const; const CITATION_DISPLAY_KEY = ["systemSettings", "citationDisplay"] as const; const LEADERBOARD_KEY = ["systemSettings", "leaderboard"] as const; const SKILL_COMPRESSION_KEY = ["systemSettings", "skillCompression"] as const; +const MODEL_CONCURRENCY_KEY = ["systemSettings", "modelConcurrency"] as const; export function usePromptPrefixSettings() { return useQuery({ @@ -238,6 +242,27 @@ export function useUpdateSkillCompressionSettings() { }); } +export function useModelConcurrencySettings() { + return useQuery({ + queryKey: MODEL_CONCURRENCY_KEY, + queryFn: () => loadModelConcurrencySettings(), + staleTime: 60_000, + refetchOnWindowFocus: false, + retry: false, + }); +} + +export function useUpdateModelConcurrencySettings() { + const queryClient = useQueryClient(); + return useMutation({ + mutationFn: (body: ModelConcurrencyUpdate) => + updateModelConcurrencySettings(body), + onSuccess: (data) => { + queryClient.setQueryData(MODEL_CONCURRENCY_KEY, data); + }, + }); +} + export function useLeaderboardSettings() { return useQuery({ queryKey: LEADERBOARD_KEY, diff --git a/frontend-web/src/pages/ModelConcurrencySettingsPage.tsx b/frontend-web/src/pages/ModelConcurrencySettingsPage.tsx new file mode 100644 index 00000000..af060286 --- /dev/null +++ b/frontend-web/src/pages/ModelConcurrencySettingsPage.tsx @@ -0,0 +1,189 @@ +"use client"; + +import { Loader2Icon } from "lucide-react"; +import { useEffect, useMemo, useRef, useState } from "react"; +import { toast } from "sonner"; + +import { Button } from "@/components/ui/button"; +import { Input } from "@/components/ui/input"; +import { Switch } from "@/components/ui/switch"; +import { useModels } from "@/core/models/hooks"; +import { + useModelConcurrencySettings, + useUpdateModelConcurrencySettings, +} from "@/core/system-settings/hooks"; + +export default function ModelConcurrencySettingsPage({ + embedded = false, +}: { + embedded?: boolean; +} = {}) { + const { data, isLoading, error } = useModelConcurrencySettings(); + const { models } = useModels(); + const update = useUpdateModelConcurrencySettings(); + + const [enabled, setEnabled] = useState(false); + const [defaultLimit, setDefaultLimit] = useState(10); + const [modelLimits, setModelLimits] = useState>({}); + const initializedRef = useRef(false); + + useEffect(() => { + if (!data || initializedRef.current) return; + setEnabled(data.enabled); + setDefaultLimit(data.default_limit); + setModelLimits( + Object.fromEntries( + Object.entries(data.model_limits ?? {}).map(([name, limit]) => [ + name, + String(limit), + ]), + ), + ); + initializedRef.current = true; + }, [data]); + + const modelNames = useMemo(() => models.map((model) => model.name), [models]); + + const handleModelLimitChange = (modelName: string, value: string) => { + setModelLimits((prev) => ({ ...prev, [modelName]: value })); + }; + + const handleSave = async () => { + const nextDefault = Math.max(1, Math.min(Math.round(defaultLimit) || 10, 10000)); + const cleaned: Record = {}; + for (const [name, rawValue] of Object.entries(modelLimits)) { + const trimmed = rawValue.trim(); + if (!trimmed) continue; + const parsed = Math.round(Number(trimmed)); + if (!Number.isFinite(parsed) || parsed < 1) { + toast.error(`模型 ${name} 的并发上限必须是大于 0 的整数`); + return; + } + cleaned[name] = Math.min(parsed, 10000); + } + + try { + await update.mutateAsync({ + enabled, + default_limit: nextDefault, + model_limits: cleaned, + }); + setDefaultLimit(nextDefault); + setModelLimits( + Object.fromEntries( + Object.entries(cleaned).map(([name, limit]) => [name, String(limit)]), + ), + ); + toast.success("已保存"); + } catch (e) { + toast.error(e instanceof Error ? e.message : "保存失败"); + } + }; + + return ( +
+
+

+ 模型并发限制 +

+

+ 按模型限制同时进行中的问答数。达到上限后，新请求会收到“当前并发已达到上限，请稍后再试”的提示。 +

+
+ + {isLoading ? ( +
+ +
+ ) : error ? ( +
+ 加载失败：{error.message} +
+ ) : ( +
+
+
+
开启限制
+
+ 关闭后，不再按模型并发数拦截问答请求。 +
+
+ +
+ +
+

默认上限

+

+ 未单独配置的模型，使用这个并发上限。 +

+ setDefaultLimit(Number(e.target.value))} + className="w-32" + /> +
+ +
+
+

按模型单独设置

+

+ 留空表示使用默认上限。 +

+
+ +
+ {modelNames.length === 0 ? ( +
+ 当前没有可配置的模型。 +
+ ) : ( + modelNames.map((modelName) => ( +
+
+
{modelName}
+
+ 留空时使用默认上限 {defaultLimit || 10} +
+
+ handleModelLimitChange(modelName, e.target.value)} + className="w-32" + placeholder={String(defaultLimit || 10)} + /> +
+ )) + )} +
+
+ +
+ +
+
+ )} +
+ ); +} diff --git a/frontend-web/src/pages/StrategyRoutes.tsx b/frontend-web/src/pages/StrategyRoutes.tsx index 1aa6498a..1fd42a8e 100644 --- a/frontend-web/src/pages/StrategyRoutes.tsx +++ b/frontend-web/src/pages/StrategyRoutes.tsx @@ -51,6 +51,7 @@ import MemoryPage from "./MemoryPage"; import NewAgentPage from "./NewAgentPage"; import PromptPrefixSettingsPage from "./PromptPrefixSettingsPage"; import PrivateSquareSettingsPage from "./PrivateSquareSettingsPage"; +import ModelConcurrencySettingsPage from "./ModelConcurrencySettingsPage"; import RecommendedQuestionsPage from "./RecommendedQuestionsPage"; import ScheduledTaskRunDetailPage from "./ScheduledTaskRunDetailPage"; import ScheduledTasksPage from "./ScheduledTasksPage"; @@ -346,6 +347,16 @@ export default function StrategyRoutes() { ) } /> + + ) : ( + + ) + } + /> + + + + ) : ( + + ) + } + /> ActiveRunsResponse: user_name=user_emails.get(user_id) or user_id or "未知用户", agent_id=assistant_id, agent_name=agent_name, + model_name=str((r.metadata or {}).get("model_name") or ""), thread_title=(meta_row.display_name if meta_row else "") or "", kind=_kind_for(metadata), multitask_strategy=r.multitask_strategy, diff --git a/offline-backend-20260512/backend/app/gateway/routers/system_settings.py b/offline-backend-20260512/backend/app/gateway/routers/system_settings.py index 1241b5cd..079fc15f 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/system_settings.py +++ b/offline-backend-20260512/backend/app/gateway/routers/system_settings.py @@ -13,6 +13,7 @@ from app.gateway.deps import get_optional_user_from_request from deerflow.config import get_app_config from deerflow.config.system_settings import ( LeaderboardSettings, + ModelConcurrencySettings, PromptPrefixSettings, PublishSquare, PublishSquaresSettings, @@ -128,6 +129,18 @@ class SkillCompressionUpdateRequest(BaseModel): keep_index_in_prompt: bool | None = None +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") + + +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 + + def _skill_compression_to_response(settings: SkillCompressionSettings) -> SkillCompressionResponse: return SkillCompressionResponse( enabled=settings.enabled, @@ -136,6 +149,26 @@ def _skill_compression_to_response(settings: SkillCompressionSettings) -> SkillC ) +def _model_concurrency_to_response(settings: ModelConcurrencySettings) -> ModelConcurrencyResponse: + cleaned: dict[str, int] = {} + for name, limit in (settings.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[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, + ) + + @router.get("/skill-compression", response_model=SkillCompressionResponse) async def get_skill_compression() -> SkillCompressionResponse: """Return the skill-compression configuration (readable by any user). @@ -174,6 +207,47 @@ async def update_skill_compression(body: SkillCompressionUpdateRequest, request: return _skill_compression_to_response(saved.skill_compression) +@router.get("/model-concurrency", response_model=ModelConcurrencyResponse) +async def get_model_concurrency(request: Request) -> ModelConcurrencyResponse: + """Return the per-model concurrency-limit configuration. Admin only.""" + await _require_admin(request) + return _model_concurrency_to_response(load_system_settings().model_concurrency) + + +@router.put("/model-concurrency", response_model=ModelConcurrencyResponse) +async def update_model_concurrency( + body: ModelConcurrencyUpdateRequest, + request: Request, +) -> ModelConcurrencyResponse: + """Update the per-model concurrency-limit configuration. Admin only.""" + await _require_admin(request) + + current = load_system_settings() + fields = body.model_dump(exclude_unset=True) + update: dict[str, object] = {} + if "enabled" in fields and fields["enabled"] is not None: + update["enabled"] = bool(fields["enabled"]) + if "default_limit" in fields and fields["default_limit"] is not None: + update["default_limit"] = max(1, min(int(fields["default_limit"]), 10_000)) + if "model_limits" in fields and fields["model_limits"] is not None: + cleaned: dict[str, int] = {} + for name, limit in (body.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[model_name] = min(normalized_limit, 10_000) + update["model_limits"] = cleaned + 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) + + def _leaderboard_to_response(settings: LeaderboardSettings) -> LeaderboardSettingsResponse: return LeaderboardSettingsResponse( exclude_user_ids=list(dict.fromkeys([uid for uid in settings.exclude_user_ids if uid])), diff --git a/offline-backend-20260512/backend/app/gateway/services.py b/offline-backend-20260512/backend/app/gateway/services.py index 406850ab..5a0d0252 100644 --- a/offline-backend-20260512/backend/app/gateway/services.py +++ b/offline-backend-20260512/backend/app/gateway/services.py @@ -15,13 +15,15 @@ import time from collections.abc import Mapping from typing import Any -from fastapi import HTTPException, Request -from langchain_core.messages import HumanMessage - -from app.gateway.deps import get_run_context, get_run_manager, get_stream_bridge -from app.gateway.utils import sanitize_log_param -from deerflow.config.system_settings import load_system_settings -from deerflow.runtime import ( +from fastapi import HTTPException, Request +from langchain_core.messages import HumanMessage + +from app.gateway.deps import get_run_context, get_run_manager, get_stream_bridge +from app.gateway.utils import sanitize_log_param +from deerflow.config import get_app_config +from deerflow.config.agents_config import load_agent_config +from deerflow.config.system_settings import load_system_settings +from deerflow.runtime import ( END_SENTINEL, HEARTBEAT_SENTINEL, ConflictError, @@ -366,7 +368,7 @@ def build_run_config( return config -def _collect_requested_agent_ids(body: Any) -> set[str]: +def _collect_requested_agent_ids(body: Any) -> set[str]: """Collect agent ids from assistant_id/config/context for access checks.""" agent_ids: set[str] = set() if body.assistant_id and body.assistant_id not in _BUILTIN_GRAPH_IDS: @@ -393,10 +395,99 @@ def _collect_requested_agent_ids(body: Any) -> set[str]: if isinstance(nested_configurable, Mapping): collect(nested_configurable) - return agent_ids - - -async def _sync_legacy_agent_records(store) -> None: + return agent_ids + + +def _extract_requested_model_name(body: Any) -> str | None: + """Best-effort read of the model override requested for this run.""" + context = getattr(body, "context", None) + if isinstance(context, Mapping): + for key in ("model_name", "model"): + value = str(context.get(key) or "").strip() + if value: + return value + + config = getattr(body, "config", None) + if isinstance(config, Mapping): + for container_key in ("context", "configurable"): + container = config.get(container_key) + if not isinstance(container, Mapping): + continue + for key in ("model_name", "model"): + value = str(container.get(key) or "").strip() + if value: + return value + return None + + +def _resolve_effective_run_model_name(body: Any) -> str: + """Resolve the model that this run will effectively use for concurrency limiting.""" + app_config = get_app_config() + 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: + if app_config.get_model_config(requested_model_name): + return requested_model_name + logger.warning( + "Requested model '%s' not found; falling back to default model '%s'.", + requested_model_name, + default_model_name, + ) + return default_model_name + + assistant_id = str(getattr(body, "assistant_id", "") or "").strip() + if assistant_id and assistant_id not in _BUILTIN_GRAPH_IDS: + try: + agent_config = load_agent_config(assistant_id) + except FileNotFoundError: + agent_config = None + except Exception: + logger.debug("Failed to load agent config for model resolution: %s", assistant_id, exc_info=True) + agent_config = None + agent_model_name = str(getattr(agent_config, "model", "") or "").strip() + if agent_model_name: + if app_config.get_model_config(agent_model_name): + return agent_model_name + logger.warning( + "Agent model '%s' for agent '%s' not found; falling back to default model '%s'.", + agent_model_name, + assistant_id, + default_model_name, + ) + + return default_model_name + + +async def _enforce_model_concurrency_limit(run_mgr: RunManager, model_name: str) -> None: + """Reject the run when the selected model has reached its concurrent-run limit.""" + settings = load_system_settings().model_concurrency + if not settings.enabled: + return + + limit = settings.model_limits.get(model_name, settings.default_limit) + try: + limit = int(limit) + except (TypeError, ValueError): + limit = int(settings.default_limit) + limit = max(1, limit) + + active = await run_mgr.list_active() + current = sum( + 1 + for record in active + if str((record.metadata or {}).get("model_name") or "").strip() == model_name + ) + if current >= limit: + raise HTTPException( + status_code=429, + detail=f"当前模型 {model_name} 并发已达到上限（{limit}），请稍后再试。", + ) + + +async def _sync_legacy_agent_records(store) -> None: """Import existing folder-backed agents as built-in records. Heavy: ``list_custom_agents()`` scans ``.deer-flow/agents/`` and reads @@ -567,19 +658,25 @@ async def start_run( # admin-configured mapping rather than from a per-user-visible assistant_id. # Those run under the "default" bucket, where a user-owned unpublished agent # would otherwise 404 the access check. - if enforce_access: - await _enforce_agent_access(body, request) - t_post_enforce = time.perf_counter() - - try: - record = await run_mgr.create_or_reject( - thread_id, - body.assistant_id, - on_disconnect=disconnect, - metadata=body.metadata or {}, - kwargs={"input": body.input, "config": body.config}, - multitask_strategy=body.multitask_strategy, - ) + if enforce_access: + await _enforce_agent_access(body, request) + t_post_enforce = time.perf_counter() + + effective_model_name = _resolve_effective_run_model_name(body) + await _enforce_model_concurrency_limit(run_mgr, effective_model_name) + + run_metadata = dict(body.metadata or {}) + run_metadata["model_name"] = effective_model_name + + try: + record = await run_mgr.create_or_reject( + thread_id, + body.assistant_id, + on_disconnect=disconnect, + metadata=run_metadata, + kwargs={"input": body.input, "config": body.config}, + multitask_strategy=body.multitask_strategy, + ) except ConflictError as exc: # 同一会话上一轮回答还没结束，又提交了新问题 → 拒绝。给用户中文提示， # 而不是把内部英文「Thread ... already has an active run」直接抛到前端。 diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/config/system_settings.py b/offline-backend-20260512/backend/packages/harness/deerflow/config/system_settings.py index 0bb40ac3..eac22b77 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/config/system_settings.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/config/system_settings.py @@ -249,6 +249,26 @@ class ConcurrencyMonitorSettings(BaseModel): model_config = ConfigDict(extra="ignore") +class ModelConcurrencySettings(BaseModel): + """Admin-configurable per-model concurrent Q&A limits.""" + + enabled: bool = Field( + default=False, + description="Whether per-model concurrency limiting is enabled.", + ) + default_limit: int = Field( + default=10, + ge=1, + le=10_000, + description="Fallback max concurrent runs for models without an explicit override.", + ) + model_limits: dict[str, int] = Field( + default_factory=dict, + description="Optional per-model overrides: model_name -> max concurrent runs.", + ) + model_config = ConfigDict(extra="ignore") + + class SystemSettings(BaseModel): """Top-level system settings document.""" @@ -262,6 +282,7 @@ class SystemSettings(BaseModel): skill_compression: SkillCompressionSettings = Field(default_factory=SkillCompressionSettings) agent_modules: AgentModuleSettings = Field(default_factory=AgentModuleSettings) concurrency_monitor: ConcurrencyMonitorSettings = Field(default_factory=ConcurrencyMonitorSettings) + model_concurrency: ModelConcurrencySettings = Field(default_factory=ModelConcurrencySettings) model_config = ConfigDict(extra="ignore") @@ -305,6 +326,15 @@ def get_concurrency_monitor_settings() -> ConcurrencyMonitorSettings: return ConcurrencyMonitorSettings() +def get_model_concurrency_settings() -> ModelConcurrencySettings: + """Convenience accessor for the model-concurrency block (safe defaults on error).""" + try: + return load_system_settings().model_concurrency + except Exception: + logger.debug("Failed to read model concurrency settings; using defaults", exc_info=True) + return ModelConcurrencySettings() + + def save_system_settings(settings: SystemSettings) -> SystemSettings: """Persist the settings document atomically, returning the saved value.""" path = _settings_path() diff --git a/offline-backend-20260512/backend/tests/test_model_concurrency.py b/offline-backend-20260512/backend/tests/test_model_concurrency.py new file mode 100644 index 00000000..0c14f59a --- /dev/null +++ b/offline-backend-20260512/backend/tests/test_model_concurrency.py @@ -0,0 +1,161 @@ +"""Tests for per-model concurrency limit settings and enforcement.""" + +from __future__ import annotations + +import json +from pathlib import Path +from types import SimpleNamespace + +import pytest +from fastapi import HTTPException + +from app.gateway.routers import system_settings as ss_router +from app.gateway.services import ( + _enforce_model_concurrency_limit, + _resolve_effective_run_model_name, +) +from deerflow.config import system_settings as ss +from deerflow.runtime.runs.manager import RunManager + + +@pytest.fixture +def tmp_runtime_home(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Path: + def fake_home() -> Path: + return tmp_path + + monkeypatch.setattr(ss, "runtime_home", fake_home) + return tmp_path + + +def _patch_admin(monkeypatch: pytest.MonkeyPatch, user: object | None = None) -> None: + admin = user or SimpleNamespace( + id="admin1", + system_role="admin", + email="admin@example.com", + ) + + async def _optional(_request): + return admin + + monkeypatch.setattr(ss_router, "get_optional_user_from_request", _optional) + + +@pytest.mark.asyncio +async def test_model_concurrency_settings_round_trip( + tmp_runtime_home: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + _patch_admin(monkeypatch) + req = SimpleNamespace() + + saved = await ss_router.update_model_concurrency( + ss_router.ModelConcurrencyUpdateRequest( + enabled=True, + default_limit=8, + model_limits={"glm5": 10, "deepseek": 4}, + ), + req, + ) + assert saved.enabled is True + assert saved.default_limit == 8 + assert saved.model_limits == {"glm5": 10, "deepseek": 4} + + 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 + + loaded = await ss_router.get_model_concurrency(req) + assert loaded.enabled is True + assert loaded.default_limit == 8 + assert loaded.model_limits["deepseek"] == 4 + + +def test_resolve_effective_run_model_name_prefers_requested_model( + monkeypatch: pytest.MonkeyPatch, +) -> None: + app_config = SimpleNamespace( + models=[SimpleNamespace(name="default-model")], + get_model_config=lambda name: {"name": name} + if name in {"default-model", "glm5", "agent-model"} + else None, + ) + monkeypatch.setattr("app.gateway.services.get_app_config", lambda: app_config) + monkeypatch.setattr( + "app.gateway.services.load_agent_config", + lambda _agent_id: SimpleNamespace(model="agent-model"), + ) + + body = SimpleNamespace( + assistant_id="custom-agent", + context={"model_name": "glm5"}, + config=None, + ) + assert _resolve_effective_run_model_name(body) == "glm5" + + +def test_resolve_effective_run_model_name_uses_agent_default_when_request_missing( + monkeypatch: pytest.MonkeyPatch, +) -> None: + app_config = SimpleNamespace( + models=[SimpleNamespace(name="default-model")], + get_model_config=lambda name: {"name": name} + if name in {"default-model", "agent-model"} + else None, + ) + monkeypatch.setattr("app.gateway.services.get_app_config", lambda: app_config) + monkeypatch.setattr( + "app.gateway.services.load_agent_config", + lambda _agent_id: SimpleNamespace(model="agent-model"), + ) + + body = SimpleNamespace( + assistant_id="custom-agent", + context={}, + config=None, + ) + assert _resolve_effective_run_model_name(body) == "agent-model" + + +@pytest.mark.asyncio +async def test_enforce_model_concurrency_limit_rejects_when_limit_reached( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + "app.gateway.services.load_system_settings", + lambda: SimpleNamespace( + model_concurrency=SimpleNamespace( + enabled=True, + default_limit=5, + model_limits={"glm5": 2}, + ) + ), + ) + run_mgr = RunManager() + await run_mgr.create("t1", metadata={"model_name": "glm5"}) + await run_mgr.create("t2", metadata={"model_name": "glm5"}) + + with pytest.raises(HTTPException) as exc: + await _enforce_model_concurrency_limit(run_mgr, "glm5") + assert exc.value.status_code == 429 + assert "并发已达到上限" in str(exc.value.detail) + + +@pytest.mark.asyncio +async def test_enforce_model_concurrency_limit_skips_when_disabled( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + "app.gateway.services.load_system_settings", + lambda: SimpleNamespace( + model_concurrency=SimpleNamespace( + enabled=False, + default_limit=1, + model_limits={"glm5": 1}, + ) + ), + ) + run_mgr = RunManager() + await run_mgr.create("t1", metadata={"model_name": "glm5"}) + await _enforce_model_concurrency_limit(run_mgr, "glm5")