Merge pull request #1 from henomis/v0.0.3

V0.0.3

examples/cmd/ingestion/main.go

@@ -8,22 +8,23 @@ import (
 )
 
 func main() {
-	l := langfuse.New()
+	l := langfuse.New(context.Background())
 
-	err := l.Trace(&model.Trace{Name: "test-trace"})
+	trace, err := l.Trace(&model.Trace{Name: "test-trace"})
 	if err != nil {
 		panic(err)
 	}
 
-	err = l.Span(&model.Span{Name: "test-span"})
+	span, err := l.Span(&model.Span{Name: "test-span", TraceID: trace.ID}, nil)
 	if err != nil {
 		panic(err)
 	}
 
-	err = l.Generation(
+	generation, err := l.Generation(
 		&model.Generation{
-			Name:  "test-generation",
-			Model: "gpt-3.5-turbo",
+			TraceID: trace.ID,
+			Name:    "test-generation",
+			Model:   "gpt-3.5-turbo",
 			ModelParameters: model.M{
 				"maxTokens":   "1000",
 				"temperature": "0.9",
@@ -42,14 +43,16 @@ func main() {
 				"key": "value",
 			},
 		},
+		&span.ID,
 	)
 	if err != nil {
 		panic(err)
 	}
 
-	err = l.Event(
+	_, err = l.Event(
 		&model.Event{
-			Name: "test-event",
+			Name:    "test-event",
+			TraceID: trace.ID,
 			Metadata: model.M{
 				"key": "value",
 			},
@@ -60,33 +63,32 @@ func main() {
 				"key": "value",
 			},
 		},
+		&generation.ID,
 	)
 	if err != nil {
 		panic(err)
 	}
 
-	err = l.GenerationEnd(
-		&model.Generation{
-			Output: model.M{
-				"completion": "The Q3 OKRs contain goals for multiple teams...",
-			},
-		},
-	)
+	generation.Output = model.M{
+		"completion": "The Q3 OKRs contain goals for multiple teams...",
+	}
+	_, err = l.GenerationEnd(generation)
 	if err != nil {
 		panic(err)
 	}
 
-	err = l.Score(
+	_, err = l.Score(
 		&model.Score{
-			Name:  "test-score",
-			Value: 0.9,
+			TraceID: trace.ID,
+			Name:    "test-score",
+			Value:   0.9,
 		},
 	)
 	if err != nil {
 		panic(err)
 	}
 
-	err = l.SpanEnd(&model.Span{})
+	_, err = l.SpanEnd(span)
 	if err != nil {
 		panic(err)
 	}

internal/pkg/observer/handler.go

@@ -1,6 +1,7 @@
 package observer
 
 import (
+	"context"
 	"time"
 )
 
@@ -37,19 +38,19 @@ func (h *handler[T]) withTick(period time.Duration) *handler[T] {
 	return h
 }
 
-func (h *handler[T]) listen() {
+func (h *handler[T]) listen(ctx context.Context) {
 	ticker := time.NewTicker(h.tickerPeriod)
 
 	for {
 		select {
 		case <-ticker.C:
-			go h.handle()
+			go h.handle(ctx)
 		case cmd, ok := <-h.commandCh:
 			if !ok {
 				return
 			}
 
-			h.handle()
+			h.handle(ctx)
 			if cmd == commandFlushAndWait {
 				ticker.Stop()
 				close(h.commandCh)
@@ -58,8 +59,8 @@ func (h *handler[T]) listen() {
 	}
 }
 
-func (h *handler[T]) handle() {
-	h.fn(h.queue.All())
+func (h *handler[T]) handle(ctx context.Context) {
+	h.fn(ctx, h.queue.All())
 }
 
 func (h *handler[T]) flush() {

internal/pkg/observer/observer.go

@@ -5,21 +5,21 @@ import (
 	"time"
 )
 
-type EventHandler[T any] func(events []T)
+type EventHandler[T any] func(ctx context.Context, events []T)
 
 type Observer[T any] struct {
 	queue   *queue[T]
 	handler *handler[T]
 }
 
-func NewObserver[T any](fn EventHandler[T]) *Observer[T] {
+func NewObserver[T any](ctx context.Context, fn EventHandler[T]) *Observer[T] {
 	queue := newQueue[T]()
 
 	o := &Observer[T]{
 		queue:   queue,
 		handler: newHandler(queue, fn),
 	}
-	go o.handler.listen()
+	go o.handler.listen(ctx)
 
 	return o
 }