expose jinja converter

Author: emranbm

event_buffer.go

@@ -0,0 +1,55 @@
+package golangfuse
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/sirupsen/logrus"
+)
+
+type EventsFlushFunc func(ctx context.Context, events []IngestionEvent) error
+
+type eventBuffer struct {
+	bufferedEvents []IngestionEvent
+	mu             sync.Mutex
+	flushHandler   EventsFlushFunc
+}
+
+func newEventBufferer(flushHandler EventsFlushFunc) *eventBuffer {
+	return &eventBuffer{flushHandler: flushHandler}
+}
+
+func (i *eventBuffer) Start(ctx context.Context, period time.Duration) {
+	ticker := time.NewTicker(period)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ticker.C:
+			i.flush(ctx)
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+func (i *eventBuffer) Add(event IngestionEvent) {
+	i.mu.Lock()
+	defer i.mu.Unlock()
+	i.bufferedEvents = append(i.bufferedEvents, event)
+}
+
+func (i *eventBuffer) flush(ctx context.Context) {
+	i.mu.Lock()
+	items := i.bufferedEvents
+	i.bufferedEvents = nil
+	i.mu.Unlock()
+	if len(items) > 0 {
+		err := i.flushHandler(ctx, items)
+		if err != nil {
+			logrus.WithError(err).Error("golangfuse: error in event flush handler")
+		} else {
+			logrus.Tracef("golangfuse: flushed %d events", len(items))
+		}
+	}
+}

langfuse.go

@@ -4,26 +4,23 @@ import (
 	"context"
 	"fmt"
 	"net/http"
-	"regexp"
 	"sync/atomic"
 	"time"
 
 	"github.com/divar-ir/golangfuse/internal/constants"
-	"github.com/divar-ir/golangfuse/internal/observer"
 	"github.com/google/uuid"
 	"resty.dev/v3"
 )
 
 type Langfuse interface {
 	StartSendingEvents(ctx context.Context, period time.Duration) error
 	Trace(input, output any, options ...TraceOption)
-	GetPromptTemplate(ctx context.Context, promptName string) (string, error)
+	GetSystemPromptTemplate(ctx context.Context, promptName string) (string, error)
 }
 
 type langfuseImpl struct {
 	restClient             *resty.Client
-	eventObserver          observer.Observer[IngestionEvent]
-	eventQueue             observer.Queue[IngestionEvent]
+	eventBuffer            *eventBuffer
 	isSendingEventsStarted atomic.Bool
 	endpoint               string
 	promptLabel            string
@@ -37,24 +34,23 @@ func NewWithHttpClient(httpClient *http.Client, endpoint, publicKey, secretKey s
 	client := resty.NewWithClient(httpClient).SetBasicAuth(publicKey, secretKey)
 	c := &langfuseImpl{
 		restClient:  client,
-		eventQueue:  observer.NewQueue[IngestionEvent](),
 		endpoint:    endpoint,
 		promptLabel: "production", // TODO: use option pattern to override this default if needed
 	}
-	c.eventObserver = observer.NewObserver[IngestionEvent](c.eventQueue, c.sendEvents)
+	c.eventBuffer = newEventBufferer(c.sendEvents)
 	return c
 }
 
 func (c *langfuseImpl) StartSendingEvents(ctx context.Context, period time.Duration) error {
 	if c.isSendingEventsStarted.CompareAndSwap(false, true) {
-		go c.eventObserver.StartObserve(ctx, period)
+		go c.eventBuffer.Start(ctx, period)
 		return nil
 	} else {
 		return AlreadyStartedErr
 	}
 }
 
-func (c *langfuseImpl) GetPromptTemplate(ctx context.Context, promptName string) (string, error) {
+func (c *langfuseImpl) GetSystemPromptTemplate(ctx context.Context, promptName string) (string, error) {
 	promptObject := ChatPrompt{}
 	resp, err := c.restClient.R().
 		SetContext(ctx).
@@ -75,7 +71,7 @@ func (c *langfuseImpl) GetPromptTemplate(ctx context.Context, promptName string)
 	if promptObject.Prompt[0].Role != "system" {
 		return "", fmt.Errorf("prompt role is not system")
 	}
-	return convertJinjaVariablesToGoTemplate(promptObject.Prompt[0].Content), nil
+	return promptObject.Prompt[0].Content, nil
 }
 
 func (c *langfuseImpl) Trace(input, output any, options ...TraceOption) {
@@ -86,7 +82,7 @@ func (c *langfuseImpl) Trace(input, output any, options ...TraceOption) {
 	for _, opt := range options {
 		opt(trace)
 	}
-	c.eventQueue.Enqueue(IngestionEvent{
+	c.eventBuffer.Add(IngestionEvent{
 		ID:        uuid.NewString(),
 		Timestamp: time.Now(),
 		Type:      constants.IngestionEventTypeTraceCreate,
@@ -108,8 +104,3 @@ func (c *langfuseImpl) sendEvents(ctx context.Context, events []IngestionEvent)
 	}
 	return nil
 }
-
-func convertJinjaVariablesToGoTemplate(prompt string) string {
-	re := regexp.MustCompile(`\{\{\s*([a-zA-Z0-9_]+)\s*\}\}`)
-	return re.ReplaceAllString(prompt, "{{.$1}}")
-}

langfuse_test.go

@@ -38,34 +38,11 @@ func (s *ClientTest) TestGetPromptTemplateShouldParseApiResponse() {
 	c := s.getLangfuseClientForTest(promptName, "This is system prompt")
 
 	// When
-	promptTemplate, err := c.GetPromptTemplate(s.ctx, promptName)
+	promptTemplate, err := c.GetSystemPromptTemplate(s.ctx, promptName)
 	s.Require().NoError(err)
 	s.Require().Equal("This is system prompt", promptTemplate)
 }
 
-func (s *ClientTest) TestGetPromptTemplateShouldReturnVariablesWithGolangFormat() {
-	// Given
-	const promptName = "test-prompt"
-	jinjaVarialbeVariants := []string{
-		"{{myVar}}",
-		"{{myVar }}",
-		"{{ myVar}}",
-		"{{ myVar }}",
-	}
-
-	for _, v := range jinjaVarialbeVariants {
-		c := s.getLangfuseClientForTest(promptName,
-			fmt.Sprintf("This is system prompt %s variable.", v))
-
-		// When
-		promptTemplate, err := c.GetPromptTemplate(s.ctx, promptName)
-
-		// Then
-		s.Require().NoError(err)
-		s.Require().Equal("This is system prompt {{.myVar}} variable.", promptTemplate)
-	}
-}
-
 func (s *ClientTest) getLangfuseClientForTest(promptName, promptContent string) golangfuse.Langfuse {
 	apiResponse := fmt.Sprintf(`{
   "id" : "id",
@@ -104,7 +81,7 @@ func (s *ClientTest) TestShouldSetBasicAuth() {
 	})
 
 	// When
-	_, err := c.GetPromptTemplate(s.ctx, "test-prompt")
+	_, err := c.GetSystemPromptTemplate(s.ctx, "test-prompt")
 	s.Require().ErrorContains(err, "http mock called")
 
 	// Then

utils.go

@@ -0,0 +1,8 @@
+package golangfuse
+
+import "regexp"
+
+func JinjaToGoTemplate(prompt string) string {
+	re := regexp.MustCompile(`\{\{\s*([a-zA-Z0-9_]+)\s*\}\}`)
+	return re.ReplaceAllString(prompt, "{{.$1}}")
+}

utils_test.go

@@ -0,0 +1,37 @@
+package golangfuse_test
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/divar-ir/golangfuse"
+	"github.com/stretchr/testify/suite"
+)
+
+type UtilsTest struct {
+	suite.Suite
+}
+
+func (s *UtilsTest) TestJinjaToGoTemplateShouldReturnVariablesWithGolangFormat() {
+	// Given
+	jinjaVarialbeVariants := []string{
+		"{{myVar}}",
+		"{{myVar }}",
+		"{{ myVar}}",
+		"{{ myVar }}",
+	}
+
+	for _, v := range jinjaVarialbeVariants {
+		source := fmt.Sprintf("This is system prompt %s variable.", v)
+
+		// When
+		target := golangfuse.JinjaToGoTemplate(source)
+
+		// Then
+		s.Require().Equal("This is system prompt {{.myVar}} variable.", target)
+	}
+}
+
+func TestUtils(t *testing.T) {
+	suite.Run(t, new(UtilsTest))
+}