diff --git a/.gitignore b/.gitignore index c9a7436..96def82 100644 --- a/.gitignore +++ b/.gitignore @@ -60,3 +60,4 @@ Thumbs.db /bench/ /Dockerfile.bench /Dockerfile.bench.v2 +.plan/ diff --git a/server/engine.go b/server/engine.go index b5643b2..e99cf17 100644 --- a/server/engine.go +++ b/server/engine.go @@ -148,6 +148,11 @@ type Engine struct { // (for example an exhausted failover chain) — there is no global-pair fallback. resolve func(t *Task) (*agent.Planner, *agent.Worker) resolveAuthoritative bool + // resolveWorker returns the Work Agent for one spawn slot. profileID 0 → unbound + // task-router worker (legacy); non-zero → worker pinned to that LLM profile. + // For profileID 0, a nil result falls back to snapshotFor; for pinned slots, + // nil means wait/retry (never switch to another model). + resolveWorker func(t *Task, profileID int64) *agent.Worker // readiness reports whether a global LLM provider is configured — the signal behind // Ready()/the llm_configured indicator. Wired once at startup; nil → not ready. readiness func() bool @@ -715,6 +720,12 @@ func (e *Engine) SetAuthoritativeAgentResolver(fn func(t *Task) (*agent.Planner, e.resolveAuthoritative = true } +// SetWorkerResolver installs the per-slot worker factory used by workerLoop when +// multi-model pools assign a profile to a session (work#N). Called once at startup. +func (e *Engine) SetWorkerResolver(fn func(t *Task, profileID int64) *agent.Worker) { + e.resolveWorker = fn +} + // snapshotFor returns the planner/worker a task should run on, from the task-router // resolver. nil,nil means the task is deliberately unavailable (e.g. an exhausted // failover chain); there is no global-pair fallback. @@ -746,7 +757,7 @@ func (e *Engine) ReadyFor(t *Task) bool { // but no-op until an LLM is configured (so a task created while idle picks up // automatically once LLM is set from the UI). func (e *Engine) Run(ctx context.Context, t *Task) { - workers := e.m.Workers() + slots := e.m.WorkerSlots() e.deleteMu.RLock() if e.IsDeleting(t.ID) { e.deleteMu.RUnlock() @@ -757,13 +768,13 @@ func (e *Engine) Run(ctx context.Context, t *Task) { t.Notify() // already running — just nudge a planning round return } - rt := e.registerTaskRoutines(ctx, t.ID, 1+workers) + rt := e.registerTaskRoutines(ctx, t.ID, 1+len(slots)) e.deleteMu.RUnlock() e.touch(t.ID) runTaskRoutine(rt, func(loopCtx context.Context) { e.plannerLoop(loopCtx, t) }) - for i := 0; i < workers; i++ { - name := fmt.Sprintf("work#%d", i+1) - runTaskRoutine(rt, func(loopCtx context.Context) { e.workerLoop(loopCtx, t, name) }) + for _, slot := range slots { + name, profileID := slot.Name, slot.ProfileID + runTaskRoutine(rt, func(loopCtx context.Context) { e.workerLoop(loopCtx, t, name, profileID) }) } e.startDeadlineCoordinator(ctx, t) // 任务级超时定时器(仅 timeout>0;去重) // 仅在「完全没有活动意图(open+running)」时才 kick 首轮规划。带种子意图的任务:种子已 @@ -913,14 +924,23 @@ func (e *Engine) plannerLoop(ctx context.Context, t *Task) { } } -func (e *Engine) workerLoop(ctx context.Context, t *Task, name string) { +func (e *Engine) workerLoop(ctx context.Context, t *Task, name string, profileID int64) { for { select { case <-ctx.Done(): return default: } - _, worker := e.snapshotFor(t) + var worker *agent.Worker + if e.resolveWorker != nil { + worker = e.resolveWorker(t, profileID) + } + // Legacy unbound slots (profileID 0) may fall back to the task-router + // snapshot. Pinned multi-model pool slots must NOT silently switch models + // when the assigned profile is missing/unavailable — wait and retry. + if worker == nil && profileID == 0 { + _, worker = e.snapshotFor(t) + } if worker == nil { if sleepCtx(ctx, 1500*time.Millisecond) { return diff --git a/server/server.go b/server/server.go index 969216c..ec51b8a 100644 --- a/server/server.go +++ b/server/server.go @@ -177,6 +177,11 @@ func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDi b := s.agentsForTask(t) return b.pl, b.wk }) + // Multi-model Worker pools (#41): each spawn slot may pin a Work Agent to a + // specific LLM profile. profileID 0 keeps the unbound task-router worker. + s.engine.SetWorkerResolver(func(t *Task, profileID int64) *agent.Worker { + return s.workerForTask(t, profileID) + }) // Global readiness (Ready()/llm_configured): a global active LLM provider is // installed. Task-level runnability is separate (ReadyFor → the resolver above). s.engine.SetReadiness(func() bool { @@ -3432,6 +3437,7 @@ func (s *Server) settingsPayload() map[string]any { "global_proxy": s.m.GlobalProxy(), // 全局出口代理(http/https/socks5)，所有目标流量走它，空=直连 "python_interpreter": strings.TrimSpace(pyStored), // 用户/自动设的值(空=用运行时检测) "workers": s.m.Workers(), // 并发工作 agent 数(默认3)；对之后启动的任务生效 + "worker_pools": workerPoolsPayload(s.m), // 多模型 Worker 池；空=走 workers 遗留模式 "task_concurrency_enabled": concOn, // 任务并发上限开关(默认关) "task_concurrency_limit": concLimit, // 同时运行任务上限(开启后默认5) // LLM 轮询(故障转移)。默认关；开启后走全局激活配置的 agent 在当前配置不可用时 @@ -3452,6 +3458,18 @@ func (s *Server) settingsPayload() map[string]any { } } + +// workerPoolsPayload always returns a non-nil slice for JSON (empty = legacy mode). +func workerPoolsPayload(m *Manager) []WorkerPoolConfig { + if m == nil { + return []WorkerPoolConfig{} + } + if pools := m.WorkerPools(); pools != nil { + return pools + } + return []WorkerPoolConfig{} +} + // notifyPublicBaseURL 读推送回链用的外部地址。 func notifyPublicBaseURL(pg *db.DB) string { v, _, _ := pg.GetSetting(settingNotifyPublicBaseURL) @@ -3506,7 +3524,8 @@ func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) { WebSearchProxy *string `json:"web_search_proxy"` // 独立出口代理(http/https/socks5)；null=不改，""=清空 GlobalProxy *string `json:"global_proxy"` // 全局出口代理(http/https/socks5)；null=不改，""=清空(直连) PythonInterp *string `json:"python_interpreter"` // 自定义脚本工具的 python 解释器路径 - Workers *int `json:"workers"` // 并发工作 agent 数(>0)；对之后启动的任务生效 + Workers *int `json:"workers"` // 并发工作 agent 数(>0)；对之后启动的任务生效 + WorkerPools *[]WorkerPoolConfig `json:"worker_pools"` // 多模型池；null=不改，[]=清空回遗留模式 // 任务并发上限:同时「运行中」的任务数上限。关闭=不限;开启后新建任务超限则排队,有空位自动启动。 ConcurrencyEnabled *bool `json:"task_concurrency_enabled"` ConcurrencyLimit *int `json:"task_concurrency_limit"` @@ -3585,6 +3604,22 @@ func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) { return } } + if req.WorkerPools != nil { + if err := s.m.SetWorkerPools(*req.WorkerPools); err != nil { + writeErr(w, 400, err.Error()) + return + } + // 有池配置时把 workers 同步成合计,方便 UI/遗留读数与真实并发一致。 + if pools := *req.WorkerPools; len(pools) > 0 { + total := 0 + for _, p := range pools { + total += p.Workers + } + if total > 0 { + _ = s.m.SetWorkers(total) + } + } + } if req.ConcurrencyEnabled != nil || req.ConcurrencyLimit != nil { // 部分 PUT:未给的字段用当前值兜底,避免只改一个把另一个重置。 curOn, curLimit := s.m.ConcurrencyLimit() diff --git a/server/task_llm.go b/server/task_llm.go index 2cae9ce..48a086f 100644 --- a/server/task_llm.go +++ b/server/task_llm.go @@ -27,6 +27,9 @@ type taskAgentBundle struct { pl *agent.Planner wk *agent.Worker main *agent.MainAgent + // workersByProfile caches Work Agents pinned to a specific LLM profile for + // multi-model pools (issue #41). Key = llm_profiles.id. + workersByProfile map[int64]*agent.Worker } type llmAuditProfile struct { @@ -51,6 +54,11 @@ type taskLLMRuntime struct { s *Server taskID string agentKey string + // pinProfileID, when non-zero, forces every LLM call onto that profile and + // skips agent-binding / task-chain / global resolution. Used by multi-model + // Worker pools so each session (work#N) stays on its assigned model. Quota + // errors do NOT advance the task failover chain (sel.profileID stays 0). + pinProfileID int64 } type taskLLMError struct { @@ -122,6 +130,15 @@ func (r *taskLLMRuntime) current() (taskLLMSelection, error) { r.s.syncTaskLLMState(pt) t, _ := r.s.m.Task(r.taskID) sel := taskLLMSelection{task: t, revision: pt.LLMChainRevision} + if r.pinProfileID != 0 { + prov, cfg, ok := r.s.providerForProfile(r.pinProfileID) + if !ok { + return sel, fmt.Errorf("pinned LLM profile #%d is missing or invalid", r.pinProfileID) + } + // profileID left 0 so streamTaskLLM/completeTaskLLM never mutate the task chain. + sel.provider, sel.retry = r.s.poolForBinding(r.pinProfileID, prov, cfg), cfg.Retry + return sel, nil + } if prov, cfg, ok := r.s.agentBindingProvider(r.agentKey); ok { sel.provider, sel.retry = prov, cfg.Retry return sel, nil @@ -152,6 +169,12 @@ func (r *taskLLMRuntime) current() (taskLLMSelection, error) { // the caller. If failover switches profiles, the change takes effect on the next // agent run (a fresh Session is built per run in captureRun). func (r *taskLLMRuntime) activeCfg() (agent.Config, bool) { + if r.pinProfileID != 0 { + if _, cfg, ok := r.s.providerForProfile(r.pinProfileID); ok { + return cfg, true + } + return agent.Config{}, false + } taskNum, err := parseTaskID(r.taskID) if err != nil { return agent.Config{}, false @@ -454,6 +477,12 @@ func isRetryableStreamError(err error) bool { // CompactionWindow mirrors current()'s precedence so the context window always // matches the provider the role will actually stream on. func (r *taskLLMRuntime) CompactionWindow() int { + if r.pinProfileID != 0 { + if _, cfg, ok := r.s.providerForProfile(r.pinProfileID); ok { + return cfg.CompactionWindow() + } + return (agent.Config{}).CompactionWindow() + } if _, cfg, ok := r.s.agentBindingProvider(r.agentKey); ok { return cfg.CompactionWindow() } @@ -560,17 +589,7 @@ func (s *Server) agentsForTask(t *Task) *taskAgentBundle { workerRuntime := &taskLLMRuntime{s: s, taskID: t.ID, agentKey: "worker"} mainRuntime := &taskLLMRuntime{s: s, taskID: t.ID, agentKey: "mainagent"} tx := transcript.NewStore(filepath.Join(s.m.dir, "transcripts")) - window := workerRuntime.CompactionWindow() - wk := agent.NewWorker(workerRuntime, "task-router", s.m.dir, tx, window, s.agentMaxTurns("worker")) - wk.SetFindingRecorder(s.evidenceStore()) - wk.SetCompactionWindowResolver(workerRuntime.CompactionWindow) - wk.SetNonStreaming(workerRuntime.nonStreaming) // 按任务当前激活 profile 的流式开关(每轮读) - wk.SetMaxTokens(workerRuntime.maxTokens) // 同上,输出上限也跟随当前激活 profile - wk.SetNoaEnabled(s.m.NoaCompactionEnabled) // 实验功能:noa 上下文压缩(平台级开关,每 run 读) - wk.SetRunTimeout(time.Duration(s.agentRunSeconds("worker")) * time.Second) - wk.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) - wk.SetWebSearch(s.webSearchFor("worker")) - wk.SetConstraintInject(s.constraintInjectWorker) // 操作约束注入 worker(可配置,默认开) + wk := s.newTaskWorker(workerRuntime, tx) pl := agent.NewPlanner(plannerRuntime, "task-router", s.m.dir, tx, plannerRuntime.CompactionWindow(), s.agentMaxTurns("planner")) pl.SetFindingRecorder(s.evidenceStore()) pl.SetCompactionWindowResolver(plannerRuntime.CompactionWindow) @@ -598,11 +617,61 @@ func (s *Server) agentsForTask(t *Task) *taskAgentBundle { bundle := &taskAgentBundle{ runtime: goalRuntime, plannerRuntime: plannerRuntime, workerRuntime: workerRuntime, mainRuntime: mainRuntime, pl: pl, wk: wk, main: main, + workersByProfile: map[int64]*agent.Worker{}, } s.taskAgents[t.ID] = bundle return bundle } + +// newTaskWorker builds a Work Agent on the given runtime (shared wiring for the +// unbound task worker and per-profile pinned workers). +func (s *Server) newTaskWorker(runtime *taskLLMRuntime, tx *transcript.Store) *agent.Worker { + wk := agent.NewWorker(runtime, "task-router", s.m.dir, tx, runtime.CompactionWindow(), s.agentMaxTurns("worker")) + wk.SetFindingRecorder(s.evidenceStore()) + wk.SetCompactionWindowResolver(runtime.CompactionWindow) + wk.SetNonStreaming(runtime.nonStreaming) + wk.SetMaxTokens(runtime.maxTokens) + wk.SetNoaEnabled(s.m.NoaCompactionEnabled) + wk.SetRunTimeout(time.Duration(s.agentRunSeconds("worker")) * time.Second) + wk.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) + wk.SetWebSearch(s.webSearchFor("worker")) + wk.SetConstraintInject(s.constraintInjectWorker) + return wk +} + +// workerForTask returns the Work Agent for one Engine spawn slot. +// profileID 0 → unbound task-router worker (requires taskRuntimeAvailable). +// Non-zero → worker pinned to that LLM profile (must build; ignores task-chain exhaustion). +func (s *Server) workerForTask(t *Task, profileID int64) *agent.Worker { + if t == nil { + return nil + } + if profileID == 0 { + if !s.taskRuntimeAvailable(t, "worker") { + return nil + } + return s.agentsForTask(t).wk + } + if _, _, ok := s.providerForProfile(profileID); !ok { + return nil + } + bundle := s.agentsForTask(t) + s.taskAgentMu.Lock() + defer s.taskAgentMu.Unlock() + if bundle.workersByProfile == nil { + bundle.workersByProfile = map[int64]*agent.Worker{} + } + if w := bundle.workersByProfile[profileID]; w != nil { + return w + } + tx := transcript.NewStore(filepath.Join(s.m.dir, "transcripts")) + runtime := &taskLLMRuntime{s: s, taskID: t.ID, agentKey: "worker", pinProfileID: profileID} + w := s.newTaskWorker(runtime, tx) + bundle.workersByProfile[profileID] = w + return w +} + func (s *Server) syncTaskLLMState(pt *db.Task) { if pt == nil { return diff --git a/server/worker_pools.go b/server/worker_pools.go new file mode 100644 index 0000000..b66c691 --- /dev/null +++ b/server/worker_pools.go @@ -0,0 +1,127 @@ +package server + +import ( + "encoding/json" + "fmt" + "strings" +) + +// settingWorkerPools stores the multi-model Work Agent pool config as a JSON array. +// Empty / unset → legacy single-pool mode using settings.workers + unbound runtime. +const settingWorkerPools = "worker_pools" + +// maxWorkerPoolTotal is the hard upper bound on sum(pool.workers), matching the +// per-task override ceiling used elsewhere so a misconfigured UI can't spawn dozens +// of LLM sessions per task. +const maxWorkerPoolTotal = 50 + +// WorkerPoolConfig is one LLM profile's share of concurrent Work Agents. +type WorkerPoolConfig struct { + ProfileID int64 `json:"profile_id"` + Workers int `json:"workers"` +} + +// WorkerSlot is one Engine.Run worker loop: Name is the claim/session id (work#N); +// ProfileID 0 means unbound (legacy task-router resolution). +type WorkerSlot struct { + Name string + ProfileID int64 +} + +// validateWorkerPools checks a pools slice before persistence. Empty is allowed +// (clears multi-model mode). Non-empty requires every entry valid and total in 1..50. +func validateWorkerPools(pools []WorkerPoolConfig) error { + if len(pools) == 0 { + return nil + } + total := 0 + for i, p := range pools { + if p.ProfileID <= 0 { + return fmt.Errorf("worker_pools[%d].profile_id 必须 >0", i) + } + if p.Workers <= 0 { + return fmt.Errorf("worker_pools[%d].workers 必须 >0", i) + } + total += p.Workers + } + if total > maxWorkerPoolTotal { + return fmt.Errorf("worker_pools 合计 Worker 数最多 %d（当前 %d）", maxWorkerPoolTotal, total) + } + return nil +} + +// expandWorkerSlots turns pool configs into a spawn plan. Empty pools → legacy: +// N unbound slots named work#1..N where N = legacyWorkers (clamped to >=1). +func expandWorkerSlots(pools []WorkerPoolConfig, legacyWorkers int) []WorkerSlot { + if len(pools) == 0 { + n := legacyWorkers + if n <= 0 { + n = defaultWorkers + } + out := make([]WorkerSlot, n) + for i := 0; i < n; i++ { + out[i] = WorkerSlot{Name: fmt.Sprintf("work#%d", i+1), ProfileID: 0} + } + return out + } + var out []WorkerSlot + for _, p := range pools { + if p.ProfileID <= 0 || p.Workers <= 0 { + continue + } + for i := 0; i < p.Workers; i++ { + out = append(out, WorkerSlot{ + Name: fmt.Sprintf("work#%d", len(out)+1), + ProfileID: p.ProfileID, + }) + } + } + if len(out) == 0 { + return expandWorkerSlots(nil, legacyWorkers) + } + return out +} + +// WorkerPools returns the configured multi-model pools (nil/empty = legacy mode). +// Invalid entries in the stored JSON are skipped. +func (m *Manager) WorkerPools() []WorkerPoolConfig { + if m == nil || m.pg == nil { + return nil + } + v, ok, err := m.pg.GetSetting(settingWorkerPools) + if err != nil || !ok || strings.TrimSpace(v) == "" { + return nil + } + var raw []WorkerPoolConfig + if json.Unmarshal([]byte(v), &raw) != nil { + return nil + } + out := make([]WorkerPoolConfig, 0, len(raw)) + for _, p := range raw { + if p.ProfileID > 0 && p.Workers > 0 { + out = append(out, p) + } + } + return out +} + +// SetWorkerPools persists the multi-model pool config. An empty slice clears the +// setting (back to legacy single-pool Workers()). Applies to tasks started afterwards. +func (m *Manager) SetWorkerPools(pools []WorkerPoolConfig) error { + if err := validateWorkerPools(pools); err != nil { + return err + } + if len(pools) == 0 { + return m.pg.SetSetting(settingWorkerPools, "") + } + b, err := json.Marshal(pools) + if err != nil { + return err + } + return m.pg.SetSetting(settingWorkerPools, string(b)) +} + +// WorkerSlots returns the expanded spawn plan for Engine.Run (read per task start). +func (m *Manager) WorkerSlots() []WorkerSlot { + return expandWorkerSlots(m.WorkerPools(), m.Workers()) +} diff --git a/server/worker_pools_test.go b/server/worker_pools_test.go new file mode 100644 index 0000000..78ca527 --- /dev/null +++ b/server/worker_pools_test.go @@ -0,0 +1,86 @@ +package server + +import ( + "testing" +) + +func TestExpandWorkerSlotsLegacy(t *testing.T) { + slots := expandWorkerSlots(nil, 3) + if len(slots) != 3 { + t.Fatalf("len=%d want 3", len(slots)) + } + for i, s := range slots { + want := "work#" + itoa(i+1) + if s.Name != want || s.ProfileID != 0 { + t.Fatalf("slot[%d]=%+v want name=%s profile=0", i, s, want) + } + } + slots = expandWorkerSlots(nil, 0) + if len(slots) != defaultWorkers { + t.Fatalf("legacy 0 → defaultWorkers: got %d", len(slots)) + } +} + +func TestExpandWorkerSlotsMulti(t *testing.T) { + slots := expandWorkerSlots([]WorkerPoolConfig{ + {ProfileID: 1, Workers: 2}, + {ProfileID: 2, Workers: 3}, + }, 3) + if len(slots) != 5 { + t.Fatalf("len=%d want 5", len(slots)) + } + want := []int64{1, 1, 2, 2, 2} + for i, s := range slots { + if s.ProfileID != want[i] || s.Name != "work#"+itoa(i+1) { + t.Fatalf("slot[%d]=%+v want profile=%d", i, s, want[i]) + } + } +} + +func TestExpandWorkerSlotsSkipsInvalidFallsBack(t *testing.T) { + slots := expandWorkerSlots([]WorkerPoolConfig{ + {ProfileID: 0, Workers: 2}, + {ProfileID: 1, Workers: 0}, + }, 4) + if len(slots) != 4 || slots[0].ProfileID != 0 { + t.Fatalf("all-invalid pools should fall back to legacy: %+v", slots) + } +} + +func TestValidateWorkerPools(t *testing.T) { + if err := validateWorkerPools(nil); err != nil { + t.Fatal(err) + } + if err := validateWorkerPools([]WorkerPoolConfig{}); err != nil { + t.Fatal(err) + } + if err := validateWorkerPools([]WorkerPoolConfig{{ProfileID: 0, Workers: 1}}); err == nil { + t.Fatal("expected profile_id error") + } + if err := validateWorkerPools([]WorkerPoolConfig{{ProfileID: 1, Workers: 0}}); err == nil { + t.Fatal("expected workers error") + } + big := make([]WorkerPoolConfig, 0, 3) + big = append(big, WorkerPoolConfig{ProfileID: 1, Workers: 40}) + big = append(big, WorkerPoolConfig{ProfileID: 2, Workers: 20}) + if err := validateWorkerPools(big); err == nil { + t.Fatal("expected sum>50 error") + } + if err := validateWorkerPools([]WorkerPoolConfig{{ProfileID: 1, Workers: 3}, {ProfileID: 2, Workers: 3}}); err != nil { + t.Fatal(err) + } +} + +func itoa(n int) string { + if n == 0 { + return "0" + } + var b [16]byte + i := len(b) + for n > 0 { + i-- + b[i] = byte('0' + n%10) + n /= 10 + } + return string(b[i:]) +} diff --git a/web/src/app/(main)/system/settings/page.tsx b/web/src/app/(main)/system/settings/page.tsx index fea869f..50420b8 100644 --- a/web/src/app/(main)/system/settings/page.tsx +++ b/web/src/app/(main)/system/settings/page.tsx @@ -2,7 +2,7 @@ import * as React from "react"; -import { CpuIcon, FlaskConicalIcon, KeyboardIcon, RadioTowerIcon, SearchIcon, ShieldAlertIcon } from "lucide-react"; +import { CpuIcon, FlaskConicalIcon, KeyboardIcon, PlusIcon, RadioTowerIcon, SearchIcon, ShieldAlertIcon, Trash2Icon } from "lucide-react"; import { toast } from "sonner"; import { Button } from "@/components/ui/button"; @@ -14,7 +14,7 @@ import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@ import { Switch } from "@/components/ui/switch"; import { api } from "@/lib/api"; import { CHAT_SEND_MODE_OPTIONS, type ChatSendMode, setChatSendMode, useChatSendMode } from "@/lib/chat-send-mode"; -import type { Settings } from "@/lib/types"; +import type { LLMProfile, Settings } from "@/lib/types"; import { BasicAuthCard } from "./_components/basic-auth-card"; import { UpdateCard } from "./_components/update-card"; @@ -45,6 +45,8 @@ export default function SystemSettingsPage() { const [pyInterp, setPyInterp] = React.useState(""); const [workers, setWorkers] = React.useState("3"); const [savingWorkers, setSavingWorkers] = React.useState(false); + const [llmProfiles, setLlmProfiles] = React.useState([]); + const [workerPools, setWorkerPools] = React.useState<{ profile_id: string; workers: string }[]>([]); // 操作约束注入范围(默认都开)。 const [injectPlanner, setInjectPlanner] = React.useState(true); const [injectWorker, setInjectWorker] = React.useState(true); @@ -67,23 +69,53 @@ export default function SystemSettingsPage() { setGlobalProxyInput(s.global_proxy ?? ""); setPyInterp(s.python_interpreter ?? ""); setWorkers(String(s.workers ?? 3)); + setWorkerPools( + (s.worker_pools ?? []).map((p) => ({ + profile_id: String(p.profile_id), + workers: String(p.workers), + })), + ); setInjectPlanner(s.constraints_inject_planner !== false); setInjectWorker(s.constraints_inject_worker !== false); setNoaCompaction(!!s.noa_compaction); }, []); const saveWorkers = () => { - const n = Number(workers); - if (!Number.isInteger(n) || n <= 0) { - toast.error("并发数必须是大于 0 的整数"); + const pools: { profile_id: number; workers: number }[] = []; + for (let i = 0; i < workerPools.length; i++) { + const row = workerPools[i]; + const pid = Number(row.profile_id); + const n = Number(row.workers); + if (!Number.isInteger(pid) || pid <= 0) { + toast.error(`第 ${i + 1} 行：请选择 LLM 配置`); + return; + } + if (!Number.isInteger(n) || n <= 0) { + toast.error(`第 ${i + 1} 行：Worker 数必须是大于 0 的整数`); + return; + } + pools.push({ profile_id: pid, workers: n }); + } + const legacy = Number(workers); + if (pools.length === 0 && (!Number.isInteger(legacy) || legacy <= 0)) { + toast.error("默认并发数必须是大于 0 的整数"); + return; + } + const total = pools.length > 0 ? pools.reduce((s, p) => s + p.workers, 0) : legacy; + if (total > 50) { + toast.error("Worker 合计最多 50"); return; } setSavingWorkers(true); api - .setSettings({ workers: n }) + .setSettings({ workers: total, worker_pools: pools }) .then((s) => { apply(s); - toast.success("已保存并发工作 agent 数（对之后启动的任务生效）"); + toast.success( + pools.length > 0 + ? `已保存多模型 Worker 池（共 ${total} 个会话；对之后启动的任务生效）` + : "已保存并发工作 agent 数（对之后启动的任务生效）", + ); }) .catch((e) => toast.error("保存失败：" + (e as Error).message)) .finally(() => setSavingWorkers(false)); @@ -110,9 +142,11 @@ export default function SystemSettingsPage() { }; React.useEffect(() => { - api - .settings() - .then(apply) + Promise.all([api.settings(), api.llmProfiles().catch(() => [] as LLMProfile[])]) + .then(([s, profiles]) => { + apply(s); + setLlmProfiles(profiles); + }) .catch(() => undefined) .finally(() => setLoaded(true)); }, [apply]); @@ -748,24 +782,114 @@ export default function SystemSettingsPage() { - 工作并发 · Work Agent 数 + 工作并发 · Work Agent / 多模型池 - 每个任务并发运行的工作 agent 数量（默认 3）。数值越大并发探测越多、消耗也越高。修改后 + 可为多个 LLM 配置各分配若干 Worker；同一任务内多模型并行执行，按会话（work#N）绑定模型。 + 未配置模型池时，使用下方「默认并发」走单一模型（与历史行为一致）。修改后 对之后启动的任务生效，正在运行的任务不受影响。 - -
+ +
+ 0} onChange={(e) => setWorkers(e.target.value)} /> + {workerPools.length > 0 ? ( +

已配置模型池时以各池合计为准，默认并发输入已锁定。

+ ) : null} +
+ +
+
+ + +
+ {llmProfiles.length === 0 ? ( +

尚未配置 LLM：请先到「系统 → LLM 配置」添加并激活。

+ ) : null} + {workerPools.map((row, idx) => ( +
+ + + setWorkerPools((rows) => + rows.map((r, i) => (i === idx ? { ...r, workers: e.target.value } : r)), + ) + } + /> + Workers + +
+ ))} +

+ 当前合计： + {workerPools.length > 0 + ? workerPools.reduce((s, r) => s + (Number(r.workers) || 0), 0) + : Number(workers) || 0}{" "} + 个 Worker 会话 +

+
+ +
diff --git a/web/src/lib/mock/data.ts b/web/src/lib/mock/data.ts index 531df7b..9b8a69c 100644 --- a/web/src/lib/mock/data.ts +++ b/web/src/lib/mock/data.ts @@ -3338,6 +3338,7 @@ export const settings: Settings = { global_proxy: "", python_interpreter: "/usr/bin/python3", workers: 3, + worker_pools: [], llm_pool_enabled: true, llm_pool_bind_fallback: false, noa_compaction: false, diff --git a/web/src/lib/types.ts b/web/src/lib/types.ts index 54e26a6..5a7188d 100644 --- a/web/src/lib/types.ts +++ b/web/src/lib/types.ts @@ -968,6 +968,8 @@ export interface Settings { global_proxy?: string; python_interpreter?: string; // 自定义脚本工具的 python 解释器路径(空=运行时检测) workers?: number; // 并发工作 agent 数(默认3)；对之后启动的任务生效 + // 多模型 Worker 池：每个 LLM 配置各自占用若干并发会话。空/缺省=走上面的 workers 遗留模式。 + worker_pools?: { profile_id: number; workers: number }[]; // 任务并发上限:同时「运行中」的任务数上限。关闭=不限;开启后新建任务超限则排队,有空位自动启动。 task_concurrency_enabled?: boolean; // 默认 false task_concurrency_limit?: number; // 开启后默认 5