pubsub

Documentation

Index

• type Broker
  • func LocalBroker() Broker
• type Chunk
  • func (c Chunk[T]) MarshalJSON() ([]byte, error)
  • func (c *Chunk[T]) UnmarshalJSON(data []byte) error
• type CompositeHook
  • func (c CompositeHook) OnAssistantChunk(ctx context.Context, ac messages.Message[messages.AssistantMessage])
  • func (c CompositeHook) OnAssistantMessage(ctx context.Context, am messages.Message[messages.AssistantMessage])
  • func (c CompositeHook) OnError(ctx context.Context, err error)
  • func (c CompositeHook) OnToolCallChunk(ctx context.Context, tc messages.Message[messages.ToolCallMessage])
  • func (c CompositeHook) OnToolCallMessage(ctx context.Context, tm messages.Message[messages.ToolCallMessage])
  • func (c CompositeHook) OnToolCallResponse(ctx context.Context, tr messages.Message[messages.ToolResponse])
  • func (c CompositeHook) OnUserPrompt(ctx context.Context, up messages.Message[messages.UserMessage])
• type Delim
  • func (d Delim) MarshalJSON() ([]byte, error)
  • func (d *Delim) UnmarshalJSON(data []byte) error
• type Error
  • func (e Error) Error() string
  • func (e Error) MarshalJSON() ([]byte, error)
  • func (e *Error) UnmarshalJSON(data []byte) error
• type Event
  • func FromStreamEvent(e provider.StreamEvent, sender string) Event
• type Hook
  • func LoggingHook() Hook
  • func NewCompositeHook(hooks ...Hook) Hook
• type Request
  • func (r Request[T]) MarshalJSON() ([]byte, error)
  • func (r *Request[T]) UnmarshalJSON(data []byte) error
• type Response
  • func (r Response[T]) MarshalJSON() ([]byte, error)
  • func (r *Response[T]) UnmarshalJSON(data []byte) error
• type Subscription
• type Topic

Types

type Broker

type Broker interface {
	Topic(context.Context, string) Topic
}

func LocalBroker

func LocalBroker() Broker

type Chunk

type Chunk[T messages.Response] struct {
	RunID     uuid.UUID       `json:"run_id"`
	TurnID    uuid.UUID       `json:"turn_id"`
	Chunk     T               `json:"chunk"`
	Sender    string          `json:"sender,omitempty"`
	Timestamp strfmt.DateTime `json:"timestamp,omitempty"`
	Meta      gjson.Result    `json:"meta,omitempty"`
}

func (Chunk[T]) MarshalJSON

func (c Chunk[T]) MarshalJSON() ([]byte, error)

MarshalJSON implements custom JSON marshaling for Chunk[T]

func (*Chunk[T]) UnmarshalJSON

func (c *Chunk[T]) UnmarshalJSON(data []byte) error

UnmarshalJSON implements custom JSON unmarshaling for Chunk[T]

type CompositeHook

type CompositeHook []Hook

CompositeHook allows combining multiple hooks into a single hook implementation. Note: This is provided as a utility for combining hooks, not as a way to avoid implementing the full interface.

func (CompositeHook) OnAssistantChunk

func (c CompositeHook) OnAssistantChunk(ctx context.Context, ac messages.Message[messages.AssistantMessage])

func (CompositeHook) OnAssistantMessage

func (c CompositeHook) OnAssistantMessage(ctx context.Context, am messages.Message[messages.AssistantMessage])

func (CompositeHook) OnError

func (c CompositeHook) OnError(ctx context.Context, err error)

func (CompositeHook) OnToolCallChunk

func (c CompositeHook) OnToolCallChunk(ctx context.Context, tc messages.Message[messages.ToolCallMessage])

func (CompositeHook) OnToolCallMessage

func (c CompositeHook) OnToolCallMessage(ctx context.Context, tm messages.Message[messages.ToolCallMessage])

func (CompositeHook) OnToolCallResponse

func (c CompositeHook) OnToolCallResponse(ctx context.Context, tr messages.Message[messages.ToolResponse])

func (CompositeHook) OnUserPrompt

func (c CompositeHook) OnUserPrompt(ctx context.Context, up messages.Message[messages.UserMessage])

type Delim

type Delim struct {
	RunID  uuid.UUID `json:"run_id"`
	TurnID uuid.UUID `json:"turn_id"`
	Delim  string    `json:"delim"`
}

func (Delim) MarshalJSON

func (d Delim) MarshalJSON() ([]byte, error)

MarshalJSON implements custom JSON marshaling for Delim

func (*Delim) UnmarshalJSON

func (d *Delim) UnmarshalJSON(data []byte) error

UnmarshalJSON implements custom JSON unmarshaling for Delim

type Error

type Error struct {
	RunID     uuid.UUID       `json:"run_id"`
	TurnID    uuid.UUID       `json:"turn_id"`
	Err       error           `json:"error"`
	Sender    string          `json:"sender,omitempty"`
	Timestamp strfmt.DateTime `json:"timestamp,omitempty"`
	Meta      gjson.Result    `json:"meta,omitempty"`
}

func (Error) Error

func (e Error) Error() string

func (Error) MarshalJSON

func (e Error) MarshalJSON() ([]byte, error)

MarshalJSON implements custom JSON marshaling for Error

func (*Error) UnmarshalJSON

func (e *Error) UnmarshalJSON(data []byte) error

UnmarshalJSON implements custom JSON unmarshaling for Error

type Event

type Event interface {
	// contains filtered or unexported methods
}

func FromStreamEvent

func FromStreamEvent(e provider.StreamEvent, sender string) Event

type Hook

type Hook interface {
	OnUserPrompt(context.Context, messages.Message[messages.UserMessage])

	OnAssistantChunk(context.Context, messages.Message[messages.AssistantMessage])

	OnToolCallChunk(context.Context, messages.Message[messages.ToolCallMessage])

	OnAssistantMessage(context.Context, messages.Message[messages.AssistantMessage])

	OnToolCallMessage(context.Context, messages.Message[messages.ToolCallMessage])

	OnToolCallResponse(context.Context, messages.Message[messages.ToolResponse])

	OnError(context.Context, error)
}

Hook defines the interface for handling all possible event types in the execution flow. This interface is deliberately designed without a base "no-op" implementation to ensure consumers make explicit decisions about handling each event type.

Design decisions:

1. All methods must be implemented: This is a conscious choice to ensure compile-time safety. When new event types are added, all implementations will need to be updated.
2. No provided no-op implementation: While it might be convenient to provide a NoOpHook, doing so would undermine the interface's primary benefit of forcing conscious decisions about event handling.
3. Complete coverage: The interface covers all possible event types to ensure no events can be accidentally missed in implementations.

Implementation guidelines:

• Implement all methods explicitly, even if some events don't require handling
• Consider logging or monitoring for events that aren't actively handled
• Be prepared for new methods to be added as the system evolves

Example implementation:

type MyHandler struct{}

func (h *MyHandler) OnUserPrompt(ctx context.Context, msg messages.Message[messages.UserMessage]) {
    // Explicit handling of user prompts
}
func (h *MyHandler) OnAssistantChunk(...) {
    // Explicit decision to not handle chunks
    log.Debug("ignoring assistant chunk")
}
// ... must implement all other methods

func LoggingHook

func LoggingHook() Hook

func NewCompositeHook

func NewCompositeHook(hooks ...Hook) Hook

type Request

type Request[T messages.Request] struct {
	RunID     uuid.UUID       `json:"run_id"`
	TurnID    uuid.UUID       `json:"turn_id"`
	Message   T               `json:"message"`
	Sender    string          `json:"sender,omitempty"`
	Timestamp strfmt.DateTime `json:"timestamp,omitempty"`
	Meta      gjson.Result    `json:"meta,omitempty"`
}

func (Request[T]) MarshalJSON

func (r Request[T]) MarshalJSON() ([]byte, error)

MarshalJSON implements custom JSON marshaling for Request[T]

func (*Request[T]) UnmarshalJSON

func (r *Request[T]) UnmarshalJSON(data []byte) error

UnmarshalJSON implements custom JSON unmarshaling for Request[T]

type Response

type Response[T any] struct {
	RunID     uuid.UUID       `json:"run_id"`
	TurnID    uuid.UUID       `json:"turn_id"`
	Response  T               `json:"response"`
	Sender    string          `json:"sender,omitempty"`
	Timestamp strfmt.DateTime `json:"timestamp,omitempty"`
	Meta      gjson.Result    `json:"meta,omitempty"`
}

func (Response[T]) MarshalJSON

func (r Response[T]) MarshalJSON() ([]byte, error)

MarshalJSON implements custom JSON marshaling for Response[T]

func (*Response[T]) UnmarshalJSON

func (r *Response[T]) UnmarshalJSON(data []byte) error

UnmarshalJSON implements custom JSON unmarshaling for Response[T]

type Subscription

type Subscription interface {
	ID() string
	Unsubscribe()
}

type Topic

type Topic interface {
	Publish(context.Context, Event) error
	Subscribe(context.Context, Hook) (Subscription, error)
}