b6d7970c5c
Implements real-time token streaming to Telegram using the sendMessageDraft API (telego v1.6.0). Instead of showing only a "Thinking..." placeholder until the full response arrives, users now see partial LLM output appear in the chat as it's generated. The streaming pipeline threads through all layers: - StreamingProvider interface (providers/types.go): opt-in ChatStream() method that receives an onChunk callback with accumulated text - OpenAI-compatible SSE streaming (openai_compat/provider.go): parses SSE events with stream:true, handles text deltas and tool call assembly - Anthropic native streaming (anthropic/provider.go): uses SDK's NewStreaming() for direct Anthropic API connections - HTTPProvider delegation (http_provider.go): delegates ChatStream to the underlying openai_compat provider - StreamingCapable + Streamer interfaces (channels/interfaces.go): opt-in channel capability like TypingCapable/PlaceholderCapable - Telegram streamer (telegram/telegram.go): BeginStream returns a telegramStreamer that throttles sendMessageDraft calls (3s/200 chars) with graceful degradation on API errors - StreamDelegate bridge (bus/bus.go): decouples agent loop from channel manager without tight imports - Manager integration (manager.go): implements StreamDelegate, tracks streamActive state, coordinates with placeholder editing - Agent loop (loop.go): uses ChatStream when both provider and channel support streaming, cancels stream on tool calls, skips PublishOutbound when Finalize already delivered the message Graceful degradation: - Bots without forum/topics mode: first sendMessageDraft error sets failed=true, subsequent Updates become no-ops, Finalize still delivers via SendMessage. User sees normal non-streaming behavior. - Non-streaming providers: type assertion fails, falls back to Chat() - Config opt-out: streaming.enabled (default true) in telegram config Closes #1098
59 lines
2.7 KiB
Go
59 lines
2.7 KiB
Go
package channels
|
|
|
|
import "context"
|
|
|
|
// TypingCapable — channels that can show a typing/thinking indicator.
|
|
// StartTyping begins the indicator and returns a stop function.
|
|
// The stop function MUST be idempotent and safe to call multiple times.
|
|
type TypingCapable interface {
|
|
StartTyping(ctx context.Context, chatID string) (stop func(), err error)
|
|
}
|
|
|
|
// MessageEditor — channels that can edit an existing message.
|
|
// messageID is always string; channels convert platform-specific types internally.
|
|
type MessageEditor interface {
|
|
EditMessage(ctx context.Context, chatID string, messageID string, content string) error
|
|
}
|
|
|
|
// ReactionCapable — channels that can add a reaction (e.g. 👀) to an inbound message.
|
|
// ReactToMessage adds a reaction and returns an undo function to remove it.
|
|
// The undo function MUST be idempotent and safe to call multiple times.
|
|
type ReactionCapable interface {
|
|
ReactToMessage(ctx context.Context, chatID, messageID string) (undo func(), err error)
|
|
}
|
|
|
|
// PlaceholderCapable — channels that can send a placeholder message
|
|
// (e.g. "Thinking... 💭") that will later be edited to the actual response.
|
|
// The channel MUST also implement MessageEditor for the placeholder to be useful.
|
|
// SendPlaceholder returns the platform message ID of the placeholder so that
|
|
// Manager.preSend can later edit it via MessageEditor.EditMessage.
|
|
type PlaceholderCapable interface {
|
|
SendPlaceholder(ctx context.Context, chatID string) (messageID string, err error)
|
|
}
|
|
|
|
// StreamingCapable — channels that can show partial LLM output in real-time.
|
|
// The channel SHOULD gracefully degrade if the platform rejects streaming
|
|
// (e.g. Telegram bot without forum mode). In that case, Update becomes a no-op
|
|
// and Finalize still delivers the final message.
|
|
type StreamingCapable interface {
|
|
BeginStream(ctx context.Context, chatID string) (Streamer, error)
|
|
}
|
|
|
|
// Streamer pushes incremental content to a streaming-capable channel.
|
|
type Streamer interface {
|
|
// Update sends accumulated partial content to the user.
|
|
Update(ctx context.Context, content string) error
|
|
// Finalize commits the final message. After this, the Streamer is done.
|
|
Finalize(ctx context.Context, content string) error
|
|
// Cancel aborts the stream (e.g. when tool calls are detected mid-stream).
|
|
Cancel(ctx context.Context)
|
|
}
|
|
|
|
// PlaceholderRecorder is injected into channels by Manager.
|
|
// Channels call these methods on inbound to register typing/placeholder state.
|
|
// Manager uses the registered state on outbound to stop typing and edit placeholders.
|
|
type PlaceholderRecorder interface {
|
|
RecordPlaceholder(channel, chatID, placeholderID string)
|
|
RecordTypingStop(channel, chatID string, stop func())
|
|
RecordReactionUndo(channel, chatID string, undo func())
|
|
}
|