using System; using System.Collections.Generic; using System.Linq; using System.Text; using System.Text.Json; using System.Threading; using System.Threading.Tasks; using Gott.Models; using Microsoft.AspNetCore.Http; namespace Gott.Services; public partial class AnthropicService { // ── Multi-Modell-Panel (Fusion / MoA) ── // // Mehrere Panel-Agenten ("Proposer") beantworten dieselbe Anfrage PARALLEL und // streamen live. Anschließend werden ihre Antworten dem Worker-Agent als // -Block vorgelegt, der daraus EINE Gesamtantwort synthetisiert // und über die unveränderte Streaming-Pipeline ausgibt. // // MoA (Mixture-of-Agents): layers>1 → mehrere Proposer-Runden; jede Runde k>1 sieht // die Antworten der Vorrunde () und verfeinert. Fusion = layers==1. // JEDE Runde wird persistiert (mit Layer-Index im JSON-Transkript → F5-sicher, Runden-Divider beim Reload). // // Member laufen READ-ONLY (Anthropic: PlanMode-Filter, CLI: --permission-mode plan) // bzw. tool-los (OpenAI), damit parallele Member keine Dateien gleichzeitig schreiben. // Member-Arbeit wird NICHT über den Loop persistiert (Persist=false) – stattdessen // wird pro Member genau EINE Proposal-Message mit Tag="panel" + [Panel: Name]-Prefix // gespeichert (aus der Worker-History gefiltert, überlebt aber F5 als eigene Bubble). private class PanelMemberResult { public int Index { get; set; } public string Name { get; set; } public string Model { get; set; } public string Text { get; set; } public TokenUsage Usage { get; set; } // Token-Verbrauch dieses Members (für die Chain-Kosten) } /// /// Führt das Panel aus (Fusion = 1 Runde, MoA = Runden): pro Runde /// alle Member parallel; ab Runde 2 sieht jeder Member die Antworten der Vorrunde. JEDE Runde /// wird persistiert (Tag="panel", mit Layer-Index); der Synthese-Block der finalen Runde wird in die letzte User-Message von /// gemergt. Danach streamt der Worker als Synthesizer ganz /// normal weiter. Gibt die Anzahl beigetragener Member der finalen Runde zurück /// (0 = nichts injiziert → Worker antwortet regulär). /// public async Task RunPanel( HttpResponse response, List panel, List workerMessages, string chainId, string sessionId, string apiToken, int layers, TokenUsage usageSink, CancellationToken workCt, CancellationToken sseCt) { if (panel == null || panel.Count == 0) return 0; layers = Math.Clamp(layers, 1, 5); // Kosten-Backstop: max. 5 Proposer-Runden // Write-Lock: alle Member streamen LIVE parallel auf DIESELBE HttpResponse → // ohne Serialisierung würden sich ihre SSE-Frames auf Byte-Ebene vermischen. // Serialisiert nur die µs-Byte-Flushes, nicht die parallele Modell-Arbeit. using var sseGate = new SemaphoreSlim(1, 1); var mode = layers > 1 ? "moa" : "fusion"; await SseHelper.TryWriteSse(response, "panel_start", new { members = panel.Count, mode, layers }, sseCt, sseGate); // Gesamt-Token aller Member über ALLE Runden in den vom Aufrufer übergebenen Sink // akkumulieren. Jeder Member sammelt seine eigene Usage (result.Usage); summiert wird // erst NACH der Task.WhenAll-Barrier → kein Lock nötig. Der Aufrufer seedet damit die // Worker-Usage → die Panel-Kosten fließen über das done-Event (Live-Badge) UND // FinalizeChain (DB) als Teil der Chain-Kosten und überleben so F5. var panelUsage = usageSink ?? new TokenUsage(); // MoA: mehrere Proposer-Runden. Runde k>1 bekommt und verfeinert. // Fusion = layers==1 (keine Vorrunde). prev füttert die nächste Runde bzw. die finale Synthese. List prev = null; for (int layer = 1; layer <= layers; layer++) { await SseHelper.TryWriteSse(response, "panel_layer_start", new { layer, total = layers }, sseCt, sseGate); string prevBlock = prev == null ? null : BuildPreviousLayerBlock(prev, layer, layers); // Concurrency begrenzen: bei CLI-Membern sind das parallele claude-Children. using var sem = new SemaphoreSlim(Math.Min(3, panel.Count)); var tasks = new List>(); for (int i = 0; i < panel.Count; i++) { var member = panel[i]; var index = i; var lyr = layer; tasks.Add(Task.Run(async () => { await sem.WaitAsync(workCt); try { return await RunPanelMember(response, member, workerMessages, index, lyr, layers, prevBlock, chainId, sessionId, apiToken, sseGate, workCt, sseCt); } finally { sem.Release(); } }, workCt)); } var results = await Task.WhenAll(tasks); // Nach der Barrier (single-threaded): Token ALLER Member dieser Runde aufsummieren – // auch von Membern ohne verwertbaren Text (sie haben trotzdem Tokens verbraucht). foreach (var r in results) if (r?.Usage != null) panelUsage.Add(r.Usage); var ordered = results .Where(r => r != null && !string.IsNullOrWhiteSpace(r.Text)) .OrderBy(r => r.Index) .ToList(); await SseHelper.TryWriteSse(response, "panel_layer_done", new { layer, total = layers, members = ordered.Count }, sseCt, sseGate); if (ordered.Count == 0) { // Keine verwertbaren Antworten in dieser Runde → ohne Synthese weiter. // Die bisher verbrauchten Member-Tokens stecken bereits im usageSink (oben summiert) // und werden vom Aufrufer in die Worker-Usage geseedet → ehrliche Kosten. await SseHelper.TryWriteSse(response, "panel_done", new { synthesizing = false, members = 0 }, sseCt, sseGate); return 0; } prev = ordered; } // Synthese-Block aus der LETZTEN Runde bauen und in die letzte User-Message einmergen. var sb = new StringBuilder(); sb.AppendLine(); sb.AppendLine("---"); sb.AppendLine("[Multi-Modell-Panel] Mehrere unabhängige KI-Agenten haben parallel auf die obige Anfrage geantwortet. " + "Deine Aufgabe: Synthetisiere daraus EINE optimale Gesamtantwort für den User. Nimm die stärksten Argumente, " + "korrektesten Fakten und besten Ideen aus allen Beiträgen, korrigiere erkennbare Fehler und vermeide bloßes " + "Aneinanderreihen. Widersprechen sich Beiträge, wäge ab und entscheide begründet. Antworte direkt und natürlich " + "dem User (du musst das Panel nicht erwähnen). Du kannst deine eigenen Tools nutzen, um zu prüfen oder die Aufgabe abzuschließen."); sb.AppendLine(); sb.AppendLine(""); foreach (var r in prev) { sb.AppendLine(); sb.AppendLine($"### Antwort von {r.Name} ({r.Model}):"); sb.AppendLine(r.Text.Trim()); } sb.AppendLine(""); MergePanelIntoLastUserText(workerMessages, sb.ToString()); await SseHelper.TryWriteSse(response, "panel_done", new { synthesizing = true, members = prev.Count }, sseCt, sseGate); return prev.Count; } /// /// Führt einen einzelnen Panel-Member aus (Provider-Weiche, read-only/tool-los), /// persistiert seine Proposal-Message (Tag="panel") und sendet panel_member_* SSE-Events. /// Fängt Member-Fehler ab (ein fehlerhafter Member darf das Panel nicht killen). /// private async Task RunPanelMember( HttpResponse response, SubAgentContext member, List workerMessages, int index, int layer, int totalLayers, string previousLayerBlock, string chainId, string sessionId, string apiToken, SemaphoreSlim sseGate, CancellationToken workCt, CancellationToken sseCt) { var name = member.Agent?.Name ?? $"Panel-Agent {index + 1}"; await SseHelper.TryWriteSse(response, "panel_member_start", new { index, name, model = member.ModelId, layer, totalLayers }, sseCt, sseGate); // Eigene, ISOLIERTE Kopie pro Member. Bei MoA (Runde > 1) wird der Vorrunden-Block in den // bestehenden Text-Block der letzten User-Message gemergt – isoliert, damit er nicht in // workerMessages oder parallele Member leckt. var memberMessages = CloneForMember(workerMessages, previousLayerBlock); string text; List segments = null; // interleavtes Transkript (Text + Tools) für die Reload-Persistenz TokenUsage memberUsage = null; // Token-Verbrauch dieses Members (null = nicht verfügbar, z.B. OpenAI) try { if (!string.IsNullOrEmpty(member.Provider) && _openAiService != null) { // OpenAI-kompatibler Member: tool-los (CompleteText), Proposer aus Kontext. // Kein Tool-Transkript → unten als einzelnes Text-Segment persistiert. var systemText = ExtractSystemPromptText(member.SystemPrompt); text = await _openAiService.CompleteText(member.ModelId, systemText, memberMessages, apiToken); } else if (string.IsNullOrEmpty(apiToken) && _cliService != null && _cliService.IsAvailable) { // Kein API-Key → CLI-Member, read-only (Plan-Modus: nur Read/Glob/Grep). // Streamt LIVE (Token + Tool-Uses) über panel_member_*-Events und liefert das Transkript. var (t, segs, u) = await RunPanelMemberViaCli(response, member, memberMessages, index, sseGate, workCt, sseCt); text = t; segments = segs; memberUsage = u; } else { // Anthropic-API-Member: read-only (PlanMode) + NICHT persistieren (Caller persistiert einmal). // Live-Streaming über den Loop: pro Tool-Runde panel_member_text / panel_member_tool. // TranscriptSink sammelt Text-/Tool-Segmente für die Reload-Persistenz. memberUsage = new TokenUsage(); segments = new List(); var (t, _) = await RunNonStreamingToolLoop( member.ModelId, member.SystemPrompt, new ToolPayloadOptions { Workspace = member.Workspace, AgentId = member.AgentId, UseMemory = member.UseMemory, WebSearchVersion = member.WebSearchVersion, ToolChoice = "auto", PlanMode = true }, memberMessages, chainId, sessionId, apiToken, memberUsage, 0, 12, workCt, new NonStreamingLoopCallbacks { Persist = false, SseResponse = response, SseCt = sseCt, SseGate = sseGate, SsePrefix = "panel_member", MemberIndex = index, AgentName = name, TranscriptSink = segments }, member.Effort); text = t; } } catch (OperationCanceledException) { throw; } catch (Exception ex) { text = $"[Fehler bei {name}: {ex.Message}]"; segments = null; } text = (text ?? "").Trim(); // Member-Transkript als EINE strukturierte Message persistieren (Tag="panel" → aus // Worker-History gefiltert). Content = JSON {name, model, layer, totalLayers, // segments:[{t:"text"|"tool",...}]} → der Reload stellt Text UND Tool-Boxen wieder her // (genau wie live) und gruppiert bei MoA nach Runden mit Divider. Fällt auf ein einzelnes // Text-Segment zurück (OpenAI-Member, Fehler oder wenn nichts gestreamt wurde). JEDE Runde // wird persistiert (Iteration=layer*100+index+1 → stabile, layer-aware Reihenfolge) → MoA- // Zwischenrunden überleben F5. if (segments == null || segments.Count == 0) { segments = new List(); if (!string.IsNullOrEmpty(text)) segments.Add(new Dictionary { ["t"] = "text", ["v"] = text }); } if (segments.Count > 0) { var content = JsonSerializer.Serialize(new { name, model = member.ModelId, layer, totalLayers, segments }); await _db.SaveMessage(new MessageRecord { ChainId = chainId, Iteration = layer * 100 + index + 1, Role = "assistant", Type = "text", Content = content, Tag = "panel" }); } await SseHelper.TryWriteSse(response, "panel_member_done", new { index, name, model = member.ModelId, text }, sseCt, sseGate); return new PanelMemberResult { Index = index, Name = name, Model = member.ModelId, Text = text, Usage = memberUsage }; } /// /// Führt einen Panel-Member über die Claude CLI aus (read-only via Plan-Modus) und /// streamt seine Arbeit LIVE: Text-Deltas (panel_member_text), Tool-Start /// (panel_member_tool_use) und Tool-Result (panel_member_tool_end) – jeweils mit /// Member-Index, damit das Frontend sie der richtigen Bubble zuordnet. Die SSE-Writes /// laufen durch den geteilten (parallele Member → sonst /// vermischen sich die Frames). Persistiert nichts – der Caller speichert eine Proposal-Message. /// private async Task<(string text, List segments, TokenUsage usage)> RunPanelMemberViaCli( HttpResponse response, SubAgentContext member, List memberMessages, int index, SemaphoreSlim sseGate, CancellationToken workCt, CancellationToken sseCt) { if (_cliService == null || !_cliService.IsAvailable) return ("", new List(), null); var systemText = ExtractSystemPromptText(member.SystemPrompt); var prompt = ExtractLastUserText(memberMessages) ?? ""; // Bild-/Datei-Anhänge der letzten User-Message als Temp-Dateien ablegen, damit der // CLI-Member sie per Read-Tool sehen kann (Screenshots, PDFs etc.) – exakt wie der Worker. // Ohne das wären die Member bei bildbasierten Fragen blind → read-only Tools nutzlos. var tempImagePaths = new List(); var imageRefs = ExtractLastUserImageRefs(memberMessages, tempImagePaths); if (!string.IsNullOrEmpty(imageRefs)) prompt = imageRefs + "\n" + prompt; var historyContext = BuildHistoryContext(memberMessages); if (!string.IsNullOrEmpty(historyContext)) prompt = $"\n{historyContext}\n\n\n\n{prompt}\n"; var fullText = new StringBuilder(); // Transkript-Segmente (Text + Tools interleaved) für die Reload-Persistenz. var segments = new List(); var segText = new StringBuilder(); void FlushText() { if (segText.Length == 0) return; segments.Add(new Dictionary { ["t"] = "text", ["v"] = segText.ToString() }); segText.Clear(); } // Tool-Uses per ID korrelieren: Ein Member kann MEHRERE Tools in einer Assistant-Message // aufrufen (ToolStart#1,#2,… ToolResult#1,#2,…). Ein einzelner "currentTool"-Zeiger würde // dann Boxen vertauschen/verwaisen lassen. Außerdem ist der Tool-Input erst bei ToolEnd // (content_block_stop) vollständig – ToolStart/ToolResult tragen ihn nicht. var toolNames = new Dictionary(); // toolId -> CLI-Name var toolInputs = new Dictionary(); // toolId -> Input-JSON CliChatResult cliResult = null; // vor dem try, damit die Usage nach dem finally lesbar ist try { cliResult = await _cliService.ChatStreamingAsync( prompt, async evt => { switch (evt.Type) { case CliEventType.TextDelta: fullText.Append(evt.Text); segText.Append(evt.Text); await SseHelper.TryWriteSse(response, "panel_member_text", new { index, text = evt.Text }, sseCt, sseGate); break; case CliEventType.ToolStart: FlushText(); // Text-Segment vor der Tool-Box schließen (Reihenfolge wie live) if (!string.IsNullOrEmpty(evt.ToolId)) toolNames[evt.ToolId] = evt.ToolName; await SseHelper.TryWriteSse(response, "panel_member_tool_use", new { index, id = evt.ToolId, name = evt.ToolName }, sseCt, sseGate); break; case CliEventType.ToolEnd: // Input streamt erst nach ToolStart ein → hier vollständig. Für ToolResult merken. if (!string.IsNullOrEmpty(evt.ToolId)) toolInputs[evt.ToolId] = evt.ToolInput; break; case CliEventType.ToolResult: var rid = evt.ToolId ?? ""; toolNames.TryGetValue(rid, out var rName); toolInputs.TryGetValue(rid, out var rInput); segments.Add(new Dictionary { ["t"] = "tool", ["name"] = rName ?? "", ["input"] = rInput ?? "", ["result"] = TruncateToolResult(evt.ToolResultContent), ["error"] = evt.ToolResultIsError }); await SseHelper.TryWriteSse(response, "panel_member_tool_end", new { index, id = rid, name = rName, input = rInput, result = evt.ToolResultContent, isError = evt.ToolResultIsError }, sseCt, sseGate); break; } }, modelId: member.ModelId, systemPrompt: systemText, workspace: member.Workspace, hasWorkspace: !string.IsNullOrEmpty(member.Workspace), planMode: true, effort: member.Effort, cancellationToken: workCt ); } finally { CleanupTempImages(tempImagePaths); } FlushText(); // restliches Text-Segment am Ende sichern var usage = cliResult == null ? null : new TokenUsage { InputTokens = cliResult.InputTokens, OutputTokens = cliResult.OutputTokens, CacheReadTokens = cliResult.CacheReadTokens, CacheCreationTokens = cliResult.CacheCreationTokens }; return (fullText.ToString(), segments, usage); } /// Extrahiert den System-Prompt-Text aus den List<object>-Blöcken eines SubAgentContext. private static string ExtractSystemPromptText(List systemPrompt) { if (systemPrompt == null) return ""; var parts = new List(); foreach (var block in systemPrompt) { var json = JsonSerializer.Serialize(block); var el = JsonSerializer.Deserialize(json); if (el.TryGetProperty("text", out var t)) parts.Add(t.GetString() ?? ""); } return string.Join("\n\n", parts); } /// /// Hängt den Panel-Synthese-Block an den BESTEHENDEN Text-Block der letzten User-Message an. /// Wichtig: kein zweiter Text-Block – sonst sieht der CLI-Worker (ExtractLastUserText liefert /// nur den ersten Text-Block) die Synthese-Anweisung nicht. So funktioniert es für API, CLI und OpenAI. /// private static void MergePanelIntoLastUserText(List messages, string panelBlock) { for (int i = messages.Count - 1; i >= 0; i--) { if (messages[i] is Dictionary d && d.TryGetValue("role", out var r) && (r as string) == "user" && d.TryGetValue("content", out var c) && c is List contentList) { for (int j = 0; j < contentList.Count; j++) { var json = JsonSerializer.Serialize(contentList[j]); var el = JsonSerializer.Deserialize(json); if (el.TryGetProperty("type", out var t) && t.GetString() == "text") { var orig = el.TryGetProperty("text", out var txt) ? txt.GetString() ?? "" : ""; contentList[j] = new { type = "text", text = orig + "\n\n" + panelBlock }; return; } } // Kein Text-Block vorhanden → einen anhängen. contentList.Add(new { type = "text", text = panelBlock }); return; } } } /// /// Liefert eine pro-Member isolierte Kopie der Worker-Messages. Ist /// gesetzt (MoA, Runde > 1), wird er in den bestehenden Text-Block der letzten User-Message /// gemergt – auf einer GEKLONTEN Message + Content-Liste, damit die Mutation NICHT in /// workerMessages oder parallele Member leckt. block==null → einfache Shallow-Kopie (wie Fusion). /// private static List CloneForMember(List messages, string block) { var copy = new List(messages); if (string.IsNullOrEmpty(block)) return copy; for (int i = copy.Count - 1; i >= 0; i--) { if (copy[i] is Dictionary d && d.TryGetValue("role", out var r) && (r as string) == "user" && d.TryGetValue("content", out var c) && c is List contentList) { var newContent = new List(contentList); var newD = new Dictionary(d) { ["content"] = newContent }; bool merged = false; for (int j = 0; j < newContent.Count; j++) { var json = JsonSerializer.Serialize(newContent[j]); var el = JsonSerializer.Deserialize(json); if (el.TryGetProperty("type", out var t) && t.GetString() == "text") { var orig = el.TryGetProperty("text", out var txt) ? txt.GetString() ?? "" : ""; newContent[j] = new { type = "text", text = orig + "\n\n" + block }; merged = true; break; } } if (!merged) newContent.Add(new { type = "text", text = block }); copy[i] = newD; return copy; } } return copy; } /// /// Baut den <previous_layer_responses>-Block für MoA: die Antworten der Vorrunde als /// Referenz, mit der Anweisung, die eigene Antwort zu VERBESSERN (nicht blind zu kopieren). /// private static string BuildPreviousLayerBlock(List prev, int layer, int total) { var sb = new StringBuilder(); sb.AppendLine(); sb.AppendLine("---"); sb.AppendLine($"[Multi-Modell-Panel · Runde {layer} von {total}] Andere KI-Agenten haben in der vorherigen Runde " + "auf dieselbe Anfrage geantwortet. Nutze ihre Antworten als zusätzliche Referenz: übernimm korrekte Fakten " + "und gute Ideen, korrigiere erkennbare Fehler und VERBESSERE deine eigene Antwort. Kopiere nicht blind – " + "denke kritisch und liefere die bestmögliche eigenständige Antwort auf die ursprüngliche Anfrage."); sb.AppendLine(); sb.AppendLine(""); foreach (var r in prev) { sb.AppendLine(); sb.AppendLine($"### Antwort von {r.Name} ({r.Model}):"); sb.AppendLine(r.Text.Trim()); } sb.AppendLine(""); return sb.ToString(); } }