core

package
v0.5.4 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: May 13, 2025 License: Apache-2.0 Imports: 17 Imported by: 0

Documentation

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 provides base error types and utilities for Genkit.

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

Index

Constants

View Source 
const (
	// CodeOK means not an error; returned on success.
	CodeOK = 0
	// CodeCancelled means the operation was cancelled, typically by the caller.
	CodeCancelled = 1
	// CodeUnknown means an unknown error occurred.
	CodeUnknown = 2
	// CodeInvalidArgument means the client specified an invalid argument.
	CodeInvalidArgument = 3
	// CodeDeadlineExceeded means the deadline expired before the operation could complete.
	CodeDeadlineExceeded = 4
	// CodeNotFound means some requested entity (e.g., file or directory) was not found.
	CodeNotFound = 5
	// CodeAlreadyExists means the entity that a client attempted to create already exists.
	CodeAlreadyExists = 6
	// CodePermissionDenied means the caller does not have permission to execute the operation.
	CodePermissionDenied = 7
	// CodeUnauthenticated means the request does not have valid authentication credentials.
	CodeUnauthenticated = 16
	// CodeResourceExhausted means some resource has been exhausted.
	CodeResourceExhausted = 8
	// CodeFailedPrecondition means the operation was rejected because the system is not in a state required.
	CodeFailedPrecondition = 9
	// CodeAborted means the operation was aborted, typically due to some issue.
	CodeAborted = 10
	// CodeOutOfRange means the operation was attempted past the valid range.
	CodeOutOfRange = 11
	// CodeUnimplemented means the operation is not implemented or not supported/enabled.
	CodeUnimplemented = 12
	// CodeInternal means internal errors. Some invariants expected by the underlying system were broken.
	CodeInternal = 13
	// CodeUnavailable means the service is currently unavailable.
	CodeUnavailable = 14
	// CodeDataLoss means unrecoverable data loss or corruption.
	CodeDataLoss = 15
)

Constants for canonical status codes (integer values).

Variables

View Source 
var StatusNameToCode = map[StatusName]int{
	OK:                  CodeOK,
	CANCELLED:           CodeCancelled,
	UNKNOWN:             CodeUnknown,
	INVALID_ARGUMENT:    CodeInvalidArgument,
	DEADLINE_EXCEEDED:   CodeDeadlineExceeded,
	NOT_FOUND:           CodeNotFound,
	ALREADY_EXISTS:      CodeAlreadyExists,
	PERMISSION_DENIED:   CodePermissionDenied,
	UNAUTHENTICATED:     CodeUnauthenticated,
	RESOURCE_EXHAUSTED:  CodeResourceExhausted,
	FAILED_PRECONDITION: CodeFailedPrecondition,
	ABORTED:             CodeAborted,
	OUT_OF_RANGE:        CodeOutOfRange,
	UNIMPLEMENTED:       CodeUnimplemented,
	INTERNAL:            CodeInternal,
	UNAVAILABLE:         CodeUnavailable,
	DATA_LOSS:           CodeDataLoss,
}

StatusNameToCode maps status names to their integer code values. Exported for potential use elsewhere if needed.

Functions

func HTTPStatusCode added in v0.5.3 

func HTTPStatusCode(name StatusName) int

HTTPStatusCode gets the corresponding HTTP status code for a given Genkit status name.

func RegisterSpanProcessor 

func RegisterSpanProcessor(r *registry.Registry, sp sdktrace.SpanProcessor)

RegisterSpanProcessor registers an OpenTelemetry SpanProcessor for tracing.

func Run added in v0.3.0 

func Run[Out any](ctx context.Context, name string, fn func() (Out, error)) (Out, error)

Run runs the function f in the context of the current flow and returns what f returns. It returns an error if no flow is active.

Each call to Run results in a new step in the flow. A step has its own span in the trace, and its result is cached so that if the flow is restarted, f will not be called a second time.

func WithActionContext added in v0.3.0 

func WithActionContext(ctx context.Context, actionCtx ActionContext) context.Context

WithActionContext returns a new Context with Action runtime context (side channel data) value set.

Types

type Action 

type Action interface {
	// Name returns the name of the action.
	Name() string
	// RunJSON runs the action with the given JSON input and streaming callback and returns the output as JSON.
	RunJSON(ctx context.Context, input json.RawMessage, cb func(context.Context, json.RawMessage) error) (json.RawMessage, error)
}

Action is the interface that all Genkit primitives (e.g. flows, models, tools) have in common.

type ActionContext added in v0.3.0 

type ActionContext = map[string]any

ActionContext is the runtime context for an Action.

func FromContext added in v0.3.0 

func FromContext(ctx context.Context) ActionContext

FromContext returns the Action runtime context (side channel data) from context.

type ActionDef added in v0.3.0 

type ActionDef[In, Out, Stream any] struct {
	// contains filtered or unexported fields
}

An ActionDef is a named, observable operation that underlies all Genkit primitives. It consists of a function that takes an input of type I and returns an output of type O, optionally streaming values of type S incrementally by invoking a callback. It optionally has other metadata, like a description and JSON Schemas for its input and output which it validates against.

Each time an ActionDef is run, it results in a new trace span.

func DefineAction 

func DefineAction[In, Out any](
	r *registry.Registry,
	provider, name string,
	atype atype.ActionType,
	metadata map[string]any,
	fn Func[In, Out],
) *ActionDef[In, Out, struct{}]

DefineAction creates a new non-streaming Action and registers it.

func DefineActionWithInputSchema 

func DefineActionWithInputSchema[Out any](
	r *registry.Registry,
	provider, name string,
	atype atype.ActionType,
	metadata map[string]any,
	inputSchema *jsonschema.Schema,
	fn Func[any, Out],
) *ActionDef[any, Out, struct{}]

DefineActionWithInputSchema creates a new Action and registers it. This differs from DefineAction in that the input schema is defined dynamically; the static input type is "any". This is used for prompts.

func DefineStreamingAction 

func DefineStreamingAction[In, Out, Stream any](
	r *registry.Registry,
	provider, name string,
	atype atype.ActionType,
	metadata map[string]any,
	fn StreamingFunc[In, Out, Stream],
) *ActionDef[In, Out, Stream]

DefineStreamingAction creates a new streaming action and registers it.

func LookupActionFor 

func LookupActionFor[In, Out, Stream any](r *registry.Registry, typ atype.ActionType, provider, name string) *ActionDef[In, Out, Stream]

LookupActionFor returns the action for the given key in the global registry, or nil if there is none. It panics if the action is of the wrong type.

func (*ActionDef[In, Out, Stream]) Desc added in v0.3.0 

func (a *ActionDef[In, Out, Stream]) Desc() action.Desc

Desc returns a description of the action.

func (*ActionDef[In, Out, Stream]) Name added in v0.3.0 

func (a *ActionDef[In, Out, Stream]) Name() string

Name returns the Action's Name.

func (*ActionDef[In, Out, Stream]) Run added in v0.3.0 

func (a *ActionDef[In, Out, Stream]) Run(ctx context.Context, input In, cb StreamCallback[Stream]) (output Out, err error)

Run executes the Action's function in a new trace span.

func (*ActionDef[In, Out, Stream]) RunJSON added in v0.3.0 

func (a *ActionDef[In, Out, Stream]) RunJSON(ctx context.Context, input json.RawMessage, cb StreamCallback[json.RawMessage]) (json.RawMessage, error)

RunJSON runs the action with a JSON input, and returns a JSON result.

type ContextProvider added in v0.3.0 

type ContextProvider = func(ctx context.Context, req RequestData) (ActionContext, error)

ContextProvider is a function that returns an ActionContext for a given request. It is used to provide additional context to the Action.

type Flow 

type Flow[In, Out, Stream any] ActionDef[In, Out, Stream]

A Flow is a user-defined Action. A Flow[In, Out, Stream] represents a function from In to Out. The Stream parameter is for flows that support streaming: providing their results incrementally.

func DefineFlow added in v0.3.0 

func DefineFlow[In, Out any](r *registry.Registry, name string, fn Func[In, Out]) *Flow[In, Out, struct{}]

DefineFlow creates a Flow that runs fn, and registers it as an action. fn takes an input of type In and returns an output of type Out.

func DefineStreamingFlow added in v0.3.0 

func DefineStreamingFlow[In, Out, Stream any](r *registry.Registry, name string, fn StreamingFunc[In, Out, Stream]) *Flow[In, Out, Stream]

DefineStreamingFlow creates a streaming Flow that runs fn, and registers it as an action.

fn takes an input of type In and returns an output of type Out, optionally streaming values of type Stream incrementally by invoking a callback.

If the function supports streaming and the callback is non-nil, it should stream the results by invoking the callback periodically, ultimately returning with a final return value that includes all the streamed data. Otherwise, it should ignore the callback and just return a result.

func (*Flow[In, Out, Stream]) Name 

func (f *Flow[In, Out, Stream]) Name() string

Name returns the name of the flow.

func (*Flow[In, Out, Stream]) Run 

func (f *Flow[In, Out, Stream]) Run(ctx context.Context, input In) (Out, error)

Run runs the flow in the context of another flow.

func (*Flow[In, Out, Stream]) RunJSON added in v0.3.0 

func (f *Flow[In, Out, Stream]) RunJSON(ctx context.Context, input json.RawMessage, cb StreamCallback[json.RawMessage]) (json.RawMessage, error)

RunJSON runs the flow with JSON input and streaming callback and returns the output as JSON.

func (*Flow[In, Out, Stream]) Stream 

func (f *Flow[In, Out, Stream]) Stream(ctx context.Context, input In) func(func(*StreamingFlowValue[Out, Stream], error) bool)

Stream runs the flow in the context of another flow and streams the output. It returns a function whose argument function (the "yield function") will be repeatedly called with the results.

If the yield function is passed a non-nil error, the flow has failed with that error; the yield function will not be called again.

If the yield function's StreamingFlowValue argument has Done == true, the value's Output field contains the final output; the yield function will not be called again.

Otherwise the Stream field of the passed StreamingFlowValue holds a streamed result.

type Func 

type Func[In, Out any] = func(context.Context, In) (Out, error)

Func is an alias for non-streaming functions with input of type In and output of type Out.

type GenkitError added in v0.5.3 

type GenkitError struct {
	Message  string         `json:"message"` // Exclude from default JSON if embedded elsewhere
	Status   StatusName     `json:"status"`
	HTTPCode int            `json:"-"`                // Exclude from default JSON
	Details  map[string]any `json:"details"`          // Use map for arbitrary details
	Source   *string        `json:"source,omitempty"` // Pointer for optional
}

GenkitError is the base error type for Genkit errors.

func NewError added in v0.5.3 

func NewError(status StatusName, message string, args ...any) *GenkitError

NewError creates a new GenkitError with a stack trace.

func (*GenkitError) Error added in v0.5.3 

func (e *GenkitError) Error() string

Error implements the standard error interface.

func (*GenkitError) ToReflectionError added in v0.5.3 

func (e *GenkitError) ToReflectionError() ReflectionError

ToReflectionError returns a JSON-serializable representation for reflection API responses.

type Middleware added in v0.3.0 

type Middleware[In, Out, Stream any] = func(StreamingFunc[In, Out, Stream]) StreamingFunc[In, Out, Stream]

Middleware is a function that wraps an action execution, similar to HTTP middleware. It can modify the input, output, and context, or perform side effects.

func ChainMiddleware added in v0.3.0 

func ChainMiddleware[In, Out, Stream any](middlewares ...Middleware[In, Out, Stream]) Middleware[In, Out, Stream]

ChainMiddleware creates a new Middleware that applies a sequence of Middlewares, so that they execute in the given order when handling action request. In other words, ChainMiddleware(m1, m2)(handler) = m1(m2(handler))

func Middlewares added in v0.3.0 

func Middlewares[In, Out, Stream any](ms ...Middleware[In, Out, Stream]) []Middleware[In, Out, Stream]

Middlewares returns an array of middlewares that are passes in as an argument. core.Middlewares(apple, banana) is identical to []core.Middleware[InputType, OutputType]{apple, banana}

type ReflectionError added in v0.5.3 

type ReflectionError struct {
	Details *ReflectionErrorDetails `json:"details,omitempty"`
	Message string                  `json:"message"`
	Code    int                     `json:"code"`
}

ReflectionError is the wire format for HTTP errors for Reflection API responses.

func ToReflectionError added in v0.5.3 

func ToReflectionError(err error) ReflectionError

ToReflectionError gets the JSON representation for reflection API Error responses.

type ReflectionErrorDetails added in v0.5.3 

type ReflectionErrorDetails struct {
	Stack   *string `json:"stack,omitempty"` // Use pointer for optional
	TraceID *string `json:"traceId,omitempty"`
}

type RequestData added in v0.3.0 

type RequestData struct {
	Method  string            // Method is the HTTP method of the request (e.g. "GET", "POST", etc.)
	Headers map[string]string // Headers is the headers of the request. The keys are the header names in lowercase.
	Input   json.RawMessage   // Input is the body of the request.
}

RequestData is the data associated with a request. It is used to provide additional context to the Action.

type Status added in v0.5.3 

type Status struct {
	Name    StatusName `json:"name"`
	Message string     `json:"message,omitempty"`
}

Status represents a status condition, typically used in responses or errors.

func NewStatus added in v0.5.3 

func NewStatus(name StatusName, message string) *Status

NewStatus creates a new Status object.

type StatusName added in v0.5.3 

type StatusName string

StatusName defines the set of canonical status names. 

const (
	OK                  StatusName = "OK"
	CANCELLED           StatusName = "CANCELLED"
	UNKNOWN             StatusName = "UNKNOWN"
	INVALID_ARGUMENT    StatusName = "INVALID_ARGUMENT"
	DEADLINE_EXCEEDED   StatusName = "DEADLINE_EXCEEDED"
	NOT_FOUND           StatusName = "NOT_FOUND"
	ALREADY_EXISTS      StatusName = "ALREADY_EXISTS"
	PERMISSION_DENIED   StatusName = "PERMISSION_DENIED"
	UNAUTHENTICATED     StatusName = "UNAUTHENTICATED"
	RESOURCE_EXHAUSTED  StatusName = "RESOURCE_EXHAUSTED"
	FAILED_PRECONDITION StatusName = "FAILED_PRECONDITION"
	ABORTED             StatusName = "ABORTED"
	OUT_OF_RANGE        StatusName = "OUT_OF_RANGE"
	UNIMPLEMENTED       StatusName = "UNIMPLEMENTED"
	INTERNAL            StatusName = "INTERNAL_SERVER_ERROR"
	UNAVAILABLE         StatusName = "UNAVAILABLE"
	DATA_LOSS           StatusName = "DATA_LOSS"
)

Constants for canonical status names.

type StreamCallback added in v0.3.0 

type StreamCallback[Stream any] = func(context.Context, Stream) error

StreamCallback is a function that is called during streaming to return the next chunk of the stream.

type StreamingFlowValue added in v0.5.0 

type StreamingFlowValue[Out, Stream any] struct {
	Done   bool
	Output Out    // valid if Done is true
	Stream Stream // valid if Done is false
}

StreamingFlowValue is either a streamed value or a final output of a flow.

type StreamingFunc added in v0.3.0 

type StreamingFunc[In, Out, Stream any] = func(context.Context, In, StreamCallback[Stream]) (Out, error)

StreamingFunc is an alias for streaming functions with input of type In, output of type Out, and streaming chunk of type Stream.

type UserFacingError added in v0.5.3 

type UserFacingError struct {
	Message string         `json:"message"` // Exclude from default JSON if embedded elsewhere
	Status  StatusName     `json:"status"`
	Details map[string]any `json:"details"` // Use map for arbitrary details
}

UserFacingError is the base error type for user facing errors.

func NewPublicError added in v0.5.3 

func NewPublicError(status StatusName, message string, details map[string]any) *UserFacingError

NewPublicError allows a web framework handler to know it is safe to return the message in a request. Other kinds of errors will result in a generic 500 message to avoid the possibility of internal exceptions being leaked to attackers.

func (*UserFacingError) Error added in v0.5.3 

func (e *UserFacingError) Error() string

Error implements the standard error interface for UserFacingError.

Source Files

View all Source files

Directories

Path Synopsis
Package logger provides a context-scoped slog.Logger.
 Package logger provides a context-scoped slog.Logger.
Package gtime provides time functionality for Go Genkit.
 Package gtime provides time functionality for Go Genkit.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.