diff --git a/backend/internal/agent/runtime/runtime.go b/backend/internal/agent/runtime/runtime.go index b38f84795..c1403c72e 100644 --- a/backend/internal/agent/runtime/runtime.go +++ b/backend/internal/agent/runtime/runtime.go @@ -222,27 +222,83 @@ func RuntimeDir() (string, error) { return "", errors.New("Agent runtime files are missing; set CANVAS_PI_RUNTIME_DIR") } -var ( - agentSlotsOnce sync.Once - agentSlots chan struct{} +const ( + defaultAgentProcessLimit = 30 + maxAgentProcessLimit = 64 ) -func acquireAgentProcess(ctx context.Context) (func(), error) { - agentSlotsOnce.Do(func() { - limit := 8 - if raw := strings.TrimSpace(os.Getenv("CANVAS_AGENT_MAX_PROCESSES")); raw != "" { - if parsed, err := strconv.Atoi(raw); err == nil && parsed > 0 && parsed <= 1000 { - limit = parsed - } +type agentProcessGate struct { + mu sync.Mutex + limit int + active int + wake chan struct{} +} + +var processGate = newAgentProcessGate() + +func newAgentProcessGate() *agentProcessGate { + limit := defaultAgentProcessLimit + if raw := strings.TrimSpace(os.Getenv("CANVAS_AGENT_MAX_PROCESSES")); raw != "" { + if parsed, err := strconv.Atoi(raw); err == nil { + limit = clampAgentProcessLimit(parsed) } - agentSlots = make(chan struct{}, limit) - }) - select { - case agentSlots <- struct{}{}: - return func() { <-agentSlots }, nil - case <-ctx.Done(): - return nil, ctx.Err() } + return &agentProcessGate{limit: limit, wake: make(chan struct{})} +} + +func clampAgentProcessLimit(value int) int { + if value < 1 { + return defaultAgentProcessLimit + } + if value > maxAgentProcessLimit { + return maxAgentProcessLimit + } + return value +} + +// SetProcessLimit changes how many embedded Agent processes may run at once. +// Lowering the limit does not stop a process that has already started. +func SetProcessLimit(value int) (active int, limit int) { + processGate.mu.Lock() + defer processGate.mu.Unlock() + processGate.limit = clampAgentProcessLimit(value) + close(processGate.wake) + processGate.wake = make(chan struct{}) + return processGate.active, processGate.limit +} + +func ProcessUsage() (active int, limit int) { + processGate.mu.Lock() + defer processGate.mu.Unlock() + return processGate.active, processGate.limit +} + +func acquireAgentProcess(ctx context.Context) (func(), error) { + for { + processGate.mu.Lock() + if processGate.active < processGate.limit { + processGate.active++ + processGate.mu.Unlock() + return releaseAgentProcess, nil + } + wake := processGate.wake + processGate.mu.Unlock() + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-wake: + } + } +} + +func releaseAgentProcess() { + processGate.mu.Lock() + if processGate.active > 0 { + processGate.active-- + } + close(processGate.wake) + processGate.wake = make(chan struct{}) + processGate.mu.Unlock() } func nodeMemoryMB() string { diff --git a/backend/internal/agent/runtime/runtime_test.go b/backend/internal/agent/runtime/runtime_test.go index 147fef974..4089f63f5 100644 --- a/backend/internal/agent/runtime/runtime_test.go +++ b/backend/internal/agent/runtime/runtime_test.go @@ -24,6 +24,25 @@ func fakeRuntime(t *testing.T, script string) { t.Setenv("CANVAS_PI_RUNTIME_DIR", dir) } +func TestSetProcessLimitDoesNotStopActiveWork(t *testing.T) { + t.Cleanup(func() { SetProcessLimit(defaultAgentProcessLimit) }) + release, err := acquireAgentProcess(context.Background()) + if err != nil { + t.Fatal(err) + } + active, limit := SetProcessLimit(1) + if active != 1 || limit != 1 { + t.Fatalf("active=%d limit=%d, want active 1 and limit 1", active, limit) + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond) + defer cancel() + if _, err := acquireAgentProcess(ctx); err == nil { + t.Fatal("lowered limit still admitted another process") + } + release() + SetProcessLimit(defaultAgentProcessLimit) +} + func noopBridge() Bridge { handler := func(context.Context, map[string]json.RawMessage) (any, error) { return map[string]any{"ok": true}, nil } return Bridge{Model: handler, Tool: handler, Event: handler} diff --git a/backend/internal/agent/yingceagent/client.go b/backend/internal/agent/yingceagent/client.go index 27f93df8b..9244d4cdf 100644 --- a/backend/internal/agent/yingceagent/client.go +++ b/backend/internal/agent/yingceagent/client.go @@ -14,6 +14,68 @@ import ( agentruntime "infinite-canvas/backend/internal/agent/runtime" ) +type Status struct { + OK bool `json:"ok"` + Active int `json:"active"` + Limit int `json:"limit"` + Queued int `json:"queued"` +} + +func Inspect(ctx context.Context, endpoint string) (Status, error) { + endpoint = strings.TrimRight(strings.TrimSpace(endpoint), "/") + if endpoint == "" { + return Status{}, fmt.Errorf("YINGCE_AGENT_URL is empty") + } + request, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint+"/healthz", nil) + if err != nil { + return Status{}, err + } + response, err := remoteClient.Do(request) + if err != nil { + return Status{}, err + } + defer response.Body.Close() + if response.StatusCode != http.StatusOK { + return Status{}, fmt.Errorf("yingce-agent health returned HTTP %d", response.StatusCode) + } + var status Status + if err := json.NewDecoder(io.LimitReader(response.Body, 4096)).Decode(&status); err != nil { + return Status{}, err + } + return status, nil +} + +func SetLimit(ctx context.Context, endpoint, serviceToken string, limit int) error { + endpoint = strings.TrimRight(strings.TrimSpace(endpoint), "/") + serviceToken = strings.TrimSpace(serviceToken) + if endpoint == "" { + return fmt.Errorf("YINGCE_AGENT_URL is empty") + } + if len(serviceToken) < 32 { + return fmt.Errorf("YINGCE_AGENT_TOKEN must contain at least 32 characters") + } + payload, err := json.Marshal(map[string]int{"maxSessions": limit}) + if err != nil { + return err + } + request, err := http.NewRequestWithContext(ctx, http.MethodPut, endpoint+"/v1/limit", bytes.NewReader(payload)) + if err != nil { + return err + } + request.Header.Set("Authorization", "Bearer "+serviceToken) + request.Header.Set("Content-Type", "application/json") + response, err := remoteClient.Do(request) + if err != nil { + return err + } + defer response.Body.Close() + if response.StatusCode != http.StatusOK { + detail, _ := io.ReadAll(io.LimitReader(response.Body, 1024)) + return fmt.Errorf("yingce-agent limit returned HTTP %d: %s", response.StatusCode, strings.TrimSpace(string(detail))) + } + return nil +} + // Run executes one Agent session on the independent yingce-agent service. // The Go process keeps the bridge, billing, and tool execution. A configured // remote endpoint never falls back to an embedded Node process: doing so can diff --git a/backend/internal/agent/yingceagent/client_test.go b/backend/internal/agent/yingceagent/client_test.go index d76302f67..e154bfc49 100644 --- a/backend/internal/agent/yingceagent/client_test.go +++ b/backend/internal/agent/yingceagent/client_test.go @@ -116,4 +116,14 @@ if (!response.ok) { if !called { t.Fatal("remote runtime did not call the Go tool bridge") } + if err := SetLimit(ctx, endpoint, token, 16); err != nil { + t.Fatal(err) + } + status, err := Inspect(ctx, endpoint) + if err != nil { + t.Fatal(err) + } + if status.Limit != 16 { + t.Fatalf("limit = %d, want 16", status.Limit) + } } diff --git a/backend/internal/app/admin_system_performance.go b/backend/internal/app/admin_system_performance.go index 2de8eecef..223df8902 100644 --- a/backend/internal/app/admin_system_performance.go +++ b/backend/internal/app/admin_system_performance.go @@ -3,14 +3,18 @@ package app import ( "context" "errors" + "fmt" "os" "runtime" "strconv" "strings" "time" + agentruntime "infinite-canvas/backend/internal/agent/runtime" + "infinite-canvas/backend/internal/agent/yingceagent" "infinite-canvas/backend/internal/buildinfo" "infinite-canvas/backend/internal/model" + "infinite-canvas/backend/internal/platform" "infinite-canvas/backend/internal/repository" "github.com/redis/go-redis/v9" @@ -28,9 +32,23 @@ type AdminSystemPerformance struct { Disk SystemPerformanceDisk `json:"disk"` Database repository.DatabaseRuntimeStats `json:"database"` Redis SystemPerformanceRedis `json:"redis"` + Agents []SystemPerformanceAgent `json:"agents"` Build buildinfo.Info `json:"build"` } +type SystemPerformanceAgent struct { + ID string `json:"id"` + Name string `json:"name"` + Mode string `json:"mode"` + Endpoint string `json:"endpoint"` + Healthy bool `json:"healthy"` + Active int `json:"active"` + Limit int `json:"limit"` + Queued int `json:"queued"` + ConfiguredLimit int `json:"configuredLimit"` + StatusMessage string `json:"statusMessage,omitempty"` +} + type SystemPerformanceHost struct { Hostname string `json:"hostname"` OS string `json:"os"` @@ -149,17 +167,98 @@ func (s *Service) AdminSystemPerformance(ctx context.Context, actor *model.User) disk := collectSystemDisk(s.dataDir) databaseStats, databaseErr := s.repo.DatabaseRuntimeStats(ctx) redisStats := s.collectRedisPerformance(ctx) + agents := s.collectAgentServices(ctx) status := "healthy" - if databaseErr != nil || !databaseStats.Connected || !disk.Available || !disk.Writable || s.ValidateRuntime() != nil || (redisStats.Configured && !redisStats.Connected) { + if databaseErr != nil || !databaseStats.Connected || !disk.Available || !disk.Writable || s.ValidateRuntime() != nil || (redisStats.Configured && !redisStats.Connected) || agentServicesDegraded(agents) { status = "degraded" } return &AdminSystemPerformance{ CollectedAt: time.Now(), Status: status, Host: SystemPerformanceHost{Hostname: hostname, OS: runtime.GOOS, Arch: runtime.GOARCH, CPUCores: runtime.NumCPU(), GOMAXPROCS: runtime.GOMAXPROCS(0), ProcessID: os.Getpid(), UptimeSeconds: int64(time.Since(systemProcessStartedAt).Seconds()), Goroutines: runtime.NumGoroutine(), ActiveWorkerTasks: s.ActiveWorkerTasks(), LoadAverage: load, LoadAverageAvailable: loadAvailable}, - Memory: memory, Disk: disk, Database: databaseStats, Redis: redisStats, Build: buildinfo.Current(), + Memory: memory, Disk: disk, Database: databaseStats, Redis: redisStats, Agents: agents, Build: buildinfo.Current(), }, nil } +type AgentSessionLimitRequest struct { + MaxSessions int `json:"maxSessions"` +} + +func (s *Service) UpdateAgentSessionLimit(ctx context.Context, actor *model.User, maxSessions int) (*SystemPerformanceAgent, error) { + if err := s.RequireAdmin(actor); err != nil { + return nil, err + } + if maxSessions < platform.MinRuntimeAgentSessions || maxSessions > platform.MaxRuntimeAgentSessions { + return nil, BadAuthRequest(fmt.Sprintf("Agent 同时对话上限必须是 %d-%d 的整数", platform.MinRuntimeAgentSessions, platform.MaxRuntimeAgentSessions)) + } + policy, err := s.RuntimePolicy() + if err != nil { + return nil, err + } + policy.Task.AgentMaxSessions = maxSessions + if _, err := s.UpdateRuntimePolicySetting(actor, policy); err != nil { + return nil, err + } + if err := s.applyAgentSessionLimit(ctx, maxSessions); err != nil { + return nil, err + } + agents := s.collectAgentServices(ctx) + if len(agents) == 0 { + return nil, fmt.Errorf("Agent 服务状态不可用") + } + return &agents[0], nil +} + +func (s *Service) applyAgentSessionLimit(ctx context.Context, maxSessions int) error { + agentruntime.SetProcessLimit(maxSessions) + endpoint := strings.TrimSpace(os.Getenv("YINGCE_AGENT_URL")) + if endpoint == "" { + return nil + } + return yingceagent.SetLimit(ctx, endpoint, os.Getenv("YINGCE_AGENT_TOKEN"), maxSessions) +} + +func (s *Service) collectAgentServices(ctx context.Context) []SystemPerformanceAgent { + configured := platform.DefaultRuntimeAgentSessions + if policy, err := s.RuntimePolicy(); err == nil && policy.Task.AgentMaxSessions > 0 { + configured = policy.Task.AgentMaxSessions + } + endpoint := strings.TrimSpace(os.Getenv("YINGCE_AGENT_URL")) + if endpoint == "" { + active, limit := agentruntime.ProcessUsage() + return []SystemPerformanceAgent{{ + ID: "embedded", Name: "画布 Agent", Mode: "embedded", Endpoint: "本进程", + Healthy: true, Active: active, Limit: limit, ConfiguredLimit: configured, + StatusMessage: "本地开发由后端进程直接启动，不单独占用容器。", + }} + } + service := SystemPerformanceAgent{ + ID: "yingce-agent", Name: "画布 Agent", Mode: "remote", Endpoint: endpoint, + ConfiguredLimit: configured, + } + status, err := yingceagent.Inspect(ctx, endpoint) + if err != nil { + service.StatusMessage = "独立 Agent 服务不可达。正在进行的对话会失败，不会自动改回本进程执行。" + return []SystemPerformanceAgent{service} + } + service.Healthy = status.OK + service.Active = status.Active + service.Limit = status.Limit + service.Queued = status.Queued + if status.Limit != configured { + service.StatusMessage = "运行中的上限与已保存配置不一致，重新保存一次即可应用。" + } + return []SystemPerformanceAgent{service} +} + +func agentServicesDegraded(agents []SystemPerformanceAgent) bool { + for _, agent := range agents { + if !agent.Healthy { + return true + } + } + return false +} + func collectSystemMemory() SystemPerformanceMemory { var stats runtime.MemStats runtime.ReadMemStats(&stats) diff --git a/backend/internal/app/cloud_agent_step_timeout_test.go b/backend/internal/app/cloud_agent_step_timeout_test.go index 461f0a68e..da4f1e8a8 100644 --- a/backend/internal/app/cloud_agent_step_timeout_test.go +++ b/backend/internal/app/cloud_agent_step_timeout_test.go @@ -161,6 +161,7 @@ func TestRuntimePolicyBackfillsAgentStepLimitsForLegacyJSON(t *testing.T) { task := value["task"].(map[string]any) delete(task, "agentStepMaxOutputTokens") delete(task, "agentStepTimeoutSeconds") + delete(task, "agentMaxSessions") trimmed, err := json.Marshal(value) if err != nil { t.Fatal(err) @@ -176,6 +177,9 @@ func TestRuntimePolicyBackfillsAgentStepLimitsForLegacyJSON(t *testing.T) { if effective.Task.AgentStepMaxOutputTokens != platform.DefaultRuntimeAgentStepOutputTokens { t.Fatalf("legacy json 未回填单步输出上限: %d", effective.Task.AgentStepMaxOutputTokens) } + if effective.Task.AgentMaxSessions != platform.DefaultRuntimeAgentSessions { + t.Fatalf("legacy json 未回填同时对话上限: %d", effective.Task.AgentMaxSessions) + } if effective.Task.AgentStepTimeoutSeconds != platform.DefaultRuntimeAgentStepTimeout { t.Fatalf("legacy json 未回填单步超时: %d", effective.Task.AgentStepTimeoutSeconds) } diff --git a/backend/internal/app/service.go b/backend/internal/app/service.go index 404d5e5e3..20746a456 100644 --- a/backend/internal/app/service.go +++ b/backend/internal/app/service.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "log/slog" "strings" "sync" "time" @@ -166,6 +167,7 @@ func (s *Service) StartWorker() { s.startResourceDeletionWorker(ctx) s.startSkillSyncWorker(ctx) s.startPaymentWorker(ctx) + go s.syncAgentSessionLimit() s.runWorkerLoop(func(ctx context.Context) { ticker := time.NewTicker(3 * time.Second) defer ticker.Stop() @@ -180,6 +182,19 @@ func (s *Service) StartWorker() { }) } +func (s *Service) syncAgentSessionLimit() { + policy, err := s.RuntimePolicy() + if err != nil { + slog.Warn("agent session limit load failed", "error", err) + return + } + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + if err := s.applyAgentSessionLimit(ctx, policy.Task.AgentMaxSessions); err != nil { + slog.Warn("agent session limit apply failed", "error", err) + } +} + func (s *Service) BeginDrain() { s.backgroundWorkers().BeginDrain() } func (s *Service) StopWorker(ctx context.Context) error { return s.backgroundWorkers().Stop(ctx) } diff --git a/backend/internal/handler/admin_system_performance.go b/backend/internal/handler/admin_system_performance.go index 869ab6c6c..b4a2d4213 100644 --- a/backend/internal/handler/admin_system_performance.go +++ b/backend/internal/handler/admin_system_performance.go @@ -23,6 +23,26 @@ func RegisterAdminSystemPerformanceRoutes(r *gin.RouterGroup, svc *service.Servi ok(c, result) }) + r.PUT("/admin/system-performance/agent-limit", func(c *gin.Context) { + actor, err := currentUser(c, svc) + if err != nil { + failService(c, err) + return + } + c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, 4<<10) + var request service.AgentSessionLimitRequest + if err := c.ShouldBindJSON(&request); err != nil { + failService(c, service.BadAuthRequest("Agent 并发请求无效")) + return + } + result, err := svc.UpdateAgentSessionLimit(c.Request.Context(), actor, request.MaxSessions) + if err != nil { + failService(c, err) + return + } + ok(c, result) + }) + r.POST("/admin/system-performance/cache/clear", func(c *gin.Context) { actor, err := currentUser(c, svc) if err != nil { diff --git a/backend/internal/handler/admin_system_performance_test.go b/backend/internal/handler/admin_system_performance_test.go index fca44ed82..82254e5f7 100644 --- a/backend/internal/handler/admin_system_performance_test.go +++ b/backend/internal/handler/admin_system_performance_test.go @@ -14,6 +14,7 @@ func TestAdminSystemPerformanceRoutesAreRegistered(t *testing.T) { RegisterAdminSystemPerformanceRoutes(router.Group("/api"), &service.Service{}) wanted := map[string]bool{ "GET /api/admin/system-performance": false, + "PUT /api/admin/system-performance/agent-limit": false, "POST /api/admin/system-performance/cache/clear": false, } for _, route := range router.Routes() { diff --git a/backend/internal/platform/runtime_policy.go b/backend/internal/platform/runtime_policy.go index 20fbd591f..1d2af3750 100644 --- a/backend/internal/platform/runtime_policy.go +++ b/backend/internal/platform/runtime_policy.go @@ -36,6 +36,9 @@ const ( MinRuntimeAgentStepTimeoutSeconds = 30 MaxRuntimeAgentStepTimeoutSeconds = 3_600 DefaultRuntimeAgentStepTimeout = 0 + MinRuntimeAgentSessions = 1 + MaxRuntimeAgentSessions = 64 + DefaultRuntimeAgentSessions = 30 ) // agentStepMaxOutputTokensDefault 读取部署级覆盖并按同一套范围夹取；环境变量无法表达"不限制"（0）， @@ -90,6 +93,8 @@ type RuntimeTaskPolicy struct { AgentStepMaxOutputTokens int `json:"agentStepMaxOutputTokens"` // AgentStepTimeoutSeconds 是画布 Agent 单步模型调用的秒级墙钟；0 表示沿用文本任务超时。 AgentStepTimeoutSeconds int `json:"agentStepTimeoutSeconds"` + // AgentMaxSessions 是同时进行中的画布 Agent 轮次上限。审批等待不占用名额。 + AgentMaxSessions int `json:"agentMaxSessions"` } type RuntimeRequestPolicy struct { @@ -173,6 +178,7 @@ func DefaultRuntimePolicy() RuntimePolicySetting { DefaultTimeoutMinutes: 10, AgentStepMaxOutputTokens: agentStepMaxOutputTokensDefault(), AgentStepTimeoutSeconds: agentStepTimeoutSecondsDefault(), + AgentMaxSessions: DefaultRuntimeAgentSessions, }, Request: RuntimeRequestPolicy{ TaskCreatePerMinute: 30, @@ -213,6 +219,7 @@ func selfUseRuntimePolicy() RuntimePolicySetting { AudioTimeoutMinutes: maxRuntimeTimeoutMinutes, VideoTimeoutMinutes: maxRuntimeTimeoutMinutes, StoryboardTimeoutMinutes: maxRuntimeTimeoutMinutes, DefaultTimeoutMinutes: maxRuntimeTimeoutMinutes, AgentStepMaxOutputTokens: MaxRuntimeAgentStepOutputTokens, AgentStepTimeoutSeconds: MaxRuntimeAgentStepTimeoutSeconds, + AgentMaxSessions: MaxRuntimeAgentSessions, } value.Request = RuntimeRequestPolicy{ TaskCreatePerMinute: maxRuntimeRate, @@ -437,6 +444,9 @@ func validateRuntimePolicy(value RuntimePolicySetting) error { if task.AgentStepTimeoutSeconds != 0 && (task.AgentStepTimeoutSeconds < MinRuntimeAgentStepTimeoutSeconds || task.AgentStepTimeoutSeconds > MaxRuntimeAgentStepTimeoutSeconds) { return kernel.BadAuthRequest(fmt.Sprintf("Agent 单步超时必须是 0 或 %d-%d 秒的整数 (0 表示沿用文本任务超时)", MinRuntimeAgentStepTimeoutSeconds, MaxRuntimeAgentStepTimeoutSeconds)) } + if task.AgentMaxSessions < MinRuntimeAgentSessions || task.AgentMaxSessions > MaxRuntimeAgentSessions { + return kernel.BadAuthRequest(fmt.Sprintf("Agent 同时对话上限必须是 %d-%d 的整数", MinRuntimeAgentSessions, MaxRuntimeAgentSessions)) + } request := value.Request for label, item := range map[string]int{ "任务创建频控": request.TaskCreatePerMinute, diff --git a/backend/internal/service/aliases_types.go b/backend/internal/service/aliases_types.go index 8ac04891b..8828ebcfc 100644 --- a/backend/internal/service/aliases_types.go +++ b/backend/internal/service/aliases_types.go @@ -27,6 +27,7 @@ type ( AdminAuditPage = app.AdminAuditPage AdminCacheClearGroupResult = app.AdminCacheClearGroupResult AdminCacheClearRequest = app.AdminCacheClearRequest + AgentSessionLimitRequest = app.AgentSessionLimitRequest AdminCacheClearResult = app.AdminCacheClearResult AdminChannelModelFetchResult = app.AdminChannelModelFetchResult AdminChannelModelImportRequest = app.AdminChannelModelImportRequest diff --git a/docker-compose.deploy.yml b/docker-compose.deploy.yml index 3daac69c5..37047b4a4 100644 --- a/docker-compose.deploy.yml +++ b/docker-compose.deploy.yml @@ -42,7 +42,7 @@ services: image: ${CANVAS_YINGCE_AGENT_IMAGE:?请先配置 CANVAS_YINGCE_AGENT_IMAGE（必须包含镜像 digest）} environment: YINGCE_AGENT_TOKEN: ${YINGCE_AGENT_TOKEN:?请先配置 YINGCE_AGENT_TOKEN} - MAX_CONCURRENT_SESSIONS: ${YINGCE_AGENT_MAX_SESSIONS:-8} + MAX_CONCURRENT_SESSIONS: ${YINGCE_AGENT_MAX_SESSIONS:-30} NODE_MAX_OLD_SPACE_SIZE: ${YINGCE_AGENT_NODE_MEMORY_MB:-512} healthcheck: test: ["CMD", "wget", "-qO-", "http://127.0.0.1:8081/healthz"] diff --git a/web/src/pages/admin/settings/runtime-policy-settings-page.tsx b/web/src/pages/admin/settings/runtime-policy-settings-page.tsx index 69e61fadd..23d2aed95 100644 --- a/web/src/pages/admin/settings/runtime-policy-settings-page.tsx +++ b/web/src/pages/admin/settings/runtime-policy-settings-page.tsx @@ -77,6 +77,14 @@ const agentFields: PolicyField[] = [ min: 0, max: 3_600, }, + { + group: "task", + name: "agentMaxSessions", + label: "同时对话上限", + extra: "同时进行中的画布 Agent 轮次，新安装默认 30。审批等待不占名额，降低上限不会中断已开始的对话。", + unit: "个", + max: 64, + }, ]; const rateFields: PolicyField[] = [ diff --git a/web/src/pages/admin/settings/system-performance-page.tsx b/web/src/pages/admin/settings/system-performance-page.tsx index 49859cc81..89adbbef9 100644 --- a/web/src/pages/admin/settings/system-performance-page.tsx +++ b/web/src/pages/admin/settings/system-performance-page.tsx @@ -1,8 +1,8 @@ -import { App, Button, Progress, Skeleton, Switch } from "antd"; -import { Activity, AlertTriangle, Cpu, Database, Gauge, HardDrive, MemoryStick, RefreshCw, Server, ShieldCheck, Trash2, Wifi } from "lucide-react"; +import { App, Button, InputNumber, Progress, Skeleton, Switch } from "antd"; +import { Activity, AlertTriangle, Bot, Cpu, Database, Gauge, HardDrive, MemoryStick, RefreshCw, Server, ShieldCheck, Trash2, Wifi } from "lucide-react"; import { useCallback, useEffect, useRef, useState, type ReactNode } from "react"; -import { clearRuntimeCache, getSystemPerformance, type SystemPerformance } from "@/services/api/system-performance"; +import { clearRuntimeCache, getSystemPerformance, updateAgentSessionLimit, type SystemPerformance } from "@/services/api/system-performance"; import { AdminPageFrame } from "../components/admin-shell"; import { AdminStatusBadge, SettingsSectionCard } from "../components/admin-ui"; @@ -61,6 +61,8 @@ export default function SystemPerformancePage() { const [loading, setLoading] = useState(true); const [refreshing, setRefreshing] = useState(false); const [clearing, setClearing] = useState(false); + const [savingAgentLimit, setSavingAgentLimit] = useState(false); + const [agentLimitDraft, setAgentLimitDraft] = useState(null); const [autoRefresh, setAutoRefresh] = useState(true); const [loadError, setLoadError] = useState(""); const mountedRef = useRef(true); @@ -149,7 +151,7 @@ export default function SystemPerformancePage() { return ( } > @@ -213,6 +215,30 @@ export default function SystemPerformancePage() { + {(data.agents || []).map((agent) => { + const draft = agentLimitDraft ?? agent.configuredLimit; + return } title="画布 Agent 服务" description="同时进行中的对话占用一个名额。审批等待不占用。降低上限不会中断已经开始的对话，也不会在这台机器上再启动容器。" status={{ label: agent.healthy ? "服务可连接" : "服务不可达", color: agent.healthy ? "success" : "error" }} footer={<>新安装默认 30，一般不用再改 setAgentLimitDraft(typeof value === "number" ? value : draft)} />}> + + + + + + + + + + ; + })} + } title="缓存维护" description="只清理可安全重建的运行时状态，保留业务数据与活动并发租约。" status={} footer={<>清理后频控与线路健康状态会重新计算}>
{(data?.redis.cacheGroups || []).map((group) =>
{group.label}{formatNumber(group.keys)}keys
)} diff --git a/web/src/services/api/auth.ts b/web/src/services/api/auth.ts index 296be5c29..a900c2fe7 100644 --- a/web/src/services/api/auth.ts +++ b/web/src/services/api/auth.ts @@ -350,6 +350,8 @@ export type RuntimeTaskPolicy = { agentStepMaxOutputTokens: number; /** 画布 Agent 单步模型调用的秒级墙钟；0 表示沿用文本任务超时。 */ agentStepTimeoutSeconds: number; + /** 同时进行中的画布 Agent 轮次上限。审批等待不占用名额。 */ + agentMaxSessions: number; }; export type RuntimeRequestPolicy = { diff --git a/web/src/services/api/system-performance.ts b/web/src/services/api/system-performance.ts index f057dc873..eb9c166a1 100644 --- a/web/src/services/api/system-performance.ts +++ b/web/src/services/api/system-performance.ts @@ -105,9 +105,23 @@ export type SystemPerformance = { cacheGroups: SystemPerformanceCacheGroup[]; statusMessage?: string; }; + agents: SystemPerformanceAgent[]; build: { version: string; commit: string; buildTime: string; goVersion: string }; }; +export type SystemPerformanceAgent = { + id: string; + name: string; + mode: "remote" | "embedded"; + endpoint: string; + healthy: boolean; + active: number; + limit: number; + queued: number; + configuredLimit: number; + statusMessage?: string; +}; + export type RuntimeCacheClearResult = { scope: "runtime"; deletedKeys: number; @@ -120,6 +134,10 @@ export function getSystemPerformance(signal?: AbortSignal) { return http.get("/admin/system-performance", { signal }); } +export function updateAgentSessionLimit(maxSessions: number) { + return http.put("/admin/system-performance/agent-limit", { maxSessions }); +} + export function clearRuntimeCache() { return http.post("/admin/system-performance/cache/clear", { scope: "runtime" }); } diff --git a/yingce-agent/server.mjs b/yingce-agent/server.mjs index c64794e34..f60d250b3 100644 --- a/yingce-agent/server.mjs +++ b/yingce-agent/server.mjs @@ -4,7 +4,7 @@ import { createServer } from "node:http"; import { dirname } from "node:path"; const token = process.env.YINGCE_AGENT_TOKEN || ""; -const maxSessions = positiveInt(process.env.MAX_CONCURRENT_SESSIONS, 8, 1000); +let maxSessions = positiveInt(process.env.MAX_CONCURRENT_SESSIONS, 30, 64); const memoryMB = positiveInt(process.env.NODE_MAX_OLD_SPACE_SIZE, 512, 8192); const runtimePath = process.env.YINGCE_AGENT_RUNTIME || "/app/agent-runtime/agent-runtime.mjs"; const port = Number(process.env.PORT || 8081); @@ -48,13 +48,21 @@ function acquire(request) { }); } +function admitWaiters() { + while (waiters.length > 0 && active < maxSessions) { + active += 1; + waiters.shift().resolve(); + } +} + function release() { - const waiter = waiters.shift(); - if (waiter) { - waiter.resolve(); + if (waiters.length > 0 && active <= maxSessions) { + waiters.shift().resolve(); + admitWaiters(); return; } active = Math.max(0, active - 1); + admitWaiters(); } function readBody(request) { @@ -85,7 +93,27 @@ function writeRuntimeError(response, message) { const server = createServer(async (request, response) => { if (request.method === "GET" && request.url === "/healthz") { response.writeHead(200, { "content-type": "application/json" }); - response.end(JSON.stringify({ ok: true, active, limit: maxSessions })); + response.end(JSON.stringify({ ok: true, active, limit: maxSessions, queued: waiters.length })); + return; + } + if (request.method === "PUT" && request.url === "/v1/limit") { + if (!authorized(request)) { + response.writeHead(401, { "content-type": "application/json" }); + response.end(JSON.stringify({ error: "unauthorized" })); + return; + } + try { + const body = JSON.parse((await readBody(request)).toString("utf8")); + const next = positiveInt(body.maxSessions, 0, 64); + if (!next) throw new Error("maxSessions must be an integer from 1 to 64"); + maxSessions = next; + admitWaiters(); + response.writeHead(200, { "content-type": "application/json" }); + response.end(JSON.stringify({ ok: true, active, limit: maxSessions, queued: waiters.length })); + } catch (error) { + response.writeHead(400, { "content-type": "application/json" }); + response.end(JSON.stringify({ error: error instanceof Error ? error.message : "invalid limit" })); + } return; } if (request.method !== "POST" || request.url !== "/v1/runs") {