aggregators

package
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Aug 30, 2026 License: BSD-2-Clause Imports: 20 Imported by: 0

Documentation

Overview

Package aggregators assembles the conversation around an LLM. The user aggregator collects transcriptions into a user message and triggers the LLM; the assistant aggregator collects the streamed response into an assistant message. Both share one LLMContext, so the conversation accrues across turns.

Place the user aggregator before the LLM and the assistant aggregator at the end of the pipeline: 

pipeline.New(input, stt, agg.User(), llm, tts, output, agg.Assistant())

By default the user turn ends when the STT service finalizes a transcription. With WithTurnTaking, the turn instead ends when a turntaking.Detector reports end-of-turn (a UserStoppedSpeakingFrame), gated on having a finalized transcript — so a Smart Turn model, not STT endpointing, decides when the bot responds. Add the turntaking.Detector right after the input transport: 

pipeline.New(input, detector, stt, agg.User(), llm, tts, output, agg.Assistant())

Index

Constants

View Source 
const (
	// EventUserTurnStarted fires when the user's turn begins, carrying the
	// turns.StartStrategy that decided it began.
	//
	//	events.On(agg.User().Events(), aggregators.EventUserTurnStarted,
	//	    func(ctx context.Context, s turns.StartStrategy) { … })
	EventUserTurnStarted = "on_user_turn_started"
	// EventUserTurnStopped fires when the user's turn ends, carrying a
	// UserTurnStopped describing what they said and which strategy decided the
	// turn was over.
	EventUserTurnStopped = "on_user_turn_stopped"
	// EventUserTurnInferenceTriggered fires when a stop strategy decides there
	// is enough to answer, which is usually but not always the moment the turn
	// ends. It carries the turns.StopStrategy that decided.
	EventUserTurnInferenceTriggered = "on_user_turn_inference_triggered"
	// EventUserTurnStopTimeout fires when a turn was closed because no stop
	// strategy decided it had ended. It carries no argument.
	EventUserTurnStopTimeout = "on_user_turn_stop_timeout"
	// EventUserTurnIdle fires when the user has said nothing for the configured
	// idle timeout. It carries no argument.
	EventUserTurnIdle = "on_user_turn_idle"
	// EventUserTurnMessageAdded fires when the user's words are written to the
	// conversation, carrying a UserTurnMessageAdded. A turn can write more than
	// once, when an early inference answers part of it, so this fires per write
	// where EventUserTurnStopped fires once per turn.
	EventUserTurnMessageAdded = "on_user_turn_message_added"
	// EventUserMuteStarted fires when the user becomes muted and their input
	// stops reaching the bot. It carries no argument.
	EventUserMuteStarted = "on_user_mute_started"
	// EventUserMuteStopped fires when the user is unmuted. It carries no
	// argument.
	EventUserMuteStopped = "on_user_mute_stopped"
)

The events the user half of a pair raises around each turn it collects.

View Source 
const (
	// EventAssistantTurnStarted fires when the bot's turn begins, which is the
	// model starting to answer or, for an utterance spoken with no answer around
	// it, that speech starting. It carries no argument.
	//
	//	events.On(agg.Assistant().Events(), aggregators.EventAssistantTurnStarted,
	//	    func(ctx context.Context, _ any) { … })
	EventAssistantTurnStarted = "on_assistant_turn_started"
	// EventAssistantTurnStopped fires when the bot's turn ends, carrying an
	// AssistantTurnStopped describing what it said.
	//
	//	events.On(agg.Assistant().Events(), aggregators.EventAssistantTurnStopped,
	//	    func(ctx context.Context, t aggregators.AssistantTurnStopped) { … })
	EventAssistantTurnStopped = "on_assistant_turn_stopped"
	// EventAssistantThought fires when a reasoning model finishes a thought,
	// carrying an AssistantThought with what it reasoned.
	EventAssistantThought = "on_assistant_thought"
)

The events the assistant half of a pair raises around each turn it writes.

View Source 
const (
	// EventRequestSummarization fires when a summary is needed and no dedicated
	// LLM was configured to produce it. Its argument is the
	// *frames.LLMContextSummaryRequestFrame to put to the pipeline's LLM. It is
	// synchronous: the summarizer has marked the request in flight and is waiting
	// for it to be sent.
	EventRequestSummarization = "on_request_summarization"
	// EventSummaryApplied fires once a summary has been written into the
	// conversation. Its argument is a SummaryApplied describing the compression.
	EventSummaryApplied = "on_summary_applied"
)

The events a Summarizer raises.

View Source 
const EventCompletion = "on_completion"

EventCompletion fires with a whole LLM response once it has been gathered. It carries a Completion. 

events.On(agg.Events(), aggregators.EventCompletion,
    func(ctx context.Context, c aggregators.Completion) { … })

Variables

This section is empty.

Functions

This section is empty.

Types

type AssistantAggregator 

type AssistantAggregator struct {
	*processor.Base
	// contains filtered or unexported fields
}

AssistantAggregator collects the LLM's streamed text into a single assistant message and appends it to the context when the response completes. If the response is interrupted (barge-in), the partial text gathered so far is committed so the context reflects what the bot actually said. The response fields are touched from both the process goroutine (text frames) and the input goroutine (the InterruptionFrame system frame), so they are mutex-guarded.

func (*AssistantAggregator) Cleanup added in v0.1.0 

func (a *AssistantAggregator) Cleanup(ctx context.Context) error

Cleanup cancels the context-updated callbacks and the summarization in flight, and waits for them to return.

func (*AssistantAggregator) HasFunctionCallsInProgress added in v0.1.0 

func (a *AssistantAggregator) HasFunctionCallsInProgress() bool

HasFunctionCallsInProgress reports whether any tool call of the current turn has yet to report a final result.

It is what tells something waiting on the whole batch, rather than on one call, that the turn's calls are done: a caller acting on a tool result in its context-updated callback reads this to know whether it is the last one, since acting while siblings are still running would act on a half-finished turn.

func (*AssistantAggregator) ProcessFrame 

func (a *AssistantAggregator) ProcessFrame(ctx context.Context, f frames.Frame, dir processor.Direction) error

ProcessFrame collects LLM text into an assistant message.

func (*AssistantAggregator) Setup added in v0.1.0 

func (a *AssistantAggregator) Setup(ctx context.Context, s processor.Setup) error

Setup opens the lifetime the context-updated callbacks run under.

func (*AssistantAggregator) Summarizer added in v0.1.0 

func (a *AssistantAggregator) Summarizer() *Summarizer

Summarizer is the conversation summarizer this aggregator owns, for attaching handlers to the events it raises.

type AssistantThought added in v0.1.0 

type AssistantThought struct {
	// Content is the thought, whole.
	Content string
	// Timestamp is when the thought began, as ISO 8601.
	Timestamp string
}

AssistantThought is one completed piece of a reasoning model's thinking.

type AssistantTurnStopped added in v0.1.0 

type AssistantTurnStopped struct {
	// Content is everything the bot said during the turn, as one message. It is
	// empty for a turn that said nothing.
	Content string
	// Interrupted reports whether the turn was cut off rather than finishing.
	Interrupted bool
	// Timestamp is when the turn began, as ISO 8601.
	Timestamp string
}

AssistantTurnStopped describes a bot turn that has just ended.

type Completion added in v0.1.0 

type Completion struct {
	// Text is everything the model said between the start and end of the
	// response.
	Text string
	// Completed reports whether the response finished. It is false for a
	// response an interruption cut short, where Text is what had been said by
	// then.
	Completed bool
}

Completion is a gathered LLM response.

type FullResponse added in v0.1.0 

type FullResponse struct {
	*processor.Base
	// contains filtered or unexported fields
}

FullResponse gathers a whole LLM response and reports it, leaving the frames it gathered from untouched.

It collects the text between an LLMFullResponseStartFrame and its matching end frame, and raises EventCompletion with the result. An interruption reports what had been gathered by then, marked as unfinished.

It is for something that wants each reply whole and off the frame path: a transcript, a moderation check, an evaluation. Nothing it sees is consumed.

func NewFullResponse added in v0.1.0 

func NewFullResponse(name string) *FullResponse

NewFullResponse builds a full-response aggregator.

func (*FullResponse) ProcessFrame added in v0.1.0 

func (a *FullResponse) ProcessFrame(ctx context.Context, f frames.Frame, dir processor.Direction) error

ProcessFrame gathers the response and forwards every frame untouched.

type GateFunc added in v0.1.0 

type GateFunc func(frames.Frame) bool

GateFunc reports whether a frame opens or closes a gate.

type Gated added in v0.1.0 

type Gated struct {
	*processor.Base
	// contains filtered or unexported fields
}

Gated holds frames back until something opens the gate, then releases what it held.

The frame that opens the gate goes first, then everything accumulated behind it, in the order it arrived. That ordering is the point: what was held is meant to arrive after whatever announced it, not before. The frame that closes the gate is held with the rest, so it is the first thing released next time.

System frames are never held, whatever the gate says, because a pipeline that cannot hear about an interruption while a gate is closed would have no way out of it. Frames traveling the other way are not this gate's business and pass untouched.

Ending the run is not a system frame, so a closed gate holds that too and the processors behind it never hear the run end. A pipeline carrying a gate has to open it before shutting down; canceling works either way, since that is a system frame.

func NewGated added in v0.1.0 

func NewGated(name string, cfg GatedConfig) *Gated

NewGated builds a gate over cfg.

func (*Gated) ProcessFrame added in v0.1.0 

func (g *Gated) ProcessFrame(
	ctx context.Context, frame frames.Frame, dir processor.Direction,
) error

ProcessFrame implements processor.Processor.

type GatedConfig added in v0.1.0 

type GatedConfig struct {
	// Open reports whether a frame opens the gate. It is consulted only while
	// the gate is closed.
	Open GateFunc
	// Close reports whether a frame closes the gate. It is consulted only while
	// the gate is open.
	Close GateFunc
	// StartOpen has the gate open before any frame arrives.
	StartOpen bool
	// Direction is the way the frames this gate holds are traveling. Frames
	// going the other way pass untouched. Downstream when unset.
	Direction processor.Direction
}

GatedConfig configures a Gated aggregator.

type GatedContext added in v0.1.0 

type GatedContext struct {
	*processor.Base
	// contains filtered or unexported fields
}

GatedContext holds the conversation back from the model until a notifier says to let it through.

It is how a turn waits on something the pipeline cannot express as a frame order: a classifier deciding whether to answer at all, a lookup that has to finish first. Only the most recent context is held, because an older one describes a conversation that has since moved on, so what is released when the notifier fires is the conversation as it stands rather than a backlog of it.

func NewGatedContext added in v0.1.0 

func NewGatedContext(name string, cfg GatedContextConfig) *GatedContext

NewGatedContext builds a gate over cfg.

func (*GatedContext) Cleanup added in v0.1.0 

func (g *GatedContext) Cleanup(ctx context.Context) error

Cleanup implements processor.Processor.

func (*GatedContext) ProcessFrame added in v0.1.0 

func (g *GatedContext) ProcessFrame(
	ctx context.Context, frame frames.Frame, dir processor.Direction,
) error

ProcessFrame implements processor.Processor.

type GatedContextConfig added in v0.1.0 

type GatedContextConfig struct {
	// Notifier releases the held conversation each time it is signaled.
	Notifier notify.Notifier
	// StartOpen lets the first conversation through without waiting, so the bot
	// can open the call before anything has had to decide.
	StartOpen bool
}

GatedContextConfig configures a GatedContext aggregator.

type LLMText added in v0.1.0 

type LLMText struct {
	*processor.Base
	// contains filtered or unexported fields
}

LLMText turns the tokens a model streams into aggregated text, using the aggregator it was built with.

It sits between the LLM and whatever reads its output, and is where the raw stream is grouped, categorized, rewritten or filtered before a synthesizer or a context aggregator sees it: grouping by sentence for a service that speaks better with whole sentences, or pulling code blocks out with a pattern aggregator so they are not read aloud.

Each LLMTextFrame is consumed and what the aggregator completes is pushed on as an AggregatedTextFrame. Whatever is left over is flushed when the response ends, so the last unit is not held back waiting for a boundary that will never arrive.

func NewLLMText added in v0.1.0 

func NewLLMText(name string) *LLMText

NewLLMText builds a processor grouping the model's output into sentences.

func NewLLMTextWith added in v0.1.0 

func NewLLMTextWith(name string, aggregator text.Aggregator) *LLMText

NewLLMTextWith builds a processor grouping the model's output with aggregator.

func (*LLMText) ProcessFrame added in v0.1.0 

func (p *LLMText) ProcessFrame(ctx context.Context, f frames.Frame, dir processor.Direction) error

ProcessFrame converts the model's tokens into aggregated text.

func (*LLMText) Reset added in v0.1.0 

func (p *LLMText) Reset()

Reset clears what the processor has gathered.

type Option 

type Option func(*options)

Option configures an aggregator Pair.

func WithMuteStrategies added in v0.1.0 

func WithMuteStrategies(strategies ...turns.MuteStrategy) Option

WithMuteStrategies suppresses the user's input while the bot is engaged: while it is speaking, or while a tool call it is waiting on runs. Input that arrives muted is dropped rather than queued, so the bot is not answered by something the user said while it was not listening.

The strategies are consulted for every frame, so each keeps up with the conversation, and any one of them asking for silence is enough. A change of state is announced to the pipeline as a UserMuteStarted or UserMuteStopped frame, and raised as an event.

It is independent of WithTurns: muting the user and deciding when their turn ended are different questions.

func WithSummarization 

func WithSummarization(cfg frames.AutoSummarizationConfig) Option

WithSummarization compresses the conversation automatically once it passes either of the configured thresholds. Older turns are replaced by a summary written into the conversation ahead of the messages kept back, so the model keeps the history without carrying every message of it.

Summarization runs beside the conversation rather than in the path of it, so it never adds latency to a turn. Without this option the conversation is still compressed on demand, whenever an frames.LLMSummarizeContextFrame is pushed into the pipeline.

The summary is generated by the pipeline's own LLM unless the configuration names one of its own, which is how summarization is routed to a cheaper model than the one carrying the conversation.

func WithToolChangeMessages added in v0.1.0 

func WithToolChangeMessages() Option

WithToolChangeMessages announces a change of toolset to the model.

On each LLMSetToolsFrame the aggregators diff the new toolset against the one the conversation currently advertises and append a developer message naming what was added and what was removed. It helps the model stay coherent across a mid-conversation toolset change, and heads off several flavors of tool-call hallucination: calling tools that have been removed, avoiding tools that have been added back, and inventing output (made-up answers, or tool-call-shaped text that is not a tool call) when no tool is available.

Only the standard tools are diffed; tools written in one provider's own format are left out of it.

Both halves of the pair take part, which is what makes it work whichever of them handles a given frame first. They share the conversation, so whichever one gets there first writes the announcement and the other one's diff is empty by the time it looks, and the message is written exactly once.

func WithTurns added in v0.1.0 

func WithTurns(cfg turns.Config) Option

WithTurns configures the turn taking the user aggregator drives.

The aggregator always decides when the user's turn began and ended; this is how the strategies it decides with, and the idle and mute settings around them, are chosen. Without it the defaults stand.

The strategies run inside the user aggregator, on the same frames and in the same order as the aggregation. That is what makes the turn's own transcript part of it: a turn ends because a transcript finalized, and were the decision made in a processor of its own, the end-of-turn frame would be a system frame racing ahead of the transcript that caused it and the user's last words would be dropped from the message the model is given. Where the decision has to be shared instead, put a turns.UserTurnProcessor in the pipeline and give this aggregator turns.ExternalStrategies so it adopts what that decides.

func WithVAD added in v0.1.0 

func WithVAD(cfg VADConfig) Option

WithVAD detects voice activity inside the user aggregator rather than in a processor of its own.

The frames it produces are queued back into the aggregator rather than pushed at a neighbor, so the turn strategies running here see the speech their own detector heard.

Use it when the aggregator is the only thing that needs the detection. Where a transport, an interruption decision or a recorder needs it too, put a vadproc.Processor after the input transport instead and leave this unset: the detection is then done once, where everything downstream of it can see it. Running both means analyzing the same audio twice.

Muted input never reaches the detector, so a muted microphone does not read as speech.

type Pair 

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

Pair is a user and assistant aggregator sharing one conversation context.

func New 

func New(ctx *frames.LLMContext, opts ...Option) *Pair

New builds a user/assistant aggregator pair around ctx.

func (*Pair) Assistant 

func (p *Pair) Assistant() *AssistantAggregator

Assistant returns the assistant-side aggregator.

func (*Pair) Context 

func (p *Pair) Context() *frames.LLMContext

Context returns the shared conversation context.

func (*Pair) User 

func (p *Pair) User() *UserAggregator

User returns the user-side aggregator.

type Sentence added in v0.1.0 

type Sentence struct {
	*processor.Base
	// contains filtered or unexported fields
}

Sentence gathers text frames into whole sentences, pushing one on only once a sentence has ended. It is for a downstream processor that needs coherent sentences rather than the fragments a model streams: 

TextFrame("Hello,")  -> nothing
TextFrame(" world.") -> TextFrame("Hello, world.")

An interim transcription is dropped: it is a guess that will be revised, so folding it into a sentence would repeat what the final one says.

func NewSentence added in v0.1.0 

func NewSentence(name string) *Sentence

NewSentence builds a sentence aggregator that finds sentence boundaries the way the rest of the framework does.

func NewSentenceWith added in v0.1.0 

func NewSentenceWith(name string, tokenizer text.SentenceTokenizer) *Sentence

NewSentenceWith builds a sentence aggregator finding sentence boundaries with tokenizer, for a caller that has one already or wants another language.

func (*Sentence) ProcessFrame added in v0.1.0 

func (s *Sentence) ProcessFrame(ctx context.Context, f frames.Frame, dir processor.Direction) error

ProcessFrame gathers text and pushes each completed sentence.

type Summarizer 

type Summarizer struct {
	events.Registry
	// contains filtered or unexported fields
}

Summarizer compresses a conversation that has grown long, either automatically once a threshold is passed or on demand when an LLMSummarizeContextFrame arrives. It watches the conversation, asks for a summary, and applies the result.

It is not a processor. The assistant aggregator owns one and hands it every frame, which is what lets the compression happen beside the conversation rather than in the path of it.

A summary is produced one of two ways. With a dedicated LLM configured it is generated directly, off the pipeline entirely. Without one, the summarizer raises EventRequestSummarization and the owner puts the request to the pipeline's own LLM, whose answer comes back as an LLMContextSummaryResultFrame.

func NewSummarizer added in v0.1.0 

func NewSummarizer(convo *frames.LLMContext, cfg frames.AutoSummarizationConfig, autoTrigger bool) *Summarizer

NewSummarizer builds a Summarizer over convo. autoTrigger enables the threshold checks; leave it false to compress only when asked with an LLMSummarizeContextFrame.

func (*Summarizer) Cleanup added in v0.1.0 

func (s *Summarizer) Cleanup(ctx context.Context)

Cleanup abandons any summarization in flight and waits for the goroutines of the dedicated-LLM path and of the asynchronous events.

func (*Summarizer) ProcessFrame added in v0.1.0 

func (s *Summarizer) ProcessFrame(ctx context.Context, f frames.Frame)

ProcessFrame lets the summarizer watch the conversation. The owner calls it for every frame, after the frame has been forwarded.

type SummaryApplied added in v0.1.0 

type SummaryApplied struct {
	// OriginalMessageCount is how many messages the conversation held before.
	OriginalMessageCount int
	// NewMessageCount is how many it holds after.
	NewMessageCount int
	// SummarizedMessageCount is how many were folded into the summary.
	SummarizedMessageCount int
	// PreservedMessageCount is how many survived uncompressed, counting a
	// preserved leading system message.
	PreservedMessageCount int
}

SummaryApplied describes a compression that has been applied to the conversation.

type UserAggregator 

type UserAggregator struct {
	*processor.Base
	// contains filtered or unexported fields
}

UserAggregator collects transcriptions into a user message and, when the user's turn ends, appends it to the context and triggers the LLM with an LLMContextFrame.

When turn strategies are configured it also drives them, and the idle watchdog, from this same processor: every frame is folded into the aggregation first and only then handed to the controllers, so a turn that ends on a finalized transcript ends with that transcript already in the message.

func (*UserAggregator) Cleanup added in v0.1.0 

func (u *UserAggregator) Cleanup(ctx context.Context) error

Cleanup releases what the controllers hold. It is the one call that happens exactly once, which is why releasing a shared detector or strategy belongs here rather than at the end of the session.

func (*UserAggregator) ProcessFrame 

func (u *UserAggregator) ProcessFrame(ctx context.Context, f frames.Frame, dir processor.Direction) error

ProcessFrame collects transcriptions and triggers the LLM.

The frame is handled first and only then given to the turn and idle controllers, so anything the frame contributes to the aggregation is already there when a strategy decides the turn ended on it.

func (*UserAggregator) Push added in v0.1.0 

func (u *UserAggregator) Push(ctx context.Context, f frames.Frame, dir processor.Direction) error

Push implements turns.Emitter.

func (*UserAggregator) Setup added in v0.1.0 

func (u *UserAggregator) Setup(ctx context.Context, s processor.Setup) error

Setup wires the controllers.

type UserTurnMessageAdded added in v0.1.0 

type UserTurnMessageAdded struct {
	// Content is the message that was written. It is the segment written now,
	// not the whole turn: a turn answered early writes what it had, then writes
	// the rest when it ends.
	Content string
	// Timestamp is when the turn began, as ISO 8601.
	Timestamp string
	// UserID identifies the speaker, and is empty when the transcription service
	// does not say who spoke.
	UserID string
}

UserTurnMessageAdded describes a user message written to the conversation.

type UserTurnStopped added in v0.1.0 

type UserTurnStopped struct {
	// Strategy is the stop strategy that decided the turn was over. It is nil
	// when nothing decided: a turn closed because no strategy did, or one closed
	// by the session ending.
	Strategy turns.StopStrategy
	// Content is everything the user said during the turn, including whatever an
	// earlier inference already answered. It is empty for a turn that said
	// nothing.
	Content string
	// Timestamp is when the turn began, as ISO 8601.
	Timestamp string
	// UserID identifies the speaker, and is empty when the transcription service
	// does not say who spoke.
	UserID string
}

UserTurnStopped describes a user turn that has just ended.

type VADConfig added in v0.1.0 

type VADConfig struct {
	// Analyzer detects voice activity in the incoming audio. Required.
	Analyzer vad.Analyzer
	// AudioIdleTimeout is how long to wait, with the user speaking and no audio
	// arriving at all, before taking the speech to have stopped. It covers the
	// audio going away mid-utterance, a muted microphone being the usual case.
	//
	// Leave it nil for one second. A zero duration turns the watch off.
	AudioIdleTimeout *time.Duration
}

VADConfig configures the voice-activity detection a user aggregator runs.

Source Files

View all Source files

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.