use parallel llm scorer

Signed-off-by: kaedwen <kaedwen@heinrich.blue>

config.example.yaml

@@ -77,11 +77,12 @@ news:
 llm_scorer:
   enabled: false              # Set to true to enable LLM-based sentiment
   endpoint: http://localhost:11434
-  model_name: mistral         # "mistral", "llama2", "llama2:70b"
+  model_name: mistral         # "mistral", "llama2", "llama2:70b", "phi3:mini", "llama3.2:1b"
   timeout: 30                 # Seconds or Go duration
   temperature: 0.3            # 0.0-1.0, lower = more deterministic
   max_retries: 2
   ensemble_weight: 0.7        # 0.0-1.0 (1.0 = LLM only, 0.7 = 70% LLM + 30% keyword)
+  max_parallel: 20            # Max parallel LLM requests (worker pool size)
 
 # Logging
 log_level: info               # debug, info, warn, error

pkg/app/news/aggregator.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/pheinrich/aitrade/pkg/config"
 	"github.com/pheinrich/aitrade/pkg/db"
+	"github.com/pheinrich/aitrade/pkg/model"
 	"golang.org/x/sync/errgroup"
 )
 
@@ -22,6 +23,10 @@ type Aggregator struct {
 	retentionDays int
 	logger        *slog.Logger
 	onNewsUpdated func() // Callback for SSE notifications
+
+	// Worker pool for LLM scoring
+	llmJobs    chan *model.NewsArticle
+	llmWorkers int
 }
 
 func NewAggregator(
@@ -32,6 +37,12 @@ func NewAggregator(
 	llmScorer *LLMScorer,
 	logger *slog.Logger,
 ) *Aggregator {
+	// Determine worker pool size
+	llmWorkers := 20 // Default
+	if llmScorer != nil && llmScorer.cfg.MaxParallel > 0 {
+		llmWorkers = llmScorer.cfg.MaxParallel
+	}
+
 	agg := &Aggregator{
 		sources:       make([]Source, 0),
 		newsRepo:      newsRepo,
@@ -41,6 +52,8 @@ func NewAggregator(
 		fetchTimeout:  fetchTimeout,
 		retentionDays: retentionDays,
 		logger:        logger,
+		llmJobs:       make(chan *model.NewsArticle, 1000), // Buffered channel for queued articles
+		llmWorkers:    llmWorkers,
 	}
 
 	return agg
@@ -123,8 +136,17 @@ func (a *Aggregator) Run(ctx context.Context) error {
 	a.logger.Info("news aggregator starting",
 		slog.Int("sources", len(a.sources)),
 		slog.Duration("poll_interval", a.pollInterval),
+		slog.Int("llm_workers", a.llmWorkers),
 	)
 
+	// Start LLM worker pool if LLM scorer is enabled
+	if a.llmScorer != nil && a.llmScorer.Enabled() {
+		for i := 0; i < a.llmWorkers; i++ {
+			go a.llmWorker(ctx, i)
+		}
+		a.logger.Info("LLM worker pool started", slog.Int("workers", a.llmWorkers))
+	}
+
 	// Fetch immediately on start
 	if err := a.fetchAllSources(ctx); err != nil {
 		a.logger.Error("initial news fetch failed", slog.Any("error", err))
@@ -137,6 +159,7 @@ func (a *Aggregator) Run(ctx context.Context) error {
 		select {
 		case <-ctx.Done():
 			a.logger.Info("news aggregator stopping")
+			close(a.llmJobs) // Signal workers to stop
 			return ctx.Err()
 		case <-ticker.C:
 			if err := a.fetchAllSources(ctx); err != nil {
@@ -146,10 +169,47 @@ func (a *Aggregator) Run(ctx context.Context) error {
 	}
 }
 
+// llmWorker processes articles from the LLM jobs queue
+func (a *Aggregator) llmWorker(ctx context.Context, workerID int) {
+	a.logger.Debug("LLM worker started", slog.Int("worker_id", workerID))
+
+	for {
+		select {
+		case <-ctx.Done():
+			a.logger.Debug("LLM worker stopping", slog.Int("worker_id", workerID))
+			return
+		case article, ok := <-a.llmJobs:
+			if !ok {
+				// Channel closed, worker exits
+				a.logger.Debug("LLM worker channel closed", slog.Int("worker_id", workerID))
+				return
+			}
+
+			// Process article with LLM scoring
+			if err := a.llmScorer.Analyze(ctx, article); err != nil {
+				a.logger.Warn("LLM scoring failed",
+					slog.Int("worker_id", workerID),
+					slog.String("article_id", fmt.Sprintf("%d", article.ID)),
+					slog.Any("error", err))
+				// Article falls back to keyword sentiment (already set)
+			}
+
+			// Store article (with LLM score or fallback keyword score)
+			if err := a.newsRepo.Create(ctx, article); err != nil {
+				a.logger.Error("failed to store article",
+					slog.Int("worker_id", workerID),
+					slog.String("url", article.URL),
+					slog.Any("error", err),
+				)
+			}
+		}
+	}
+}
+
 func (a *Aggregator) fetchAllSources(ctx context.Context) error {
 	a.logger.Debug("fetching from all news sources")
 
-	var totalFetched, totalStored atomic.Int64
+	var totalFetched atomic.Int64
 
 	// Use errgroup for parallel fetching
 	g, gCtx := errgroup.WithContext(ctx)
@@ -173,33 +233,36 @@ func (a *Aggregator) fetchAllSources(ctx context.Context) error {
 
 			totalFetched.Add(int64(len(articles)))
 
-			// Analyze sentiment and store
+			// Process articles: keyword analysis + queue for LLM if enabled
 			for _, article := range articles {
-				// Always use keyword analyzer first (fast)
+				// Always use keyword analyzer first (fast, synchronous)
 				a.analyzer.Analyze(article)
 
-				// If LLM scorer enabled, use it (may fallback to keyword)
 				if a.llmScorer != nil && a.llmScorer.Enabled() {
-					if err := a.llmScorer.Analyze(sourceCtx, article); err != nil {
+					// Send to LLM worker pool for async processing
-						a.logger.Warn("LLM scoring failed",
+					select {
-							slog.String("url", article.URL),
+					case a.llmJobs <- article:
-							slog.Any("error", err))
+						// Article queued successfully for LLM processing + DB insert
-						// Article already has keyword sentiment, continue
+					case <-sourceCtx.Done():
+						// Context canceled, store with keyword score only
+						article.SentimentMethod = "keyword"
+						if err := a.newsRepo.Create(sourceCtx, article); err != nil {
+							a.logger.Error("failed to store article",
+								slog.String("url", article.URL),
+								slog.Any("error", err),
+							)
+						}
 					}
 				} else {
-					// No LLM scorer, mark as keyword-only
+					// No LLM scorer, store immediately with keyword score
 					article.SentimentMethod = "keyword"
+					if err := a.newsRepo.Create(sourceCtx, article); err != nil {
+						a.logger.Error("failed to store article",
+							slog.String("url", article.URL),
+							slog.Any("error", err),
+						)
+					}
 				}
-
-				if err := a.newsRepo.Create(sourceCtx, article); err != nil {
-					a.logger.Error("failed to store article",
-						slog.String("url", article.URL),
-						slog.Any("error", err),
-					)
-					continue
-				}
-
-				totalStored.Add(1)
 			}
 
 			return nil
@@ -212,16 +275,15 @@ func (a *Aggregator) fetchAllSources(ctx context.Context) error {
 	}
 
 	fetched := int(totalFetched.Load())
-	stored := int(totalStored.Load())
 
 	a.logger.Info("news fetch completed",
 		slog.Int("fetched", fetched),
-		slog.Int("stored", stored),
+		slog.Int("queued_for_llm", fetched), // All articles are queued for processing
-		slog.Int("duplicates", fetched-stored),
 	)
 
-	// Notify listeners if new articles were stored
+	// Notify listeners (workers will update DB asynchronously)
-	if stored > 0 && a.onNewsUpdated != nil {
+	if fetched > 0 && a.onNewsUpdated != nil {
+		// Note: This fires immediately after fetch, not after all DB inserts
 		a.onNewsUpdated()
 	}
 

pkg/config/config.go

@@ -102,6 +102,7 @@ type LLMScorerConfig struct {
 	Temperature    float64  `yaml:"temperature"`
 	MaxRetries     int      `yaml:"max_retries"`
 	EnsembleWeight float64  `yaml:"ensemble_weight"` // 0.0-1.0, 1.0 = LLM only
+	MaxParallel    int      `yaml:"max_parallel"`    // Max parallel LLM requests (default: 20)
 }
 
 func Load() (*Config, error) {
@@ -165,6 +166,7 @@ func Load() (*Config, error) {
 			Temperature:    getEnvFloat("LLM_SCORER_TEMPERATURE", 0.3),
 			MaxRetries:     getEnvInt("LLM_SCORER_MAX_RETRIES", 2),
 			EnsembleWeight: getEnvFloat("LLM_SCORER_ENSEMBLE_WEIGHT", 0.7),
+			MaxParallel:    getEnvInt("LLM_SCORER_MAX_PARALLEL", 20),
 		},
 		LogLevel: getEnv("LOG_LEVEL", "info"),
 	}