Skip to content

Adaptive concurrency through llmgate's limiter; parallel documents and page batches (HAL-1372) #65

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hallelx2 merged 1 commit into from
Sep 18, 2026
Merged

Adaptive concurrency through llmgate's limiter; parallel documents and page batches (HAL-1372) #65

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 25 additions & 3 deletions cmd/engine/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ import (

"github.com/hallelx2/llmgate"
"github.com/hallelx2/llmgate/judge/typesafe"
"github.com/hallelx2/llmgate/middleware/limit"
"github.com/hallelx2/llmgate/middleware/retry"
"github.com/hallelx2/llmgate/pricing"
"github.com/hallelx2/llmgate/provider/anthropic"
Expand Down Expand Up @@ -127,7 +128,10 @@ func run() error {
return fmt.Errorf("init llm: %w", err)
}
}
judge, err := buildJudge(cfg.LLM.Judge)
if llmClient != nil {
llmClient = limit.Client(newLimiter("llm", cfg.LLM.Concurrency, logger))(llmClient)
}
judge, err := buildJudge(cfg.LLM.Judge, newLimiter("judge", cfg.LLM.Concurrency, logger))
if err != nil {
logger.Error("judge: config invalid", "err", err)
os.Exit(1)
Expand Down Expand Up @@ -416,7 +420,23 @@ func modelFor(c config.LLMConfig) string {
// call; a Judge request that fails past them is handled by the TOC
// builder, which keeps extraction's pages rather than degrading
// silently (HAL-1369).
func buildJudge(c config.JudgeBlock) (llmgate.Judge, error) {
// newLimiter builds one adaptive limiter for a provider and logs every
// change it makes: a throttled run must be visible, never silent.
func newLimiter(name string, c config.ConcurrencyBlock, logger *slog.Logger) *limit.Limiter {
return limit.New(limit.Config{
Initial: c.Initial,
Max: c.Max,
OnChange: func(e limit.Event) {
if e.Cause == "success" {
logger.Info("limiter: widened", "provider", name, "from", e.From, "to", e.To)
return
}
logger.Warn("limiter: narrowed", "provider", name, "cause", e.Cause, "from", e.From, "to", e.To, "paused_for", e.PausedFor, "err", e.Err)
},
})
}

func buildJudge(c config.JudgeBlock, lim *limit.Limiter) (llmgate.Judge, error) {
if c.TypeSafe.APIKey == "" {
return nil, nil
}
Expand All @@ -428,7 +448,9 @@ func buildJudge(c config.JudgeBlock) (llmgate.Judge, error) {
if err != nil {
return nil, err
}
return retry.NewJudge(retry.Config{MaxRetries: 3})(j), nil
// The limiter sits inside retry: each attempt takes a slot, and the
// failure that triggers a retry has already narrowed the limit.
return retry.NewJudge(retry.Config{MaxRetries: 3})(limit.Judge(lim)(j)), nil
}

func buildLLM(c config.LLMConfig) (llmgate.Client, error) {
Expand Down
134 changes: 82 additions & 52 deletions cmd/navbench/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,9 +19,11 @@ import (
"path/filepath"
"sort"
"strings"
"sync"
"time"

"github.com/hallelx2/llmgate/judge/typesafe"
"github.com/hallelx2/llmgate/middleware/limit"
"github.com/hallelx2/llmgate/middleware/retry"

"github.com/hallelx2/vectorless-engine/pkg/ingest"
Expand Down Expand Up @@ -71,7 +73,8 @@ func main() {
out := flag.String("out", "", "JSONL of per-question outcomes")
maxLeaves := flag.Int("leaves", 3, "sections read per question")
maxPages := flag.Int("pages", 40, "pages judged per question")
limit := flag.Int("limit", 0, "stop after this many questions (0 = all)")
limitQ := flag.Int("limit", 0, "stop after this many questions (0 = all)")
parallel := flag.Int("parallel", 1, "questions in flight at once; the provider's adaptive limiter governs requests")
flag.Parse()
if *qPath == "" || *trees == "" || *pdfs == "" {
fmt.Fprintln(os.Stderr, "usage: navbench -questions q.jsonl -trees dir -pdfs dir [-out o.jsonl]")
Expand All @@ -84,11 +87,14 @@ func main() {
fmt.Fprintln(os.Stderr, "judge:", err)
os.Exit(1)
}
nav := &retrieval.JudgeNavigator{Judge: retry.NewJudge(retry.Config{MaxRetries: 3})(tj), MaxLeaves: *maxLeaves, MaxPages: *maxPages}
lim := limit.New(limit.Config{Initial: 4, OnChange: func(e limit.Event) {
fmt.Fprintf(os.Stderr, " limiter %s %d -> %d %v\n", e.Cause, e.From, e.To, e.Err)
}})
nav := &retrieval.JudgeNavigator{Judge: retry.NewJudge(retry.Config{MaxRetries: 3})(limit.Judge(lim)(tj)), MaxLeaves: *maxLeaves, MaxPages: *maxPages}

qs := readQuestions(*qPath)
if *limit > 0 && len(qs) > *limit {
qs = qs[:*limit]
if *limitQ > 0 && len(qs) > *limitQ {
qs = qs[:*limitQ]
}
var of *os.File
if *out != "" {
Expand All @@ -98,60 +104,84 @@ func main() {

pageCache := map[string][]ingest.PageText{}
leafCache := map[string][]retrieval.NavLeaf{}
var results []outcome
// Documents are parsed once, up front and sequentially, so the
// parallel part is only Judge traffic.
for _, q := range qs {
o := outcome{ID: q.ID, Doc: q.Doc, Question: q.Question, Gold: q.Evidence}
leaves, pages, err := load(q.Doc, *trees, *pdfs, leafCache, pageCache)
if err != nil {
o.Err = err.Error()
results = append(results, o)
report(o)
continue
}
byNum := map[int]string{}
for _, p := range pages {
byNum[p.PageNumber] = p.Text
if _, ok := leafCache[q.Doc]; !ok {
_, _, _ = load(q.Doc, *trees, *pdfs, leafCache, pageCache)
}
loadPages := func(_ context.Context, l retrieval.NavLeaf) ([]retrieval.NavPage, error) {
var ps []retrieval.NavPage
for n := l.Start; n <= l.End; n++ {
if t, ok := byNum[n]; ok {
ps = append(ps, retrieval.NavPage{Number: n, Text: t})
}
runStart := time.Now()
results := make([]outcome, len(qs))
if *parallel < 1 {
*parallel = 1
}
sem := make(chan struct{}, *parallel)
var wg sync.WaitGroup
var outMu sync.Mutex
for i, q := range qs {
wg.Add(1)
sem <- struct{}{}
go func(i int, q question) {
defer wg.Done()
defer func() { <-sem }()
o := outcome{ID: q.ID, Doc: q.Doc, Question: q.Question, Gold: q.Evidence}
leaves, pages, err := load(q.Doc, *trees, *pdfs, leafCache, pageCache)
if err != nil {
o.Err = err.Error()
results[i] = o
outMu.Lock()
report(o)
outMu.Unlock()
return
}
byNum := map[int]string{}
for _, p := range pages {
byNum[p.PageNumber] = p.Text
}
loadPages := func(_ context.Context, l retrieval.NavLeaf) ([]retrieval.NavPage, error) {
var ps []retrieval.NavPage
for n := l.Start; n <= l.End; n++ {
if t, ok := byNum[n]; ok {
ps = append(ps, retrieval.NavPage{Number: n, Text: t})
}
}
return ps, nil
}
return ps, nil
}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
start := time.Now()
res, err := nav.Navigate(ctx, q.Question, leaves, loadPages)
cancel()
o.Seconds = time.Since(start).Seconds()
if err != nil {
o.Err = err.Error()
results = append(results, o)
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
start := time.Now()
res, err := nav.Navigate(ctx, q.Question, leaves, loadPages)
cancel()
o.Seconds = time.Since(start).Seconds()
if err != nil {
o.Err = err.Error()
} else {
for _, l := range res.Selected {
o.Selected = append(o.Selected, l.Title)
o.SelectedPages = append(o.SelectedPages, [2]int{l.Start, l.End})
}
for _, e := range res.Evidence {
o.Evidence = append(o.Evidence, e.Page.Number)
o.EvidenceP = append(o.EvidenceP, e.P)
}
o.PagesRead = len(res.Pages)
o.Requests, o.InTokens, o.CostUSD = res.Requests, res.Usage.InputTokens, res.Usage.CostUSD
o.LeafHit = allInRanges(q.Evidence, o.SelectedPages)
o.Recall = recall(q.Evidence, o.Evidence)
o.Hit = o.Recall == 1
}
results[i] = o
outMu.Lock()
report(o)
continue
}
for _, l := range res.Selected {
o.Selected = append(o.Selected, l.Title)
o.SelectedPages = append(o.SelectedPages, [2]int{l.Start, l.End})
}
for _, e := range res.Evidence {
o.Evidence = append(o.Evidence, e.Page.Number)
o.EvidenceP = append(o.EvidenceP, e.P)
}
o.PagesRead = len(res.Pages)
o.Requests, o.InTokens, o.CostUSD = res.Requests, res.Usage.InputTokens, res.Usage.CostUSD
o.LeafHit = allInRanges(q.Evidence, o.SelectedPages)
o.Recall = recall(q.Evidence, o.Evidence)
o.Hit = o.Recall == 1
results = append(results, o)
report(o)
if of != nil {
b, _ := json.Marshal(o)
of.Write(append(b, '\n'))
}
if of != nil {
b, _ := json.Marshal(o)
of.Write(append(b, '\n'))
}
outMu.Unlock()
}(i, q)
}
wg.Wait()
fmt.Printf("\nwall %.1fs for %d questions at parallel=%d; limiter now %d\n", time.Since(runStart).Seconds(), len(qs), *parallel, lim.Limit())
summarise(results)
}

Expand Down
26 changes: 23 additions & 3 deletions cmd/server/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ import (

"github.com/hallelx2/llmgate"
"github.com/hallelx2/llmgate/judge/typesafe"
"github.com/hallelx2/llmgate/middleware/limit"
"github.com/hallelx2/llmgate/middleware/retry"
"github.com/hallelx2/llmgate/pricing"
"github.com/hallelx2/llmgate/provider/anthropic"
Expand Down Expand Up @@ -144,7 +145,8 @@ func run() error {
if err != nil {
return fmt.Errorf("init llm: %w", err)
}
judge, err := buildJudge(cfg.Engine.LLM.Judge)
llmClient = limit.Client(newLimiter("llm", cfg.Engine.LLM.Concurrency, logger))(llmClient)
judge, err := buildJudge(cfg.Engine.LLM.Judge, newLimiter("judge", cfg.Engine.LLM.Concurrency, logger))
if err != nil {
logger.Error("judge: config invalid", "err", err)
os.Exit(1)
Expand Down Expand Up @@ -416,7 +418,23 @@ func modelFor(c enginecfg.LLMConfig) string {
// call; a Judge request that fails past them is handled by the TOC
// builder, which keeps extraction's pages rather than degrading
// silently (HAL-1369).
func buildJudge(c enginecfg.JudgeBlock) (llmgate.Judge, error) {
// newLimiter builds one adaptive limiter for a provider and logs every
// change it makes: a throttled run must be visible, never silent.
func newLimiter(name string, c enginecfg.ConcurrencyBlock, logger *slog.Logger) *limit.Limiter {
return limit.New(limit.Config{
Initial: c.Initial,
Max: c.Max,
OnChange: func(e limit.Event) {
if e.Cause == "success" {
logger.Info("limiter: widened", "provider", name, "from", e.From, "to", e.To)
return
}
logger.Warn("limiter: narrowed", "provider", name, "cause", e.Cause, "from", e.From, "to", e.To, "paused_for", e.PausedFor, "err", e.Err)
},
})
}

func buildJudge(c enginecfg.JudgeBlock, lim *limit.Limiter) (llmgate.Judge, error) {
if c.TypeSafe.APIKey == "" {
return nil, nil
}
Expand All @@ -428,7 +446,9 @@ func buildJudge(c enginecfg.JudgeBlock) (llmgate.Judge, error) {
if err != nil {
return nil, err
}
return retry.NewJudge(retry.Config{MaxRetries: 3})(j), nil
// The limiter sits inside retry: each attempt takes a slot, and the
// failure that triggers a retry has already narrowed the limit.
return retry.NewJudge(retry.Config{MaxRetries: 3})(limit.Judge(lim)(j)), nil
}

func buildLLM(c enginecfg.LLMConfig) (llmgate.Client, error) {
Expand Down
Loading
Loading