consensus

Documentation

Index

• Constants
• func TruncateResponses(responses map[string]string, maxTotalChars int) map[string]string
• type CancelCallback
• type ChunkCallback
• type ConsensusAnalysis
  • func ParseConsensusAnalysis(rawOutput string) *ConsensusAnalysis
• type Engine
  • func NewEngine(primary provider.Provider, timeout time.Duration, minResponses int) *Engine
  • func (e *Engine) BuildConsensusPrompt(originalPrompt string, responses map[string]string, mode SynthesisMode, ...) []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
  • func FanOutWithTools(ctx context.Context, providers []provider.Provider, ...) *FanOutResult
• type FanOutToolExecutor
• type MinorityReport
• 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)
  • func (p *Pipeline) SetCancelCallback(cb CancelCallback)
  • func (p *Pipeline) SetChunkCallback(cb ChunkCallback)
  • func (p *Pipeline) SetFanOutTools(tools []provider.ToolDefinition, exec FanOutToolExecutor, ...)
• type SynthesisMode

Constants

const MaxFanOutToolRounds = 25

MaxFanOutToolRounds limits how many tool-call iterations a fan-out provider can perform before being forcibly stopped. Prevents runaway API token burn.

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 CancelCallback

type CancelCallback func(cancels map[string]context.CancelFunc)

Pipeline orchestrates the full consensus workflow: fan-out query to all providers, threshold check, and synthesis via the primary provider. CancelCallback is called during fan-out with the per-provider cancel functions, allowing external code to cancel individual providers mid-query.

type ChunkCallback

type ChunkCallback func(providerID string, chunk provider.StreamChunk)

ChunkCallback is called for each streaming chunk from a provider during fan-out. It receives the provider ID and the chunk. Called from provider goroutines — implementations must be safe for concurrent use.

type ConsensusAnalysis

type ConsensusAnalysis struct {
	Recommendation  string
	Confidence      string // "high", "medium", "low", or ""
	Agreements      []string
	MinorityReports []MinorityReport
	Evidence        []string
	Raw             string // the full original synthesis text
}

ConsensusAnalysis holds the structured breakdown of a consensus synthesis.

func ParseConsensusAnalysis

func ParseConsensusAnalysis(rawOutput string) *ConsensusAnalysis

ParseConsensusAnalysis extracts structured sections from the primary model's synthesis output. If the output does not contain structured headers (## ), the full text is returned as both Recommendation and Raw (graceful degradation).

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, mode SynthesisMode, history ...provider.Message) []provider.Message

BuildConsensusPrompt constructs the message slice sent to the primary provider for synthesis. The mode controls the depth of analysis requested.

func (*Engine) Synthesize

func (e *Engine) Synthesize(
	ctx context.Context,
	originalPrompt string,
	responses map[string]string,
	opts provider.QueryOpts,
	history ...provider.Message,
) (<-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
	// Latencies maps provider ID to the response wall-clock duration.
	Latencies map[string]time.Duration
	// Skipped lists provider IDs that were skipped due to context limits.
	Skipped []string
	// CancelFuncs maps provider ID to a function that cancels that provider's context.
	// Available during fan-out for preemptive cancellation.
	CancelFuncs map[string]context.CancelFunc
}

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,
	onChunk ChunkCallback,
) *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 onChunk is non-nil, it is called for every streaming chunk as it arrives from each provider, enabling real-time display of individual provider output.

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

func FanOutWithTools

func FanOutWithTools(
	ctx context.Context,
	providers []provider.Provider,
	messages []provider.Message,
	opts provider.QueryOpts,
	timeout time.Duration,
	tracker *tokens.TokenTracker,
	onChunk ChunkCallback,
	readOnlyTools []provider.ToolDefinition,
	toolExec FanOutToolExecutor,
	toolCapable map[string]bool,
	onCancel CancelCallback,
) *FanOutResult

FanOutWithTools is like FanOut but allows read-only tools during fan-out. readOnlyTools are the tool definitions sent to providers (e.g., file_read). toolExec executes tool calls; if nil, tools are stripped from the request. toolCapable lists provider IDs that support structured tool calling. Providers not in this set receive no tools (even if readOnlyTools is set). If toolCapable is nil, all providers get tools (backward compat).

type FanOutToolExecutor

type FanOutToolExecutor func(call provider.ToolCall) (output string, err error)

FanOutToolExecutor executes a single tool call during fan-out. Only read-only tools should be wired here. It is called from concurrent provider goroutines, so the implementation must be safe for concurrent use.

type MinorityReport

type MinorityReport struct {
	ProviderID string
	Position   string
	Reasoning  string
}

MinorityReport captures a dissenting view from the consensus.

type Pipeline

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

func NewPipeline

func NewPipeline(
	providers []provider.Provider,
	primary provider.Provider,
	timeout time.Duration,
	minResponses int,
	tracker *tokens.TokenTracker,
	mode SynthesisMode,
) *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.
• mode controls synthesis depth (quick/balanced/thorough).

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.

func (*Pipeline) SetCancelCallback

func (p *Pipeline) SetCancelCallback(cb CancelCallback)

SetCancelCallback sets a callback that receives per-provider cancel functions during fan-out, enabling external cancellation of individual providers.

func (*Pipeline) SetChunkCallback

func (p *Pipeline) SetChunkCallback(cb ChunkCallback)

SetChunkCallback sets a callback that fires for each streaming chunk from individual providers during fan-out. This enables real-time display of individual provider output while the fan-out is in progress.

func (*Pipeline) SetFanOutTools

func (p *Pipeline) SetFanOutTools(tools []provider.ToolDefinition, exec FanOutToolExecutor, toolCapable map[string]bool)

SetFanOutTools configures read-only tools available to providers during fan-out. The executor handles the actual tool execution (e.g., file_read). toolCapable maps provider IDs that support structured tool calling — others won't receive tools. Pass nil to send tools to all providers.

type SynthesisMode

type SynthesisMode string

SynthesisMode controls the depth of the consensus synthesis prompt.

const (
	// SynthesisQuick produces a concise, direct answer without structured sections.
	SynthesisQuick SynthesisMode = "quick"
	// SynthesisBalanced produces a structured synthesis with confidence, agreements,
	// minority reports, and evidence sections.
	SynthesisBalanced SynthesisMode = "balanced"
	// SynthesisThorough produces deep analysis with extended reasoning, trade-off
	// analysis, step-by-step verification, and alternative approaches.
	SynthesisThorough SynthesisMode = "thorough"
)