Panel agents answer the user prompt in parallel (read-only, non-streaming); their responses are merged into the worker's last user message as a <panel_responses> block, and the session agent synthesizes one final streamed answer. The RunPanel engine is built layer-ready for future MoA. - New session fields: multi_model_mode (off/fusion), panel_agent_ids (CSV) with idempotent column migration - New partial Services/AnthropicService.Panel.cs holding the RunPanel engine - Provider-agnostic members: Anthropic RunNonStreamingToolLoop / OpenAI CompleteText / CLI plan-mode; all forced read-only, persisted with Tag=panel - SSE events: panel_start, panel_member_start, panel_member_done, panel_done - Frontend: multi-agent picker, live panel bubbles, history rendering - Graceful degradation: panel errors fall back to worker-only Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
267 lines
12 KiB
C#
267 lines
12 KiB
C#
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) ──
|
||
//
|
||
// Mehrere Panel-Agenten ("Proposer") beantworten dieselbe Anfrage PARALLEL und
|
||
// non-streaming. Anschließend werden ihre Antworten dem Worker-Agent als
|
||
// <panel_responses>-Block vorgelegt, der daraus EINE Gesamtantwort synthetisiert
|
||
// und über die unveränderte Streaming-Pipeline ausgibt.
|
||
//
|
||
// 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; }
|
||
}
|
||
|
||
/// <summary>
|
||
/// Führt das Panel aus (Fusion): alle Member parallel, sammelt ihre Antworten,
|
||
/// persistiert sie (Tag="panel") und merged den Synthese-Block in die letzte
|
||
/// User-Message von <paramref name="workerMessages"/> ein. Danach streamt der Worker
|
||
/// als Synthesizer ganz normal weiter. Gibt die Anzahl beigetragener Member zurück
|
||
/// (0 = nichts injiziert → Worker antwortet regulär).
|
||
/// </summary>
|
||
public async Task<int> RunPanel(
|
||
HttpResponse response,
|
||
List<SubAgentContext> panel,
|
||
List<object> workerMessages,
|
||
string chainId,
|
||
string sessionId,
|
||
string apiToken,
|
||
CancellationToken workCt,
|
||
CancellationToken sseCt)
|
||
{
|
||
if (panel == null || panel.Count == 0) return 0;
|
||
|
||
await SseHelper.TryWriteSse(response, "panel_start", new { members = panel.Count, mode = "fusion" }, sseCt);
|
||
|
||
// Concurrency begrenzen: bei CLI-Membern sind das parallele claude-Children.
|
||
using var sem = new SemaphoreSlim(Math.Min(3, panel.Count));
|
||
var tasks = new List<Task<PanelMemberResult>>();
|
||
for (int i = 0; i < panel.Count; i++)
|
||
{
|
||
var member = panel[i];
|
||
var index = i;
|
||
tasks.Add(Task.Run(async () =>
|
||
{
|
||
await sem.WaitAsync(workCt);
|
||
try { return await RunPanelMember(response, member, workerMessages, index, chainId, sessionId, apiToken, workCt, sseCt); }
|
||
finally { sem.Release(); }
|
||
}, workCt));
|
||
}
|
||
|
||
var results = await Task.WhenAll(tasks);
|
||
|
||
var ordered = results
|
||
.Where(r => r != null && !string.IsNullOrWhiteSpace(r.Text))
|
||
.OrderBy(r => r.Index)
|
||
.ToList();
|
||
|
||
if (ordered.Count == 0)
|
||
{
|
||
await SseHelper.TryWriteSse(response, "panel_done", new { synthesizing = false, members = 0 }, sseCt);
|
||
return 0;
|
||
}
|
||
|
||
// Synthese-Block 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("<panel_responses>");
|
||
foreach (var r in ordered)
|
||
{
|
||
sb.AppendLine();
|
||
sb.AppendLine($"### Antwort von {r.Name} ({r.Model}):");
|
||
sb.AppendLine(r.Text.Trim());
|
||
}
|
||
sb.AppendLine("</panel_responses>");
|
||
|
||
MergePanelIntoLastUserText(workerMessages, sb.ToString());
|
||
|
||
await SseHelper.TryWriteSse(response, "panel_done", new { synthesizing = true, members = ordered.Count }, sseCt);
|
||
return ordered.Count;
|
||
}
|
||
|
||
/// <summary>
|
||
/// 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).
|
||
/// </summary>
|
||
private async Task<PanelMemberResult> RunPanelMember(
|
||
HttpResponse response,
|
||
SubAgentContext member,
|
||
List<object> workerMessages,
|
||
int index,
|
||
string chainId,
|
||
string sessionId,
|
||
string apiToken,
|
||
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 }, sseCt);
|
||
|
||
var memberMessages = new List<object>(workerMessages); // eigene Kopie pro Member
|
||
string text;
|
||
try
|
||
{
|
||
if (!string.IsNullOrEmpty(member.Provider) && _openAiService != null)
|
||
{
|
||
// OpenAI-kompatibler Member: tool-los (CompleteText), Proposer aus Kontext.
|
||
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).
|
||
text = await RunPanelMemberViaCli(member, memberMessages, workCt);
|
||
}
|
||
else
|
||
{
|
||
// Anthropic-API-Member: read-only (PlanMode) + NICHT persistieren (Caller persistiert einmal).
|
||
var memberUsage = new TokenUsage();
|
||
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 },
|
||
member.Effort);
|
||
text = t;
|
||
}
|
||
}
|
||
catch (OperationCanceledException) { throw; }
|
||
catch (Exception ex) { text = $"[Fehler bei {name}: {ex.Message}]"; }
|
||
|
||
text = (text ?? "").Trim();
|
||
|
||
// Proposal als EINE Message persistieren: Tag="panel" → aus Worker-History gefiltert,
|
||
// [Panel: Name]-Prefix → Label beim History-Rendering. Iteration=index+1 → stabile Reihenfolge.
|
||
if (!string.IsNullOrEmpty(text))
|
||
{
|
||
await _db.SaveMessage(new MessageRecord
|
||
{
|
||
ChainId = chainId, Iteration = index + 1,
|
||
Role = "assistant", Type = "text",
|
||
Content = $"[Panel: {name}]\n{text}", Tag = "panel"
|
||
});
|
||
}
|
||
|
||
await SseHelper.TryWriteSse(response, "panel_member_done",
|
||
new { index, name, model = member.ModelId, text }, sseCt);
|
||
|
||
return new PanelMemberResult { Index = index, Name = name, Model = member.ModelId, Text = text };
|
||
}
|
||
|
||
/// <summary>
|
||
/// Führt einen Panel-Member über die Claude CLI aus (read-only via Plan-Modus).
|
||
/// Sammelt nur den Text, persistiert nichts und sendet keine Text-Deltas
|
||
/// (parallele Member → Deltas würden interleaven; der volle Text kommt via panel_member_done).
|
||
/// </summary>
|
||
private async Task<string> RunPanelMemberViaCli(SubAgentContext member, List<object> memberMessages, CancellationToken workCt)
|
||
{
|
||
if (_cliService == null || !_cliService.IsAvailable) return "";
|
||
|
||
var systemText = ExtractSystemPromptText(member.SystemPrompt);
|
||
var prompt = ExtractLastUserText(memberMessages) ?? "";
|
||
var historyContext = BuildHistoryContext(memberMessages);
|
||
if (!string.IsNullOrEmpty(historyContext))
|
||
prompt = $"<conversation_history>\n{historyContext}\n</conversation_history>\n\n<current_request>\n{prompt}\n</current_request>";
|
||
|
||
var fullText = new StringBuilder();
|
||
await _cliService.ChatStreamingAsync(
|
||
prompt,
|
||
evt =>
|
||
{
|
||
if (evt.Type == CliEventType.TextDelta)
|
||
fullText.Append(evt.Text);
|
||
return Task.CompletedTask;
|
||
},
|
||
modelId: member.ModelId,
|
||
systemPrompt: systemText,
|
||
workspace: member.Workspace,
|
||
hasWorkspace: !string.IsNullOrEmpty(member.Workspace),
|
||
planMode: true,
|
||
effort: member.Effort,
|
||
cancellationToken: workCt
|
||
);
|
||
return fullText.ToString();
|
||
}
|
||
|
||
/// <summary>Extrahiert den System-Prompt-Text aus den List<object>-Blöcken eines SubAgentContext.</summary>
|
||
private static string ExtractSystemPromptText(List<object> systemPrompt)
|
||
{
|
||
if (systemPrompt == null) return "";
|
||
var parts = new List<string>();
|
||
foreach (var block in systemPrompt)
|
||
{
|
||
var json = JsonSerializer.Serialize(block);
|
||
var el = JsonSerializer.Deserialize<JsonElement>(json);
|
||
if (el.TryGetProperty("text", out var t))
|
||
parts.Add(t.GetString() ?? "");
|
||
}
|
||
return string.Join("\n\n", parts);
|
||
}
|
||
|
||
/// <summary>
|
||
/// 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.
|
||
/// </summary>
|
||
private static void MergePanelIntoLastUserText(List<object> messages, string panelBlock)
|
||
{
|
||
for (int i = messages.Count - 1; i >= 0; i--)
|
||
{
|
||
if (messages[i] is Dictionary<string, object> d &&
|
||
d.TryGetValue("role", out var r) && (r as string) == "user" &&
|
||
d.TryGetValue("content", out var c) && c is List<object> contentList)
|
||
{
|
||
for (int j = 0; j < contentList.Count; j++)
|
||
{
|
||
var json = JsonSerializer.Serialize(contentList[j]);
|
||
var el = JsonSerializer.Deserialize<JsonElement>(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;
|
||
}
|
||
}
|
||
}
|
||
}
|