Blizzard
f9c849b14b
- specialistTool 加 timeout 字段(默认 specialistTimeout=3min，专家可多轮 react+工具故给宽) - InvokableRun 用 WithTimeout 包裹专家派发；超时(DeadlineExceeded)作为"观察" 跳过该专家(err=nil)，lead 据其余专家继续综合，不中断整个协调 - 2 单测：卡死专家 ~50ms 跳过并返回超时观察 / 正常专家不受影响 - timeout=0 时不包裹(向后兼容既有 specialistTool 构造) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
262 lines
10 KiB
Go
262 lines
10 KiB
Go
package eino
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"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
|
||
}
|
||
|
||
// 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
|
||
for _, 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: 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)
|
||
}
|