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, 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, 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.pool.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.pool.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.pool.ToolCallingModel() if tcm == nil || len(specs) == 0 { tr.info(node, "system", "协调者降级", "模型不支持函数调用或未配置专家,退回普通对话") o.runAgent(ctx, taskID, b, system, tr, node) 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) 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) 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) }