44// parse — bytes → hierarchical outline (parser.Registry)
55// build tree — outline → sections persisted in Postgres + object store
66// summarize — every section gets an LLM-written summary
7+ // hyde — every leaf gets a list of HyDE candidate questions
8+ //
9+ // After parse + persist, the summarize and hyde stages run CONCURRENTLY:
10+ // HyDE operates from a section's title + content (the summary, when
11+ // available, is a nice-to-have), so it has no hard ordering dependency
12+ // on summarize. Running them in parallel roughly halves wall time on
13+ // long documents. Total LLM-in-flight is capped by an optional shared
14+ // semaphore (Pipeline.GlobalLLMConcurrency) so we don't oversubscribe
15+ // the provider.
716//
817// The pipeline is driven by a queue job of kind queue.KindIngestDocument.
918// Each stage is idempotent so a retry from any point leaves the document
@@ -87,6 +96,24 @@ type Pipeline struct {
8796 // HyDEConcurrency bounds parallel LLM calls during the HyDE stage.
8897 // Default: 4.
8998 HyDEConcurrency int
99+
100+ // GlobalLLMConcurrency, when > 0, caps the total number of LLM calls
101+ // in flight across BOTH the summarize and HyDE stages combined.
102+ // Each stage still respects its own per-stage cap
103+ // (SummaryConcurrency / HyDEConcurrency), but neither can push the
104+ // shared counter above this ceiling. Useful because summarize and
105+ // HyDE now run concurrently — without this, total in-flight load is
106+ // SummaryConcurrency + HyDEConcurrency, which may exceed the
107+ // provider's per-tenant rate limit.
108+ //
109+ // 0 disables the global cap (each stage is bounded only by its own
110+ // per-stage semaphore). Default applied by NewPipeline: 12.
111+ GlobalLLMConcurrency int
112+
113+ // globalLLMSem is the lazily-initialized shared semaphore enforcing
114+ // GlobalLLMConcurrency. nil means "no global cap" — callers fall back
115+ // to per-stage limits only.
116+ globalLLMSem chan struct {}
90117}
91118
92119// NewPipeline returns a Pipeline with sensible defaults filled in.
@@ -103,12 +130,41 @@ func NewPipeline(p Pipeline) *Pipeline {
103130 if p .HyDEConcurrency <= 0 {
104131 p .HyDEConcurrency = 4
105132 }
133+ // Default the global cap to a value that comfortably exceeds the
134+ // sum of the two default per-stage caps (4 + 4 = 8) while leaving
135+ // some headroom — but stays well below typical provider per-tenant
136+ // concurrency limits.
137+ if p .GlobalLLMConcurrency < 0 {
138+ p .GlobalLLMConcurrency = 0
139+ }
140+ if p .GlobalLLMConcurrency == 0 {
141+ p .GlobalLLMConcurrency = 12
142+ }
143+ if p .GlobalLLMConcurrency > 0 {
144+ p .globalLLMSem = make (chan struct {}, p .GlobalLLMConcurrency )
145+ }
106146 if p .Logger == nil {
107147 p .Logger = slog .Default ()
108148 }
109149 return & p
110150}
111151
152+ // acquireGlobalLLM blocks until a global-LLM-concurrency slot is free,
153+ // or returns false if ctx is canceled first. Returns a release func the
154+ // caller must invoke (typically deferred). Safe to call when the global
155+ // semaphore is disabled — the returned release is a no-op.
156+ func (p * Pipeline ) acquireGlobalLLM (ctx context.Context ) (release func (), ok bool ) {
157+ if p .globalLLMSem == nil {
158+ return func () {}, true
159+ }
160+ select {
161+ case p .globalLLMSem <- struct {}{}:
162+ return func () { <- p .globalLLMSem }, true
163+ case <- ctx .Done ():
164+ return func () {}, false
165+ }
166+ }
167+
112168// Handler returns a queue.Handler suitable for queue.KindIngestDocument.
113169func (p * Pipeline ) Handler () queue.Handler {
114170 return func (ctx context.Context , j queue.Job ) error {
@@ -144,22 +200,32 @@ func (p *Pipeline) Run(ctx context.Context, pl Payload) error {
144200 if err := p .DB .SetDocumentStatus (ctx , pl .DocumentID , db .StatusSummarizing , "" ); err != nil {
145201 return err
146202 }
147- if err := p .summarize (ctx , pl .DocumentID , pl .Profile ); err != nil {
203+
204+ stageStart := time .Now ()
205+ summarizeFn := func (ctx context.Context ) error {
206+ return p .summarize (ctx , pl .DocumentID , pl .Profile )
207+ }
208+ var hydeFn func (ctx context.Context ) error
209+ if p .HyDEEnabled {
210+ hydeFn = func (ctx context.Context ) error {
211+ return p .generateCandidateQuestions (ctx , pl .DocumentID , pl .Profile )
212+ }
213+ }
214+ sumErr , hydeErr := runParallelStages (ctx , summarizeFn , hydeFn )
215+ if sumErr != nil {
148216 // Summarization failures are recoverable — a section without a
149217 // summary is still query-able, just less efficient. We log and
150218 // proceed rather than dead-letter the document.
151- log .Warn ("ingest: summarize had errors" , "err" , err )
219+ log .Warn ("ingest: summarize had errors" , "err" , sumErr )
152220 }
153-
154- if p .HyDEEnabled {
155- if err := p .generateCandidateQuestions (ctx , pl .DocumentID , pl .Profile ); err != nil {
156- // HyDE is a retrieval-quality booster, not a correctness
157- // requirement. Failures here leave the document fully usable
158- // (just with less recall on lexically-distant queries), so we
159- // log and proceed.
160- log .Warn ("ingest: hyde had errors" , "err" , err )
161- }
221+ if hydeErr != nil {
222+ // HyDE is a retrieval-quality booster, not a correctness
223+ // requirement. Failures here leave the document fully usable
224+ // (just with less recall on lexically-distant queries), so we
225+ // log and proceed.
226+ log .Warn ("ingest: hyde had errors" , "err" , hydeErr )
162227 }
228+ log .Info ("ingest: summarize+hyde complete" , "elapsed" , time .Since (stageStart ))
163229
164230 if err := p .DB .SetDocumentStatus (ctx , pl .DocumentID , db .StatusReady , "" ); err != nil {
165231 return err
@@ -168,6 +234,32 @@ func (p *Pipeline) Run(ctx context.Context, pl Payload) error {
168234 return nil
169235}
170236
237+ // runParallelStages runs summarize and HyDE concurrently, returning each
238+ // stage's error independently so callers can log them separately. A nil
239+ // hydeFn skips the HyDE stage (returns nil for hydeErr).
240+ //
241+ // Extracted so the interleave behaviour is testable without touching the
242+ // real DB-backed summarize/HyDE entry points.
243+ func runParallelStages (ctx context.Context , summarizeFn , hydeFn func (context.Context ) error ) (summarizeErr , hydeErr error ) {
244+ var wg sync.WaitGroup
245+ wg .Add (1 )
246+ go func () {
247+ defer wg .Done ()
248+ if summarizeFn != nil {
249+ summarizeErr = summarizeFn (ctx )
250+ }
251+ }()
252+ if hydeFn != nil {
253+ wg .Add (1 )
254+ go func () {
255+ defer wg .Done ()
256+ hydeErr = hydeFn (ctx )
257+ }()
258+ }
259+ wg .Wait ()
260+ return summarizeErr , hydeErr
261+ }
262+
171263func (p * Pipeline ) parse (ctx context.Context , pl Payload ) (* parser.ParsedDoc , error ) {
172264 rc , _ , err := p .Storage .Get (ctx , pl .SourceRef )
173265 if err != nil {
@@ -336,7 +428,14 @@ func (p *Pipeline) summarize(ctx context.Context, docID tree.DocumentID, profile
336428 mu .Unlock ()
337429 }
338430
431+ // Global cap on total LLM-in-flight across summarize+HyDE.
432+ // Released the moment the LLM call returns.
433+ release , ok := p .acquireGlobalLLM (gctx )
434+ if ! ok {
435+ return nil
436+ }
339437 summary , err := p .summaryFor (gctx , s , childLines , profile )
438+ release ()
340439 if err != nil {
341440 mu .Lock ()
342441 errs = append (errs , fmt .Errorf ("section %s: %w" , s .ID , err ))
0 commit comments