using System; using System.IO; using System.Collections.Concurrent; using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; using Dapper; using Microsoft.Data.Sqlite; using Microsoft.Extensions.Hosting; namespace Gott.Services; /// /// Background-Service der Messages embedded: /// 1. Beim Start: Backfill aller bestehenden Text-Messages /// 2. Laufend: Neue Messages über Queue verarbeiten /// public class EmbeddingBackgroundService : BackgroundService { readonly VectorSearchService _vectorSearch; readonly EmbeddingService _embedding; readonly string _dbConnectionString; readonly ConcurrentQueue _queue = new(); readonly SemaphoreSlim _signal = new(0); readonly string _dataPath; public EmbeddingBackgroundService(VectorSearchService vectorSearch, EmbeddingService embedding, string dbPath) { _vectorSearch = vectorSearch; _embedding = embedding; _dbConnectionString = $"Data Source={dbPath}"; _dataPath = Path.GetDirectoryName(dbPath); } /// Enqueue eine neue Message zum Embedden. public void Enqueue(long messageId, string chainId, string sessionId, string agentId, string role, string text) { _queue.Enqueue(new EmbeddingJob(messageId, chainId, sessionId, agentId, role, text)); _signal.Release(); } protected override async Task ExecuteAsync(CancellationToken ct) { if (!_embedding.IsAvailable) { Console.WriteLine("[EmbeddingBG] Embedding model not available – service disabled."); return; } // Kurz warten bis App vollständig gestartet await Task.Delay(3000, ct); // Backfill bestehender Messages await BackfillAsync(ct); // Memory-Dateien embedden await BackfillMemoryFilesAsync(ct); // Live-Embedding neuer Messages Console.WriteLine("[EmbeddingBG] Waiting for new messages..."); while (!ct.IsCancellationRequested) { try { await _signal.WaitAsync(ct); while (_queue.TryDequeue(out var job)) { if (ct.IsCancellationRequested) break; try { if (job.Role.StartsWith("memory:")) { // Memory-File-Embedding var filePath = job.Role.Substring("memory:".Length); await _vectorSearch.EmbedMemoryFileAsync(job.AgentId, filePath, job.Text); } else { await _vectorSearch.EmbedMessageAsync(job.MessageId, job.ChainId, job.SessionId, job.AgentId, job.Role, job.Text); } } catch (Exception ex) { Console.WriteLine($"[EmbeddingBG] Error embedding job: {ex.Message}"); } } } catch (OperationCanceledException) { break; } } } async Task BackfillAsync(CancellationToken ct) { Console.WriteLine("[EmbeddingBG] Starting backfill of existing messages..."); var existingIds = await _vectorSearch.GetEmbeddedMessageIdsAsync(); Console.WriteLine($"[EmbeddingBG] Already embedded: {existingIds.Count} messages"); using var conn = new SqliteConnection(_dbConnectionString); await conn.OpenAsync(ct); // Alle text-Messages laden (user + assistant, inkl. archivierte Sessions) var messages = await conn.QueryAsync(@" SELECT m.id, m.chain_id, c.session_id, c.agent_id, m.role, m.content, m.type FROM messages m JOIN chains c ON m.chain_id = c.chain_id WHERE m.type = 'text' AND m.role IN ('user', 'assistant') AND m.content IS NOT NULL AND LENGTH(m.content) >= 10 AND m.deleted_on IS NULL ORDER BY m.id ASC "); int total = 0; int embedded = 0; int skipped = 0; foreach (var msg in messages) { if (ct.IsCancellationRequested) break; total++; if (existingIds.Contains(msg.Id)) { skipped++; continue; } try { // SessionId kann auch aus archive kommen var sessionId = msg.SessionId ?? ""; await _vectorSearch.EmbedMessageAsync(msg.Id, msg.ChainId, sessionId, msg.AgentId, msg.Role, msg.Content); embedded++; if (embedded % 100 == 0) Console.WriteLine($"[EmbeddingBG] Backfill progress: {embedded} embedded, {skipped} skipped, {total} total"); } catch (Exception ex) { Console.WriteLine($"[EmbeddingBG] Error backfilling message {msg.Id}: {ex.Message}"); } } Console.WriteLine($"[EmbeddingBG] Backfill complete: {embedded} new embeddings, {skipped} skipped, {total} total messages."); } async Task BackfillMemoryFilesAsync(CancellationToken ct) { Console.WriteLine("[EmbeddingBG] Starting memory file backfill..."); var agentsDir = Path.Combine(_dataPath, "agents"); if (!Directory.Exists(agentsDir)) { Console.WriteLine("[EmbeddingBG] No agents directory found, skipping memory backfill."); return; } int total = 0; foreach (var agentDir in Directory.GetDirectories(agentsDir)) { if (ct.IsCancellationRequested) break; var agentId = Path.GetFileName(agentDir); // Alle .md Dateien im Agent-Verzeichnis (rekursiv) var mdFiles = Directory.GetFiles(agentDir, "*.md", SearchOption.AllDirectories); foreach (var mdFile in mdFiles) { if (ct.IsCancellationRequested) break; try { var text = await File.ReadAllTextAsync(mdFile, ct); if (string.IsNullOrWhiteSpace(text) || text.Length < 10) continue; var relativePath = Path.GetRelativePath(agentDir, mdFile); await _vectorSearch.EmbedMemoryFileAsync(agentId, relativePath, text); total++; } catch (Exception ex) { Console.WriteLine($"[EmbeddingBG] Error embedding memory file {mdFile}: {ex.Message}"); } } } Console.WriteLine($"[EmbeddingBG] Memory file backfill complete: {total} files embedded."); } /// Embeddet eine einzelne Memory-Datei (nach memory_write). public void EnqueueMemoryFile(string agentId, string filePath, string text) { _queue.Enqueue(new EmbeddingJob(0, "", "", agentId, "memory:" + filePath, text)); _signal.Release(); } record EmbeddingJob(long MessageId, string ChainId, string SessionId, string AgentId, string Role, string Text); class BackfillRow { public long Id { get; set; } public string ChainId { get; set; } public string SessionId { get; set; } public string AgentId { get; set; } public string Role { get; set; } public string Content { get; set; } public string Type { get; set; } } }