Eino Compose 模块详解

graph TB
    subgraph "编排层 (Orchestration Layer)"
        Chain[Chain 链式编排]
        Graph[Graph 图式编排]
        Workflow[Workflow 工作流编排]
    end
    
    subgraph "编译层 (Compilation Layer)"
        Compiler[Graph Compiler]
        TypeChecker[Type Checker]
        Optimizer[Execution Optimizer]
    end
    
    subgraph "执行层 (Execution Layer)"
        Runnable[Runnable Interface]
        Runner[Graph Runner]
        TaskManager[Task Manager]
        ChannelManager[Channel Manager]
    end
    
    subgraph "组件层 (Component Layer)"
        Lambda[Lambda Functions]
        Components[Eino Components]
        SubGraphs[Sub Graphs]
    end
    
    Chain --> Compiler
    Graph --> Compiler
    Workflow --> Compiler
    
    Compiler --> TypeChecker
    Compiler --> Optimizer
    
    TypeChecker --> Runnable
    Optimizer --> Runnable
    
    Runnable --> Runner
    Runner --> TaskManager
    Runner --> ChannelManager
    
    Lambda --> Components
    Components --> SubGraphs
    
    style Chain fill:#e8f5e8
    style Graph fill:#fff3e0
    style Workflow fill:#f3e5f5
    style Runnable fill:#e3f2fd

// compose/types.go

type component = components.Component

// 内置组件类型
const (
	ComponentOfUnknown     component = "Unknown"
	ComponentOfGraph       component = "Graph"
	ComponentOfWorkflow    component = "Workflow"
	ComponentOfChain       component = "Chain"
	ComponentOfPassthrough component = "Passthrough"
	ComponentOfToolsNode   component = "ToolsNode"
	ComponentOfLambda      component = "Lambda"
)

// NodeTriggerMode 控制图节点的触发模式
type NodeTriggerMode string

const (
	// AnyPredecessor 当任何前驱节点完成时触发
	// 适用于 Pregel 模式，支持循环和迭代
	AnyPredecessor NodeTriggerMode = "any_predecessor"
	
	// AllPredecessor 当所有前驱节点完成时才触发
	// 适用于 DAG 模式，确保严格的依赖顺序
	AllPredecessor NodeTriggerMode = "all_predecessor"
)

// compose/types_composable.go

// AnyGraph 可组合图的统一接口
// 由 Graph、Chain、Workflow 实现
type AnyGraph interface {
	getGenericHelper() *genericHelper
	compile(ctx context.Context, options *graphCompileOptions) (*composableRunnable, error)
	inputType() reflect.Type
	outputType() reflect.Type
	component() component
}

// compose/runnable.go

// Runnable 是可执行对象的接口。Graph、Chain 可以编译为 Runnable。
// runnable 是 eino 的核心概念，我们对四种数据流模式做降级兼容，
// 可以自动连接只实现一种或多种方法的组件。
// Runnable 是框架的核心抽象，定义了四种数据流模式
// 详细的接口定义请参考 [核心API深度分析](/posts/eino-03-core-api-analysis/)
type Runnable[I, O any] interface {
	// 省略具体方法定义 - 请参考核心API文档
}

graph LR
    subgraph "输入类型"
        SingleInput[单值输入 I]
        StreamInput[流式输入 StreamReader<I>]
    end
    
    subgraph "输出类型"
        SingleOutput[单值输出 O]
        StreamOutput[流式输出 StreamReader<O>]
    end
    
    subgraph "四种模式"
        Invoke[Invoke: I → O]
        Stream[Stream: I → StreamReader<O>]
        Collect[Collect: StreamReader<I> → O]
        Transform[Transform: StreamReader<I> → StreamReader<O>]
    end
    
    SingleInput --> Invoke
    SingleInput --> Stream
    StreamInput --> Collect
    StreamInput --> Transform
    
    Invoke --> SingleOutput
    Stream --> StreamOutput
    Collect --> SingleOutput
    Transform --> StreamOutput
    
    style Invoke fill:#e8f5e8
    style Stream fill:#fff3e0
    style Collect fill:#f3e5f5
    style Transform fill:#e3f2fd

// compose/runnable.go 内部实现

// composableRunnable 是 Runnable 的具体实现
type composableRunnable struct {
	invoke    func(context.Context, any, ...Option) (any, error)
	stream    func(context.Context, any, ...Option) (*schema.StreamReader[any], error)
	collect   func(context.Context, *schema.StreamReader[any], ...Option) (any, error)
	transform func(context.Context, *schema.StreamReader[any], ...Option) (*schema.StreamReader[any], error)
	
	inputType  reflect.Type
	outputType reflect.Type
	
	genericHelper *genericHelper
	nodeInfo      *nodeInfo
	meta          *executorMeta
}

// 自动适配示例：通过 Stream 实现 Invoke
func (cr *composableRunnable) invokeByStream(ctx context.Context, input any, opts ...Option) (any, error) {
	stream, err := cr.stream(ctx, input, opts...)
	if err != nil {
		return nil, err
	}
	defer stream.Close()
	
	var result any
	for {
		chunk, err := stream.Recv()
		if err == io.EOF {
			break
		}
		if err != nil {
			return nil, err
		}
		result = chunk // 保留最后一个值
	}
	
	return result, nil
}

// 自动适配示例：通过 Invoke 实现 Stream
func (cr *composableRunnable) streamByInvoke(ctx context.Context, input any, opts ...Option) (*schema.StreamReader[any], error) {
	result, err := cr.invoke(ctx, input, opts...)
	if err != nil {
		return nil, err
	}
	
	// 将单个结果包装为只有一个元素的流
	return schema.StreamReaderFromArray([]any{result}), nil
}

// compose/chain.go

// Chain 是组件链
// Chain 节点可以是并行/分支/序列组件
// Chain 设计为构建器模式使用（使用前应该 Compile()）
// 接口是 `Chain 风格`，可以像这样使用：`chain.AppendXX(...).AppendXX(...)`
type Chain[I, O any] struct {
	gg          *Graph[I, O]  // 内部使用 Graph 实现
	preNodeKeys []string      // 前一个节点的键列表
	err         error         // 构建过程中的错误
}

// NewChain 创建具有输入/输出类型的链
func NewChain[I, O any](opts ...NewGraphOption) *Chain[I, O] {
	ch := &Chain[I, O]{
		gg: NewGraph[I, O](opts...),
	}
	ch.gg.cmp = ComponentOfChain
	return ch
}

// 聊天模型组件
func (c *Chain[I, O]) AppendChatModel(node model.BaseChatModel, opts ...GraphAddNodeOpt) *Chain[I, O] {
	gNode, options := toChatModelNode(node, opts...)
	c.addNode(gNode, options)
	return c
}

// Lambda 函数组件
func (c *Chain[I, O]) AppendLambda(node *Lambda, opts ...GraphAddNodeOpt) *Chain[I, O] {
	gNode, options := toLambdaNode(node, opts...)
	c.addNode(gNode, options)
	return c
}

// 检索器组件
func (c *Chain[I, O]) AppendRetriever(node retriever.Retriever, opts ...GraphAddNodeOpt) *Chain[I, O] {
	gNode, options := toRetrieverNode(node, opts...)
	c.addNode(gNode, options)
	return c
}

// 并行结构
func (c *Chain[I, O]) AppendParallel(p *Parallel) *Chain[I, O] {
	gNode, options := toParallelNode(p)
	c.addNode(gNode, options)
	return c
}

// 分支结构
func (c *Chain[I, O]) AppendBranch(b *ChainBranch) *Chain[I, O] {
	gNode, options := toBranchNode(b)
	c.addNode(gNode, options)
	return c
}

// compose/chain.go

func (c *Chain[I, O]) addNode(node *graphNode, options *graphAddNodeOpts) {
	if c.err != nil {
		return
	}
	if c.gg.compiled {
		c.reportError(ErrChainCompiled)
		return
	}
	if node == nil {
		c.reportError(fmt.Errorf("chain add node invalid, node is nil"))
		return
	}
	
	// 生成节点键
	nodeKey := options.nodeOptions.nodeKey
	defaultNodeKey := c.nextNodeKey()
	if nodeKey == "" {
		nodeKey = defaultNodeKey
	}
	
	// 添加节点到底层图
	err := c.gg.addNode(nodeKey, node, options)
	if err != nil {
		c.reportError(err)
		return
	}
	
	// 自动连接前驱节点
	if len(c.preNodeKeys) == 0 {
		c.preNodeKeys = append(c.preNodeKeys, START)
	}
	for _, preNodeKey := range c.preNodeKeys {
		e := c.gg.AddEdge(preNodeKey, nodeKey)
		if e != nil {
			c.reportError(e)
			return
		}
	}
	
	// 更新前驱节点列表
	c.preNodeKeys = []string{nodeKey}
}

package main

import (
	"context"
	"fmt"
	
	"github.com/cloudwego/eino/compose"
	"github.com/cloudwego/eino/schema"
)

func chainExample() {
	ctx := context.Background()
	
	// 创建处理器 Lambda
	preprocessor := compose.InvokableLambda(func(ctx context.Context, input string) (map[string]any, error) {
		return map[string]any{
			"query": input,
			"timestamp": time.Now(),
		}, nil
	})
	
	// 创建格式化器 Lambda
	formatter := compose.InvokableLambda(func(ctx context.Context, input map[string]any) (*schema.Message, error) {
		query := input["query"].(string)
		return &schema.Message{
			Role:    schema.Assistant,
			Content: fmt.Sprintf("处理查询: %s", query),
		}, nil
	})
	
	// 构建链
	chain := compose.NewChain[string, *schema.Message]().
		AppendLambda("preprocessor", preprocessor).
		AppendLambda("formatter", formatter)
	
	// 编译
	runnable, err := chain.Compile(ctx)
	if err != nil {
		panic(err)
	}
	
	// 执行
	result, err := runnable.Invoke(ctx, "Hello Eino")
	if err != nil {
		panic(err)
	}
	
	fmt.Printf("结果: %s\n", result.Content)
}

// 并行处理示例
func parallelChainExample() {
	ctx := context.Background()
	
	// 创建并行处理器
	parallel := compose.NewParallel().
		AppendLambda("processor1", compose.InvokableLambda(func(ctx context.Context, input string) (string, error) {
			return "处理器1: " + input, nil
		})).
		AppendLambda("processor2", compose.InvokableLambda(func(ctx context.Context, input string) (string, error) {
			return "处理器2: " + input, nil
		}))
	
	// 创建合并器
	merger := compose.InvokableLambda(func(ctx context.Context, input map[string]any) (string, error) {
		results := make([]string, 0)
		for _, v := range input {
			if str, ok := v.(string); ok {
				results = append(results, str)
			}
		}
		return strings.Join(results, " | "), nil
	})
	
	// 构建链
	chain := compose.NewChain[string, string]().
		AppendParallel(parallel).
		AppendLambda("merger", merger)
	
	// 编译并执行
	runnable, err := chain.Compile(ctx)
	if err != nil {
		panic(err)
	}
	
	result, err := runnable.Invoke(ctx, "测试数据")
	if err != nil {
		panic(err)
	}
	
	fmt.Printf("并行处理结果: %s\n", result)
}

// compose/graph.go

type graph struct {
	nodes        map[string]*graphNode    // 节点映射
	dataEdges    map[string][]string      // 数据流边
	controlEdges map[string][]string      // 控制流边
	branches     map[string][]*GraphBranch // 条件分支
	
	startNodes []string  // 起始节点
	endNodes   []string  // 结束节点
	
	// 类型验证相关
	toValidateMap       map[string][]struct {
		endNode  string
		mappings []*FieldMapping
	}
	fieldMappingRecords map[string][]*FieldMapping
	
	// 编译状态
	compiled   bool
	buildError error
	
	// 类型信息
	expectedInputType  reflect.Type
	expectedOutputType reflect.Type
	genericHelper      *genericHelper
	
	// 组件类型
	cmp component
	
	// 状态管理
	stateType      reflect.Type
	stateGenerator func() any
	
	// 处理器
	handlerOnEdges   map[string]map[string][]handlerPair
	handlerPreNode   map[string][]handlerPair
	handlerPreBranch map[string][][]handlerPair
}

// 添加聊天模型节点
func (g *Graph[I, O]) AddChatModelNode(key string, node model.BaseChatModel, opts ...GraphAddNodeOpt) error {
	gNode, options := toChatModelNode(node, opts...)
	return g.g.addNode(key, gNode, options)
}

// 添加 Lambda 节点
func (g *Graph[I, O]) AddLambdaNode(key string, node *Lambda, opts ...GraphAddNodeOpt) error {
	gNode, options := toLambdaNode(node, opts...)
	return g.g.addNode(key, gNode, options)
}

// 添加子图节点
func (g *Graph[I, O]) AddGraphNode(key string, node AnyGraph, opts ...GraphAddNodeOpt) error {
	gNode, options := toGraphNode(node, opts...)
	return g.g.addNode(key, gNode, options)
}

// 添加直通节点（用于连接多个分支）
func (g *Graph[I, O]) AddPassthroughNode(key string, opts ...GraphAddNodeOpt) error {
	gNode, options := toPassthroughNode(opts...)
	return g.g.addNode(key, gNode, options)
}

// 添加数据流和控制流边
func (g *Graph[I, O]) AddEdge(startNode, endNode string, mappings ...*FieldMapping) error {
	return g.g.addEdgeWithMappings(startNode, endNode, false, false, mappings...)
}

// 添加仅控制流边（不传递数据）
func (g *Graph[I, O]) AddControlEdge(startNode, endNode string) error {
	return g.g.addEdgeWithMappings(startNode, endNode, false, true)
}

// 添加仅数据流边（不控制执行顺序）
func (g *Graph[I, O]) AddDataEdge(startNode, endNode string, mappings ...*FieldMapping) error {
	return g.g.addEdgeWithMappings(startNode, endNode, true, false, mappings...)
}

// 添加条件分支
func (g *Graph[I, O]) AddBranch(startNode string, branch *GraphBranch) error {
	return g.g.addBranch(startNode, branch, false)
}

// compose/field_mapping.go

// FieldMapping 定义字段映射关系
type FieldMapping struct {
	from string  // 源字段路径
	to   string  // 目标字段路径
}

// FieldPath 字段路径类型
type FieldPath []string

// 创建字段映射
func NewFieldMapping(from, to string) *FieldMapping {
	return &FieldMapping{from: from, to: to}
}

// 从字段路径创建映射
func FromFieldPath(from FieldPath) *FieldMapping {
	return &FieldMapping{from: joinFieldPath(from)}
}

// 到字段路径创建映射
func ToFieldPath(to FieldPath) *FieldMapping {
	return &FieldMapping{to: joinFieldPath(to)}
}

// 字段映射应用示例
func applyFieldMappings(input map[string]any, mappings []*FieldMapping) (map[string]any, error) {
	result := make(map[string]any)
	
	for _, mapping := range mappings {
		// 从输入中提取值
		value, err := extractValue(input, mapping.from)
		if err != nil {
			return nil, err
		}
		
		// 设置到目标路径
		err = setValue(result, mapping.to, value)
		if err != nil {
			return nil, err
		}
	}
	
	return result, nil
}

// compose/branch.go

// GraphBranch 图分支定义
type GraphBranch struct {
	condition  func(ctx context.Context, input any) (string, error)  // 分支条件函数
	endNodes   map[string]bool                                       // 目标节点集合
	noDataFlow bool                                                  // 是否传递数据
}

// 创建条件分支
func NewGraphBranch(condition func(ctx context.Context, input any) (string, error)) *GraphBranch {
	return &GraphBranch{
		condition: condition,
		endNodes:  make(map[string]bool),
	}
}

// 添加分支目标
func (gb *GraphBranch) AddTarget(nodeKey string) *GraphBranch {
	gb.endNodes[nodeKey] = true
	return gb
}

// 设置不传递数据
func (gb *GraphBranch) WithNoDataFlow() *GraphBranch {
	gb.noDataFlow = true
	return gb
}

// 分支执行逻辑
func (gb *GraphBranch) execute(ctx context.Context, input any) ([]string, error) {
	target, err := gb.condition(ctx, input)
	if err != nil {
		return nil, err
	}
	
	if gb.endNodes[target] {
		return []string{target}, nil
	}
	
	return nil, fmt.Errorf("branch target %s not found", target)
}

package main

import (
	"context"
	"fmt"
	"strings"
	
	"github.com/cloudwego/eino/compose"
	"github.com/cloudwego/eino/schema"
)

func graphExample() {
	ctx := context.Background()
	
	// 创建图
	graph := compose.NewGraph[string, string]()
	
	// 添加预处理节点
	err := graph.AddLambdaNode("preprocessor", compose.InvokableLambda(func(ctx context.Context, input string) (map[string]any, error) {
		return map[string]any{
			"text": strings.ToLower(input),
			"length": len(input),
		}, nil
	}))
	if err != nil {
		panic(err)
	}
	
	// 添加条件分支
	branch := compose.NewGraphBranch(func(ctx context.Context, input any) (string, error) {
		data := input.(map[string]any)
		length := data["length"].(int)
		if length > 10 {
			return "long_processor", nil
		}
		return "short_processor", nil
	}).AddTarget("long_processor").AddTarget("short_processor")
	
	err = graph.AddBranch("preprocessor", branch)
	if err != nil {
		panic(err)
	}
	
	// 添加长文本处理器
	err = graph.AddLambdaNode("long_processor", compose.InvokableLambda(func(ctx context.Context, input map[string]any) (string, error) {
		text := input["text"].(string)
		return fmt.Sprintf("长文本处理: %s", text), nil
	}))
	if err != nil {
		panic(err)
	}
	
	// 添加短文本处理器
	err = graph.AddLambdaNode("short_processor", compose.InvokableLambda(func(ctx context.Context, input map[string]any) (string, error) {
		text := input["text"].(string)
		return fmt.Sprintf("短文本处理: %s", text), nil
	}))
	if err != nil {
		panic(err)
	}
	
	// 添加合并节点
	err = graph.AddPassthroughNode("merger")
	if err != nil {
		panic(err)
	}
	
	// 连接边
	err = graph.AddEdge("long_processor", "merger")
	if err != nil {
		panic(err)
	}
	
	err = graph.AddEdge("short_processor", "merger")
	if err != nil {
		panic(err)
	}
	
	// 编译并执行
	runnable, err := graph.Compile(ctx)
	if err != nil {
		panic(err)
	}
	
	// 测试短文本
	result1, err := runnable.Invoke(ctx, "Hello")
	if err != nil {
		panic(err)
	}
	fmt.Printf("短文本结果: %s\n", result1)
	
	// 测试长文本
	result2, err := runnable.Invoke(ctx, "This is a very long text for testing")
	if err != nil {
		panic(err)
	}
	fmt.Printf("长文本结果: %s\n", result2)
}

// 字段映射示例
func fieldMappingExample() {
	ctx := context.Background()
	
	graph := compose.NewGraph[map[string]any, map[string]any]()
	
	// 添加数据处理节点
	err := graph.AddLambdaNode("processor", compose.InvokableLambda(func(ctx context.Context, input map[string]any) (map[string]any, error) {
		name := input["user_name"].(string)
		age := input["user_age"].(int)
		
		return map[string]any{
			"processed_name": strings.ToUpper(name),
			"processed_age":  age * 2,
		}, nil
	}))
	if err != nil {
		panic(err)
	}
	
	// 添加格式化节点，使用字段映射
	err = graph.AddLambdaNode("formatter", compose.InvokableLambda(func(ctx context.Context, input map[string]any) (map[string]any, error) {
		return map[string]any{
			"result": fmt.Sprintf("Name: %s, Age: %v", input["name"], input["age"]),
		}, nil
	}))
	if err != nil {
		panic(err)
	}
	
	// 添加带字段映射的边
	err = graph.AddEdge("processor", "formatter",
		compose.NewFieldMapping("processed_name", "name"),
		compose.NewFieldMapping("processed_age", "age"),
	)
	if err != nil {
		panic(err)
	}
	
	// 编译并执行
	runnable, err := graph.Compile(ctx)
	if err != nil {
		panic(err)
	}
	
	input := map[string]any{
		"user_name": "alice",
		"user_age":  25,
	}
	
	result, err := runnable.Invoke(ctx, input)
	if err != nil {
		panic(err)
	}
	
	fmt.Printf("字段映射结果: %v\n", result)
}

// compose/workflow.go

// WorkflowNode 工作流节点
type WorkflowNode struct {
	g                *graph                    // 底层图引用
	key              string                    // 节点键
	addInputs        []func() error            // 待添加的输入函数列表
	staticValues     map[string]any            // 静态值映射
	dependencySetter func(string, dependencyType) // 依赖设置器
	mappedFieldPath  map[string]any            // 映射字段路径
}

// Workflow 是 graph 的包装器，用声明依赖和字段映射替代 AddEdge
// 底层使用 NodeTriggerMode(AllPredecessor)，因此不支持循环
type Workflow[I, O any] struct {
	g                *graph                           // 底层图
	workflowNodes    map[string]*WorkflowNode         // 工作流节点映射
	workflowBranches []*WorkflowBranch               // 工作流分支
	dependencies     map[string]map[string]dependencyType // 依赖关系映射
}

// 依赖类型
type dependencyType int

const (
	normalDependency   dependencyType = iota  // 普通依赖（数据+控制）
	noDirectDependency                        // 无直接依赖（仅控制）
	branchDependency                          // 分支依赖
)

// 创建聊天模型节点
func (wf *Workflow[I, O]) AddChatModelNode(key string, chatModel model.BaseChatModel, opts ...GraphAddNodeOpt) *WorkflowNode {
	_ = wf.g.AddChatModelNode(key, chatModel, opts...)
	return wf.initNode(key)
}

// 创建 Lambda 节点
func (wf *Workflow[I, O]) AddLambdaNode(key string, lambda *Lambda, opts ...GraphAddNodeOpt) *WorkflowNode {
	_ = wf.g.AddLambdaNode(key, lambda, opts...)
	return wf.initNode(key)
}

// 初始化工作流节点
func (wf *Workflow[I, O]) initNode(key string) *WorkflowNode {
	n := &WorkflowNode{
		g:            wf.g,
		key:          key,
		staticValues: make(map[string]any),
		dependencySetter: func(fromNodeKey string, typ dependencyType) {
			if _, ok := wf.dependencies[key]; !ok {
				wf.dependencies[key] = make(map[string]dependencyType)
			}
			wf.dependencies[key][fromNodeKey] = typ
		},
		mappedFieldPath: make(map[string]any),
	}
	wf.workflowNodes[key] = n
	return n
}

// 添加输入依赖（数据流 + 控制流）
func (n *WorkflowNode) AddInput(fromNodeKey string, inputs ...*FieldMapping) *WorkflowNode {
	return n.addDependencyRelation(fromNodeKey, inputs, &workflowAddInputOpts{})
}

// 添加输入依赖（带选项）
func (n *WorkflowNode) AddInputWithOptions(fromNodeKey string, inputs []*FieldMapping, opts ...WorkflowAddInputOpt) *WorkflowNode {
	options := &workflowAddInputOpts{}
	for _, opt := range opts {
		opt(options)
	}
	return n.addDependencyRelation(fromNodeKey, inputs, options)
}

// 添加控制依赖（仅控制流）
func (n *WorkflowNode) AddDependency(fromNodeKey string) *WorkflowNode {
	return n.addDependencyRelation(fromNodeKey, nil, &workflowAddInputOpts{dependencyWithoutInput: true})
}

// 设置静态值
func (n *WorkflowNode) SetStaticValue(path FieldPath, value any) *WorkflowNode {
	pathStr := joinFieldPath(path)
	n.staticValues[pathStr] = value
	return n
}

// 内部依赖关系添加逻辑
func (n *WorkflowNode) addDependencyRelation(fromNodeKey string, inputs []*FieldMapping, opts *workflowAddInputOpts) *WorkflowNode {
	n.addInputs = append(n.addInputs, func() error {
		// 根据选项确定边类型
		noControl := opts.noDirectDependency
		noData := opts.dependencyWithoutInput || len(inputs) == 0
		
		// 添加边到底层图
		err := n.g.addEdgeWithMappings(fromNodeKey, n.key, noControl, noData, inputs...)
		if err != nil {
			return err
		}
		
		// 记录依赖类型
		if opts.noDirectDependency {
			n.dependencySetter(fromNodeKey, noDirectDependency)
		} else if opts.dependencyWithoutInput {
			n.dependencySetter(fromNodeKey, normalDependency)
		} else {
			n.dependencySetter(fromNodeKey, normalDependency)
		}
		
		// 验证和记录字段映射
		if len(inputs) > 0 {
			paths := make([]FieldPath, len(inputs))
			for i, input := range inputs {
				paths[i] = splitFieldPath(input.to)
			}
			return n.checkAndAddMappedPath(paths)
		}
		
		return nil
	})
	
	return n
}