eino-compose

编排概览

github.com/cloudwego/eino/compose 包提供三种编排 API：

API	拓扑	循环	类型对齐
Graph	有向图	是（Pregel 模式）/ 否（DAG 模式）	整体输入/输出
Chain	线性序列	否	整体输入/输出
Workflow	DAG	否	字段级映射

*Chain 基于 Pregel 模式下的 Graph 实现，但会强制采用线性拓扑。

这三种 API 都会编译为 Runnable[I, O]，它提供 Invoke、Stream、Collect 和 Transform 方法。

import "github.com/cloudwego/eino/compose"

选择 API

• Chain —— 顺序流水线：提示词 -> 模型 -> 解析器。API 最简单。
• Graph —— 需要分支、循环（ReAct Agent）或扇出/扇入时使用。
• Workflow —— 需要在不同结构体类型的节点之间进行字段级映射时使用；仅支持 DAG。

Graph 快速参考

g := compose.NewGraph[InputType, OutputType]()

// Add nodes
g.AddChatModelNode("model", chatModel)
g.AddChatTemplateNode("tmpl", tmpl)
g.AddToolsNode("tools", toolsNode)
g.AddLambdaNode("fn", compose.InvokableLambda(myFunc))
g.AddPassthroughNode("pass")
g.AddGraphNode("sub", subGraph)

// Connect nodes
g.AddEdge(compose.START, "tmpl")
g.AddEdge("tmpl", "model")
g.AddEdge("model", compose.END)

// Branch (conditional routing)
branch := compose.NewGraphBranch(conditionFn, map[string]bool{"a": true, "b": true})
g.AddBranch("model", branch)

// Compile and run
r, err := g.Compile(ctx)
out, err := r.Invoke(ctx, input)

Chain 快速参考

chain := compose.NewChain[InputType, OutputType]()
chain.
    AppendChatTemplate(tmpl).
    AppendChatModel(model).
    AppendLambda(compose.InvokableLambda(parseFn))

r, err := chain.Compile(ctx)
out, err := r.Invoke(ctx, input)

追加方法：AppendChatModel、AppendChatTemplate、AppendToolsNode、AppendAgenticToolsNode、AppendLambda、AppendGraph、AppendParallel、AppendBranch、AppendPassthrough、AppendRetriever、AppendEmbedding、AppendLoader、AppendIndexer、AppendDocumentTransformer。

Workflow 快速参考

wf := compose.NewWorkflow[InputStruct, OutputStruct]()

wf.AddLambdaNode("node1", compose.InvokableLambda(fn1)).
    AddInput(compose.START, compose.MapFields("FieldA", "InputField"))

wf.AddLambdaNode("node2", compose.InvokableLambda(fn2)).
    AddInput("node1", compose.ToField("Result"))

wf.End().AddInput("node2")

r, err := wf.Compile(ctx)

字段映射辅助方法：MapFields、ToField、FromField、MapFieldPaths、ToFieldPath、FromFieldPath。

流式编程

Runnable[I, O] 的四种交互模式：

模式	输入	输出	Lambda 构造函数
Invoke	I	O	compose.InvokableLambda
Stream	I	*StreamReader[O]	compose.StreamableLambda
Collect	*StreamReader[I]	O	compose.CollectableLambda
Transform	*StreamReader[I]	*StreamReader[O]	compose.TransformableLambda

框架会在不同模式之间自动转换：

• Invoke 调用：所有内部节点都以 Invoke 模式运行。
• Stream/Collect/Transform 调用：所有内部节点都以 Transform 模式运行；缺少的模式会自动补齐。

流式处理基础组件位于 github.com/cloudwego/eino/schema：

sr, sw := schema.Pipe[T](capacity)
// sw.Send(chunk, nil); sw.Close()
// chunk, err := sr.Recv(); sr.Close()

AgenticToolsNode

AgenticToolsNode 是 ToolsNode 的 Agentic 对应版本。它操作的是 *schema.AgenticMessage，而非 *schema.Message，会提取 FunctionToolCall 内容块，并返回 FunctionToolResult 内容块。

import "github.com/cloudwego/eino/compose"

// Create from the same ToolsNodeConfig
agenticToolsNode, err := compose.NewAgenticToolsNode(ctx, &compose.ToolsNodeConfig{
    Tools: []tool.BaseTool{myTool1, myTool2},
})

// Use in Graph
g := compose.NewGraph[*schema.AgenticMessage, []*schema.AgenticMessage]()
g.AddAgenticToolsNode("tools", agenticToolsNode)

// Use in Chain
chain := compose.NewChain[*schema.AgenticMessage, []*schema.AgenticMessage]()
chain.AppendAgenticToolsNode(agenticToolsNode)

与 ToolsNode 的主要区别：

• 输入：*schema.AgenticMessage（读取 ContentBlockTypeFunctionToolCall 块）
• 输出：[]*schema.AgenticMessage（写入 ContentBlockTypeFunctionToolResult 块）
• 支持多模态工具结果（文本、图像、音频、视频、文件）
• 支持处理用于动态工具发现的 ToolSearchFunctionToolResult
• 复用相同的 ToolsNodeConfig——所有现有工具均可照常使用

编译与运行

r, err := g.Compile(ctx,
    compose.WithGraphName("my_graph"),
    compose.WithNodeTriggerMode(compose.AllPredecessor), // DAG mode
)

// Non-streaming
out, err := r.Invoke(ctx, input)

// Streaming
stream, err := r.Stream(ctx, input)
defer stream.Close()
for {
    chunk, err := stream.Recv()
    if err == io.EOF { break }
    if err != nil { return err }
    process(chunk)
}

状态图

在单个请求内的节点之间共享状态：

g := compose.NewGraph[string, string](compose.WithGenLocalState(func(ctx context.Context) *MyState {
    return &MyState{}
}))

g.AddLambdaNode("node", lambda,
    compose.WithStatePreHandler(func(ctx context.Context, in string, state *MyState) (string, error) {
        // read/write state before node executes
        return in, nil
    }),
    compose.WithStatePostHandler(func(ctx context.Context, out string, state *MyState) (string, error) {
        // read/write state after node executes
        return out, nil
    }),
)

给 Agent 的说明

帮助用户构建编排时：

1. 大多数情况下默认使用 Graph。仅在简单的线性流水线中使用 Chain。需要在不同结构体类型之间进行字段级映射时使用 Workflow。
2. 始终展示 Compile 步骤——g.Compile(ctx) 会返回 Runnable[I,O]。
3. 始终关闭 StreamReader——获取流后立即使用 defer sr.Close()。
4. 上游输出类型必须与下游输入类型匹配（或使用 WithInputKey/WithOutputKey 转换 map）。
5. 对于循环图（例如 ReAct Agent），使用默认 Pregel 模式（AnyPredecessor）。对于 DAG，设置 AllPredecessor。
6. 使用 compose.WithCallbacks(handler) 在运行时注入日志记录/跟踪。
7. 对于暂停/恢复工作流，结合中断节点使用 compose.WithCheckPointStore(store)。

参考文件

按需阅读以下文件，了解详细 API、示例和高级用法：

• reference/graph.md —— 完整的 Graph API、分支、状态图、循环图和完整的 ReAct 示例
• reference/chain.md —— Chain API、适用场景、链中的并行和分支
• reference/workflow.md —— Workflow API、字段级映射辅助工具和约束
• reference/stream.md —— StreamReader/Writer、Pipe/Copy/Merge、lambda 构造函数和自动转换规则
• reference/callback.md —— Callback 时序、处理程序注册、触发规则和追踪示例
• reference/call-option.md —— 按请求设置的 CallOption、组件类型选项和节点定位
• reference/checkpoint-and-state.md —— CheckPointStore、中断/恢复和状态管理