mirror of
https://github.com/RuoJi6/ARTEX.git
synced 2026-10-27 17:03:49 +08:00
chore: reset #41 multi-model worker files to origin/main for Codex rewrite
This commit is contained in:
1 parent
6f09b313d6
commit
6ffe04cf3d
9 files changed
+34
-499
No files matched your search
@@ -60,4 +60,3 @@ Thumbs.db
|
||||
/bench/
|
||||
/Dockerfile.bench
|
||||
/Dockerfile.bench.v2
|
||||
.plan/
|
||||
+7
-27
@@ -148,11 +148,6 @@ 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
|
||||
@@ -720,12 +715,6 @@ 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.
|
||||
@@ -757,7 +746,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) {
|
||||
slots := e.m.WorkerSlots()
|
||||
workers := e.m.Workers()
|
||||
e.deleteMu.RLock()
|
||||
if e.IsDeleting(t.ID) {
|
||||
e.deleteMu.RUnlock()
|
||||
@@ -768,13 +757,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+len(slots))
|
||||
rt := e.registerTaskRoutines(ctx, t.ID, 1+workers)
|
||||
e.deleteMu.RUnlock()
|
||||
e.touch(t.ID)
|
||||
runTaskRoutine(rt, func(loopCtx context.Context) { e.plannerLoop(loopCtx, t) })
|
||||
for _, slot := range slots {
|
||||
name, profileID := slot.Name, slot.ProfileID
|
||||
runTaskRoutine(rt, func(loopCtx context.Context) { e.workerLoop(loopCtx, t, name, profileID) })
|
||||
for i := 0; i < workers; i++ {
|
||||
name := fmt.Sprintf("work#%d", i+1)
|
||||
runTaskRoutine(rt, func(loopCtx context.Context) { e.workerLoop(loopCtx, t, name) })
|
||||
}
|
||||
e.startDeadlineCoordinator(ctx, t) // 任务级超时定时器(仅 timeout>0;去重)
|
||||
// 仅在「完全没有活动意图(open+running)」时才 kick 首轮规划。带种子意图的任务:种子已
|
||||
@@ -924,23 +913,14 @@ func (e *Engine) plannerLoop(ctx context.Context, t *Task) {
|
||||
}
|
||||
}
|
||||
|
||||
func (e *Engine) workerLoop(ctx context.Context, t *Task, name string, profileID int64) {
|
||||
func (e *Engine) workerLoop(ctx context.Context, t *Task, name string) {
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
default:
|
||||
}
|
||||
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)
|
||||
}
|
||||
_, worker := e.snapshotFor(t)
|
||||
if worker == nil {
|
||||
if sleepCtx(ctx, 1500*time.Millisecond) {
|
||||
return
|
||||
|
||||
+1
-36
@@ -177,11 +177,6 @@ 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 {
|
||||
@@ -3437,7 +3432,6 @@ 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 在当前配置不可用时
|
||||
@@ -3458,18 +3452,6 @@ 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)
|
||||
@@ -3524,8 +3506,7 @@ 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);对之后启动的任务生效
|
||||
WorkerPools *[]WorkerPoolConfig `json:"worker_pools"` // 多模型池;null=不改,[]=清空回遗留模式
|
||||
Workers *int `json:"workers"` // 并发工作 agent 数(>0);对之后启动的任务生效
|
||||
// 任务并发上限:同时「运行中」的任务数上限。关闭=不限;开启后新建任务超限则排队,有空位自动启动。
|
||||
ConcurrencyEnabled *bool `json:"task_concurrency_enabled"`
|
||||
ConcurrencyLimit *int `json:"task_concurrency_limit"`
|
||||
@@ -3604,22 +3585,6 @@ 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()
|
||||
|
||||
+11
-80
@@ -27,9 +27,6 @@ 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 {
|
||||
@@ -54,11 +51,6 @@ 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 {
|
||||
@@ -130,15 +122,6 @@ 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
|
||||
@@ -169,12 +152,6 @@ 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
|
||||
@@ -477,12 +454,6 @@ 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()
|
||||
}
|
||||
@@ -589,7 +560,17 @@ 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"))
|
||||
wk := s.newTaskWorker(workerRuntime, tx)
|
||||
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(可配置,默认开)
|
||||
pl := agent.NewPlanner(plannerRuntime, "task-router", s.m.dir, tx, plannerRuntime.CompactionWindow(), s.agentMaxTurns("planner"))
|
||||
pl.SetFindingRecorder(s.evidenceStore())
|
||||
pl.SetCompactionWindowResolver(plannerRuntime.CompactionWindow)
|
||||
@@ -617,61 +598,11 @@ 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
|
||||
|
||||
@@ -1,127 +0,0 @@
|
||||
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())
|
||||
}
|
||||
@@ -1,86 +0,0 @@
|
||||
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:])
|
||||
}
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
import * as React from "react";
|
||||
|
||||
import { CpuIcon, FlaskConicalIcon, KeyboardIcon, PlusIcon, RadioTowerIcon, SearchIcon, ShieldAlertIcon, Trash2Icon } from "lucide-react";
|
||||
import { CpuIcon, FlaskConicalIcon, KeyboardIcon, RadioTowerIcon, SearchIcon, ShieldAlertIcon } 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 { LLMProfile, Settings } from "@/lib/types";
|
||||
import type { Settings } from "@/lib/types";
|
||||
|
||||
import { BasicAuthCard } from "./_components/basic-auth-card";
|
||||
import { UpdateCard } from "./_components/update-card";
|
||||
@@ -45,8 +45,6 @@ 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<LLMProfile[]>([]);
|
||||
const [workerPools, setWorkerPools] = React.useState<{ profile_id: string; workers: string }[]>([]);
|
||||
// 操作约束注入范围(默认都开)。
|
||||
const [injectPlanner, setInjectPlanner] = React.useState(true);
|
||||
const [injectWorker, setInjectWorker] = React.useState(true);
|
||||
@@ -69,53 +67,23 @@ 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 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");
|
||||
const n = Number(workers);
|
||||
if (!Number.isInteger(n) || n <= 0) {
|
||||
toast.error("并发数必须是大于 0 的整数");
|
||||
return;
|
||||
}
|
||||
setSavingWorkers(true);
|
||||
api
|
||||
.setSettings({ workers: total, worker_pools: pools })
|
||||
.setSettings({ workers: n })
|
||||
.then((s) => {
|
||||
apply(s);
|
||||
toast.success(
|
||||
pools.length > 0
|
||||
? `已保存多模型 Worker 池(共 ${total} 个会话;对之后启动的任务生效)`
|
||||
: "已保存并发工作 agent 数(对之后启动的任务生效)",
|
||||
);
|
||||
toast.success("已保存并发工作 agent 数(对之后启动的任务生效)");
|
||||
})
|
||||
.catch((e) => toast.error("保存失败:" + (e as Error).message))
|
||||
.finally(() => setSavingWorkers(false));
|
||||
@@ -142,11 +110,9 @@ export default function SystemSettingsPage() {
|
||||
};
|
||||
|
||||
React.useEffect(() => {
|
||||
Promise.all([api.settings(), api.llmProfiles().catch(() => [] as LLMProfile[])])
|
||||
.then(([s, profiles]) => {
|
||||
apply(s);
|
||||
setLlmProfiles(profiles);
|
||||
})
|
||||
api
|
||||
.settings()
|
||||
.then(apply)
|
||||
.catch(() => undefined)
|
||||
.finally(() => setLoaded(true));
|
||||
}, [apply]);
|
||||
@@ -782,114 +748,24 @@ export default function SystemSettingsPage() {
|
||||
<CardHeader>
|
||||
<CardTitle className="flex items-center gap-2 text-base">
|
||||
<CpuIcon className="size-4" />
|
||||
工作并发 · Work Agent / 多模型池
|
||||
工作并发 · Work Agent 数
|
||||
</CardTitle>
|
||||
<CardDescription>
|
||||
可为多个 LLM 配置各分配若干 Worker;同一任务内多模型并行执行,按会话(work#N)绑定模型。
|
||||
未配置模型池时,使用下方「默认并发」走单一模型(与历史行为一致)。修改后
|
||||
每个任务并发运行的工作 agent 数量(默认 3)。数值越大并发探测越多、消耗也越高。修改后
|
||||
<b>对之后启动的任务生效</b>,正在运行的任务不受影响。
|
||||
</CardDescription>
|
||||
</CardHeader>
|
||||
<CardContent className="flex flex-col gap-4">
|
||||
<div className="flex flex-col gap-2">
|
||||
<Label className="text-sm font-normal text-muted-foreground">
|
||||
默认并发(未配置多模型池时使用)
|
||||
</Label>
|
||||
<CardContent className="flex flex-col gap-3">
|
||||
<div className="flex items-center gap-2">
|
||||
<Input
|
||||
type="number"
|
||||
min={1}
|
||||
className="w-32 font-mono text-sm"
|
||||
placeholder="3"
|
||||
value={workers}
|
||||
disabled={!loaded || savingWorkers || workerPools.length > 0}
|
||||
disabled={!loaded || savingWorkers}
|
||||
onChange={(e) => setWorkers(e.target.value)}
|
||||
/>
|
||||
{workerPools.length > 0 ? (
|
||||
<p className="text-muted-foreground text-xs">已配置模型池时以各池合计为准,默认并发输入已锁定。</p>
|
||||
) : null}
|
||||
</div>
|
||||
|
||||
<div className="flex flex-col gap-2 border-t pt-4">
|
||||
<div className="flex items-center justify-between gap-2">
|
||||
<Label className="text-sm font-normal text-muted-foreground">多模型 Worker 池</Label>
|
||||
<Button
|
||||
type="button"
|
||||
variant="outline"
|
||||
size="sm"
|
||||
disabled={!loaded || savingWorkers || llmProfiles.length === 0}
|
||||
onClick={() =>
|
||||
setWorkerPools((rows) => [
|
||||
...rows,
|
||||
{
|
||||
profile_id: llmProfiles[0]?.id ?? "",
|
||||
workers: "1",
|
||||
},
|
||||
])
|
||||
}
|
||||
>
|
||||
<PlusIcon className="size-3.5" />
|
||||
添加模型池
|
||||
</Button>
|
||||
</div>
|
||||
{llmProfiles.length === 0 ? (
|
||||
<p className="text-xs text-amber-500">尚未配置 LLM:请先到「系统 → LLM 配置」添加并激活。</p>
|
||||
) : null}
|
||||
{workerPools.map((row, idx) => (
|
||||
<div key={idx} className="flex flex-wrap items-center gap-2">
|
||||
<Select
|
||||
value={row.profile_id}
|
||||
disabled={!loaded || savingWorkers}
|
||||
onValueChange={(v) =>
|
||||
setWorkerPools((rows) => rows.map((r, i) => (i === idx ? { ...r, profile_id: v } : r)))
|
||||
}
|
||||
>
|
||||
<SelectTrigger className="w-56">
|
||||
<SelectValue placeholder="选择 LLM 配置" />
|
||||
</SelectTrigger>
|
||||
<SelectContent>
|
||||
{llmProfiles.map((p) => (
|
||||
<SelectItem key={p.id} value={p.id}>
|
||||
{p.name || p.model}
|
||||
{p.is_default ? " · 激活" : ""}
|
||||
</SelectItem>
|
||||
))}
|
||||
</SelectContent>
|
||||
</Select>
|
||||
<Input
|
||||
type="number"
|
||||
min={1}
|
||||
className="w-24 font-mono text-sm"
|
||||
value={row.workers}
|
||||
disabled={!loaded || savingWorkers}
|
||||
onChange={(e) =>
|
||||
setWorkerPools((rows) =>
|
||||
rows.map((r, i) => (i === idx ? { ...r, workers: e.target.value } : r)),
|
||||
)
|
||||
}
|
||||
/>
|
||||
<span className="text-muted-foreground text-xs">Workers</span>
|
||||
<Button
|
||||
type="button"
|
||||
variant="ghost"
|
||||
size="icon"
|
||||
disabled={!loaded || savingWorkers}
|
||||
onClick={() => setWorkerPools((rows) => rows.filter((_, i) => i !== idx))}
|
||||
aria-label="移除模型池"
|
||||
>
|
||||
<Trash2Icon className="size-4 text-destructive" />
|
||||
</Button>
|
||||
</div>
|
||||
))}
|
||||
<p className="text-muted-foreground text-xs">
|
||||
当前合计:
|
||||
{workerPools.length > 0
|
||||
? workerPools.reduce((s, r) => s + (Number(r.workers) || 0), 0)
|
||||
: Number(workers) || 0}{" "}
|
||||
个 Worker 会话
|
||||
</p>
|
||||
</div>
|
||||
|
||||
<div>
|
||||
<Button onClick={saveWorkers} disabled={!loaded || savingWorkers}>
|
||||
保存
|
||||
</Button>
|
||||
|
||||
@@ -3338,7 +3338,6 @@ 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,
|
||||
|
||||
@@ -968,8 +968,6 @@ 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
|
||||
|
||||
Reference in new issue
Block a user