feat: revamp Instill Format (#774)

Because
- Instill Format needs a more flexible, user-friendly way to handle
diverse data types and conversions.
- Binary data support and format standardization improve functionality
and allow smoother developer and user experiences.

This commit
- Introduces support for unstructured data formats, with auto-conversion
for image, audio, and video codec.
 - Enables data attributes access in file-type data.
- Adds support for default values in the variable section, allowing
fields to be optionally omitted in requests if a default is set.
- Adds URL-based file upload support with a flexible workflow for base64
and URL uploads.
- Separates data structs and format interfaces, ensuring Instill Format
effectively differentiates data content from usage intent.
- Updates components (openai, anthropic, mistral, text, document, image,
collection) to align with the new interface structures.

.github/workflows/coverage.yml

@@ -22,8 +22,6 @@ jobs:
         ports:
           - 5432:5432
     steps:
-      - name: Install operator dependencies
-        run: sudo apt-get update -y && sudo apt-get install poppler-utils wv unrtf tidy tesseract-ocr libtesseract-dev -y
 
       - uses: actions/checkout@v3
         with:
@@ -42,7 +40,7 @@ jobs:
 
       - name: Generate coverage report
         run: |
-          go mod tidy
+          make build-dev
           make coverage DBTEST=true OCR=true
 
       - name: Upload coverage report

.golangci.yaml

@@ -5,3 +5,5 @@ linters:
 linters-settings:
   stylecheck:
     checks: [ "all" ]
+  staticcheck:
+    checks: [ "all", "-SA1019" ]

Dockerfile.dev

@@ -21,6 +21,7 @@ RUN --mount=target=. --mount=type=cache,target=/root/.cache/go-build --mount=typ
 
 # k6
 RUN go install go.k6.io/xk6/cmd/xk6@v${XK6_VERSION}
+RUN go install github.com/mfridman/[email protected]
 RUN xk6 build v${K6_VERSION} --with github.com/grafana/xk6-sql --output /usr/bin/k6
 
 # -- set up Go

Makefile

@@ -76,19 +76,40 @@ dbtest-pre:
 .PHONY: coverage
 coverage:
 	@if [ "${DBTEST}" = "true" ]; then  make dbtest-pre; fi
-	@${GOTEST_FLAGS} go test -v -race ${GOTEST_TAGS} -coverpkg=./... -coverprofile=coverage.out -covermode=atomic ./...
+	@docker run --rm \
+		-v $(PWD):/${SERVICE_NAME} \
+		-e GOTEST_FLAGS="${GOTEST_FLAGS}" \
+		--user $(id -u):$(id -g) \
+		--entrypoint= \
+		instill/${SERVICE_NAME}:dev \
+			go test -v -race ${GOTEST_TAGS} -coverpkg=./... -coverprofile=coverage.out -covermode=atomic -timeout 30m ./...
 	@if [ "${HTML}" = "true" ]; then  \
-		go tool cover -func=coverage.out && \
-		go tool cover -html=coverage.out && \
-		rm coverage.out; \
+		docker run --rm \
+			-v $(PWD):/${SERVICE_NAME} \
+			--user $(id -u):$(id -g) \
+			--entrypoint= \
+			instill/${SERVICE_NAME}:dev \
+				go tool cover -func=coverage.out && \
+				go tool cover -html=coverage.out && \
+				rm coverage.out; \
 	fi
 
 .PHONY: test
 test:
 	@if [ "${OCR}" = "true" ]; then \
-		make test-ocr; \
+		docker run --rm \
+			-v $(PWD):/${SERVICE_NAME} \
+			--user $(id -u):$(id -g) \
+			--entrypoint= \
+			instill/${SERVICE_NAME}:dev \
+				make test-ocr; \
 	else \
-		go test -v ./... -json | tparse --notests --all; \
+		docker run --rm \
+			-v $(PWD):/${SERVICE_NAME} \
+			--user $(id -u):$(id -g) \
+			--entrypoint= \
+			instill/${SERVICE_NAME}:dev \
+				go test -v ./... -json | tparse --notests --all;  \
 	fi
 
 .PHONY: test-ocr

go.mod

@@ -60,7 +60,6 @@ require (
 	github.com/openfga/api/proto v0.0.0-20240318145204-66b9e5cb403c
 	github.com/pkoukk/tiktoken-go v0.1.7
 	github.com/redis/go-redis/v9 v9.5.1
-	github.com/samber/lo v1.47.0
 	github.com/sijms/go-ora v1.3.2
 	github.com/slack-go/slack v0.12.5
 	github.com/tmc/langchaingo v0.1.10

go.sum

@@ -1776,8 +1776,6 @@ github.com/safchain/ethtool v0.0.0-20210803160452-9aa261dae9b1/go.mod h1:Z0q5wiB
 github.com/saintfish/chardet v0.0.0-20120816061221-3af4cd4741ca/go.mod h1:uugorj2VCxiV1x+LzaIdVa9b4S4qGAcH6cbhh4qVxOU=
 github.com/saintfish/chardet v0.0.0-20230101081208-5e3ef4b5456d h1:hrujxIzL1woJ7AwssoOcM/tq5JjjG2yYOc8odClEiXA=
 github.com/saintfish/chardet v0.0.0-20230101081208-5e3ef4b5456d/go.mod h1:uugorj2VCxiV1x+LzaIdVa9b4S4qGAcH6cbhh4qVxOU=
-github.com/samber/lo v1.47.0 h1:z7RynLwP5nbyRscyvcD043DWYoOcYRv3mV8lBeqOCLc=
-github.com/samber/lo v1.47.0/go.mod h1:RmDH9Ct32Qy3gduHQuKJ3gW1fMHAnE/fAzQuf6He5cU=
 github.com/santhosh-tekuri/jsonschema/v5 v5.3.1 h1:lZUw3E0/J3roVtGQ+SCrUrg3ON6NgVqpn3+iol9aGu4=
 github.com/santhosh-tekuri/jsonschema/v5 v5.3.1/go.mod h1:uToXkOrWAZ6/Oc07xWQrPOhJotwFIyu2bBVN41fcDUY=
 github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=

pkg/component/ai/anthropic/v0/io.go

@@ -0,0 +1,38 @@
+package anthropic
+
+type MessagesInput struct {
+	ChatHistory  []ChatMessage `instill:"chat-history"`
+	MaxNewTokens int           `instill:"max-new-tokens"`
+	ModelName    string        `instill:"model-name"`
+	Prompt       string        `instill:"prompt"`
+	PromptImages []string      `instill:"prompt-images"`
+	Seed         int           `instill:"seed"`
+	SystemMsg    string        `instill:"system-message"`
+	Temperature  float32       `instill:"temperature"`
+	TopK         int           `instill:"top-k"`
+}
+
+type ChatMessage struct {
+	Role    string              `instill:"role"`
+	Content []MultiModalContent `instill:"content"`
+}
+
+type MultiModalContent struct {
+	ImageURL URL    `instill:"image-url"`
+	Text     string `instill:"text"`
+	Type     string `instill:"type"`
+}
+
+type URL struct {
+	URL string `instill:"url"`
+}
+
+type MessagesOutput struct {
+	Text  string        `instill:"text"`
+	Usage messagesUsage `instill:"usage"`
+}
+
+type messagesUsage struct {
+	InputTokens  int `instill:"input-tokens"`
+	OutputTokens int `instill:"output-tokens"`
+}

pkg/component/ai/anthropic/v0/main.go

@@ -3,20 +3,19 @@ package anthropic
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
 	"slices"
 	"strings"
 	"sync"
 
 	_ "embed"
 
-	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/types/known/structpb"
 
 	anthropicsdk "github.com/anthropics/anthropic-sdk-go"
 
 	"github.com/instill-ai/pipeline-backend/pkg/component/base"
+	"github.com/instill-ai/pipeline-backend/pkg/component/internal/util"
 )
 
 const (
@@ -75,43 +74,6 @@ type messagesReq struct {
 	TopP          float32     `json:"top_p,omitempty"`
 }
 
-type MessagesInput struct {
-	ChatHistory  []ChatMessage `json:"chat-history"`
-	MaxNewTokens int           `json:"max-new-tokens"`
-	ModelName    string        `json:"model-name"`
-	Prompt       string        `json:"prompt"`
-	PromptImages []string      `json:"prompt-images"`
-	Seed         int           `json:"seed"`
-	SystemMsg    string        `json:"system-message"`
-	Temperature  float32       `json:"temperature"`
-	TopK         int           `json:"top-k"`
-}
-
-type ChatMessage struct {
-	Role    string              `json:"role"`
-	Content []MultiModalContent `json:"content"`
-}
-
-type MultiModalContent struct {
-	ImageURL URL    `json:"image-url"`
-	Text     string `json:"text"`
-	Type     string `json:"type"`
-}
-
-type URL struct {
-	URL string `json:"url"`
-}
-
-type MessagesOutput struct {
-	Text  string        `json:"text"`
-	Usage messagesUsage `json:"usage"`
-}
-
-type messagesUsage struct {
-	InputTokens  int `json:"input-tokens"`
-	OutputTokens int `json:"output-tokens"`
-}
-
 type message struct {
 	Role    string    `json:"role"`
 	Content []content `json:"content"`
@@ -149,7 +111,7 @@ func Init(bc base.Component) *component {
 type execution struct {
 	base.ComponentExecution
 
-	execute                func(*structpb.Struct, *base.Job, context.Context) (*structpb.Struct, error)
+	execute                func(context.Context, *base.Job) error
 	client                 *anthropicsdk.Client
 	usesInstillCredentials bool
 }
@@ -217,33 +179,15 @@ func (e *execution) Execute(ctx context.Context, jobs []*base.Job) error {
 	return base.ConcurrentExecutor(ctx, jobs, e.execute)
 }
 
-func (e *execution) generateText(_ *structpb.Struct, job *base.Job, ctx context.Context) (*structpb.Struct, error) {
-
-	input, err := job.Input.Read(ctx)
-	if err != nil {
-		job.Error.Error(ctx, err)
-		return nil, err
-	}
+func (e *execution) generateText(ctx context.Context, job *base.Job) error {
 
 	var inputStruct MessagesInput
-	err = base.ConvertFromStructpb(input, &inputStruct)
+	err := job.Input.ReadData(ctx, &inputStruct)
 	if err != nil {
 		job.Error.Error(ctx, err)
-		return nil, err
+		return err
 	}
 
-	// type MessagesInput struct {
-	// 	ChatHistory  []ChatMessage `json:"chat-history"`
-	// 	MaxNewTokens int           `json:"max-new-tokens"`
-	// 	ModelName    string        `json:"model-name"`
-	// 	Prompt       string        `json:"prompt"`
-	// 	PromptImages []string      `json:"prompt-images"`
-	// 	Seed         int           `json:"seed"`
-	// 	SystemMsg    string        `json:"system-message"`
-	// 	Temperature  float32       `json:"temperature"`
-	// 	TopK         int           `json:"top-k"`
-	// }
-
 	messageParams := anthropicsdk.MessageNewParams{
 		Model:     anthropicsdk.F(inputStruct.ModelName),
 		MaxTokens: anthropicsdk.Int(int64(inputStruct.MaxNewTokens)),
@@ -284,9 +228,9 @@ func (e *execution) generateText(_ *structpb.Struct, job *base.Job, ctx context.
 		// check if the image extension is supported
 		if !slices.Contains(supportedImageExtensions, extension) {
 			job.Error.Error(ctx, err)
-			return nil, fmt.Errorf("unsupported image extension, expected one of: %v , got %s", supportedImageExtensions, extension)
+			return fmt.Errorf("unsupported image extension, expected one of: %v , got %s", supportedImageExtensions, extension)
 		}
-		blocks = append(blocks, anthropicsdk.NewImageBlockBase64(fmt.Sprintf("image/%s", extension), base.TrimBase64Mime(image)))
+		blocks = append(blocks, anthropicsdk.NewImageBlockBase64(fmt.Sprintf("image/%s", extension), util.TrimBase64Mime(image)))
 	}
 	messages = append(messages, anthropicsdk.NewUserMessage(blocks...))
 	messageParams.Messages = anthropicsdk.F(messages)
@@ -308,35 +252,19 @@ func (e *execution) generateText(_ *structpb.Struct, job *base.Job, ctx context.
 		err = message.Accumulate(event)
 		if err != nil {
 			job.Error.Error(ctx, err)
-			return nil, err
+			return err
 		}
 
 		switch delta := event.Delta.(type) {
 		case anthropicsdk.ContentBlockDeltaEventDelta:
 
 			if delta.Text != "" {
-				fmt.Println("delta.Text")
-				fmt.Println(delta.Text)
 				text += delta.Text
 				outputStruct.Text = text
-
-				output := &structpb.Struct{}
-				outputJSON, err := json.Marshal(outputStruct)
+				err = job.Output.WriteData(ctx, outputStruct)
 				if err != nil {
 					job.Error.Error(ctx, err)
-					return nil, err
-				}
-
-				err = protojson.Unmarshal(outputJSON, output)
-				if err != nil {
-					job.Error.Error(ctx, err)
-					return nil, err
-				}
-
-				err = job.Output.Write(ctx, output)
-				if err != nil {
-					job.Error.Error(ctx, err)
-					return nil, err
+					return err
 				}
 			}
 		}
@@ -345,24 +273,11 @@ func (e *execution) generateText(_ *structpb.Struct, job *base.Job, ctx context.
 	outputStruct.Usage.InputTokens = int(message.Usage.InputTokens)
 	outputStruct.Usage.OutputTokens = int(message.Usage.OutputTokens)
 
-	outputJSON, err := json.Marshal(outputStruct)
+	err = job.Output.WriteData(ctx, outputStruct)
 	if err != nil {
 		job.Error.Error(ctx, err)
-		return nil, err
-	}
-
-	output := &structpb.Struct{}
-	err = protojson.Unmarshal(outputJSON, output)
-	if err != nil {
-		job.Error.Error(ctx, err)
-		return nil, err
-	}
-
-	err = job.Output.Write(ctx, output)
-	if err != nil {
-		job.Error.Error(ctx, err)
-		return nil, err
+		return err
 	}
 
-	return output, nil
+	return nil
 }

pkg/component/ai/huggingface/v0/main.go

@@ -14,6 +14,7 @@ import (
 	"google.golang.org/protobuf/types/known/structpb"
 
 	"github.com/instill-ai/pipeline-backend/pkg/component/base"
+	"github.com/instill-ai/pipeline-backend/pkg/component/internal/util"
 	"github.com/instill-ai/x/errmsg"
 )
 
@@ -433,7 +434,7 @@ func (e *execution) Execute(ctx context.Context, jobs []*base.Job) error {
 				continue
 			}
 
-			b, err := base64.StdEncoding.DecodeString(base.TrimBase64Mime(inputStruct.Image))
+			b, err := base64.StdEncoding.DecodeString(util.TrimBase64Mime(inputStruct.Image))
 			if err != nil {
 				job.Error.Error(ctx, err)
 				continue
@@ -459,7 +460,7 @@ func (e *execution) Execute(ctx context.Context, jobs []*base.Job) error {
 				continue
 			}
 
-			b, err := base64.StdEncoding.DecodeString(base.TrimBase64Mime(inputStruct.Image))
+			b, err := base64.StdEncoding.DecodeString(util.TrimBase64Mime(inputStruct.Image))
 			if err != nil {
 				job.Error.Error(ctx, err)
 				continue
@@ -504,7 +505,7 @@ func (e *execution) Execute(ctx context.Context, jobs []*base.Job) error {
 				continue
 			}
 
-			b, err := base64.StdEncoding.DecodeString(base.TrimBase64Mime(inputStruct.Image))
+			b, err := base64.StdEncoding.DecodeString(util.TrimBase64Mime(inputStruct.Image))
 			if err != nil {
 				job.Error.Error(ctx, err)
 				continue
@@ -530,7 +531,7 @@ func (e *execution) Execute(ctx context.Context, jobs []*base.Job) error {
 				continue
 			}
 
-			b, err := base64.StdEncoding.DecodeString(base.TrimBase64Mime(inputStruct.Image))
+			b, err := base64.StdEncoding.DecodeString(util.TrimBase64Mime(inputStruct.Image))
 			if err != nil {
 				job.Error.Error(ctx, err)
 				continue
@@ -562,7 +563,7 @@ func (e *execution) Execute(ctx context.Context, jobs []*base.Job) error {
 				continue
 			}
 
-			b, err := base64.StdEncoding.DecodeString(base.TrimBase64Mime(inputStruct.Audio))
+			b, err := base64.StdEncoding.DecodeString(util.TrimBase64Mime(inputStruct.Audio))
 			if err != nil {
 				job.Error.Error(ctx, err)
 				continue
@@ -589,7 +590,7 @@ func (e *execution) Execute(ctx context.Context, jobs []*base.Job) error {
 				continue
 			}
 
-			b, err := base64.StdEncoding.DecodeString(base.TrimBase64Mime(inputStruct.Audio))
+			b, err := base64.StdEncoding.DecodeString(util.TrimBase64Mime(inputStruct.Audio))
 			if err != nil {
 				job.Error.Error(ctx, err)
 				continue