Autumn-27
587ba422b4
依赖 norma v0.3.0 的 Provider.Complete。为每个 llm_profile 增加 streaming 开关:开(默认)走流式 SSE;关走真·非流式(stream:false、一次性返回完整 JSON),可绕开部分网关糟糕的 SSE 实现(空帧、思考字段丢帧),代价是失去 运行中的实时进度与实时 token 计数。 - agent.Config.Stream → agentcore.Options.NonStreaming = !Stream;worker / planner / mainagent / chat / goals 经 SetNonStreaming resolver 注入,任务 路由按当前激活 profile 动态取值 - llmpool.Pool / llmrec.Recorder / taskLLMRuntime 三个 Provider 包装均实现 Complete(分别带故障转移 / 录制 / 同 provider 重试与额度 failover),使 输出、token 统计、操作统计、LLM 录制四条链路在非流式下保持兼容 - db: llm_profiles 新增 streaming 列 + ALTER 补旧库(默认 true,旧配置无感) - 系统 → LLM 配置表单增加「流式输出」开关
653 lines
23 KiB
Go
653 lines
23 KiB
Go
package server
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"iter"
|
||
"log"
|
||
"path/filepath"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"github.com/Autumn-27/artex/agent"
|
||
"github.com/Autumn-27/artex/db"
|
||
"github.com/Autumn-27/artex/llmrec"
|
||
"github.com/Autumn-27/norma/llm"
|
||
"github.com/Autumn-27/norma/memory"
|
||
"github.com/Autumn-27/norma/transcript"
|
||
)
|
||
|
||
type taskAgentBundle struct {
|
||
runtime *taskLLMRuntime // goal decomposition runtime
|
||
plannerRuntime *taskLLMRuntime
|
||
workerRuntime *taskLLMRuntime
|
||
mainRuntime *taskLLMRuntime
|
||
pl *agent.Planner
|
||
wk *agent.Worker
|
||
main *agent.MainAgent
|
||
}
|
||
|
||
type llmAuditProfile struct {
|
||
ID int64 `json:"id"`
|
||
Name string `json:"name"`
|
||
Format string `json:"format"`
|
||
Model string `json:"model"`
|
||
}
|
||
|
||
type llmTransitionAudit struct {
|
||
Mode string `json:"mode"` // automatic | manual | exhausted
|
||
Reason string `json:"reason"`
|
||
Previous *llmAuditProfile `json:"previous,omitempty"`
|
||
Next *llmAuditProfile `json:"next,omitempty"`
|
||
}
|
||
|
||
type llmActivityMetadata struct {
|
||
LLMTransition llmTransitionAudit `json:"llm_transition"`
|
||
}
|
||
|
||
type taskLLMRuntime struct {
|
||
s *Server
|
||
taskID string
|
||
agentKey string
|
||
}
|
||
|
||
type taskLLMError struct {
|
||
taskID string
|
||
chainExhausted bool
|
||
cause error
|
||
}
|
||
|
||
func (e *taskLLMError) Error() string {
|
||
if e.chainExhausted {
|
||
return fmt.Sprintf("task %s LLM profile chain exhausted: %v", e.taskID, e.cause)
|
||
}
|
||
return e.cause.Error()
|
||
}
|
||
|
||
func (e *taskLLMError) Unwrap() error { return e.cause }
|
||
|
||
func isTaskLLMRuntimeError(err error) bool {
|
||
var target *taskLLMError
|
||
return errors.As(err, &target)
|
||
}
|
||
|
||
func isTaskLLMChainExhausted(err error) bool {
|
||
var target *taskLLMError
|
||
return errors.As(err, &target) && target.chainExhausted
|
||
}
|
||
|
||
// isQuotaExhaustedError is intentionally strict. A generic 429, auth error,
|
||
// network failure, or 5xx does not rotate providers; the response must explicitly
|
||
// identify quota, credits, billing balance, or payment exhaustion.
|
||
func isQuotaExhaustedError(err error) bool {
|
||
return err != nil && agent.IsQuotaExhaustedMessage(err.Error())
|
||
}
|
||
|
||
type taskLLMSelection struct {
|
||
task *Task
|
||
profileID int64
|
||
revision int64
|
||
provider llm.Provider
|
||
}
|
||
|
||
type taskLLMStreamHooks struct {
|
||
current func() (taskLLMSelection, error)
|
||
exhaust func(taskLLMSelection, error) (db.TaskLLMTransition, error)
|
||
transition func(taskLLMSelection, db.TaskLLMTransition, error)
|
||
}
|
||
|
||
// current resolves the provider this role runs on, by precedence:
|
||
// Agent 绑定 → 任务 LLM 配置链 → 全局/环境配置。
|
||
// 绑定优先于任务链：某个角色被显式指定了模型，就一直跑在那个模型上；绑定不存在或
|
||
// 构建失败时才降级到任务链，任务链为空时再降级到全局配置。
|
||
// 返回的 profile id 只有走任务链时才非零 —— streamTaskLLM 以此判断额度错误是否
|
||
// 应该推进任务的故障转移状态（绑定/全局路径不改任务链状态，沿用既有语义）。
|
||
func (r *taskLLMRuntime) current() (*Task, int64, int64, llm.Provider, error) {
|
||
taskNum, err := parseTaskID(r.taskID)
|
||
if err != nil {
|
||
return nil, 0, 0, nil, err
|
||
}
|
||
pt, err := r.s.m.pg.GetTask(taskNum)
|
||
if err != nil {
|
||
return nil, 0, 0, nil, err
|
||
}
|
||
if pt == nil {
|
||
return nil, 0, 0, nil, fmt.Errorf("task %s not found", r.taskID)
|
||
}
|
||
r.s.syncTaskLLMState(pt)
|
||
t, _ := r.s.m.Task(r.taskID)
|
||
if prov, _, ok := r.s.agentBindingProvider(r.agentKey); ok {
|
||
return t, 0, pt.LLMChainRevision, prov, nil
|
||
}
|
||
if len(pt.LLMProfileIDs) > 0 {
|
||
if pt.ActiveLLMProfileID == nil {
|
||
return t, 0, pt.LLMChainRevision, nil, &taskLLMError{taskID: r.taskID, chainExhausted: true, cause: errors.New("all selected profiles are quota exhausted")}
|
||
}
|
||
prov, _, ok := r.s.providerForProfile(*pt.ActiveLLMProfileID)
|
||
if !ok {
|
||
return t, *pt.ActiveLLMProfileID, pt.LLMChainRevision, nil, fmt.Errorf("LLM profile #%d is missing or invalid", *pt.ActiveLLMProfileID)
|
||
}
|
||
return t, *pt.ActiveLLMProfileID, pt.LLMChainRevision, prov, nil
|
||
}
|
||
prov, _, ok := r.s.globalProvider()
|
||
if !ok {
|
||
return t, 0, pt.LLMChainRevision, nil, fmt.Errorf("task %s has no available fallback LLM provider", r.taskID)
|
||
}
|
||
return t, 0, pt.LLMChainRevision, prov, nil
|
||
}
|
||
|
||
// nonStreaming reports whether the task's currently-active LLM source is set to
|
||
// non-streaming. It mirrors current()'s source precedence (agent binding →
|
||
// active chain profile → global) but reads only the streaming flag. Read-only
|
||
// and best-effort: any resolution error falls back to streaming (false), the
|
||
// safe default. If failover switches to a profile with a different streaming
|
||
// setting, the change takes effect on the next agent run (a fresh Session is
|
||
// built per run in captureRun).
|
||
func (r *taskLLMRuntime) nonStreaming() bool {
|
||
taskNum, err := parseTaskID(r.taskID)
|
||
if err != nil {
|
||
return false
|
||
}
|
||
pt, err := r.s.m.pg.GetTask(taskNum)
|
||
if err != nil || pt == nil {
|
||
return false
|
||
}
|
||
if _, cfg, ok := r.s.agentBindingProvider(r.agentKey); ok {
|
||
return !cfg.Stream
|
||
}
|
||
if len(pt.LLMProfileIDs) > 0 && pt.ActiveLLMProfileID != nil {
|
||
if _, cfg, ok := r.s.providerForProfile(*pt.ActiveLLMProfileID); ok {
|
||
return !cfg.Stream
|
||
}
|
||
}
|
||
if _, cfg, ok := r.s.globalProvider(); ok {
|
||
return !cfg.Stream
|
||
}
|
||
return false
|
||
}
|
||
|
||
func parseTaskID(id string) (int64, error) {
|
||
n, err := strconv.ParseInt(id, 10, 64)
|
||
if err != nil || n <= 0 {
|
||
return 0, fmt.Errorf("invalid task id %q", id)
|
||
}
|
||
return n, nil
|
||
}
|
||
|
||
// streamHooks builds the failover/exhaustion callbacks shared by Stream and
|
||
// Complete: how to read the current profile selection, how to mark it quota
|
||
// exhausted, and how to emit a failover transition.
|
||
func (r *taskLLMRuntime) streamHooks() taskLLMStreamHooks {
|
||
return taskLLMStreamHooks{
|
||
current: func() (taskLLMSelection, error) {
|
||
task, profileID, revision, provider, err := r.current()
|
||
return taskLLMSelection{task: task, profileID: profileID, revision: revision, provider: provider}, err
|
||
},
|
||
exhaust: func(selection taskLLMSelection, cause error) (db.TaskLLMTransition, error) {
|
||
taskNum, _ := parseTaskID(r.taskID)
|
||
transition, err := r.s.m.pg.MarkTaskLLMProfileQuotaExhaustedAtRevision(taskNum, selection.profileID, selection.revision, cause.Error())
|
||
if err != nil {
|
||
return transition, err
|
||
}
|
||
if pt, getErr := r.s.m.pg.GetTask(taskNum); getErr == nil && pt != nil {
|
||
r.s.syncTaskLLMState(pt)
|
||
}
|
||
return transition, nil
|
||
},
|
||
transition: func(selection taskLLMSelection, transition db.TaskLLMTransition, cause error) {
|
||
r.s.emitTaskLLMTransition(selection.task, transition, cause)
|
||
},
|
||
}
|
||
}
|
||
|
||
func (r *taskLLMRuntime) Stream(ctx context.Context, req llm.CompletionRequest) iter.Seq2[llm.StreamEvent, error] {
|
||
ctx = llmrec.WithTaskID(ctx, r.taskID)
|
||
return streamTaskLLM(ctx, r.taskID, req, r.streamHooks())
|
||
}
|
||
|
||
// Complete is the non-streaming counterpart of Stream. A non-streaming call is
|
||
// atomic — it never delivers partial output — so every failure is safe to retry
|
||
// on the same provider or fail over to the next profile without risking
|
||
// duplicated model output or tool execution (the "committed" bookkeeping the
|
||
// streaming path needs is unnecessary here).
|
||
func (r *taskLLMRuntime) Complete(ctx context.Context, req llm.CompletionRequest) (llm.Message, string, llm.Usage, error) {
|
||
ctx = llmrec.WithTaskID(ctx, r.taskID)
|
||
return completeTaskLLM(ctx, r.taskID, req, r.streamHooks())
|
||
}
|
||
|
||
func completeTaskLLM(ctx context.Context, taskID string, req llm.CompletionRequest, hooks taskLLMStreamHooks) (llm.Message, string, llm.Usage, error) {
|
||
for {
|
||
selection, err := hooks.current()
|
||
if err != nil {
|
||
return llm.Message{}, "", llm.Usage{}, err
|
||
}
|
||
var (
|
||
msg llm.Message
|
||
sr string
|
||
usage llm.Usage
|
||
callErr error
|
||
)
|
||
// 同 provider 安全窗口重试:非流式调用要么整体成功、要么整体失败,没有
|
||
// 中途已交付输出的问题,所以任何瞬时失败都可原样重试。
|
||
for attempt := 0; ; attempt++ {
|
||
msg, sr, usage, callErr = selection.provider.Complete(ctx, req)
|
||
if callErr != nil && ctx.Err() == nil &&
|
||
attempt < sameProviderStreamRetries && isRetryableStreamError(callErr) {
|
||
backoff := sameProviderRetryBackoff(attempt)
|
||
log.Printf("[task-llm] task %s 非流式调用失败,%v 后同 provider 重试 (%d/%d): %v",
|
||
taskID, backoff, attempt+1, sameProviderStreamRetries, callErr)
|
||
if sleepCtx(ctx, backoff) {
|
||
break // 退避期间 ctx 取消 → 停止重试
|
||
}
|
||
continue
|
||
}
|
||
break
|
||
}
|
||
if callErr == nil {
|
||
return msg, sr, usage, nil
|
||
}
|
||
// profileID=0: 显式链已被清空;非额度错误:透传。二者都不改任务 failover 状态。
|
||
if selection.profileID == 0 || !isQuotaExhaustedError(callErr) {
|
||
return llm.Message{}, "", llm.Usage{}, callErr
|
||
}
|
||
transition, markErr := hooks.exhaust(selection, callErr)
|
||
if markErr != nil {
|
||
return llm.Message{}, "", llm.Usage{}, fmt.Errorf("mark profile quota exhausted after %v: %w", callErr, markErr)
|
||
}
|
||
if transition.Advanced && !transition.Stale && hooks.transition != nil {
|
||
hooks.transition(selection, transition, callErr)
|
||
}
|
||
if !transition.Stale && transition.NextProfileID == nil {
|
||
return llm.Message{}, "", llm.Usage{}, &taskLLMError{taskID: taskID, chainExhausted: transition.ChainExhausted, cause: callErr}
|
||
}
|
||
// 没有任何输出交付给调用方,换下一个 profile 重放同一逻辑请求是安全的。
|
||
}
|
||
}
|
||
|
||
func streamTaskLLM(ctx context.Context, taskID string, req llm.CompletionRequest, hooks taskLLMStreamHooks) iter.Seq2[llm.StreamEvent, error] {
|
||
return func(yield func(llm.StreamEvent, error) bool) {
|
||
for {
|
||
selection, err := hooks.current()
|
||
if err != nil {
|
||
yield(llm.StreamEvent{}, err)
|
||
return
|
||
}
|
||
committed := false
|
||
var pending []llm.StreamEvent
|
||
var streamErr error
|
||
// 同 provider 安全窗口重试:committed 之前(还没向调用方交付任何输出)
|
||
// 的瞬时失败可以原样重放,不会重复模型输出或工具执行。committed 之后、
|
||
// ctx 取消、或确定性/额度错误则跳出,交给下方原有的透传/故障转移逻辑。
|
||
for attempt := 0; ; attempt++ {
|
||
committed = false
|
||
pending = nil
|
||
streamErr = nil
|
||
for event, err := range selection.provider.Stream(ctx, req) {
|
||
if err != nil {
|
||
streamErr = err
|
||
break
|
||
}
|
||
if !committed && !streamEventCommitsOutput(event) {
|
||
pending = append(pending, event)
|
||
continue
|
||
}
|
||
if !committed {
|
||
for _, buffered := range pending {
|
||
if !yield(buffered, nil) {
|
||
return
|
||
}
|
||
}
|
||
pending = nil
|
||
committed = true
|
||
}
|
||
if !yield(event, nil) {
|
||
return
|
||
}
|
||
}
|
||
if streamErr != nil && !committed && ctx.Err() == nil &&
|
||
attempt < sameProviderStreamRetries && isRetryableStreamError(streamErr) {
|
||
backoff := sameProviderRetryBackoff(attempt)
|
||
log.Printf("[task-llm] task %s 提交前流失败,%v 后同 provider 重试 (%d/%d): %v",
|
||
taskID, backoff, attempt+1, sameProviderStreamRetries, streamErr)
|
||
if sleepCtx(ctx, backoff) {
|
||
break // 退避期间 ctx 取消 → 停止重试
|
||
}
|
||
continue
|
||
}
|
||
break
|
||
}
|
||
if streamErr == nil {
|
||
for _, buffered := range pending {
|
||
if !yield(buffered, nil) {
|
||
return
|
||
}
|
||
}
|
||
return
|
||
}
|
||
// profileID=0 means the explicit chain was cleared while this stable task
|
||
// bundle was still in use. Agent/global fallback errors follow the legacy
|
||
// behavior and never mutate task failover state.
|
||
if selection.profileID == 0 || !isQuotaExhaustedError(streamErr) {
|
||
for _, buffered := range pending {
|
||
if !yield(buffered, nil) {
|
||
return
|
||
}
|
||
}
|
||
yield(llm.StreamEvent{}, streamErr)
|
||
return
|
||
}
|
||
transition, markErr := hooks.exhaust(selection, streamErr)
|
||
if markErr != nil {
|
||
cause := fmt.Errorf("mark profile quota exhausted after %v: %w", streamErr, markErr)
|
||
if committed {
|
||
// Output may already have driven tool execution. Report the persistence
|
||
// failure, but classify it as router-handled so the worker does not
|
||
// replay the entire intent and duplicate those side effects.
|
||
yield(llm.StreamEvent{}, &taskLLMError{taskID: taskID, cause: cause})
|
||
} else {
|
||
yield(llm.StreamEvent{}, cause)
|
||
}
|
||
return
|
||
}
|
||
if transition.Advanced && !transition.Stale && hooks.transition != nil {
|
||
hooks.transition(selection, transition, streamErr)
|
||
}
|
||
if committed || (!transition.Stale && transition.NextProfileID == nil) {
|
||
yield(llm.StreamEvent{}, &taskLLMError{taskID: taskID, chainExhausted: transition.ChainExhausted, cause: streamErr})
|
||
return
|
||
}
|
||
// No event reached the caller, so replaying the same logical request on
|
||
// the next profile cannot duplicate model output or tool execution.
|
||
}
|
||
}
|
||
}
|
||
|
||
func streamEventCommitsOutput(event llm.StreamEvent) bool {
|
||
switch event.Type {
|
||
case llm.SETextDelta, llm.SEThinkingDelta, llm.SEToolInputJSON:
|
||
return event.Text != ""
|
||
case llm.SEToolUseStart, llm.SEMessageDelta, llm.SEMessageStop:
|
||
return true
|
||
default:
|
||
return false
|
||
}
|
||
}
|
||
|
||
// 提交前安全窗口内、对同一 provider 的重试次数。SDK 的 doStream 只重试建连阶段
|
||
// (拿到 200 之前);流一旦开始,中途断流 / overloaded / 流内 429 等瞬时故障会直接
|
||
// 冒泡成 model_error,零重试。只要一个 token 都还没交给调用方(!committed),重放
|
||
// 完全相同的请求就不会重复模型输出或工具副作用,因此这里补一层同 provider 退避重试,
|
||
// 把这类抖动挡在意图整体重跑之前。
|
||
const sameProviderStreamRetries = 2
|
||
|
||
// sameProviderRetryBackoff 是第 attempt 次重试前的退避(0.5s、1s…,上限 4s),
|
||
// 与 SDK backoffSleep 同风格但封顶更小,避免拖住 worker 的收尾/取消响应。
|
||
// 以变量形式暴露,便于测试将退避置零。
|
||
var sameProviderRetryBackoff = func(attempt int) time.Duration {
|
||
return min(500*time.Millisecond*(1<<attempt), 4*time.Second)
|
||
}
|
||
|
||
// isRetryableStreamError 判断「提交前的流失败」是否值得在同一 provider 上重放。
|
||
// 瞬时的传输中断 / 供应商过载 / 限流会自行恢复,可安全重试;而以下三类不重试:
|
||
// - 额度耗尽:交给 profile 故障转移处理,别在这里白烧重试
|
||
// - 上下文过长:相同请求重放也没用,交给 harness 的 reactive 压缩兜底
|
||
// - 4xx 确定性拒绝(400/401/403/404/422):到哪个 provider 都一样会失败
|
||
func isRetryableStreamError(err error) bool {
|
||
if err == nil {
|
||
return false
|
||
}
|
||
if isQuotaExhaustedError(err) {
|
||
return false
|
||
}
|
||
s := strings.ToLower(err.Error())
|
||
if strings.Contains(s, "too long") || strings.Contains(s, "context length") ||
|
||
strings.Contains(s, "context_length") || strings.Contains(s, "maximum context") ||
|
||
strings.Contains(s, "status 413") {
|
||
return false
|
||
}
|
||
for _, code := range []string{"status 400", "status 401", "status 403", "status 404", "status 422"} {
|
||
if strings.Contains(s, code) {
|
||
return false
|
||
}
|
||
}
|
||
// 其余(传输 reset/EOF/timeout、408/429/5xx、流内 error 事件如 anthropic
|
||
// overloaded_error 等)一律视为瞬时,允许重试。
|
||
return true
|
||
}
|
||
|
||
// 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 _, cfg, ok := r.s.agentBindingProvider(r.agentKey); ok {
|
||
return cfg.CompactionWindow()
|
||
}
|
||
taskNum, err := parseTaskID(r.taskID)
|
||
if err != nil {
|
||
return (agent.Config{}).CompactionWindow()
|
||
}
|
||
chain, err := r.s.m.pg.TaskLLMProfiles(taskNum)
|
||
if err != nil {
|
||
return (agent.Config{}).CompactionWindow()
|
||
}
|
||
if len(chain) == 0 {
|
||
if _, cfg, ok := r.s.globalProvider(); ok {
|
||
return cfg.CompactionWindow()
|
||
}
|
||
return (agent.Config{}).CompactionWindow()
|
||
}
|
||
minimum := 0
|
||
for _, entry := range chain {
|
||
if cfg, ok := r.s.loadProfileConfig(entry.ProfileID); ok {
|
||
window := cfg.CompactionWindow()
|
||
if minimum == 0 || window < minimum {
|
||
minimum = window
|
||
}
|
||
}
|
||
}
|
||
if minimum == 0 {
|
||
return (agent.Config{}).CompactionWindow()
|
||
}
|
||
return minimum
|
||
}
|
||
|
||
// agentBindingProvider resolves the profile a role is explicitly bound to
|
||
// (agents.llm_profile_id) — the highest-precedence level for task agents. ok=false
|
||
// when the role has no binding or the bound profile no longer builds, so callers
|
||
// fall through to the task chain. A bound profile stays exclusive unless
|
||
// llm_pool_bind_fallback is on, which is what poolForBinding encodes.
|
||
func (s *Server) agentBindingProvider(agentKey string) (llm.Provider, agent.Config, bool) {
|
||
id := s.effectiveProfileForAgent(agentKey, nil)
|
||
if id == nil {
|
||
return nil, agent.Config{}, false
|
||
}
|
||
prov, cfg, ok := s.providerForProfile(*id)
|
||
if !ok {
|
||
return nil, agent.Config{}, false
|
||
}
|
||
return s.poolForBinding(*id, prov, cfg), cfg, true
|
||
}
|
||
|
||
// globalProvider returns the process-wide provider (persisted active profile or
|
||
// environment config) — the last resort once a role has neither a binding nor a
|
||
// task chain.
|
||
func (s *Server) globalProvider() (llm.Provider, agent.Config, bool) {
|
||
s.cfgMu.Lock()
|
||
defer s.cfgMu.Unlock()
|
||
if !s.llmOn || s.llmProv == nil {
|
||
return nil, agent.Config{}, false
|
||
}
|
||
return s.llmProv, s.llmCfg, true
|
||
}
|
||
|
||
// taskRuntimeAvailable reports whether every listed role can resolve a provider
|
||
// under the runtime precedence in current(): the role's own binding first, then
|
||
// the task chain, then global. An exhausted chain is a hard stop for unbound
|
||
// roles rather than a silent fall through to global — same as current().
|
||
func (s *Server) taskRuntimeAvailable(t *Task, agentKeys ...string) bool {
|
||
if t == nil || len(agentKeys) == 0 {
|
||
return false
|
||
}
|
||
state := t.llmStateSnapshot()
|
||
unboundReady := false
|
||
if len(state.ProfileIDs) > 0 {
|
||
if state.ActiveID != nil {
|
||
_, _, unboundReady = s.providerForProfile(*state.ActiveID)
|
||
}
|
||
} else {
|
||
_, _, unboundReady = s.globalProvider()
|
||
}
|
||
for _, key := range agentKeys {
|
||
if _, _, ok := s.agentBindingProvider(key); ok {
|
||
continue
|
||
}
|
||
if !unboundReady {
|
||
return false
|
||
}
|
||
}
|
||
return true
|
||
}
|
||
|
||
func (s *Server) invalidateTaskAgents() {
|
||
s.taskAgentMu.Lock()
|
||
s.taskAgents = map[string]*taskAgentBundle{}
|
||
s.taskAgentMu.Unlock()
|
||
}
|
||
|
||
func (s *Server) agentsForTask(t *Task) *taskAgentBundle {
|
||
s.taskAgentMu.Lock()
|
||
defer s.taskAgentMu.Unlock()
|
||
if bundle := s.taskAgents[t.ID]; bundle != nil {
|
||
return bundle
|
||
}
|
||
goalRuntime := &taskLLMRuntime{s: s, taskID: t.ID, agentKey: "goals"}
|
||
plannerRuntime := &taskLLMRuntime{s: s, taskID: t.ID, agentKey: "planner"}
|
||
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.SetCompactionWindowResolver(workerRuntime.CompactionWindow)
|
||
wk.SetNonStreaming(workerRuntime.nonStreaming) // 按任务当前激活 profile 的流式开关(每轮读)
|
||
wk.SetRunTimeout(time.Duration(s.agentRunSeconds("worker")) * time.Second)
|
||
wk.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert())
|
||
wk.SetMemory(memory.NewStore(filepath.Join(s.m.dir, "memory")))
|
||
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.SetCompactionWindowResolver(plannerRuntime.CompactionWindow)
|
||
pl.SetNonStreaming(plannerRuntime.nonStreaming)
|
||
pl.SetKillWork(s.engine.KillWork)
|
||
pl.SetSteerWork(s.engine.SteerWork)
|
||
pl.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert())
|
||
pl.SetWebSearch(s.webSearchFor("planner"))
|
||
pl.SetConstraintInject(s.constraintInjectPlanner) // 操作约束注入 planner(可配置,默认开)
|
||
main := agent.NewMainAgent(mainRuntime, "task-router", s.m.dir, tx, mainRuntime.CompactionWindow(), s.agentMaxTurns("mainagent"))
|
||
main.SetCompactionWindowResolver(mainRuntime.CompactionWindow)
|
||
main.SetNonStreaming(mainRuntime.nonStreaming)
|
||
main.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert())
|
||
main.SetWebSearch(s.webSearchFor("mainagent"))
|
||
main.SetSteerWork(s.engine.SteerWork) // steer_work：人对运行中 work 实时纠偏
|
||
bundle := &taskAgentBundle{
|
||
runtime: goalRuntime, plannerRuntime: plannerRuntime, workerRuntime: workerRuntime,
|
||
mainRuntime: mainRuntime, pl: pl, wk: wk, main: main,
|
||
}
|
||
s.taskAgents[t.ID] = bundle
|
||
return bundle
|
||
}
|
||
|
||
func (s *Server) syncTaskLLMState(pt *db.Task) {
|
||
if pt == nil {
|
||
return
|
||
}
|
||
id := fmt.Sprintf("%d", pt.ID)
|
||
s.m.mu.Lock()
|
||
if task := s.m.tasks[id]; task != nil {
|
||
task.setLLMState(pt.LLMProfileID, pt.ActiveLLMProfileID, pt.LLMProfileIDs, pt.LLMChainRevision, pt.LLMFailoverState, pt.LLMFailoverReason)
|
||
}
|
||
s.m.mu.Unlock()
|
||
}
|
||
|
||
func (s *Server) emitTaskLLMTransition(t *Task, transition db.TaskLLMTransition, cause error) {
|
||
if t == nil {
|
||
return
|
||
}
|
||
previous := s.llmAuditProfile(transition.PreviousProfileID)
|
||
var next *llmAuditProfile
|
||
if transition.NextProfileID != nil {
|
||
next = s.llmAuditProfile(*transition.NextProfileID)
|
||
}
|
||
mode := "automatic"
|
||
kind := "llm_switch"
|
||
summary := fmt.Sprintf("%s 额度不足", llmAuditProfileLabel(previous))
|
||
if transition.NextProfileID != nil {
|
||
summary += fmt.Sprintf("，后续调用切换到 %s", llmAuditProfileLabel(next))
|
||
} else {
|
||
mode = "exhausted"
|
||
kind = "llm_failover"
|
||
summary += "，配置链已耗尽"
|
||
}
|
||
metadata, _ := json.Marshal(llmActivityMetadata{LLMTransition: llmTransitionAudit{
|
||
Mode: mode, Reason: cause.Error(), Previous: previous, Next: next,
|
||
}})
|
||
s.engine.emitActivity(t, db.Activity{Worker: "system", Kind: kind, IsError: transition.ChainExhausted, Summary: summary, Detail: cause.Error(), Metadata: metadata})
|
||
log.Printf("[llm-failover] task %s: %s", t.ID, summary)
|
||
}
|
||
|
||
func (s *Server) llmAuditProfile(id int64) *llmAuditProfile {
|
||
if id <= 0 || s.m == nil || s.m.pg == nil {
|
||
return nil
|
||
}
|
||
p, err := s.m.pg.ProfileByID(id)
|
||
if err != nil || p == nil {
|
||
return &llmAuditProfile{ID: id, Name: fmt.Sprintf("配置 #%d", id)}
|
||
}
|
||
return &llmAuditProfile{ID: p.ID, Name: p.Name, Format: p.Format, Model: p.Model}
|
||
}
|
||
|
||
func llmAuditProfileLabel(profile *llmAuditProfile) string {
|
||
if profile == nil {
|
||
return "默认配置"
|
||
}
|
||
name := profile.Name
|
||
if name == "" {
|
||
name = fmt.Sprintf("配置 #%d", profile.ID)
|
||
}
|
||
detail := []string{}
|
||
if profile.Format != "" {
|
||
detail = append(detail, profile.Format)
|
||
}
|
||
if profile.Model != "" {
|
||
detail = append(detail, profile.Model)
|
||
}
|
||
if len(detail) == 0 {
|
||
return name
|
||
}
|
||
return fmt.Sprintf("%s（%s）", name, strings.Join(detail, " / "))
|
||
}
|
||
|
||
func sameOptionalID(a, b *int64) bool {
|
||
if a == nil || b == nil {
|
||
return a == nil && b == nil
|
||
}
|
||
return *a == *b
|
||
}
|
||
|
||
func (s *Server) emitManualTaskLLMSwitch(t *Task, previousID, nextID *int64) db.Activity {
|
||
previous := (*llmAuditProfile)(nil)
|
||
next := (*llmAuditProfile)(nil)
|
||
if previousID != nil {
|
||
previous = s.llmAuditProfile(*previousID)
|
||
}
|
||
if nextID != nil {
|
||
next = s.llmAuditProfile(*nextID)
|
||
}
|
||
summary := fmt.Sprintf("用户手动将任务 LLM 从 %s 切换到 %s", llmAuditProfileLabel(previous), llmAuditProfileLabel(next))
|
||
metadata, _ := json.Marshal(llmActivityMetadata{LLMTransition: llmTransitionAudit{
|
||
Mode: "manual", Reason: "用户手动切换任务 LLM", Previous: previous, Next: next,
|
||
}})
|
||
return s.engine.emitActivity(t, db.Activity{Worker: "system", Kind: "llm_switch", Summary: summary, Detail: summary, Metadata: metadata})
|
||
}
|