consensus

Documentation

Index

• func TruncateResponses(responses map[string]string, maxTotalChars int) map[string]string
• type Engine
  • func NewEngine(primary provider.Provider, timeout time.Duration, minResponses int) *Engine
  • func (e *Engine) BuildConsensusPrompt(originalPrompt string, responses map[string]string) []provider.Message
  • func (e *Engine) Synthesize(ctx context.Context, originalPrompt string, responses map[string]string, ...) (<-chan provider.StreamChunk, error)
• type FanOutResult
  • func FanOut(ctx context.Context, providers []provider.Provider, ...) *FanOutResult
• type Pipeline
  • func NewPipeline(providers []provider.Provider, primary provider.Provider, ...) *Pipeline
  • func (p *Pipeline) Run(ctx context.Context, messages []provider.Message, opts provider.QueryOpts) (<-chan provider.StreamChunk, *FanOutResult, error)

Functions

func TruncateResponses

func TruncateResponses(responses map[string]string, maxTotalChars int) map[string]string

TruncateResponses returns a copy of responses where, if the total character count exceeds maxTotalChars, the longest responses are proportionally truncated so the combined length fits within the budget. Truncated entries have "[truncated]" appended.

Types

type Engine

type Engine struct {
	// contains filtered or unexported fields
}

Engine drives the consensus synthesis step. It takes the collected responses from a fan-out query and asks a primary provider to synthesize them into a single authoritative answer.

func NewEngine

func NewEngine(primary provider.Provider, timeout time.Duration, minResponses int) *Engine

NewEngine creates a consensus Engine.

• primary is the provider used to synthesize the final answer.
• timeout is the maximum time allowed for the synthesis query.
• minResponses is the minimum number of successful responses required before synthesis can proceed.

func (*Engine) BuildConsensusPrompt

func (e *Engine) BuildConsensusPrompt(originalPrompt string, responses map[string]string) []provider.Message

BuildConsensusPrompt constructs the message slice sent to the primary provider for synthesis.

func (*Engine) Synthesize

func (e *Engine) Synthesize(
	ctx context.Context,
	originalPrompt string,
	responses map[string]string,
	opts provider.QueryOpts,
) (<-chan provider.StreamChunk, error)

Synthesize sends the consensus prompt to the primary provider and returns the streaming response channel. The caller is responsible for consuming the channel.

type FanOutResult

type FanOutResult struct {
	// Responses maps provider ID to the full assembled response text.
	Responses map[string]string
	// Errors maps provider ID to any error encountered during the query.
	Errors map[string]error
	// Usage maps provider ID to the token usage reported by that provider.
	Usage map[string]tokens.Usage
	// Skipped lists provider IDs that were skipped due to context limits.
	Skipped []string
}

FanOutResult holds the collected responses and errors from a fan-out query to multiple providers.

func FanOut

func FanOut(
	ctx context.Context,
	providers []provider.Provider,
	messages []provider.Message,
	opts provider.QueryOpts,
	timeout time.Duration,
	tracker *tokens.TokenTracker,
) *FanOutResult

FanOut dispatches a query to all providers concurrently, collects their streaming responses into complete strings, and returns once every provider has finished or the timeout is reached.

If a tracker is provided, providers that would exceed their context limit are skipped (recorded in result.Skipped).

type Pipeline

type Pipeline struct {
	// contains filtered or unexported fields
}

Pipeline orchestrates the full consensus workflow: fan-out query to all providers, threshold check, and synthesis via the primary provider.

func NewPipeline

func NewPipeline(
	providers []provider.Provider,
	primary provider.Provider,
	timeout time.Duration,
	minResponses int,
	tracker *tokens.TokenTracker,
) *Pipeline

NewPipeline creates a Pipeline.

• providers is the full set of providers to fan-out to.
• primary is the provider used for the synthesis step.
• timeout is the per-phase timeout (fan-out and synthesis each get this).
• minResponses is the minimum number of successful fan-out responses required before synthesis proceeds.
• tracker is optional (may be nil) for token usage tracking and limit enforcement.

func (*Pipeline) Run

func (p *Pipeline) Run(
	ctx context.Context,
	messages []provider.Message,
	opts provider.QueryOpts,
) (<-chan provider.StreamChunk, *FanOutResult, error)

Run executes the full consensus pipeline:

1. Fan-out the query to every provider.
2. Check the minimum-response threshold.
3. If only the primary provider responded, return its response directly without synthesis.
4. Otherwise, synthesize the collected responses through the primary.

It returns the streaming consensus channel, the raw fan-out results (so the TUI can display individual responses), and any error.