stages

type ArenaAssertionStage ¶

type ArenaAssertionStage struct {
	stage.BaseStage
	// contains filtered or unexported fields
}

ArenaAssertionStage validates assertions after LLM responses.

func NewArenaAssertionStage(
	registry *runtimeValidators.Registry,
	assertionConfigs []assertions.AssertionConfig,
) *ArenaAssertionStage

func (s *ArenaAssertionStage) Process(
	ctx context.Context,
	input <-chan stage.StreamElement,
	output chan<- stage.StreamElement,
) error

type ArenaStateStoreSaveStage ¶

type ArenaStateStoreSaveStage struct {
	stage.BaseStage
	// contains filtered or unexported fields
}

ArenaStateStoreSaveStage saves conversation state with telemetry to ArenaStateStore. This stage captures validation results, turn metrics, and cost information for Arena testing and analysis.

func NewArenaStateStoreSaveStage(config *pipeline.StateStoreConfig) *ArenaStateStoreSaveStage

func (s *ArenaStateStoreSaveStage) Process(ctx context.Context, input <-chan stage.StreamElement, output chan<- stage.StreamElement) error

type HistoryInjectionStage ¶

type HistoryInjectionStage struct {
	stage.BaseStage
	// contains filtered or unexported fields
}

HistoryInjectionStage prepends conversation history to the pipeline stream. This is useful for self-play scenarios where history needs to be explicitly provided before the LLM generates the next user message.

The stage emits all history messages first, then forwards any incoming elements.

func NewHistoryInjectionStage(history []types.Message) *HistoryInjectionStage

func (s *HistoryInjectionStage) Process(ctx context.Context, input <-chan stage.StreamElement, output chan<- stage.StreamElement) error

type MetadataInjectionStage ¶

type MetadataInjectionStage struct {
	stage.BaseStage
	// contains filtered or unexported fields
}

MetadataInjectionStage injects metadata into stream element metadata.

func NewMetadataInjectionStage(metadata map[string]interface{}) *MetadataInjectionStage

func (s *MetadataInjectionStage) Process(
	ctx context.Context,
	input <-chan stage.StreamElement,
	output chan<- stage.StreamElement,
) error

type MockScenarioContextStage ¶

type MockScenarioContextStage struct {
	stage.BaseStage
	// contains filtered or unexported fields
}

MockScenarioContextStage adds scenario context to the stream elements for MockProvider to use scenario-specific responses.

This stage should be placed before ProviderStage in the pipeline when using MockProvider to ensure scenario context is available.

func NewMockScenarioContextStage(scenario *config.Scenario) *MockScenarioContextStage

func (s *MockScenarioContextStage) Process(ctx context.Context, input <-chan stage.StreamElement, output chan<- stage.StreamElement) error

type PersonaAssemblyStage ¶

type PersonaAssemblyStage struct {
	stage.BaseStage
	// contains filtered or unexported fields
}

PersonaAssemblyStage assembles persona prompts using the same fragment/template system as PromptAssemblyStage. It enriches elements with the persona's assembled prompt and variables.

This stage mirrors the behavior of PromptAssemblyMiddleware but for personas: - Uses persona's BuildSystemPrompt() which handles fragment assembly - Supports template variable substitution with {{variable}} syntax - Injects persona-specific variables (goals, constraints, style) - Sets base variables for downstream template stage

func NewPersonaAssemblyStage(
	persona *config.UserPersonaPack,
	region string,
	baseVariables map[string]string,
) *PersonaAssemblyStage

func (s *PersonaAssemblyStage) Process(ctx context.Context, input <-chan stage.StreamElement, output chan<- stage.StreamElement) error

type ScenarioContextExtractionStage ¶

type ScenarioContextExtractionStage struct {
	stage.BaseStage
	// contains filtered or unexported fields
}

ScenarioContextExtractionStage extracts context using scenario metadata and conversation history. This is designed for Arena use where rich scenario metadata is available.

When scenario metadata is present, it uses: - Scenario metadata variables (domain, user role from scenario definition) - Scenario description and task type - Message analysis as fallback

Extracted variables are merged into element metadata, allowing templates to use them.

func NewScenarioContextExtractionStage(scenario *config.Scenario) *ScenarioContextExtractionStage

func (s *ScenarioContextExtractionStage) Process(ctx context.Context, input <-chan stage.StreamElement, output chan<- stage.StreamElement) error

type SelfPlayUserTurnContextStage ¶

type SelfPlayUserTurnContextStage struct {
	stage.BaseStage
	// contains filtered or unexported fields
}

SelfPlayUserTurnContextStage adds scenario context for the NEXT user turn (completed user turns + 1). Intended only for self-play user generation.

This stage enriches elements with metadata that MockProvider uses to select the appropriate mock response based on scenario and turn number.

func NewSelfPlayUserTurnContextStage(scenario *config.Scenario) *SelfPlayUserTurnContextStage

func NewSelfPlayUserTurnContextStageWithHint(
	scenario *config.Scenario,
	turnIndexHint int,
) *SelfPlayUserTurnContextStage

func (s *SelfPlayUserTurnContextStage) Process(ctx context.Context, input <-chan stage.StreamElement, output chan<- stage.StreamElement) error

type StripToolMessagesStage ¶

type StripToolMessagesStage struct {
	stage.BaseStage
}

StripToolMessagesStage removes tool role messages from the stream. This is used in self-play scenarios before calling the self-play provider.

func NewStripToolMessagesStage() *StripToolMessagesStage

func (s *StripToolMessagesStage) Process(
	ctx context.Context,
	input <-chan stage.StreamElement,
	output chan<- stage.StreamElement,
) error

type TurnIndexStage ¶

type TurnIndexStage struct {
	stage.BaseStage
}

TurnIndexStage computes role-specific turn counters from accumulated messages. It sets clear, role-specific metadata keys that other stages can consume: - arena_user_completed_turns: number of completed user messages - arena_user_next_turn: completed user messages + 1 (next user turn to generate) - arena_assistant_completed_turns: number of completed assistant messages - arena_assistant_next_turn: completed assistant messages + 1

This stage enriches all elements with turn count metadata.

func NewTurnIndexStage() *TurnIndexStage

func (s *TurnIndexStage) Process(ctx context.Context, input <-chan stage.StreamElement, output chan<- stage.StreamElement) error

type VariableInjectionStage ¶

type VariableInjectionStage struct {
	stage.BaseStage
	// contains filtered or unexported fields
}

VariableInjectionStage injects variables into stream element metadata.

func NewVariableInjectionStage(variables map[string]string) *VariableInjectionStage

func (s *VariableInjectionStage) Process(
	ctx context.Context,
	input <-chan stage.StreamElement,
	output chan<- stage.StreamElement,
) error