diff --git a/compose/graph_manager.go b/compose/graph_manager.go
index 46df3488e..2681774c8 100644
--- a/compose/graph_manager.go
+++ b/compose/graph_manager.go
@@ -325,6 +325,7 @@ func (t *taskManager) submit(tasks []*task) error {
 			i--
 			t.num++
 			t.done.Send(currentTask)
+			continue
 		}
 
 		t.runningTasks[currentTask.nodeKey] = currentTask
diff --git a/internal/concat.go b/internal/concat.go
index fd9b8abc5..4d8929ac9 100644
--- a/internal/concat.go
+++ b/internal/concat.go
@@ -20,13 +20,15 @@ import (
 	"fmt"
 	"reflect"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/cloudwego/eino/internal/generic"
 )
 
 var (
-	concatFuncs = map[reflect.Type]any{
+	concatFuncsMu sync.RWMutex
+	concatFuncs   = map[reflect.Type]any{
 		generic.TypeOf[string]():        concatStrings,
 		generic.TypeOf[int8]():          useLast[int8],
 		generic.TypeOf[int16]():         useLast[int16],
@@ -69,11 +71,16 @@ func concatStrings(ss []string) (string, error) {
 }
 
 func RegisterStreamChunkConcatFunc[T any](fn func([]T) (T, error)) {
+	concatFuncsMu.Lock()
 	concatFuncs[generic.TypeOf[T]()] = fn
+	concatFuncsMu.Unlock()
 }
 
 func GetConcatFunc(typ reflect.Type) func(reflect.Value) (reflect.Value, error) {
-	if fn, ok := concatFuncs[typ]; ok {
+	concatFuncsMu.RLock()
+	fn, ok := concatFuncs[typ]
+	concatFuncsMu.RUnlock()
+	if ok {
 		return func(a reflect.Value) (reflect.Value, error) {
 			rvs := reflect.ValueOf(fn).Call([]reflect.Value{a})
 			var err error
diff --git a/internal/merge.go b/internal/merge.go
index 6d0f94de9..bf5fb58de 100644
--- a/internal/merge.go
+++ b/internal/merge.go
@@ -19,18 +19,27 @@ package internal
 import (
 	"fmt"
 	"reflect"
+	"sync"
 
 	"github.com/cloudwego/eino/internal/generic"
 )
 
-var mergeFuncs = map[reflect.Type]any{}
+var (
+	mergeFuncsMu sync.RWMutex
+	mergeFuncs   = map[reflect.Type]any{}
+)
 
 func RegisterValuesMergeFunc[T any](fn func([]T) (T, error)) {
+	mergeFuncsMu.Lock()
 	mergeFuncs[generic.TypeOf[T]()] = fn
+	mergeFuncsMu.Unlock()
 }
 
 func GetMergeFunc(typ reflect.Type) func([]any) (any, error) {
-	if fn, ok := mergeFuncs[typ]; ok {
+	mergeFuncsMu.RLock()
+	fn, ok := mergeFuncs[typ]
+	mergeFuncsMu.RUnlock()
+	if ok {
 		return func(vs []any) (any, error) {
 			rvs := reflect.MakeSlice(reflect.SliceOf(typ), 0, len(vs))
 			for _, v := range vs {