sundynix-agentix/sundynix-dispatcher/internal/eino/coordinator.go

package eino

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"regexp"
	"strings"
	"time"

	"github.com/cloudwego/eino/components/tool"
	"github.com/cloudwego/eino/compose"
	"github.com/cloudwego/eino/flow/agent/react"
	"github.com/cloudwego/eino/schema"

	"github.com/sundynix/sundynix-dispatcher/internal/dsl"
	"github.com/sundynix/sundynix-dispatcher/internal/harness"
	"github.com/sundynix/sundynix-shared/prompts"
)

// 多智能体协同（Eino × Anthropic 融合，详见仓库 MULTI_AGENT.md）：
// orchestrator = react.Agent（机器=Eino），编排认知按 Anthropic orchestrator-worker 配方；
// 专家 = 包成工具的子 agent（agent-as-tool），lead 给每个专家写定制简报（brief）后并行派发、综合。

// leadOrchestratorPrompt 是协调者(lead)的编排提示词 —— Anthropic 配方：先分解、按复杂度定额（别过度派发）、
// 给每个专家写明确简报、并行派发、收齐后综合消解冲突，不直接拼接专家原文。
const leadOrchestratorPrompt = `你是多智能体协调者(lead)。你手下有若干专家，每个专家是你可以调用的一个工具。
工作方式：
1. 先想清楚任务能否拆分、需要哪几个专家、各花多少功夫——简单任务用 1 个专家甚至自己直接答，复杂任务才并行派多个，不要过度派发。
2. 调用专家时，在 brief 参数里给它写一份明确的简报：要它完成的子任务、期望的输出形式、边界与约束；不要把原问题原样丢给它。
3. 可以在一轮里并行调用多个专家。收齐各专家的结论后，由你综合成一段连贯、完整的最终答复，消解专家之间的冲突，必要时点明哪个结论来自哪个专家。
不要把专家的原始返回直接拼接给用户——要消化、综合。`

// specialistCondensedSuffix 追加到每个专家的系统提示词：要求精炼返回，省 orchestrator 综合时的上下文。
const specialistCondensedSuffix = "\n\n返回要求：只给结论与关键依据要点，简洁，不要堆砌原始材料或冗长背景。"

// specialistSpec 是协调者节点配置里的一个子智能体(专家)定义。
type specialistSpec struct {
	Name   string
	Use    string   // 用途/擅长，作为工具描述供 lead 判断何时调
	System string   // 专家人设
	Tools  []string // 该专家可用的 MCP 工具名子集（空=纯对话专家）
}

// parseSpecialists 从协调者节点 config 的 agents 列表解析出专家定义。
func parseSpecialists(cfg map[string]any) []specialistSpec {
	raw, _ := cfg["agents"].([]any)
	var out []specialistSpec
	for _, item := range raw {
		m, ok := item.(map[string]any)
		if !ok {
			continue
		}
		s := specialistSpec{Name: cstr(m, "name"), Use: cstr(m, "use"), System: cstr(m, "system")}
		if s.Name == "" {
			continue
		}
		if tools, ok := m["tools"].([]any); ok {
			for _, t := range tools {
				if ts, ok := t.(string); ok && strings.TrimSpace(ts) != "" {
					s.Tools = append(s.Tools, strings.TrimSpace(ts))
				}
			}
		}
		out = append(out, s)
	}
	return out
}

// funcNameSafe 匹配 OpenAI 兼容 API 对 tools[].function.name 的约束：^[a-zA-Z0-9_-]+$。
var funcNameSafe = regexp.MustCompile(`[^a-zA-Z0-9_-]+`)

// toolFuncName 把专家名规范化成合法的函数名（给模型看）。
// 关键：专家名常是中文（"条款专家"），直接当函数名会被 OpenAI 兼容 API 400 拒绝
// （Invalid 'tools[0].function.name': does not match '^[a-zA-Z0-9_-]+$'），整个多智能体协调直接挂。
// 非法字符 → 下划线；清空则退化为 expert_N；used 保证同一批内唯一。展示名/轨迹仍用原名。
// 模型靠 Desc(spec.Use) 判断何时调用它，函数名本身不承载语义，故退化命名不影响派发质量。
func toolFuncName(name string, idx int, used map[string]bool) string {
	s := strings.Trim(funcNameSafe.ReplaceAllString(name, "_"), "_-")
	if s == "" {
		s = fmt.Sprintf("expert_%d", idx+1)
	}
	if used[s] {
		s = fmt.Sprintf("%s_%d", s, idx+1)
	}
	used[s] = true
	return s
}

// specialistTool 把一个专家(react.Agent 或 ChatModel)包成 Eino InvokableTool（agent-as-tool）：
// 入参 brief = lead 写给它的定制简报；返回专家的精炼结论。每次派发落一条 agent 轨迹。
type specialistTool struct {
	info    *schema.ToolInfo
	name    string
	run     func(ctx context.Context, brief string) (string, error)
	tr      *execTracer
	timeout time.Duration // 单次派发上限；0=不限。超时作为"观察"跳过该专家，不中断协调。
}

func (s *specialistTool) Info(context.Context) (*schema.ToolInfo, error) { return s.info, nil }

// InvokableRun 跑一次专家派发：取 lead 写的 brief → 执行子 agent → 返回精炼结论给 orchestrator。
// 专家失败作为"观察"返回（不中断协调）；专家 I/O 计入共享预算（Anthropic 的成本克制由 Budget 兜底）。
func (s *specialistTool) InvokableRun(ctx context.Context, argsJSON string, _ ...tool.Option) (string, error) {
	var args struct {
		Brief string `json:"brief"`
	}
	if argsJSON != "" {
		_ = json.Unmarshal([]byte(argsJSON), &args)
	}
	brief := strings.TrimSpace(args.Brief)
	if brief == "" {
		brief = "（协调者未给简报，请按你的职责处理当前任务）"
	}
	if bud := harness.BudgetFrom(ctx); bud != nil {
		bud.AddPrompt(brief)
	}
	end := s.tr.span("agent:"+s.name, "agent", "派发专家 "+s.name)
	rctx := ctx
	if s.timeout > 0 {
		var cancel context.CancelFunc
		rctx, cancel = context.WithTimeout(ctx, s.timeout)
		defer cancel()
	}
	out, err := s.run(rctx, brief)
	if err != nil {
		// 超时单独提示：作为"观察"跳过该专家，lead 据其余专家继续综合（不中断协调）。
		if s.timeout > 0 && errors.Is(err, context.DeadlineExceeded) {
			end("专家超时", err)
			return fmt.Sprintf("专家 %s 超时（>%s），本轮跳过其结论。", s.name, s.timeout), nil
		}
		end("专家执行失败", err)
		return "专家 " + s.name + " 执行失败：" + err.Error(), nil
	}
	if bud := harness.BudgetFrom(ctx); bud != nil {
		bud.AddComplete(out)
	}
	end("简报 "+truncate(brief, 100)+" → "+truncate(out, 160), nil)
	return out, nil
}

// buildSpecialists 据配置建专家工具集。带工具→react.Agent(过滤后的 MCP 工具子集)；不带→纯 ChatModel。
// 无可用模型的专家跳过。专家是内联叶子（非图节点），天然不能再是 coordinator → 杜绝套娃递归。
func (o *Orchestrator) buildSpecialists(ctx context.Context, specs []specialistSpec, b *board, taskID string, tr *execTracer) []tool.BaseTool {
	allTools := o.agentTools(b, taskID, tr) // 全量 MCP 工具，下面按 spec.Tools 过滤
	byName := map[string]tool.BaseTool{}
	for _, t := range allTools {
		if info, err := t.Info(ctx); err == nil {
			byName[info.Name] = t
		}
	}
	var out []tool.BaseTool
	usedFuncNames := map[string]bool{}
	for i, spec := range specs {
		sys := firstNonEmpty(spec.System, defaultAgentSystem) + specialistCondensedSuffix
		run := o.specialistRunner(ctx, b, spec, sys, byName)
		if run == nil {
			tr.info("coordinator", "system", "专家跳过", "无可用模型："+spec.Name)
			continue
		}
		out = append(out, &specialistTool{
			name: spec.Name, tr: tr, run: run, timeout: specialistTimeout,
			info: &schema.ToolInfo{
				Name: toolFuncName(spec.Name, i, usedFuncNames), // 给模型的函数名须 ASCII；展示/轨迹仍用 spec.Name
				Desc: firstNonEmpty(spec.Use, "专家 "+spec.Name),
				ParamsOneOf: schema.NewParamsOneOfByParams(map[string]*schema.ParameterInfo{
					"brief": {Type: schema.String, Desc: "给该专家的明确简报：子任务、期望输出、边界约束", Required: true},
				}),
			},
		})
	}
	return out
}

// specialistRunner 构造专家执行闭包：带工具且模型支持函数调用→react.Agent.Generate；否则→ChatModel.Generate。
// 无可用模型返回 nil（该专家被跳过）。
func (o *Orchestrator) specialistRunner(ctx context.Context, b *board, spec specialistSpec, sys string, byName map[string]tool.BaseTool) func(context.Context, string) (string, error) {
	var tools []tool.BaseTool
	for _, tn := range spec.Tools {
		if t, ok := byName[tn]; ok {
			tools = append(tools, t)
		}
	}
	if len(tools) > 0 {
		if tcm := o.agentPool(b).ToolCallingModel(); tcm != nil {
			ag, err := react.NewAgent(ctx, &react.AgentConfig{
				ToolCallingModel:      tcm,
				ToolsConfig:           compose.ToolsNodeConfig{Tools: tools},
				MaxStep:               reactMaxStep(),
				StreamToolCallChecker: streamHasToolCall,
			})
			if err == nil {
				return func(c context.Context, brief string) (string, error) {
					msg, gerr := ag.Generate(c, []*schema.Message{schema.SystemMessage(sys), schema.UserMessage(brief)})
					if gerr != nil {
						return "", gerr
					}
					return msg.Content, nil
				}
			}
		}
		// 工具型专家但模型不支持函数调用 → 退纯对话（下方）
	}
	cm := o.agentPool(b).ChatModel()
	if cm == nil {
		return nil
	}
	return func(c context.Context, brief string) (string, error) {
		msg, gerr := cm.Generate(c, []*schema.Message{schema.SystemMessage(sys), schema.UserMessage(brief)})
		if gerr != nil {
			return "", gerr
		}
		return msg.Content, nil
	}
}

// runCoordinator 执行多智能体协调节点：orchestrator = ReAct（专家=agent-as-tool），lead 提示词按
// Anthropic 配方（分解→定制简报→并行派发→综合）。无 ToolCallingModel / 0 可用专家 → 降级单 agent。
func (o *Orchestrator) runCoordinator(ctx context.Context, taskID string, b *board, system string, n dsl.Node, tr *execTracer, node string) {
	specs := parseSpecialists(n.Config)
	tcm := o.agentPool(b).ToolCallingModel()
	if tcm == nil || len(specs) == 0 {
		tr.info(node, "system", "协调者降级", "模型不支持函数调用或未配置专家，退回普通对话")
		o.runAgent(ctx, taskID, b, system, tr, node, labelOf(n, "多智能体协调"))
		return
	}
	specialists := o.buildSpecialists(ctx, specs, b, taskID, tr)
	if len(specialists) == 0 {
		tr.info(node, "system", "协调者降级", "无可用专家，退回普通对话")
		o.runAgent(ctx, taskID, b, system, tr, node, labelOf(n, "多智能体协调"))
		return
	}

	ag, err := react.NewAgent(ctx, &react.AgentConfig{
		ToolCallingModel:      tcm,
		ToolsConfig:           compose.ToolsNodeConfig{Tools: specialists},
		MaxStep:               reactMaxStep(),
		StreamToolCallChecker: streamHasToolCall,
	})
	if err != nil {
		tr.emit(node, "model", "error", "构建协调者", err.Error(), 0)
		o.runAgent(ctx, taskID, b, system, tr, node, labelOf(n, "多智能体协调"))
		return
	}

	// lead 提示词 = Anthropic 编排配方 + 节点自定义任务背景。
	leadSys := prompts.Get(prompts.CoordinatorLead)
	if s := strings.TrimSpace(system); s != "" && s != defaultAgentSystem {
		leadSys += "\n\n任务背景：" + s
	}
	rc := &RunCtx{
		UserID: b.uid, SessionID: b.sid,
		System:   leadSys,
		Query:    b.query,
		History:  b.history,
		ToolOut:  append(append([]string{}, b.toolOut...), b.refs...),
		Upstream: append([]string{}, b.agentOut...), // 前序协作 agent 产出 → 接力
	}
	msgs, _ := buildMessages(ctx, rc)
	if bud := harness.BudgetFrom(ctx); bud != nil {
		for _, m := range msgs {
			bud.AddPrompt(m.Content)
		}
		if bud.Exceeded() {
			if b.fatalErr == nil {
				b.fatalErr = errBudget
			}
			tr.emit(node, "system", "error", "token 预算", "已达单任务预算上限，中止", 0)
			return
		}
	}

	tr.emit(node, "model", "start", "多智能体协调", fmt.Sprintf("%d 个专家可派发", len(specialists)), 0)
	t0 := time.Now()
	sr, err := ag.Stream(ctx, msgs)
	if err != nil {
		tr.emit(node, "model", "error", "多智能体协调", err.Error(), time.Since(t0).Milliseconds())
		return
	}
	defer sr.Close()
	o.streamAgentReply(ctx, taskID, b, sr, tr, node, "多智能体协调", t0)
}