mirror of
https://github.com/RuoJi6/ARTEX.git
synced 2026-10-27 17:03:49 +08:00
refactor(engine): 移除从不执行的全局 planner/worker 空壳,就绪信号改用谓词
权威解析器(SetAuthoritativeAgentResolver)落地后,引擎每轮规划/worker 都走 per-task 的 agentsForTask 路由;applyLLM 经 buildPlannerWorker 构建、UseLLM 装入 的全局 planner/worker 从不被执行,仅充当 Ready() 的'已配置'探针。这层空壳是 功能容易被错误接线的温床(冷节点压缩此前就误接在它上面)。 改动: - 删除 buildPlannerWorker、providerForAgent(仅前者调用)、Engine.UseLLM/snapshot/ Mode、Engine.planner/worker/mu 字段。 - Ready() 改为读 SetReadiness 注入的谓词(基于 s.llmOn);snapshotFor 去掉全局 兜底,权威解析器返回 nil 即表示该任务不可用(语义不变)。 - applyLLM 不再构建全局 pair;NewServer 注入就绪谓词。 - 相应更新 3 处依赖旧注入方式的测试(改用权威解析器注入)。 生产行为不变(全局 pair 本就从不执行);go build/vet 通过,server 测试失败集合 与基线一致(均为需 DB 的既有用例),无新增失败。
This commit is contained in:
1 parent
cd5e779a24
commit
3b60191e86
5 files changed
+35
-94
No files matched your search
@@ -150,7 +150,9 @@ func TestChatMentionWorkerReceivesServerDetails(t *testing.T) {
|
||||
return llm.Message{Role: llm.RoleAssistant, Content: []llm.ContentBlock{llm.TextBlock("已读取引用")}}, "end_turn", llm.Usage{}, nil
|
||||
}}, "test", m.dir, nil, 10000, 1)
|
||||
worker.SetNonStreaming(func() bool { return true })
|
||||
s.engine.UseLLM(nil, worker)
|
||||
s.engine.SetAuthoritativeAgentResolver(func(*Task) (*agent.Planner, *agent.Worker) {
|
||||
return nil, worker
|
||||
})
|
||||
send := func(message string) *httptest.ResponseRecorder {
|
||||
body, _ := json.Marshal(map[string]string{"message": message, "request_id": "worker-mention-test"})
|
||||
r := httptest.NewRequest("POST", "/", strings.NewReader(string(body)))
|
||||
|
||||
+16
-33
@@ -96,9 +96,6 @@ type Engine struct {
|
||||
|
||||
bc *Broadcaster // live activity pub/sub (SSE)
|
||||
|
||||
mu sync.RWMutex
|
||||
planner *agent.Planner
|
||||
worker *agent.Worker
|
||||
started sync.Map // taskID -> bool, so Run is idempotent per task
|
||||
lastAct sync.Map // taskID -> int64 unix, last planner/worker activity (heartbeat)
|
||||
llmCalls sync.Map // taskID -> *int64, actual planner/worker/main-agent LLM calls
|
||||
@@ -146,11 +143,14 @@ type Engine struct {
|
||||
inflight sync.Map // taskID -> *int64, 在跑的 planner.Plan + worker.Execute 计数(用于 drain)
|
||||
coordStarted sync.Map // taskID -> bool, deadline 协调器是否已启动(Run/reload 去重)
|
||||
|
||||
// resolve, if set, returns a task's dedicated planner/worker. In legacy mode
|
||||
// nil,nil falls back to the global pair; an authoritative resolver uses nil,nil
|
||||
// to mean this task is deliberately unavailable (for example an exhausted chain).
|
||||
// resolve returns a task's dedicated planner/worker (wired by the server as the
|
||||
// authoritative task-router). nil,nil means this task is deliberately unavailable
|
||||
// (for example an exhausted failover chain) — there is no global-pair fallback.
|
||||
resolve func(t *Task) (*agent.Planner, *agent.Worker)
|
||||
resolveAuthoritative bool
|
||||
// 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
|
||||
}
|
||||
|
||||
type taskRuntime struct {
|
||||
@@ -696,18 +696,9 @@ func (e *Engine) appendActivity(t *Task, r db.Activity) (int64, error) {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
// UseLLM installs (or replaces) the LLM planner + work agent at runtime.
|
||||
func (e *Engine) UseLLM(p *agent.Planner, w *agent.Worker) {
|
||||
e.mu.Lock()
|
||||
e.planner, e.worker = p, w
|
||||
e.mu.Unlock()
|
||||
}
|
||||
|
||||
func (e *Engine) snapshot() (*agent.Planner, *agent.Worker) {
|
||||
e.mu.RLock()
|
||||
defer e.mu.RUnlock()
|
||||
return e.planner, e.worker
|
||||
}
|
||||
// SetReadiness wires the global "an LLM provider is configured" predicate (read by
|
||||
// Ready() / the llm_configured indicator). Called once at startup.
|
||||
func (e *Engine) SetReadiness(fn func() bool) { e.readiness = fn }
|
||||
|
||||
// SetAgentResolver installs a per-task planner/worker resolver (wired by the server).
|
||||
// Called once at startup before any task loop runs, so no lock is needed on reads.
|
||||
@@ -724,9 +715,9 @@ func (e *Engine) SetAuthoritativeAgentResolver(fn func(t *Task) (*agent.Planner,
|
||||
e.resolveAuthoritative = true
|
||||
}
|
||||
|
||||
// snapshotFor returns the planner/worker a task should run on: the task's dedicated
|
||||
// pair when it pins a specific LLM profile (via the resolver), else the global active
|
||||
// pair. The resolver returning nil means "no override" (fall back to global).
|
||||
// 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.
|
||||
func (e *Engine) snapshotFor(t *Task) (*agent.Planner, *agent.Worker) {
|
||||
if e.resolve != nil {
|
||||
p, w := e.resolve(t)
|
||||
@@ -734,13 +725,13 @@ func (e *Engine) snapshotFor(t *Task) (*agent.Planner, *agent.Worker) {
|
||||
return p, w
|
||||
}
|
||||
}
|
||||
return e.snapshot()
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
// Ready reports whether an LLM provider is configured.
|
||||
// Ready reports whether a global LLM provider is configured (via the readiness
|
||||
// predicate wired at startup).
|
||||
func (e *Engine) Ready() bool {
|
||||
p, w := e.snapshot()
|
||||
return p != nil && w != nil
|
||||
return e.readiness != nil && e.readiness()
|
||||
}
|
||||
|
||||
// ReadyFor reports whether a specific task can resolve a planner/worker pair.
|
||||
@@ -751,14 +742,6 @@ func (e *Engine) ReadyFor(t *Task) bool {
|
||||
return p != nil && w != nil
|
||||
}
|
||||
|
||||
// Mode reports "llm" when ready, else "idle".
|
||||
func (e *Engine) Mode() string {
|
||||
if e.Ready() {
|
||||
return "llm"
|
||||
}
|
||||
return "idle"
|
||||
}
|
||||
|
||||
// Run starts the planner loop + N worker loops for a task. The loops always run
|
||||
// but no-op until an LLM is configured (so a task created while idle picks up
|
||||
// automatically once LLM is set from the UI).
|
||||
|
||||
+10
-55
@@ -164,6 +164,13 @@ func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDi
|
||||
b := s.agentsForTask(t)
|
||||
return b.pl, b.wk
|
||||
})
|
||||
// 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 {
|
||||
s.cfgMu.Lock()
|
||||
defer s.cfgMu.Unlock()
|
||||
return s.llmOn
|
||||
})
|
||||
// Wire DB-stored prompt templates into the agents (新版方案 §3.3 / §5a). With no
|
||||
// override row, agents keep their built-in defaults — behavior is unchanged.
|
||||
if m.pg != nil {
|
||||
@@ -398,44 +405,6 @@ func (s *Server) webSearchFor(key string) agent.WebSearchOpts {
|
||||
return o
|
||||
}
|
||||
|
||||
// buildPlannerWorker builds a planner+worker pair. Each agent resolves its OWN LLM by
|
||||
// precedence agent-binding → pin → the passed global fallback (gProv/gCfg), so planner
|
||||
// and worker can run on different models (e.g. a stronger planner, a cheaper worker).
|
||||
// pinID is the task's pinned profile (nil on the global active path). With no agent
|
||||
// binding and no pin, both fall back to gProv/gCfg — identical to the previous single-
|
||||
// provider behavior. Called by applyLLM (global active pair).
|
||||
func (s *Server) buildPlannerWorker(pinID *int64, gProv llm.Provider, gCfg agent.Config) (*agent.Planner, *agent.Worker) {
|
||||
tx := transcript.NewStore(filepath.Join(s.m.dir, "transcripts")) // raw LLM conversation logs
|
||||
// traffic host tools flow through ToolAugment for every agent and are filtered by
|
||||
// the tools-table binding (default = worker), so worker behavior is unchanged.
|
||||
wProv, wCfg := s.providerForAgent("worker", pinID, gProv, gCfg)
|
||||
wk := agent.NewWorker(wProv, wCfg.Model, s.m.dir, tx, wCfg.CompactionWindow(), s.agentMaxTurns("worker"))
|
||||
wk.SetFindingRecorder(s.evidenceStore())
|
||||
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.SetNonStreaming(nonStreamingResolver(wCfg)) // 该 profile 选非流式时走 Provider.Complete
|
||||
wk.SetMaxTokens(maxTokensResolver(wCfg)) // 单次回复输出上限(0 = 不发)
|
||||
wk.SetNoaEnabled(s.m.NoaCompactionEnabled) // 实验功能:noa 上下文压缩(每 run 读)
|
||||
pProv, pCfg := s.providerForAgent("planner", pinID, gProv, gCfg)
|
||||
pl := agent.NewPlanner(pProv, pCfg.Model, s.m.dir, tx, pCfg.CompactionWindow(), s.agentMaxTurns("planner"))
|
||||
pl.SetFindingRecorder(s.evidenceStore())
|
||||
pl.SetKillWork(s.engine.KillWork) // planner kill_work → terminate a running work
|
||||
pl.SetSteerWork(s.engine.SteerWork) // planner steer_work → inject mid-run course-correction
|
||||
pl.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) // WebFetch through the recording proxy
|
||||
pl.SetWebSearch(s.webSearchFor("planner"))
|
||||
pl.SetConstraintInject(s.constraintInjectPlanner) // 操作约束注入 planner(可配置,默认开;每轮读)
|
||||
pl.SetNonStreaming(nonStreamingResolver(pCfg)) // 该 profile 选非流式时走 Provider.Complete
|
||||
pl.SetMaxTokens(maxTokensResolver(pCfg)) // 单次回复输出上限(0 = 不发)
|
||||
pl.SetNoaEnabled(s.m.NoaCompactionEnabled) // 实验功能:noa 上下文压缩(每 run 读)
|
||||
// cold-digest §7: background cold-node compaction, on the planner's provider/
|
||||
// model (§4 uses the running agent's model). Uses Complete (non-streaming) for
|
||||
// the one-shot body summarization.
|
||||
pl.SetCompactor(agent.NewCompactor(pProv, pCfg.Model))
|
||||
return pl, wk
|
||||
}
|
||||
|
||||
// nonStreamingResolver returns a resolver capturing a profile's streaming choice.
|
||||
// The agents read it per run; a profile change rebuilds the agents (applyLLM),
|
||||
// so the captured value is always the one in effect for this build.
|
||||
@@ -483,8 +452,9 @@ func (s *Server) applyLLM(cfg agent.Config) error {
|
||||
if act, err := s.m.pg.ActiveProfile(); err == nil && act != nil {
|
||||
prov = s.poolForActive(act.ID, prov, cfg)
|
||||
}
|
||||
pl, wk := s.buildPlannerWorker(nil, prov, cfg)
|
||||
s.engine.UseLLM(pl, wk)
|
||||
// No global planner/worker pair: every task runs on its own task-routed pair
|
||||
// (agentsForTask), resolved through the engine's authoritative resolver. Global
|
||||
// readiness (Ready()/llm_configured) is reported from s.llmOn, set below.
|
||||
|
||||
tx := transcript.NewStore(filepath.Join(s.m.dir, "transcripts"))
|
||||
win := cfg.CompactionWindow()
|
||||
@@ -622,21 +592,6 @@ func (s *Server) providerForProfile(id int64) (llm.Provider, agent.Config, bool)
|
||||
return prov, cfg, true
|
||||
}
|
||||
|
||||
// providerForAgent returns the provider+cfg one agent should run on: its bound/pinned
|
||||
// profile if that resolves, else the passed global fallback (gProv/gCfg).
|
||||
// A bound/pinned profile is EXCLUSIVE by default: it does not fail over, so an
|
||||
// agent deliberately put on a specific model never silently runs on another one.
|
||||
// Turning on llm_pool_bind_fallback relaxes that — poolForBinding then appends the
|
||||
// failover chain behind it.
|
||||
func (s *Server) providerForAgent(agentKey string, pinID *int64, gProv llm.Provider, gCfg agent.Config) (llm.Provider, agent.Config) {
|
||||
if eff := s.effectiveProfileForAgent(agentKey, pinID); eff != nil {
|
||||
if prov, cfg, ok := s.providerForProfile(*eff); ok {
|
||||
return s.poolForBinding(*eff, prov, cfg), cfg
|
||||
}
|
||||
}
|
||||
return gProv, gCfg
|
||||
}
|
||||
|
||||
// chatAgentForProfile returns a ChatAgent built from a specific LLM profile, cached
|
||||
// per profile id. Returns nil if the profile is missing or has no API key.
|
||||
func (s *Server) chatAgentForProfile(id int64) *agent.ChatAgent {
|
||||
|
||||
+2
-3
@@ -583,9 +583,8 @@ func (s *Server) agentsForTask(t *Task) *taskAgentBundle {
|
||||
pl.SetWebSearch(s.webSearchFor("planner"))
|
||||
pl.SetConstraintInject(s.constraintInjectPlanner) // 操作约束注入 planner(可配置,默认开)
|
||||
// cold-digest §7: 冷节点后台压缩。引擎经权威解析器实际驱动的就是这套 per-task planner
|
||||
// (agentsForTask),所以 Compactor 必须接在这里;buildPlannerWorker 那套全局 pair 被解析器
|
||||
// 旁路、从不跑规划轮,接在那里等于不生效。走任务路由的 planner provider(§4:与 agent 同模型),
|
||||
// 压缩用 Complete 一次性生成 body。
|
||||
// (agentsForTask),Compactor 必须接在这里。走任务路由的 planner provider(§4:与 agent
|
||||
// 同模型,随任务 LLM 链解析),压缩用 Complete 一次性生成 body。
|
||||
pl.SetCompactor(agent.NewCompactor(plannerRuntime, "task-router"))
|
||||
main := agent.NewMainAgent(mainRuntime, "task-router", s.m.dir, tx, mainRuntime.CompactionWindow(), s.agentMaxTurns("mainagent"))
|
||||
main.SetFindingRecorder(s.evidenceStore())
|
||||
|
||||
@@ -601,9 +601,11 @@ func TestTaskRuntimeAvailableUnboundRoleFollowsChainThenGlobal(t *testing.T) {
|
||||
func TestAuthoritativeTaskResolverDoesNotFallBackWhenChainUnavailable(t *testing.T) {
|
||||
t.Parallel()
|
||||
e := &Engine{}
|
||||
e.UseLLM(&agent.Planner{}, &agent.Worker{})
|
||||
e.SetAuthoritativeAgentResolver(func(*Task) (*agent.Planner, *agent.Worker) {
|
||||
return &agent.Planner{}, &agent.Worker{}
|
||||
})
|
||||
if !e.ReadyFor(&Task{ID: "7"}) {
|
||||
t.Fatal("global provider should initially make task ready")
|
||||
t.Fatal("an available authoritative chain should make the task ready")
|
||||
}
|
||||
e.SetAuthoritativeAgentResolver(func(*Task) (*agent.Planner, *agent.Worker) {
|
||||
return nil, nil
|
||||
|
||||
Reference in new issue
Block a user