Workflow evaluations

Go workflow evaluations are in public preview and can change before reaching general availability.

Requires Go SDK v0.18.0 or later. The Go SDK saves run state in its own format, so start and resume a run with the Go SDK.

Workflow evaluations let you submit tasks or scorers to an asynchronous provider API, such as the OpenAI Batch API, and collect the results later. Use them for evaluations that run through a batch API, which providers often offer at a discount, or that need to continue across process exits. You write functions that submit requests, check whether they’re done, and fetch the results. The SDK calls them, saves progress in a store, and records the outputs and scores in one Braintrust experiment.

How workflow evaluations work

A workflow evaluation has three stages:

Define an evaluation: Build an eval.WorkflowEval with a task, scorers, and a store, and attach your Braintrust client with braintrust.NewWorkflowEval. It works like eval.Eval, except that the task or scorers can send their work to a provider’s asynchronous API.
Start the run: Call Start with your dataset to submit the requests. It returns a run ID right away, without waiting for the provider.
Resume until completion: Pass the run ID to Poll on a schedule, or pass a provider’s completion event to ProcessSubmissionResult from a webhook handler. Each call collects the results that are ready and scores them. The run is done when its status is eval.WorkflowRunCompleted.

The following example walks through each of these stages.

Example: OpenAI Batch API

This example sends questions about capital cities through the OpenAI Batch API and scores each answer against the expected city. Here’s what the program does:

Submits the cases to OpenAI together, in one batch.
Polls OpenAI until the batch finishes, then collects each answer.
Scores each answer with exactMatch, an ordinary scorer, and logs the results to Braintrust.

To get started, create a Go module, add the Braintrust and OpenAI SDKs, and set your BRAINTRUST_API_KEY and OPENAI_API_KEY environment variables:

go mod init capital-cities
go get github.com/braintrustdata/braintrust-sdk-go@v0.18.0 github.com/openai/openai-go
export BRAINTRUST_API_KEY="your-braintrust-api-key"
export OPENAI_API_KEY="your-openai-api-key"

Then, save the evaluation and its callbacks in evaluation.go. They live in their own file so the Redis and webhook programs later on this page can reuse them:

evaluation.go

package main

import (
	"bufio"
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"strings"

	"github.com/openai/openai-go"
	"github.com/openai/openai-go/responses"

	"github.com/braintrustdata/braintrust-sdk-go"
	"github.com/braintrustdata/braintrust-sdk-go/eval"
)

// Data your Submit callback returns and the SDK saves between calls. The SDK
// passes it back to the polling and Collect callbacks. Here it holds the
// OpenAI batch ID. You choose the fields. It must be JSON.
type submission struct {
	BatchID string `json:"batch_id"`
}

// One request in a batch. CustomID is the ID the SDK gives each item in a
// batch. It goes to OpenAI as custom_id, so results can be matched to items.
type batchRequest struct {
	CustomID string
	Prompt   string
}

// The fields this example reads from one line of a batch's output or error
// file.
type batchResultLine struct {
	CustomID string          `json:"custom_id"`
	Error    json.RawMessage `json:"error"`
	Response *struct {
		StatusCode int                `json:"status_code"`
		Body       responses.Response `json:"body"`
	} `json:"response"`
}

// Sends the prompts to OpenAI as one batch.
func submitBatch(ctx context.Context, client openai.Client, requests []batchRequest, info eval.WorkflowSubmissionContext) (json.RawMessage, error) {
	// The Batch API reads its requests from an uploaded JSONL file.
	var lines bytes.Buffer
	encoder := json.NewEncoder(&lines)
	for _, request := range requests {
		err := encoder.Encode(map[string]any{
			"custom_id": request.CustomID,
			"method":    "POST",
			"url":       "/v1/responses",
			"body":      map[string]any{"model": "gpt-5-mini", "input": request.Prompt},
		})
		if err != nil {
			return nil, err
		}
	}
	file, err := client.Files.New(ctx, openai.FileNewParams{
		File:    openai.File(&lines, "requests.jsonl", "application/jsonl"),
		Purpose: openai.FilePurposeBatch,
	})
	if err != nil {
		return nil, err
	}
	batch, err := client.Batches.New(ctx, openai.BatchNewParams{
		InputFileID:      file.ID,
		Endpoint:         openai.BatchNewParamsEndpointV1Responses,
		CompletionWindow: openai.BatchNewParamsCompletionWindow24h,
		// Records which run and submission the batch belongs to.
		Metadata: map[string]string{"braintrust_run_id": info.RunID, "braintrust_submission_id": info.SubmissionID},
	})
	if err != nil {
		return nil, err
	}
	return json.Marshal(submission{BatchID: batch.ID})
}

// The polling callback. Called during each Poll until the batch reports
// complete. It reports complete once OpenAI stops working on the batch,
// including when the batch fails or expires, so Collect can record the
// requests that have no result as failed and the run can finish.
func pollBatch(client openai.Client) func(context.Context, json.RawMessage, eval.WorkflowSubmissionContext) (eval.WorkflowPollResult, error) {
	return func(ctx context.Context, data json.RawMessage, _ eval.WorkflowSubmissionContext) (eval.WorkflowPollResult, error) {
		var sub submission
		if err := json.Unmarshal(data, &sub); err != nil {
			return eval.WorkflowPollResult{}, err
		}
		batch, err := client.Batches.Get(ctx, sub.BatchID)
		if err != nil {
			return eval.WorkflowPollResult{}, err
		}
		switch batch.Status {
		case openai.BatchStatusCompleted, openai.BatchStatusFailed, openai.BatchStatusExpired, openai.BatchStatusCancelled:
			return eval.WorkflowPollResult{Status: eval.WorkflowComplete}, nil
		default:
			return eval.WorkflowPollResult{Status: eval.WorkflowPending}, nil
		}
	}
}

// Reads each request's answer, or its error, from the batch's output and
// error files, and passes it to yield.
func collectText(ctx context.Context, client openai.Client, data json.RawMessage, yield func(customID, text string, err error) error) error {
	var sub submission
	if err := json.Unmarshal(data, &sub); err != nil {
		return err
	}
	batch, err := client.Batches.Get(ctx, sub.BatchID)
	if err != nil {
		return err
	}
	for _, fileID := range []string{batch.OutputFileID, batch.ErrorFileID} {
		if fileID == "" {
			continue
		}
		content, err := client.Files.Content(ctx, fileID)
		if err != nil {
			return err
		}
		defer content.Body.Close()
		scanner := bufio.NewScanner(content.Body)
		scanner.Buffer(make([]byte, 64*1024), 16*1024*1024)
		for scanner.Scan() {
			if len(bytes.TrimSpace(scanner.Bytes())) == 0 {
				continue
			}
			var line batchResultLine
			// Stops on a line that can't be matched to an item. The SDK calls
			// Collect again on the next Poll or webhook delivery.
			if err := json.Unmarshal(scanner.Bytes(), &line); err != nil || line.CustomID == "" {
				return fmt.Errorf("unexpected line in batch %s results: %s", sub.BatchID, scanner.Text())
			}
			var text string
			var itemErr error
			if line.Response == nil || line.Response.StatusCode != 200 {
				itemErr = fmt.Errorf("request failed: %s", line.Error)
			} else {
				text = line.Response.Body.OutputText()
			}
			// Returns as soon as yield returns an error, as the SDK requires.
			if err := yield(line.CustomID, text, itemErr); err != nil {
				return err
			}
		}
		if err := scanner.Err(); err != nil {
			return err
		}
	}
	return nil
}

// An ordinary scorer, like the ones you pass to eval.Eval. The SDK runs it on
// each case's collected output.
var exactMatch = eval.NewScorer("exact_match", func(_ context.Context, result eval.TaskResult[string, string]) (eval.Scores, error) {
	if strings.EqualFold(strings.TrimSpace(result.Output), result.Expected) {
		return eval.S(1), nil
	}
	return eval.S(0), nil
})

// The test cases. Each case's ID becomes its case ID in the experiment rows.
func capitalCities() eval.Dataset[string, string] {
	return eval.NewDataset([]eval.Case[string, string]{
		{ID: "france", Input: "Capital of France?", Expected: "Paris"},
		{ID: "japan", Input: "Capital of Japan?", Expected: "Tokyo"},
	})
}

// The store and completion mode are parameters so the Redis and webhook
// examples later on this page can reuse this function.
func newEvaluation(bt *braintrust.Client, client openai.Client, store eval.WorkflowEvalStore, webhook bool) *eval.WorkflowEval[string, string] {
	// How the SDK learns that a batch finished: by polling, or from a webhook
	// that the SDK matches to the batch ID GetExternalID returns.
	completion := eval.WorkflowCompletion{Poll: pollBatch(client)}
	if webhook {
		completion = eval.WorkflowCompletion{GetExternalID: func(_ context.Context, data json.RawMessage, _ eval.WorkflowSubmissionContext) (string, error) {
			var sub submission
			err := json.Unmarshal(data, &sub)
			return sub.BatchID, err
		}}
	}
	// The type arguments are the input type and the output type, which is
	// also the expected output type.
	return braintrust.NewWorkflowEval(bt, &eval.WorkflowEval[string, string]{
		Name:        "capital-cities-eval",
		ProjectName: "capital-cities-eval",
		Store:       store,
		Task: eval.WorkflowBatchTask[string, string]{
			// Puts up to 100 cases in each OpenAI batch, so this evaluation's
			// two cases go in one batch. Set TrialCount in
			// eval.WorkflowStartOpts to run each case more than once.
			Batching: eval.WorkflowBatching{MaxSize: 100},
			// Called once for each batch, with one item for each case in it.
			Submit: func(ctx context.Context, items []eval.WorkflowBatchItem[eval.WorkflowTaskItem[string, string]], info eval.WorkflowSubmissionContext) (json.RawMessage, error) {
				requests := make([]batchRequest, len(items))
				for i, item := range items {
					requests[i] = batchRequest{CustomID: item.CustomID, Prompt: item.Item.Input + " Reply with only the city name."}
				}
				return submitBatch(ctx, client, requests, info)
			},
			Completion: completion,
			// Called after the batch completes. Yields one result for each
			// item, matched by CustomID. The SDK records an item without a
			// result as failed.
			Collect: func(ctx context.Context, data json.RawMessage, _ eval.WorkflowSubmissionContext, yield func(eval.WorkflowBatchResult[eval.WorkflowTaskResult[string]]) error) error {
				return collectText(ctx, client, data, func(customID, text string, err error) error {
					return yield(eval.WorkflowBatchResult[eval.WorkflowTaskResult[string]]{
						CustomID: customID,
						Result:   eval.WorkflowTaskResult[string]{Output: text},
						Error:    err,
					})
				})
			},
		},
		// Ordinary scorers go in Scorers. eval.WorkflowScorer and
		// eval.WorkflowBatchScorer values go in WorkflowScorers.
		Scorers: []eval.Scorer[string, string]{exactMatch},
	})
}

Next, save the program that runs the evaluation in main.go:

main.go

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/openai/openai-go"
	"go.opentelemetry.io/otel/sdk/trace"

	"github.com/braintrustdata/braintrust-sdk-go"
	"github.com/braintrustdata/braintrust-sdk-go/eval"
)

func main() {
	ctx := context.Background()
	tp := trace.NewTracerProvider()
	defer tp.Shutdown(ctx)
	bt, err := braintrust.New(tp, braintrust.WithBlockingLogin(true))
	if err != nil {
		log.Fatal(err)
	}
	// The memory store works in this example because the whole run happens
	// in one process.
	evaluation := newEvaluation(bt, openai.NewClient(), eval.NewWorkflowEvalMemoryStore(), false)
	// Creates the experiment, submits the cases in one batch, and returns
	// right away with status "waiting".
	result, err := evaluation.Start(ctx, eval.WorkflowStartOpts[string, string]{Dataset: capitalCities()})
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println(result.RunID, result.Status)
	// In an application, schedule Poll instead of sleeping in one process.
	for result.Status == eval.WorkflowRunWaiting {
		time.Sleep(time.Minute)
		result, err = evaluation.Poll(ctx, result.RunID)
		if err != nil {
			log.Fatal(err)
		}
		// PendingPoll counts submitted batches that haven't been collected.
		fmt.Println(result.Status, result.PendingPoll)
	}
	// Failed counts cases whose request or scorer failed.
	fmt.Printf("failed=%d scores=%v\n%s\n", result.Failed, result.Scores, result.Permalink)
}

Run it with go run evaluation.go main.go. OpenAI batches can take minutes or hours, so the program can run for a while. After each Poll, it prints the run’s status and how many batches are still pending. When it finishes, the Braintrust experiment contains two rows, each with an exact_match score. The rest of this section walks through the example’s three stages.

Define an evaluation

First, newEvaluation in evaluation.go defines the evaluation. In outline, the definition looks like this:

braintrust.NewWorkflowEval(bt, &eval.WorkflowEval[string, string]{
	Name:        "capital-cities-eval",
	ProjectName: "capital-cities-eval",
	Store:       store,
	Task:        eval.WorkflowBatchTask[string, string]{ /* ... */ },
	Scorers:     []eval.Scorer[string, string]{exactMatch},
})

The type arguments are the input type and the output type. The expected output has the same type as the output. The following sections describe each field. For every field, see braintrust.NewWorkflowEval.

Project, experiment, and client

ProjectName sets the Braintrust project, "capital-cities-eval" in the example. Braintrust creates the project if it doesn’t exist. Without ProjectName, the evaluation uses the project configured on the client. Each run logs to a new experiment named after Name, unless you set Experiment in eval.WorkflowStartOpts.braintrust.NewWorkflowEval attaches your Braintrust client, so runs create an experiment and log results. Every process that starts or resumes a run must use a definition attached this way. To run without Braintrust, set NoSendLogs in eval.WorkflowStartOpts. The run state is still saved to the store.Name also identifies the run’s records in the store, so keep it, the scorer names, and their order unchanged while runs are active. A process that loads a run with a different definition returns an error.

Data

The dataset goes to Start in eval.WorkflowStartOpts, not in the definition. It accepts the same eval.Dataset values as eval.Eval, such as one built with eval.NewDataset. The SDK saves every case to the store when the run starts, so later calls don’t need the dataset.Each case’s ID becomes its case ID in the experiment rows. When a case has no ID, the SDK generates one when the run starts. Cases can also set Metadata, Tags, and TrialCount. Inputs, expected values, and metadata must be JSON-serializable, so the store can save them.

Task

The Task field accepts an ordinary task function built with eval.T, an eval.WorkflowBatchTask, or an eval.WorkflowTask. Use an eval.WorkflowBatchTask when the provider accepts many requests at once and returns their results later, as batch APIs do. The example’s task has these fields:

Batching: MaxSize sets the most items in one batch. Start submits every case and trial right away, in batches of up to MaxSize items.
Submit: Sends one batch to the provider. Receives a slice of eval.WorkflowBatchItem values, one for each case and trial, each with the case’s fields in Item and a CustomID that identifies the item. Use the CustomID as the provider’s request ID, such as OpenAI’s custom_id. Returns a json.RawMessage handle, such as the provider’s batch ID. The SDK saves it and passes it to the Completion and Collect callbacks.
Completion: How the SDK learns the batch is done: by polling with a function such as the example’s pollBatch, or from a webhook. Set exactly one of its Poll and GetExternalID fields. See eval.WorkflowCompletion.
Collect: Fetches the finished batch’s results and calls yield once for each item with an eval.WorkflowBatchResult, which has the CustomID and either a Result with the task’s Output or an Error. Return right away when yield returns an error. The SDK records an item that has an error or no result as failed: its row shows the error, and its scorers don’t run.

The handle must be valid JSON and must not contain clients or credentials, because the store saves it. In the example, it’s {"batch_id": "..."}.Every callback also receives an eval.WorkflowSubmissionContext with the run ID and a SubmissionID. The SubmissionID stays the same if the SDK calls Submit again for the same work, so use it as the provider’s idempotency key if the provider supports one. The example stores both IDs in the batch’s metadata.To send each case as its own request instead, use an eval.WorkflowTask. See Submit one request per case.

Scorers

Ordinary scorers, like the example’s exactMatch, go in Scorers, and classifiers go in Classifiers. They run in your process as soon as a case’s output is collected. Asynchronous scorers go in WorkflowScorers, which accepts eval.WorkflowBatchScorer and eval.WorkflowScorer values. Use an eval.WorkflowBatchScorer when scoring needs requests that finish later, such as LLM-as-a-judge calls through a batch API.An eval.WorkflowBatchScorer takes the same fields as an eval.WorkflowBatchTask, plus a Name. Its items are eval.WorkflowScorerItem values, which include the task’s Output along with the case’s fields, and each result’s Result is an eval.Scores. A failed item omits that score and records the error in the row’s scorer_errors metadata. See eval.WorkflowBatchScorer.An eval.WorkflowBatchScorer submits a batch once MaxSize items have task outputs, or once no more items can get one. To submit partial batches sooner, set Batching.MaxWait. The SDK submits a partial batch that has waited that long on the next Start, Poll, or ProcessSubmissionResult call.Example: Score with a batch requestThe following newLLMJudge scorer asks a model to grade each answer through the OpenAI Batch API. It reuses the example’s submitBatch, pollBatch, and collectText. To try it, add it to the end of evaluation.go:

// Grades each answer with a model, through the OpenAI Batch API.
func newLLMJudge(client openai.Client) eval.WorkflowBatchScorer[string, string] {
	return eval.WorkflowBatchScorer[string, string]{
		// Names the score in Braintrust. Keep it the same across resumes.
		Name: "llm_judge",
		// Waits for up to 100 task outputs, or until no more can arrive, then
		// submits them in one grading batch.
		Batching: eval.WorkflowBatching{MaxSize: 100},
		Submit: func(ctx context.Context, items []eval.WorkflowBatchItem[eval.WorkflowScorerItem[string, string]], info eval.WorkflowSubmissionContext) (json.RawMessage, error) {
			requests := make([]batchRequest, len(items))
			for i, item := range items {
				requests[i] = batchRequest{CustomID: item.CustomID, Prompt: strings.Join([]string{
					"Does the answer name the expected city? Reply with only 1 for yes or 0 for no.",
					"Question: " + item.Item.Input,
					"Expected: " + item.Item.Expected,
					"Answer: " + item.Item.Output,
				}, "\n")}
			}
			return submitBatch(ctx, client, requests, info)
		},
		// Reuses the task's polling callback, since both are OpenAI batches.
		Completion: eval.WorkflowCompletion{Poll: pollBatch(client)},
		// Called after pollBatch reports the grading batch complete.
		Collect: func(ctx context.Context, data json.RawMessage, _ eval.WorkflowSubmissionContext, yield func(eval.WorkflowBatchResult[eval.Scores]) error) error {
			return collectText(ctx, client, data, func(customID, text string, err error) error {
				result := eval.WorkflowBatchResult[eval.Scores]{CustomID: customID, Error: err}
				// Records an unexpected reply as a failed score for that case.
				switch verdict := strings.TrimSpace(text); {
				case err != nil:
				case verdict == "1":
					result.Result = eval.S(1)
				case verdict == "0":
					result.Result = eval.S(0)
				default:
					result.Error = fmt.Errorf("unexpected judge response: %s", verdict)
				}
				return yield(result)
			})
		},
	}
}

Then add WorkflowScorers: []eval.WorkflowScorerDefinition[string, string]{newLLMJudge(client)} to the definition in newEvaluation and run the program again. Your exactMatch scorer runs as soon as the answers arrive. The same Poll submits one grading batch for both answers, and a later Poll collects the grades.

Store

The Store field saves the run’s progress between calls:

eval.NewWorkflowEvalMemoryStore(): Keeps progress in memory. Use it when the whole run happens in one process, as in the example.
eval.NewWorkflowEvalRedisStore(): Keeps progress in Redis, through a function that runs Redis commands with your Redis client. Use it when the run must survive process exits or be resumed by another process. Records expire after seven days by default.

For Redis options and custom stores, see Workflow stores. Every process that works on a run must use the same store, key prefix, and evaluation definition.

Start the run

Next, main in main.go starts a run by calling Start, which creates the experiment, submits the first batches, and returns an *eval.WorkflowResult with the RunID, without waiting for the provider. With an ordinary task function, Start runs the task and the ordinary scorers for every case before it returns. Save the RunID. You pass it to Poll and Status. If a callback fails after the run is saved, Start returns the run ID along with the error, so you can still resume the run.

Calling Start again creates a new run and submits every request to the provider again. To continue an existing run, pass its run ID to Poll, or pass a completion event to ProcessSubmissionResult.

Resume until completion

Finally, the example calls Poll in a loop until every case is scored. A waiting run can resume in either of two ways, or both:

Polling: Call evaluation.Poll(ctx, runID) on a schedule.
Webhooks: Call evaluation.ProcessSubmissionResult() when the provider sends a completion event.

Each task and scorer chooses its own method. If the task uses webhooks and a scorer uses polling, you still need to schedule Poll to collect the scorer results. Serialize the calls that advance a run, including across processes. The store’s operations are atomic, but the SDK doesn’t lock a run while it advances it. Different runs can advance at the same time. Check progress. Each call returns a result whose Status stays eval.WorkflowRunWaiting until the run completes, and whose PendingPoll and PendingWebhook fields count the batches not yet collected. To check without advancing the run, call evaluation.Status(ctx, runID). Get the results. When the status is eval.WorkflowRunCompleted, the result’s Scores has the average of each score, Failed counts the cases with a failed request or scorer, and Permalink links to the experiment.

Resume with polling

The main example runs in one process that waits until the run finishes, and it keeps progress in memory.For runs that take hours, an alternative is a separate program, such as the following resume.go, that runs in place of main.go. Each time it runs, it does one step and exits, and it saves progress in Redis so the next run continues where the last one stopped. It reuses newEvaluation from evaluation.go. It requires the go-redis client (go get github.com/redis/go-redis/v9) and a REDIS_URL environment variable, such as redis://localhost:6379:

resume.go

package main

import (
	"context"
	"errors"
	"fmt"
	"log"
	"os"
	"time"

	"github.com/openai/openai-go"
	"github.com/redis/go-redis/v9"
	"go.opentelemetry.io/otel/sdk/trace"

	"github.com/braintrustdata/braintrust-sdk-go"
	"github.com/braintrustdata/braintrust-sdk-go/eval"
)

func main() {
	ctx := context.Background()
	tp := trace.NewTracerProvider()
	defer tp.Shutdown(ctx)
	bt, err := braintrust.New(tp, braintrust.WithBlockingLogin(true))
	if err != nil {
		log.Fatal(err)
	}
	options, err := redis.ParseURL(os.Getenv("REDIS_URL"))
	if err != nil {
		log.Fatal(err)
	}
	rdb := redis.NewClient(options)
	defer rdb.Close()
	// Saves run state in Redis, so any process with the same Redis URL and
	// evaluation definition can resume the run.
	store, err := eval.NewWorkflowEvalRedisStore(func(ctx context.Context, args ...any) (any, error) {
		value, err := rdb.Do(ctx, args...).Result()
		// The store expects nil, nil for a missing key.
		if errors.Is(err, redis.Nil) {
			return nil, nil
		}
		return value, err
	}, "capital-cities:", 7*24*time.Hour)
	if err != nil {
		log.Fatal(err)
	}
	evaluation := newEvaluation(bt, openai.NewClient(), store, false)
	// With no argument, starts a new run and prints its run ID. With a run
	// ID, checks the run's pending batches once and advances any cases whose
	// results are ready.
	var result *eval.WorkflowResult
	if len(os.Args) > 1 {
		result, err = evaluation.Poll(ctx, os.Args[1])
	} else {
		result, err = evaluation.Start(ctx, eval.WorkflowStartOpts[string, string]{Dataset: capitalCities()})
	}
	// Start returns the run ID even if a callback fails after the run is
	// saved, so print it before checking the error.
	if result != nil {
		fmt.Println(result.RunID, result.Status)
	}
	if err != nil {
		log.Fatal(err)
	}
}

To use it:

Running go run evaluation.go resume.go with no arguments submits the requests and prints the run ID.
Running go run evaluation.go resume.go <run-id> advances the run once. A scheduled job can run it periodically until it prints completed.

Resume from provider webhooks

Use webhook completion when your provider sends an event when a request finishes. For OpenAI, configure a webhook endpoint for batch.completed, batch.failed, batch.expired, and batch.cancelled events, and use a shared Redis store. The following webhook.go reuses newEvaluation from evaluation.go, with webhook completion:

webhook.go

package main

import (
	"context"
	"errors"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"time"

	"github.com/openai/openai-go"
	"github.com/redis/go-redis/v9"
	"go.opentelemetry.io/otel/sdk/trace"

	"github.com/braintrustdata/braintrust-sdk-go"
	"github.com/braintrustdata/braintrust-sdk-go/eval"
)

func main() {
	ctx := context.Background()
	tp := trace.NewTracerProvider()
	defer tp.Shutdown(ctx)
	bt, err := braintrust.New(tp, braintrust.WithBlockingLogin(true))
	if err != nil {
		log.Fatal(err)
	}
	options, err := redis.ParseURL(os.Getenv("REDIS_URL"))
	if err != nil {
		log.Fatal(err)
	}
	rdb := redis.NewClient(options)
	defer rdb.Close()
	store, err := eval.NewWorkflowEvalRedisStore(func(ctx context.Context, args ...any) (any, error) {
		value, err := rdb.Do(ctx, args...).Result()
		if errors.Is(err, redis.Nil) {
			return nil, nil
		}
		return value, err
	}, "capital-cities:", 7*24*time.Hour)
	if err != nil {
		log.Fatal(err)
	}
	// Reads the signing secret from OPENAI_WEBHOOK_SECRET.
	client := openai.NewClient()
	// Uses webhook completion. After Submit, the SDK saves the batch ID, so a
	// later webhook can find the submission.
	evaluation := newEvaluation(bt, client, store, true)

	// Running with the start argument starts a run that completes through
	// webhooks, then exits.
	if len(os.Args) > 1 && os.Args[1] == "start" {
		result, err := evaluation.Start(ctx, eval.WorkflowStartOpts[string, string]{Dataset: capitalCities()})
		if result != nil {
			fmt.Println(result.RunID, result.Status)
		}
		if err != nil {
			log.Fatal(err)
		}
		return
	}

	http.HandleFunc("POST /webhooks/openai", func(w http.ResponseWriter, r *http.Request) {
		body, err := io.ReadAll(r.Body)
		if err != nil {
			http.Error(w, err.Error(), http.StatusBadRequest)
			return
		}
		// Checks that the request came from OpenAI and parses the event.
		event, err := client.Webhooks.Unwrap(body, r.Header)
		if err != nil {
			http.Error(w, err.Error(), http.StatusBadRequest)
			return
		}
		// Handles the events OpenAI sends when it stops working on a batch,
		// and ignores the rest.
		switch event.Type {
		case "batch.completed", "batch.failed", "batch.expired", "batch.cancelled":
		default:
			return
		}
		// The SDK finds the run and the submission by batch ID, then calls
		// the task's Collect callback. It records each request without a
		// result, such as in an expired batch, as failed.
		result, err := evaluation.ProcessSubmissionResult(r.Context(), eval.WorkflowSubmissionResult{ExternalID: event.Data.ID})
		// Skips batches that this evaluation didn't submit, and batches that
		// use polling.
		if result == nil && err != nil && err.Error() == "unknown workflow external ID" {
			return
		}
		if err != nil {
			// Returns an error status so OpenAI delivers the event again.
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}
		log.Println(result.RunID, result.Status)
	})
	log.Fatal(http.ListenAndServe(":8080", nil))
}

Set OPENAI_WEBHOOK_SECRET to the endpoint’s signing secret from OpenAI. Start a run with go run evaluation.go webhook.go start, then serve the endpoint with go run evaluation.go webhook.go. The handler returns an error status when ProcessSubmissionResult fails, so OpenAI redelivers the event.Keep in mind:

Failed batches: The example passes batch.failed, batch.expired, and batch.cancelled events to ProcessSubmissionResult along with batch.completed. The SDK records each request without a result as failed, so the run still completes.
Run ID: The example passes only the batch ID as ExternalID. That works when each batch ID belongs to one run of the evaluation, as OpenAI batch IDs do. If several runs can share an external ID, also set RunID.
Security: ProcessSubmissionResult doesn’t check that the event came from the provider. In the example, Unwrap does.
Repeated events: The same event can arrive more than once. The SDK ignores a repeated event for a submission it already collected, but your Collect callbacks can run more than once, so keep them free of side effects that can’t repeat.

For how the SDK matches an event to a submission, see eval.WorkflowSubmissionResult.

Handle failures and concurrency

Concurrency: MaxConcurrency limits how many of your callbacks the SDK runs at once during each Start, Poll, or ProcessSubmissionResult call. It defaults to 10. For a batch task or scorer, each Submit call sends one batch.
Provider failures: When your polling callback returns eval.WorkflowFailed, Poll returns an error. The SDK checks the batch again on the next Poll, so a permanent failure makes every later Poll return an error, and the run never completes. Start a new run to recover. The example’s pollBatch avoids this by reporting eval.WorkflowComplete for failed, expired, and cancelled batches, so Collect records the requests without results as failed. The SDK doesn’t resubmit or cancel requests that the provider accepted, so retries and cancellation are up to your integration.
Errors in your callbacks: For polling, the SDK calls your polling callback and your Collect callback again on the next Poll. For webhooks, Collect runs again only when ProcessSubmissionResult is called again, such as when the provider redelivers the event. If your Submit callback returns an error, the next Poll calls it again with the same items and SubmissionID, for both polling and webhook completion. A crash after the provider accepts a request but before the SDK saves the handle can also call Submit again, so use the SubmissionID as an idempotency key if your provider supports one.
Task output: Task outputs must be JSON-serializable. An ordinary task function that returns eval.TaskOutput with UserData fails, because the store can’t save it. Use metadata instead.

Submit one request per case

An eval.WorkflowTask or eval.WorkflowScorer submits one request for each case and trial instead of batching them. Use one when your provider’s asynchronous API takes one request per job, or when each case’s scorers should start as soon as that case’s output is ready. The following task sends each case to OpenAI as its own single-request batch. To try it, replace the eval.WorkflowBatchTask in newEvaluation with this task, and add "errors" to the imports in evaluation.go:

Task: eval.WorkflowTask[string, string]{
	// Called once for each case, or once per trial if you set
	// TrialCount. Sends the case as a single-request batch.
	Submit: func(ctx context.Context, item eval.WorkflowTaskItem[string, string], info eval.WorkflowSubmissionContext) (json.RawMessage, error) {
		return submitBatch(ctx, client, []batchRequest{{
			CustomID: info.SubmissionID,
			Prompt:   item.Input + " Reply with only the city name.",
		}}, info)
	},
	Completion: completion,
	// Returns the batch's only result. Returning an error makes the SDK
	// call Collect again on the next Poll.
	Collect: func(ctx context.Context, data json.RawMessage, _ eval.WorkflowSubmissionContext) (eval.WorkflowTaskResult[string], error) {
		var result eval.WorkflowTaskResult[string]
		found := false
		err := collectText(ctx, client, data, func(_, text string, err error) error {
			found = true
			result.Output = text
			return err
		})
		if err == nil && !found {
			err = errors.New("batch has no result")
		}
		return result, err
	},
},

It differs from an eval.WorkflowBatchTask in these ways:

Items: Submit receives one eval.WorkflowTaskItem, without a CustomID. The SubmissionID is unique to each case and trial, so the example uses it as OpenAI’s custom_id.
Results: Collect returns one result, an eval.WorkflowTaskResult for an eval.WorkflowTask or an eval.Scores for an eval.WorkflowScorer, instead of calling yield.
Failures: If Collect returns an error, the SDK calls it again on the next Poll, so a request that can’t succeed, such as one in an expired batch, keeps the run waiting. Start a new run to recover.
Scorers: An eval.WorkflowScorer submits each case’s request as soon as that case’s task output is ready.

See eval.WorkflowTask and eval.WorkflowScorer.

Next steps

Review braintrust.NewWorkflowEval and the WorkflowEval methods.
Learn how to run experiments in code.
Add custom scorers to measure task quality.

Use the SDK

Features

Reference

Workflow evaluations

How workflow evaluations work

Example: OpenAI Batch API

Define an evaluation

Start the run

Resume until completion

Submit one request per case

Next steps

​How workflow evaluations work

​Example: OpenAI Batch API

​Define an evaluation

​Start the run

​Resume until completion

​Submit one request per case

​Next steps

How workflow evaluations work

Example: OpenAI Batch API

Define an evaluation

Start the run

Resume until completion

Submit one request per case

Next steps