Documentation
¶
Overview ¶
Package x provides experimental Genkit APIs.
APIs in this package are under active development and may change in any minor version release. Use with caution in production environments.
When these APIs stabilize, they will be moved to the genkit package and these exports will be deprecated.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func DefineStreamingFlow ¶
func DefineStreamingFlow[In, Out, Stream any](g *genkit.Genkit, name string, fn StreamingFunc[In, Out, Stream]) *core.Flow[In, Out, Stream]
DefineStreamingFlow defines a streaming flow that uses a channel for streaming, registers it as a core.Action of type Flow, and returns a core.Flow runner.
Unlike genkit.DefineStreamingFlow which uses a callback, this function accepts a StreamingFunc that writes stream chunks to a channel. This can be more ergonomic when integrating with other channel-based APIs or when the streaming logic is more naturally expressed with channels.
The channel passed to the function is unbuffered and managed by the framework. The function should NOT close the channel - it will be closed automatically after the function returns.
Example:
countdown := x.DefineStreamingFlow(g, "countdown",
func(ctx context.Context, start int, streamCh chan<- int) (string, error) {
for i := start; i > 0; i-- {
select {
case streamCh <- i:
case <-ctx.Done():
return "", ctx.Err()
}
}
return "liftoff!", nil
})
// Run with streaming
for val, err := range countdown.Stream(ctx, 5) {
if err != nil {
log.Fatal(err)
}
if val.Done {
fmt.Println(val.Output) // "liftoff!"
} else {
fmt.Println(val.Stream) // 5, 4, 3, 2, 1
}
}
Types ¶
type StreamingFunc ¶
type StreamingFunc[In, Out, Stream any] = func(ctx context.Context, input In, streamCh chan<- Stream) (Out, error)
StreamingFunc is a streaming function that uses a channel instead of a callback.
The function receives a send-only channel to which it should write stream chunks. The channel is managed by the framework and will be closed automatically after the function returns. The function should NOT close the channel itself.
When writing to the channel, the function should respect context cancellation:
select {
case streamCh <- chunk:
case <-ctx.Done():
return zero, ctx.Err()
}
Source Files