eino-compose

オーケストレーションの概要

github.com/cloudwego/eino/compose パッケージは、3 つのオーケストレーション API を提供します。

API	トポロジー	サイクル	型の整合性
Graph	有向グラフ	あり（Pregel モード）／なし（DAG モード）	入出力全体
Chain	線形シーケンス	なし	入出力全体
Workflow	DAG	なし	フィールドレベルのマッピング

*Chain は Graph の Pregel モード上に実装されていますが、トポロジーは線形に制限されます。

3 つすべてが Runnable[I, O] にコンパイルされ、Invoke、Stream、Collect、Transform を提供します。

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

API の選び方

• Chain -- 順次実行するパイプライン：プロンプト -> モデル -> パーサー。最もシンプルな API です。
• Graph -- 分岐、ループ（ReAct エージェント）、またはファンアウト／ファンインが必要な場合に使用します。
• 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] には 4 つのインタラクションモードがあります。

モード	入力	出力	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 に対応するエージェント型のノードです。*schema.Message ではなく *schema.AgenticMessage を処理し、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)
}

State Graph

単一のリクエスト内でノード間の状態を共有します。

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
    }),
)

エージェントへの指示

オーケストレーションの構築をユーザーに支援する場合：

1. ほとんどのユースケースでは、デフォルトで Graph を使用します。単純な線形パイプラインには Chain を使用します。異なる構造体型間のフィールドレベルのマッピングが必要な場合は Workflow を使用します。
2. 常に Compile の手順を示します。g.Compile(ctx) は Runnable[I,O] を返します。
3. 常に StreamReader をクローズします。ストリームを取得したら、すぐに defer sr.Close() を使用します。
4. 上流の出力型は下流の入力型と一致させます（一致しない場合は、マップ変換に WithInputKey／WithOutputKey を使用します）。
5. サイクルのあるグラフ（例：ReAct エージェント）では、デフォルトの Pregel モード（AnyPredecessor）を使用します。DAG では AllPredecessor を設定します。
6. compose.WithCallbacks(handler) を使用して、実行時にロギング／トレーシングを注入します。
7. 中断ノードとともに compose.WithCheckPointStore(store) を使用して、一時停止／再開できるワークフローを構築します。

参考ファイル

API、例、応用的な使い方の詳細については、必要に応じて以下のファイルを参照してください。

• reference/graph.md -- Graph API、ブランチ、ステートグラフ、循環グラフ、完全なReActの例
• reference/chain.md -- Chain API、使用する場面、Chainでの並列処理とブランチ
• reference/workflow.md -- Workflow API、フィールド単位のマッピングヘルパー、制約
• reference/stream.md -- StreamReader/Writer、Pipe/Copy/Merge、ラムダコンストラクター、自動変換ルール
• reference/callback.md -- コールバックのタイミング、ハンドラーの登録、トリガールール、トレースの例
• reference/call-option.md -- リクエスト単位のCallOption、コンポーネントタイプごとのオプション、ノードの指定
• reference/checkpoint-and-state.md -- CheckPointStore、中断と再開、状態管理