openai

Documentation

Index

• func ProcessRecordField[T any](ctx context.Context, rec opencdc.Record, resolver sdk.ReferenceResolver, ...) (opencdc.Record, error)
• type EmbeddingsProcessor
  • func NewEmbeddingsProcessor(log.CtxLogger) *EmbeddingsProcessor
  • func (p *EmbeddingsProcessor) Configure(ctx context.Context, cfg config.Config) error
  • func (p *EmbeddingsProcessor) Process(ctx context.Context, recs []opencdc.Record) []sdk.ProcessedRecord
  • func (p *EmbeddingsProcessor) Specification() (sdk.Specification, error)
• type TextgenProcessor
  • func NewTextgenProcessor(log.CtxLogger) *TextgenProcessor
  • func (p *TextgenProcessor) Configure(ctx context.Context, cfg config.Config) error
  • func (p *TextgenProcessor) Process(ctx context.Context, recs []opencdc.Record) []sdk.ProcessedRecord
  • func (p *TextgenProcessor) Specification() (sdk.Specification, error)

Examples

• EmbeddingsProcessor
• TextgenProcessor

Functions

func ProcessRecordField

func ProcessRecordField[T any](
	ctx context.Context,
	rec opencdc.Record,
	resolver sdk.ReferenceResolver,
	processor func(ctx context.Context, input string) (T, error),
	formatter func(T) ([]byte, error),
) (opencdc.Record, error)

Types

type EmbeddingsProcessor

type EmbeddingsProcessor struct {
	sdk.UnimplementedProcessor
	// contains filtered or unexported fields
}

Example

var (
	testVector  = []float32{0.1, 0.2, 0.3, 0.4, 0.5}
	testVectorS = "[0.1,0.2,0.3,0.4,0.5]"
)

processor := NewEmbeddingsProcessor(log.Nop())
processor.call = &mockEmbeddingsCaller{Embeddings: testVector}

exampleutil.RunExample(processor, exampleutil.Example{
	Summary: "Generate embeddings for text",
	Description: `This example generates embeddings for the text stored in
` + "`.Payload.After`" + `. The embeddings are returned as a JSON array of floating point numbers.
These embeddings can be used for semantic search, clustering, or other machine learning tasks.`,
	Config: config.Config{
		"api_key": "your-openai-api-key",
		"model":   "text-embedding-3-small",
		"field":   ".Payload.After",
	},
	Have: opencdc.Record{
		Position:  opencdc.Position("test-position"),
		Operation: opencdc.OperationCreate,
		Metadata:  map[string]string{"key1": "val1"},
		Key:       opencdc.RawData("test-key"),
		Payload: opencdc.Change{
			After: opencdc.RawData("This is a sample text to generate embeddings for."),
		},
	},
	Want: sdk.SingleRecord{
		Position:  opencdc.Position("test-position"),
		Operation: opencdc.OperationCreate,
		Metadata:  map[string]string{"key1": "val1"},
		Key:       opencdc.RawData("test-key"),
		Payload: opencdc.Change{
			After: opencdc.RawData(testVectorS),
		},
	},
})

Output:

processor transformed record:
--- before
+++ after
@@ -1,12 +1,12 @@
 {
   "position": "dGVzdC1wb3NpdGlvbg==",
   "operation": "create",
   "metadata": {
     "key1": "val1"
   },
   "key": "test-key",
   "payload": {
     "before": null,
-    "after": "This is a sample text to generate embeddings for."
+    "after": "[0.1,0.2,0.3,0.4,0.5]"
   }
 }

func NewEmbeddingsProcessor

func NewEmbeddingsProcessor(log.CtxLogger) *EmbeddingsProcessor

func (*EmbeddingsProcessor) Configure

func (p *EmbeddingsProcessor) Configure(ctx context.Context, cfg config.Config) error

func (*EmbeddingsProcessor) Process

func (p *EmbeddingsProcessor) Process(ctx context.Context, recs []opencdc.Record) []sdk.ProcessedRecord

func (*EmbeddingsProcessor) Specification

func (p *EmbeddingsProcessor) Specification() (sdk.Specification, error)

type TextgenProcessor

type TextgenProcessor struct {
	sdk.UnimplementedProcessor
	// contains filtered or unexported fields
}

Example

p := NewTextgenProcessor(log.Nop())
p.call = &mockOpenAICaller{}

exampleutil.RunExample(p, exampleutil.Example{
	Summary: `Transform text using OpenAI models`,
	Description: `
This example shows how to use the OpenAI text generation processor to transform a record's ` + "`.Payload.After`" + ` field
using an OpenAI model. The processor will send the content of the field to OpenAI and replace it with the response.

In this example, we're using a system message that instructs the model to convert the input text to uppercase.`,
	Config: config.Config{
		"api_key":           "fake-api-key",
		"model":             openai.GPT4oMini,
		"developer_message": "You will receive a payload. Your task is to output back the payload in uppercase.",
		"temperature":       "0",
	},
	Have: opencdc.Record{
		Operation: opencdc.OperationCreate,
		Position:  opencdc.Position("pos-1"),
		Payload: opencdc.Change{
			After: opencdc.RawData("hello world"),
		},
	},
	Want: sdk.SingleRecord{
		Operation: opencdc.OperationCreate,
		Position:  opencdc.Position("pos-1"),
		Payload: opencdc.Change{
			After: opencdc.RawData("HELLO WORLD"),
		},
	},
})

Output:

processor transformed record:
--- before
+++ after
@@ -1,10 +1,10 @@
 {
   "position": "cG9zLTE=",
   "operation": "create",
   "metadata": null,
   "key": null,
   "payload": {
     "before": null,
-    "after": "hello world"
+    "after": "HELLO WORLD"
   }
 }

func NewTextgenProcessor

func NewTextgenProcessor(log.CtxLogger) *TextgenProcessor

func (*TextgenProcessor) Configure

func (p *TextgenProcessor) Configure(ctx context.Context, cfg config.Config) error

func (*TextgenProcessor) Process

func (p *TextgenProcessor) Process(ctx context.Context, recs []opencdc.Record) []sdk.ProcessedRecord

func (*TextgenProcessor) Specification

func (p *TextgenProcessor) Specification() (sdk.Specification, error)