llmgateway

Documentation

Overview

Package llmgateway is the single chokepoint through which every LLM call in the process must pass. Its job is to bound provider concurrency (--max-concurrent-llm-requests) across every code path (direct analysis, queue-driven analysis, digest dedupe, digest summary) so the flag actually means what it says.

A Gateway owns one semaphore (`chan struct{}`) sized at construction time. Generate and Stream acquire a slot, invoke the provider, and release in a deferred call so slot leaks can't happen on panic or context cancel. One call == one slot. A 5-task analysis pipeline holds the slot 5 separate times, leaving gaps between tasks available to other callers.

Index

• type CallOption
  • func WithLabel(label string) CallOption
• type Gateway
  • func New(maxConcurrent int) *Gateway
  • func (g *Gateway) Generate(ctx context.Context, p llmprovider.Provider, prompt string, opts ...CallOption) (string, error)
  • func (g *Gateway) MaxConcurrent() int
  • func (g *Gateway) Stats() Stats
  • func (g *Gateway) Stream(ctx context.Context, p llmprovider.ChatModelProvider, ...) (resp string, err error)
• type Stats

Types

type CallOption

type CallOption func(*callConfig)

CallOption configures a single call (Generate or Stream).

func WithLabel

func WithLabel(label string) CallOption

WithLabel attaches a human-readable label to the call for log correlation (e.g. "analyze:task=key_points", "digest:dedupe", "digest:summary").

type Gateway

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

Gateway throttles LLM calls globally.

func New

func New(maxConcurrent int) *Gateway

New creates a Gateway that allows at most maxConcurrent simultaneous LLM calls. maxConcurrent < 1 is coerced to 1.

func (*Gateway) Generate

func (g *Gateway) Generate(
	ctx context.Context,
	p llmprovider.Provider,
	prompt string,
	opts ...CallOption,
) (string, error)

Generate makes a one-shot (non-streaming) LLM call through the gateway.

func (*Gateway) MaxConcurrent

func (g *Gateway) MaxConcurrent() int

MaxConcurrent returns the configured cap.

func (*Gateway) Stats

func (g *Gateway) Stats() Stats

Stats returns a snapshot of runtime counters.

func (*Gateway) Stream

func (g *Gateway) Stream(
	ctx context.Context,
	p llmprovider.ChatModelProvider,
	messages []*schema.Message,
	onChunk func(chunk *schema.Message) error,
	opts ...CallOption,
) (resp string, err error)

Stream opens a streaming LLM call through the gateway, drains the reader, and invokes onChunk for each message chunk as it arrives. The slot is held for the entire duration of the stream (reader open → io.EOF / error / cancel).

onChunk may be nil; in that case chunks are still accumulated and returned as the full response string, but no per-token callback is invoked. Returning a non-nil error from onChunk aborts the stream and is returned.

type Stats

type Stats struct {
	MaxConcurrent int
	InFlight      int64
	Waiting       int64
	TotalCalls    int64
}

Stats is a snapshot of the gateway's counters.