DeepSeek + Function Call：基于 Eino 的“计划——执行”多智能体范式实战

package plan_execute
import (    "github.com/cloudwego/eino/components/model"    "github.com/cloudwego/eino/compose"    "github.com/cloudwego/eino/schema")
// Config “计划——执行”多智能体的配置.type Config struct {    PlannerModel        model.ChatModel // planner 智能体使用的大模型    PlannerSystemPrompt string          // planner 智能体的 system prompt
    ExecutorModel        model.ChatModel         // executor 智能体使用的大模型    ToolsConfig          compose.ToolsNodeConfig // executor 智能体使用的工具执行器配置    ExecutorSystemPrompt string                  // executor 智能体的 system prompt
    ReviserModel        model.ChatModel // reviser 智能体使用的大模型    ReviserSystemPrompt string          // reviser 智能体的 system prompt
    MaxStep int // 多智能体的最大执行步骤数，避免无限循环}
// PlanExecuteMultiAgent “计划——执行”多智能体.type PlanExecuteMultiAgent struct {    // 图编排后的可执行体，输入是 Message 数组，输出是单条 Message    runnable compose.Runnable[[]*schema.Message, *schema.Message]}

// NewMultiAgent 根据配置编排一个“计划——执行”多智能体.func NewMultiAgent(ctx context.Context, config *Config) (*PlanExecuteMultiAgent, error) {    var (       toolInfos      []*schema.ToolInfo       toolsNode      *compose.ToolsNode       err            error       plannerPrompt  = config.PlannerSystemPrompt       executorPrompt = config.ExecutorSystemPrompt       reviserPrompt  = config.ReviserSystemPrompt       maxStep        = config.MaxStep    )    if len(plannerPrompt) == 0 {       plannerPrompt = defaultPlannerPrompt    }    if len(executorPrompt) == 0 {       executorPrompt = defaultExecutorPrompt    }    if len(reviserPrompt) == 0 {       reviserPrompt = defaultReviserPrompt    }    if maxStep == 0 {       maxStep = defaultMaxStep    }    if toolInfos, err = genToolInfos(ctx, config.ToolsConfig); err != nil {       return nil, err    }    // 为 Executor 配置工具    if err = config.ExecutorModel.BindTools(toolInfos); err != nil {       return nil, err    }    // 初始化 Tool 执行器节点，传入可执行的工具    if toolsNode, err = compose.NewToolNode(ctx, &config.ToolsConfig); err != nil {       return nil, err    }    // 创建一个待编排的 graph，规定整体的输入输出类型    graph := compose.NewGraph[[]*schema.Message, *schema.Message]()    // 定义 Executor 后的分支判断用的条件函数。该函数的输出是运行时选中的 NodeKey    executorPostBranchCondition := func(_ context.Context, msg *schema.Message) (endNode string, err error) {        if len(msg.ToolCalls) == 0 {           return nodeKeyExecutorToList, nil        }        return nodeKeyTools, nil    }    // 定义 Reviser 后的分支判断用的条件函数。    reviserPostBranchCondition := func(_ context.Context, sr *schema.StreamReader[*schema.Message]) (endNode string, err error) {       defer sr.Close()       var content string       for {          msg, err := sr.Recv()          if err != nil {             if err == io.EOF {                return nodeKeyReviserToList, nil             }             return "", err          }          content += msg.Content          if strings.Contains(content, "最终答案") {             return compose.END, nil          }          if len(content) > 20 {             return nodeKeyReviserToList, nil          }       }    }    // 添加 Planner 节点    _ = graph.AddChatModelNode(nodeKeyPlanner, config.PlannerModel, compose.WithNodeName(nodeKeyPlanner))    // 添加 Executor 节点    _ = graph.AddChatModelNode(nodeKeyExecutor, config.ExecutorModel, compose.WithNodeName(nodeKeyExecutor))    // 添加 Reviser 节点    _ = graph.AddChatModelNode(nodeKeyReviser, config.ReviserModel, compose.WithNodeName(nodeKeyReviser))    // 添加 Tool 执行器节点    _ = graph.AddToolsNode(nodeKeyTools, toolsNode)    // 添加三个 ToList 转换节点    _ = graph.AddLambdaNode(nodeKeyPlannerToList, compose.ToList[*schema.Message]())    _ = graph.AddLambdaNode(nodeKeyExecutorToList, compose.ToList[*schema.Message]())    _ = graph.AddLambdaNode(nodeKeyReviserToList, compose.ToList[*schema.Message]())    // 添加节点之间的边和分支    _ = graph.AddEdge(compose.START, nodeKeyPlanner)    _ = graph.AddEdge(nodeKeyPlanner, nodeKeyPlannerToList)    _ = graph.AddEdge(nodeKeyPlannerToList, nodeKeyExecutor)    _ = graph.AddBranch(nodeKeyExecutor, compose.NewStreamGraphBranch(executorPostBranchCondition, map[string]bool{       nodeKeyTools:          true,       nodeKeyExecutorToList: true,    }))    _ = graph.AddEdge(nodeKeyTools, nodeKeyExecutor)    _ = graph.AddEdge(nodeKeyExecutorToList, nodeKeyReviser)    _ = graph.AddBranch(nodeKeyReviser, compose.NewStreamGraphBranch(reviserPostBranchCondition, map[string]bool{       nodeKeyReviserToList: true,       compose.END:          true,    }))    _ = graph.AddEdge(nodeKeyReviserToList, nodeKeyExecutor)    // 编译 graph，将节点、边、分支转化为面向运行时的结构。由于 graph 中存在环，使用 AnyPredecessor 模式，同时设置运行时最大步数。    runnable, err := graph.Compile(ctx, compose.WithNodeTriggerMode(compose.AnyPredecessor), compose.WithMaxRunSteps(maxStep))    if err != nil {       return nil, err    }    return &PlanExecuteMultiAgent{       runnable: runnable,    }, nil}

type ActivityType string
const (    ActivityTypeAttraction  ActivityType = "attraction"    ActivityTypePerformance ActivityType = "performance"    ActivityTypeRestaurant  ActivityType = "restaurant"    ActivityTypeOther       ActivityType = "other")
// Activity 主题乐园中的一个项目，可以是游乐设施、表演或餐厅.type Activity struct {    Name               string       `json:"name"`    Desc               string       `json:"desc"`    Type               ActivityType `json:"type"`    Location           string       `json:"location" jsonschema:"description:项目所属的区域"`    MinHeight          int          `json:"min_height,omitempty" jsonschema:"description:参加游乐设施需要的最小身高，单位是厘米。如果为空，则没有身高要求"`    Duration           int          `json:"duration,omitempty" jsonschema:"description:一个项目参加一次需要的时间，注意不包括排队的时间。如果为空，则缺少具体的时间信息"`    TimeTable          []string     `json:"time_table,omitempty" jsonschema:"description:一个演出的时间表。如果为空，则使用 OpenTime 和 CloseTime 来表示这个项目的运营时间范围"`    OpenTime           string       `json:"open_time,omitempty" jsonschema:"description:一个项目开始运营的时间"`    CloseTime          string       `json:"close_time,omitempty" jsonschema:"description:一个项目结束运营的时间"`    RequireBooking     bool         `json:"require_booking,omitempty" jsonschema:"description:一个餐厅是否需要提前预约"`    HasPriorityAccess  bool         `json:"has_priority_access,omitempty" jsonschema:"description:一个项目是否有高速票服务"`    PriorityAccessCost int          `json:"priority_access_cost,omitempty" jsonschema:"description:一个项目如果有高速票服务，则一个人的高速票需要花多少钱"`    QueueTime          int          `json:"queue_time,omitempty" jsonschema:"description:一个项目常规需要的排队时间，单位是分钟。如果为空，则这个项目一般不需要排队"`}

// GetAttractionInfo 获取游乐设施信息.func GetAttractionInfo(_ context.Context, in *ListAttractionRequest) (out *ListAttractionResponse, err error) {    if len(in.Name) > 0 && in.Name != "all" {       for _, a := range attractions {          if a.Name == in.Name {             return &ListAttractionResponse{                Attractions: []Activity{                   a,                },             }, nil          }       }    }
    if len(in.Location) > 0 {       locationAttractions := make([]Activity, 0)       for _, a := range attractions {          if a.Location == in.Location {             locationAttractions = append(locationAttractions, a)             return &ListAttractionResponse{                Attractions: locationAttractions,             }, nil          }       }    }
    return &ListAttractionResponse{       Attractions: attractions,    }, nil}

func GetTools(ctx context.Context) (tools []tool.BaseTool, err error) {    queryTimeTool, err := utils.InferTool("query_theme_park_opening_hour", "查询乐园 A 的整体营业时间", GetParkHour)    if err != nil {       return nil, err    }    tools = append(tools, queryTimeTool)        // 以下省略多个 Tool        return}

// state 以多智能体一次运行为 scope 的全局状态，用于记录上下文type state struct {    messages            []*schema.Message}
func NewMultiAgent(ctx context.Context, config *Config) (*PlanExecuteMultiAgent, error) {    // ... 省略 N 行 ...     // 创建一个待编排的 graph，规定整体的输入输出类型，配置全局状态的初始化方法    graph := compose.NewGraph[[]*schema.Message, *schema.Message](compose.WithGenLocalState(func(ctx context.Context) *state {       return &state{}    }))    // 在大模型执行之前，向全局状态中保存上下文，并组装本次的上下文modelPreHandle := func(systemPrompt string, isDeepSeek bool) compose.StatePreHandler[[]*schema.Message, *state] {    return func(ctx context.Context, input []*schema.Message, state *state) ([]*schema.Message, error) {       for _, msg := range input {          state.messages = append(state.messages, msg)       }
       if isDeepSeek {          return append([]*schema.Message{schema.SystemMessage(systemPrompt)}, convertMessagesForDeepSeek(state.messages)...), nil       }
       return append([]*schema.Message{schema.SystemMessage(systemPrompt)}, state.messages...), nil    }}    // ... 省略 N 行 ...    // 添加 Planner 节点，同时添加 StatePreHandler 读写上下文    _ = graph.AddChatModelNode(nodeKeyPlanner, config.PlannerModel, compose.WithStatePreHandler(modelPreHandle(plannerPrompt, true)), compose.WithNodeName(nodeKeyPlanner))    // 添加 Executor 节点，同时添加 StatePreHandler 读写上下文    _ = graph.AddChatModelNode(nodeKeyExecutor, config.ExecutorModel, compose.WithStatePreHandler(modelPreHandle(executorPrompt, false)), compose.WithNodeName(nodeKeyExecutor))    // 添加 Reviser 节点，同时添加 StatePreHandler 读写上下文    _ = graph.AddChatModelNode(nodeKeyReviser, config.ReviserModel, compose.WithStatePreHandler(modelPreHandle(reviserPrompt, true)), compose.WithNodeName(nodeKeyReviser))    // 添加 Tool 执行器节点，同时添加 StatePreHandler 读写上下文    _ = graph.AddToolsNode(nodeKeyTools, toolsNode, compose.WithStatePreHandler(func(ctx context.Context, in *schema.Message, state *state) (*schema.Message, error) {        state.messages = append(state.messages, in)        return in, nil    }))    // ... 省略 N 行 ...}

func main() {    ctx := context.Background()
    deepSeekModel, err := deepseek.NewChatModel(ctx, &deepseek.ChatModelConfig{       Model:   os.Getenv("DEEPSEEK_MODEL_ID"),       APIKey:  os.Getenv("DEEPSEEK_API_KEY"),       BaseURL: os.Getenv("DEEPSEEK_BASE_URL"),    })    if err != nil {       log.Fatalf("new DeepSeek model failed: %v", err)    }
    arkModel, err := ark.NewChatModel(ctx, &ark.ChatModelConfig{       APIKey: os.Getenv("ARK_API_KEY"),       Model:  os.Getenv("ARK_MODEL_ID"),    })    if err != nil {       log.Fatalf("new Ark model failed: %v", err)    }
    toolsConfig, err := tools.GetTools(ctx)    if err != nil {       log.Fatalf("get tools config failed: %v", err)    }
    // 创建多智能体的配置，system prompt 都用默认值    config := &Config{       // planner 在调试时大部分场景不需要真的去生成，可以用 mock 输出替代       PlannerModel: &debug.ChatModelDebugDecorator{          Model: deepSeekModel,       },       ExecutorModel: arkModel,       ToolsConfig:   compose.ToolsNodeConfig{Tools: toolsConfig},       ReviserModel: &debug.ChatModelDebugDecorator{          Model: deepSeekModel,       },    }
    planExecuteAgent, err := NewMultiAgent(ctx, config)    if err != nil {       log.Fatalf("new plan execute multi agent failed: %v", err)    }
    printer := newIntermediateOutputPrinter() // 创建一个中间结果打印器    printer.printStream()                     // 开始异步输出到 console    handler := printer.toCallbackHandler()    // 转化为 Eino 框架的 callback handler
    // 以流式方式调用多智能体，实际的 OutputStream 不再需要关注，因为所有输出都由 intermediateOutputPrinter 处理了    _, err = planExecuteAgent.Stream(ctx, []*schema.Message{schema.UserMessage("我们一家三口去乐园玩，孩子身高 120 cm，园内预算 2000 元，最爱的是各种表演，游乐设施比较偏爱刺激项目，希望能在一天内尽可能多体验不同的活动，帮忙规划一个行程。我们会在园区开门之后立刻入园，在园区关闭之后再离开。")},       agent.WithComposeOptions(compose.WithCallbacks(handler)), // 将中间结果打印的 callback handler 注入进来       // 给 planner 指定 mock 输出       //agent.WithComposeOptions(compose.WithChatModelOption(debug.WithDebugOutput(schema.AssistantMessage(debug.PlannerOutput, nil))).DesignateNode(nodeKeyPlanner)),       // 给 reviser 指定 mock 输出       //agent.WithComposeOptions(compose.WithChatModelOption(debug.WithDebugOutput(schema.AssistantMessage("最终答案", nil))).DesignateNode(nodeKeyReviser)),    )    if err != nil {       log.Fatalf("stream error: %v", err)    }
    printer.wait() // 等待所有输出都处理完再结束}

相关链接