编排框架（Compose）

Eino

编排框架（Compose）

compose 包将组件连接为可执行的图，自动管理类型检查、流式处理、并发调度与 Callback 注入。它是 Eino 的"编译器 + 运行时"。

三层 API（由简到强）

┌──────────────────────────────────────────────────────────┐
│  Workflow API                                            │
│  字段级映射 · 控制流/数据流分离 · 无环 · 默认并行        │
│  ┌──────────────────────────────────────────────────┐   │
│  │  Chain API                                       │   │
│  │  纯线性管道 · 内部封装 Graph · Appender 链式构建  │   │
│  │  ┌──────────────────────────────────────────┐   │   │
│  │  │  Graph API                               │   │   │
│  │  │  全功能有向图 · 分支/并行/环 · 状态管理  │   │   │
│  │  └──────────────────────────────────────────┘   │   │
│  └──────────────────────────────────────────────────┘   │
└──────────────────────────────────────────────────────────┘

API	环（循环）	并行	数据映射	典型场景
Graph	✔	✔	全量传递	ReAct Agent、复杂 DAG
Chain	✗	✔（`AppendParallel`）	全量传递	简单线性管道
Workflow	✗	✔（默认）	字段级	多源数据聚合

Graph API — 全功能有向图

核心概念

Graph 由节点（Node）和边（Edge）组成。编译后变为 Runnable[I, O]。

    START
      │
      ▼
  ┌───────┐    正向边    ┌───────┐
  │ 验证   │──────────→│  LLM  │
  └───────┘             └───┬───┘
                            │
              ┌─────────────┼─────────────┐
              │ 分支：根据 LLM 输出路由    │
              ▼                           ▼
        ┌──────────┐               ┌──────────┐
        │  Tools   │               │   END    │
        └────┬─────┘               └──────────┘
             │ 工具结果回传（形成 ReAct 环）
             ▼
        ┌──────────┐
        │   LLM    │ ← 回到 LLM 节点（带工具结果）
        └──────────┘

创建与编译

// Graph[Input, Output] — 编译期类型安全
graph := compose.NewGraph[*Input, *Output]()

// 添加节点
graph.AddChatModelNode("llm", chatModel)
graph.AddToolsNode("tools", toolsNode)
graph.AddLambdaNode("validate", compose.InvokableLambda(validateFn))
graph.AddRetrieverNode("retrieve", retriever)
graph.AddGraphNode("sub", subGraph)       // 嵌套子图

// 边：控制流 = 数据流（上游输出直接给下游输入）
graph.AddEdge(compose.START, "validate")
graph.AddEdge("validate", "llm")

// 分支：动态路由（构建 ReAct 环的核心）
graph.AddBranch("llm", compose.NewGraphBranch(
    func(ctx context.Context, msg *schema.Message) (string, error) {
        if len(msg.ToolCalls) > 0 {
            return "tools", nil    // 模型要求调工具
        }
        return compose.END, nil    // 模型直接回复
    },
    map[string]bool{"tools": true, compose.END: true},
))
graph.AddEdge("tools", "llm")   // 工具结果回传 LLM

// 编译
runnable, err := graph.Compile(ctx,
    compose.WithGraphName("react_agent"),
    compose.WithMaxRunSteps(100), // 防止无限循环
)

四种调用模式

Invoke:   Input ──────────────────────────→ Output         （一次性）
Stream:   Input ──────────────────────────→ Stream[Output] （入一出流）
Transform: Stream[Input] ─────────────────→ Stream[Output] （入流出流）
Collect:  Stream[Input] ──────────────────→ Output         （入流一出）

// 同步
result, _ := runnable.Invoke(ctx, input)

// 流式输出（打字机效果）
stream, _ := runnable.Stream(ctx, input)
defer stream.Close()

// 流 → 流（管道模式，上游流式 → 下游流式）
outStream, _ := runnable.Transform(ctx, inStream)

// 流 → 值（等待全部 chunk → 一次性返回）
result, _ := runnable.Collect(ctx, inStream)

🔬 框架根据节点能力自动插入适配器。比如上游只支持 Invoke（同步），下游要 Stream（流式），框架自动把同步输出装箱为单元素流。

节点触发模式：Pregel vs DAG

// DAG 模式（默认）：所有前驱完成才触发
graph.AddEdge("A", "C")
graph.AddEdge("B", "C")  // C 等 A 和 B 都完成

// Pregel 模式：任意一个前驱完成就触发（适合开发中的迭代式逻辑）
graph := compose.NewGraph[I, O](compose.WithPregelMode())

DAG (AllPredecessor):              Pregel (AnyPredecessor):
    A ──┐                              A ──┐
        ├──→ C（等 A 和 B）                ╞══→ C（A 或 B 任一完成即触发）
    B ──┘                              B ──┘

图中共享状态（State）

每轮 Run 独立的、跨节点共享的 KV + 结构体状态：

type myState struct {
    Round   int
    Msgs    []*schema.Message
}

graph := compose.NewGraph[[]*schema.Message, *schema.Message](
    compose.WithGenLocalState(func(ctx context.Context) *myState {
        return &myState{Round: 0}
    }),
)

// 节点执行前：修改输入 + 读写状态
graph.AddChatModelNode("llm", model,
    compose.WithStatePreHandler(func(ctx context.Context, in []*schema.Message, state *myState) ([]*schema.Message, error) {
        state.Round++
        log.Printf("第 %d 轮模型调用", state.Round)
        return in, nil
    }),
)

// 节点执行后：读输出 + 写状态
graph.AddLambdaNode("log", compose.InvokableLambda(logFn),
    compose.WithStatePostHandler(func(ctx context.Context, out string, state *myState) (string, error) {
        state.Msgs = append(state.Msgs, schema.UserMessage(out))
        return out, nil
    }),
)

分支中访问状态

graph.AddBranch("router", compose.NewGraphMultiBranch(
    func(ctx context.Context, in string) (map[string]bool, error) {
        if shouldContinue {
            return map[string]bool{"next": true}, nil
        }
        return map[string]bool{compose.END: true}, nil
    },
    map[string]bool{"next": true, compose.END: true},
))

完整示例：含状态的 ReAct 图

type agentState struct {
    iteration int
    maxIter   int
}

func buildReActGraph(model model.BaseChatModel, tools []tool.BaseTool) (compose.Runnable[[]*schema.Message, *schema.Message], error) {
    graph := compose.NewGraph[[]*schema.Message, *schema.Message](
        compose.WithGenLocalState(func(ctx context.Context) *agentState {
            return &agentState{maxIter: 20}
        }),
    )

    // 工具节点
    toolsNode, err := compose.NewToolNode(ctx, &compose.ToolsNodeConfig{Tools: tools})
    if err != nil {
        return nil, err
    }

    // 添加 LLM 节点（带状态预处理：注入 system prompt）
    graph.AddChatModelNode("model", model,
        compose.WithStatePreHandler(func(ctx context.Context, in []*schema.Message, state *agentState) ([]*schema.Message, error) {
            state.iteration++
            if state.iteration > state.maxIter {
                return nil, fmt.Errorf("超过最大迭代次数 %d", state.maxIter)
            }
            // 首次调用时在前面插入 system prompt
            if state.iteration == 1 {
                return append([]*schema.Message{schema.SystemMessage("你是有用AI助手")}, in...), nil
            }
            return in, nil
        }),
    )
    graph.AddToolsNode("tools", toolsNode)
    graph.AddEdge(compose.START, "model")

    // 分支：模型返回 ToolCall → tools；否则 → END
    graph.AddBranch("model", compose.NewGraphBranch(
        func(ctx context.Context, msg *schema.Message) (string, error) {
            if len(msg.ToolCalls) > 0 {
                return "tools", nil
            }
            return compose.END, nil
        },
        map[string]bool{"tools": true, compose.END: true},
    ))
    graph.AddEdge("tools", "model") // 工具结果回传形成环

    return graph.Compile(context.Background(),
        compose.WithGraphName("custom_react"),
        compose.WithMaxRunSteps(100),
    )
}

Chain API — 线性管道

最简洁的 API，适合无分支的直线流程。

Chain 运行模型：
  START → Prompt → ChatModel → Lambda(format) → END

chain, err := compose.NewChain[map[string]any, *schema.Message]().
    AppendChatTemplate(template).
    AppendChatModel(model).
    AppendLambda(compose.InvokableLambda(func(ctx context.Context, msg *schema.Message) (string, error) {
        return msg.Content, nil
    })).
    Compile(ctx)

result, err := chain.Invoke(ctx, map[string]any{"topic": "Go"})

AppendParallel — Chain 内并行

                  ┌─→ translateToEnglish ─┐
  START → prompt ─┼─→ translateToFrench  ─┼─→ merge → END
                  └─→ translateToChinese ─┘

chain, _ := compose.NewChain[string, []string]().
    AppendLambda(splitFn).
    AppendParallel(
        compose.NewChain[string, string]().
            AppendChatModel(enModel).Compile(ctx),
        compose.NewChain[string, string]().
            AppendChatModel(frModel).Compile(ctx),
        compose.NewChain[string, string]().
            AppendChatModel(cnModel).Compile(ctx),
    ).
    AppendLambda(mergeFn).
    Compile(ctx)

💡 适合同一输入并发交给多个下游处理的场景（翻译、多维度分析、A/B 测试）。

Workflow API — 控制流与数据流分离

最精细的 API。节点间"谁先谁后"（控制）和"哪个字段给谁"（数据）独立声明。

基础

wf := compose.NewWorkflow[int, string]()

wf.AddLambdaNode("step1", compose.InvokableLambda(fn1)).
    AddInput(compose.START)               // step1 从 START 取全部数据

wf.AddLambdaNode("step2", compose.InvokableLambda(fn2)).
    AddInput("step1")                      // step2 从 step1 取全部数据

wf.End().AddInput("step2")                // step2 的全部输出 → END

run, _ := wf.Compile(ctx)
run.Invoke(ctx, 42)

字段级映射

START ──→ step1 ──→ step2 ──→ END
                   ↑
                   │  (仅取 step1.Result 字段，映射到 step2.InputData)

type Step1Output struct {
    Status  string
    Result  string
    Code    int
}
type Step2Input struct {
    InputData string
    Context   string
}

wf.AddLambdaNode("step2", step2Fn).
    AddInput("step1",
        compose.MapField("Result", "InputData"),    // step1.Result → step2.InputData
        compose.MapField("Status", "Context"),       // step1.Status → step2.Context
    )
// step1.Code 不会被传入 step2

仅控制依赖（不传数据）

wf.AddLambdaNode("step3", step3Fn).
    AddDependency("step2")              // step2 完成后才执行，但不接收其数据
    AddInput(compose.START)             // 数据直接从 START 取

Graph vs Workflow 对比

Graph:                          Workflow:
  边 = 控制 + 数据（一体）        边 = 控制（何时执行）
  A ──(data)──→ B               AddInput = 数据（取什么字段）
                                 节点输入 = 从多个上游"挑选字段"组合

流式处理自动化

编排框架在编译期分析上下游能力，自动插入转换适配器：

场景 1：上游 Streamable、下游仅 Invokable
  Stream[Chunk] ──→ [框架自动 Collect] ──→ 完整值 ──→ Invokable Node

场景 2：上游 Invokable、下游需 Streamable
  值 ──→ [框架自动装箱为单元素流] ──→ Stream[Value] ──→ Streamable Node

场景 3：Copy 一对多
  StreamReader ──→ Copy(3) ──→ ┌─ Reader₁ → Consumer A
                                ├─ Reader₂ → Consumer B
                                └─ Reader₃ → Consumer C

Node 类型速查

Add 方法	对应组件	说明
`AddChatModelNode`	ChatModel	LLM 调用
`AddToolsNode`	ToolsNode	工具执行
`AddRetrieverNode`	Retriever	文档检索
`AddLambdaNode`	Lambda	任意 Go 函数
`AddGraphNode`	Nested Graph	子图复用
`AddEmbeddingNode`	Embedder	向量化
`AddIndexerNode`	Indexer	文档索引入库
`AddDocumentTransformerNode`	Transformer	文档拆分/过滤
`AddLoaderNode`	Loader	文档加载
`AddBranch`	Branch	条件路由

常见陷阱

陷阱	说明
🚨 环内无退出条件	分支必须保留通往 `END` 的路径，否则无限循环（可用 `WithMaxRunSteps` 兜底）
🚨 Stream 未 Copy 就并发读	一个 StreamReader 只能一个消费者，多消费者需先 `Copy(n)`
🚨 Graph 类型不匹配	编译期校验，上游 Output 必须 = 下游 Input 类型；不兼容时报编译错误
🚨 Workflow 中回环	Workflow 仅支持无环图，需要循环场景用 Graph API
🚨 State PreHandler 中修改了 input 但未返回	返回值是新 input，被修改的 input 必须放在 return 中
🚨 大图直接 Add 不检查 err	`AddXXXNode` 返回的 error 累积，到 `Compile()` 时统一返回；每步都检查可提前定位

Schema 与消息类型 ADK 核心：Agent / Runner / Tur...