core

Overview ¶

Package core implements Genkit actions and other essential machinery. This package is primarily intended for Genkit internals and for plugins. Genkit applications should use the genkit package.

Package core implements Genkit's foundational action system and runtime machinery.

This package is primarily intended for plugin developers and Genkit internals. Application developers should use the genkit package instead, which provides a higher-level, more convenient API.

Actions ¶

Actions are the fundamental building blocks of Genkit. Every operation - flows, model calls, tool invocations, retrieval - is implemented as an action. Actions provide:

• Type-safe input/output with JSON schema validation
• Automatic tracing and observability
• Consistent error handling
• Registration in the action registry

Define a non-streaming action:

action := core.DefineAction(registry, "myAction",
	func(ctx context.Context, input string) (string, error) {
		return "processed: " + input, nil
	},
)

result, err := action.Run(context.Background(), "hello")

Define a streaming action that sends chunks during execution:

streamingAction := core.DefineStreamingAction(registry, "countdown",
	func(ctx context.Context, start int, cb core.StreamCallback[string]) (string, error) {
		for i := start; i > 0; i-- {
			if cb != nil {
				if err := cb(ctx, fmt.Sprintf("T-%d", i)); err != nil {
					return "", err
				}
			}
			time.Sleep(time.Second)
		}
		return "Liftoff!", nil
	},
)

Flows ¶

Flows are user-defined actions that orchestrate AI operations. They are the primary way application developers define business logic in Genkit:

flow := core.DefineFlow(registry, "myFlow",
	func(ctx context.Context, input string) (string, error) {
		// Use Run to create traced sub-steps
		result, err := core.Run(ctx, "step1", func() (string, error) {
			return process(input), nil
		})
		if err != nil {
			return "", err
		}
		return result, nil
	},
)

Streaming flows can send intermediate results to callers:

streamingFlow := core.DefineStreamingFlow(registry, "generateReport",
	func(ctx context.Context, input Input, cb core.StreamCallback[Progress]) (Report, error) {
		for i := 0; i < 100; i += 10 {
			if cb != nil {
				cb(ctx, Progress{Percent: i})
			}
			// ... work ...
		}
		return Report{...}, nil
	},
)

Traced Steps with Run ¶

Use Run within flows to create traced sub-operations. Each Run call creates a span in the trace that's visible in the Genkit Developer UI:

result, err := core.Run(ctx, "fetchData", func() (Data, error) {
	return fetchFromAPI()
})

processed, err := core.Run(ctx, "processData", func() (Result, error) {
	return process(result)
})

Middleware ¶

Actions support middleware for cross-cutting concerns like logging, metrics, or authentication:

loggingMiddleware := func(next core.StreamingFunc[string, string, struct{}]) core.StreamingFunc[string, string, struct{}] {
	return func(ctx context.Context, input string, cb core.StreamCallback[struct{}]) (string, error) {
		log.Printf("Input: %s", input)
		output, err := next(ctx, input, cb)
		log.Printf("Output: %s, Error: %v", output, err)
		return output, err
	}
}

Chain multiple middleware together:

combined := core.ChainMiddleware(loggingMiddleware, metricsMiddleware)
wrappedFn := combined(originalFunc)

Schema Management ¶

Register JSON schemas for use in prompts and validation:

// Define a schema from a map
core.DefineSchema(registry, "Person", map[string]any{
	"type": "object",
	"properties": map[string]any{
		"name": map[string]any{"type": "string"},
		"age":  map[string]any{"type": "integer"},
	},
	"required": []any{"name"},
})

// Define a schema from a Go type (recommended)
core.DefineSchemaFor[Person](registry)

Schemas can be referenced in .prompt files by name.

Plugin Development ¶

Plugins extend Genkit's functionality by providing models, tools, retrievers, and other capabilities. Implement the api.Plugin interface:

type MyPlugin struct {
	APIKey string
}

func (p *MyPlugin) Name() string {
	return "myplugin"
}

func (p *MyPlugin) Init(ctx context.Context) []api.Action {
	// Initialize the plugin and return actions to register
	model := ai.DefineModel(...)
	tool := ai.DefineTool(...)
	return []api.Action{model, tool}
}

For plugins that resolve actions dynamically (e.g., listing available models from an API), implement api.DynamicPlugin:

type DynamicModelPlugin struct{}

func (p *DynamicModelPlugin) ListActions(ctx context.Context) []api.ActionDesc {
	// Return descriptors of available actions
	return []api.ActionDesc{
		{Key: "/model/myplugin/model-a", Name: "model-a"},
		{Key: "/model/myplugin/model-b", Name: "model-b"},
	}
}

func (p *DynamicModelPlugin) ResolveAction(atype api.ActionType, name string) api.Action {
	// Create and return the action on demand
	return createModel(name)
}

Background Actions ¶

For long-running operations, use background actions that return immediately with an operation ID that can be polled for completion:

bgAction := core.DefineBackgroundAction(registry, "longTask",
	func(ctx context.Context, input Input) (Output, error) {
		// Start the operation
		return startLongOperation(input)
	},
	func(ctx context.Context, op *core.Operation[Output]) (*core.Operation[Output], error) {
		// Check operation status
		return checkOperationStatus(op)
	},
)

Error Handling ¶

Return user-facing errors with appropriate status codes:

if err := validate(input); err != nil {
	return nil, core.NewPublicError(core.INVALID_ARGUMENT, "Invalid input", map[string]any{
		"field": "email",
		"error": err.Error(),
	})
}

For internal errors that should be logged but not exposed to users:

return nil, core.NewError(core.INTERNAL, "database connection failed: %v", err)

Context ¶

Access action context for metadata and configuration:

ctx := core.FromContext(ctx)
if ctx != nil {
	// Access action-specific context values
}

Set action context for nested operations:

ctx = core.WithActionContext(ctx, core.ActionContext{
	"requestId": requestID,
})

For more information, see https://genkit.dev/docs/plugins

Package core provides base error types and utilities for Genkit.

Package status defines canonical status codes, names, and related types inspired by gRPC status codes.