refactor: AnthropicService partial-class Aufteilung + ToolPayloadOptions + Non-Streaming Deduplizierung

- AnthropicService.cs (337 Zeilen): Felder, Konstruktor, Setters, Utilities
- AnthropicService.Streaming.cs (750 Zeilen): StreamChat, ProcessStream, SSE-Parsing
- AnthropicService.Agents.cs (680 Zeilen): RunAgentChat, ChatNonStreaming, Review/Preparer/Finalizer
- AnthropicService.Cli.cs (530 Zeilen): StreamChatViaCli, MCP-Config, CLI-Helpers
- AddToolsToPayload: 12 Bool-Parameter → ToolPayloadOptions Record
- RunAgentChat + ChatNonStreaming → gemeinsame RunNonStreamingToolLoop-Methode

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
MelekiandClaude Opus 4.6 committed 2026-03-18 14:54:39 +01:00
1 parent 92ecaf9dac
commit 1c622abb37
5 files changed
+2025 -2116

No files matched your search

File renamed without changes.
+680
View File
@@ -0,0 +1,680 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Net.Http;
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
{
// ── Preparer-Agent Logik ──
public class PreparerResult
{
public bool ShouldForward { get; set; }
public string ForwardedPrompt { get; set; }
public string ResponseText { get; set; }
public int IterationCount { get; set; }
}
/// <summary>
/// Führt den Preparer-Agent aus. Der Preparer bekommt die User-Nachricht und die bisherige Konversation.
/// Antwortet er mit [FORWARD] → der Text nach [FORWARD] wird an den Session-Agent weitergegeben.
/// Sonst → seine Antwort wird als Rückfrage an den User gestreamt.
/// </summary>
public async Task<PreparerResult> RunPreparer(
HttpResponse response,
SubAgentContext preparer,
List<object> historyMessages,
string chainId,
string workspace,
string sessionId,
string apiToken,
CancellationToken cancellationToken)
{
// Preparer läuft VOR dem Worker und nutzt den Request-CancellationToken direkt.
// Bei Client-Disconnect wird der Preparer abgebrochen (kein Sinn weiterzumachen,
// da der Worker noch nicht gestartet hat).
cancellationToken.ThrowIfCancellationRequested();
await SseHelper.TryWriteSse(response, "preparer_start", new
{
agentId = preparer.AgentId,
agentName = preparer.Agent?.Name ?? "Vorbereiter-Agent"
}, cancellationToken);
// Preparer bekommt die History als Kontext
var preparerMessages = new List<object>(historyMessages);
var usage = new TokenUsage();
string text;
int iterations;
if (!string.IsNullOrEmpty(preparer.Provider) && _openAiService != null)
{
// OpenAI-kompatibler Preparer
var subCtx = BuildSubAgentChatContext(preparer, chainId, sessionId, apiToken);
text = await _openAiService.ChatNonStreaming(subCtx, preparerMessages);
iterations = 1;
}
else
{
(text, iterations) = await RunAgentChat(
response, preparer.ModelId, preparer.SystemPrompt, preparer.AgentId,
preparer.Workspace, preparer.UseMemory, preparer.WebSearchVersion,
preparer.ToolChoice, preparerMessages, chainId, sessionId, apiToken, usage,
0, "preparer", preparer.Agent?.Name ?? "Vorbereiter-Agent",
cancellationToken, cancellationToken);
}
var result = new PreparerResult { IterationCount = iterations, ResponseText = text };
// Logik: Standardmäßig wird IMMER weitergeleitet.
// Nur wenn der Preparer explizit [QUESTION] nutzt, wird gestoppt (= Rückfrage an User).
// [FORWARD] wird weiterhin erkannt und entfernt, ist aber nicht mehr zwingend nötig.
if (!string.IsNullOrEmpty(text) && text.Contains("[QUESTION]"))
{
// Explizite Rückfrage an User
var questionText = text.Replace("[QUESTION]", "").Trim();
result.ShouldForward = false;
result.ResponseText = questionText;
await SseHelper.TryWriteSse(response, "preparer_done", new
{
agentName = preparer.Agent?.Name ?? "Vorbereiter-Agent",
forwarded = false,
responseText = questionText
}, cancellationToken);
}
else
{
// Weiterleitung (Standard): [FORWARD]-Tag entfernen falls vorhanden
var forwardedText = text ?? "";
if (forwardedText.Contains("[FORWARD]"))
{
var idx = forwardedText.IndexOf("[FORWARD]");
forwardedText = forwardedText.Substring(idx + "[FORWARD]".Length).Trim();
}
result.ShouldForward = true;
result.ForwardedPrompt = forwardedText;
await SseHelper.TryWriteSse(response, "preparer_done", new
{
agentName = preparer.Agent?.Name ?? "Vorbereiter-Agent",
forwarded = true
}, cancellationToken);
}
return result;
}
// ── Review-Agent Logik ──
private class ReviewResult
{
public bool HasFeedback { get; set; }
public string Feedback { get; set; }
public int ReviewIterationCount { get; set; }
}
/// <summary>
/// Führt den Review-Agent aus. Der Review-Agent bekommt die Text-Konversation
/// und kann Tools nutzen (z.B. read_file um die Arbeit zu prüfen).
/// Default = Approved. Nur bei explizitem [FEEDBACK]-Tag wird Feedback an den Worker zurückgegeben.
/// reviewRound = aktuelle Runde (0-basiert), wird vom Caller hochgezählt.
/// </summary>
private async Task<ReviewResult> RunReview(
HttpResponse response,
SubAgentContext review,
List<object> workerMessages,
string chainId,
string workspace,
string sessionId,
string apiToken,
TokenUsage usage,
int baseIterations,
int reviewRound,
CancellationToken workCt,
CancellationToken sseCt)
{
await SseHelper.TryWriteSse(response, "review_start", new
{
agentId = review.AgentId,
agentName = review.Agent?.Name ?? "Review-Agent",
round = reviewRound
}, sseCt);
// Review-Agent bekommt die Worker-Konversation als Kontext
var reviewMessages = BuildReviewMessages(workerMessages);
string text;
int iterations;
if (!string.IsNullOrEmpty(review.Provider) && _openAiService != null)
{
// OpenAI-kompatibler Reviewer
var subCtx = BuildSubAgentChatContext(review, chainId, sessionId, apiToken);
text = await _openAiService.ChatNonStreaming(subCtx, reviewMessages);
iterations = 1;
}
else
{
(text, iterations) = await RunAgentChat(
response, review.ModelId, review.SystemPrompt, review.AgentId,
review.Workspace, review.UseMemory, review.WebSearchVersion,
review.ToolChoice, reviewMessages, chainId, sessionId, apiToken, usage,
baseIterations, "review", review.Agent?.Name ?? "Review-Agent",
workCt, sseCt);
}
var result = new ReviewResult { ReviewIterationCount = iterations };
if (!string.IsNullOrEmpty(text) && text.Contains("[FEEDBACK]"))
{
var feedbackText = text.Substring(text.IndexOf("[FEEDBACK]") + "[FEEDBACK]".Length).Trim();
result.HasFeedback = true;
result.Feedback = $"[Review-Feedback, Runde {reviewRound + 1}]: {feedbackText}";
}
await SseHelper.TryWriteSse(response, "review_done", new
{
agentName = review.Agent?.Name ?? "Review-Agent",
round = reviewRound,
hasFeedback = result.HasFeedback
}, sseCt);
return result;
}
/// <summary>
/// Führt den Finalizer-Agent aus. Wird einmal nach der Review-Phase (oder direkt nach dem Worker) aufgerufen.
/// Der Finalizer kann Tools nutzen, gibt aber kein Feedback an den Worker zurück.
/// </summary>
private async Task<int> RunFinalizer(
HttpResponse response,
SubAgentContext finalizer,
List<object> workerMessages,
string chainId,
string workspace,
string sessionId,
string apiToken,
TokenUsage usage,
int baseIterations,
CancellationToken workCt,
CancellationToken sseCt)
{
await SseHelper.TryWriteSse(response, "finalizer_start", new
{
agentId = finalizer.AgentId,
agentName = finalizer.Agent?.Name ?? "Finalizer-Agent"
}, sseCt);
// Finalizer bekommt die Konversation als Kontext (gleiche Methode wie Review)
var finalizerMessages = BuildReviewMessages(workerMessages);
string text;
int iterations;
if (!string.IsNullOrEmpty(finalizer.Provider) && _openAiService != null)
{
var subCtx = BuildSubAgentChatContext(finalizer, chainId, sessionId, apiToken);
text = await _openAiService.ChatNonStreaming(subCtx, finalizerMessages);
iterations = 1;
}
else
{
(text, iterations) = await RunAgentChat(
response, finalizer.ModelId, finalizer.SystemPrompt, finalizer.AgentId,
finalizer.Workspace, finalizer.UseMemory, finalizer.WebSearchVersion,
finalizer.ToolChoice, finalizerMessages, chainId, sessionId, apiToken, usage,
baseIterations, "finalizer", finalizer.Agent?.Name ?? "Finalizer-Agent",
workCt, sseCt);
}
await SseHelper.TryWriteSse(response, "finalizer_done", new
{
agentName = finalizer.Agent?.Name ?? "Finalizer-Agent"
}, sseCt);
return iterations;
}
/// <summary>
/// Baut die Messages für den Review-Agent auf.
/// Der Review-Agent bekommt die komplette Worker-Konversation als eine User-Message.
/// </summary>
private List<object> BuildReviewMessages(List<object> workerMessages)
{
// Nur die Text-Konversation extrahieren – Tool-Calls und Tool-Results weglassen.
// Der Review/Finalizer-Agent hat eigene Tools und kann bei Bedarf selbst Dateien lesen.
// Aufeinanderfolgende Messages gleicher Rolle werden zusammengeführt,
// da die Anthropic API abwechselnde user/assistant-Rollen erfordert.
var messages = new List<object>();
string lastRole = null;
var currentParts = new List<string>();
foreach (var msg in workerMessages)
{
var json = JsonSerializer.Serialize(msg);
var el = JsonSerializer.Deserialize<JsonElement>(json);
if (!el.TryGetProperty("role", out var role)) continue;
var roleStr = role.GetString() ?? "";
if (!el.TryGetProperty("content", out var content)) continue;
string textContent = null;
if (content.ValueKind == JsonValueKind.String)
{
textContent = content.GetString();
}
else if (content.ValueKind == JsonValueKind.Array)
{
var parts = new List<string>();
foreach (var block in content.EnumerateArray())
{
var blockType = block.TryGetProperty("type", out var bt) ? bt.GetString() ?? "" : "";
if (blockType == "text" && block.TryGetProperty("text", out var txt))
parts.Add(txt.GetString() ?? "");
}
if (parts.Count > 0)
textContent = string.Join("\n", parts);
}
if (string.IsNullOrEmpty(textContent)) continue;
if (roleStr == lastRole)
{
// Gleiche Rolle → Text anhängen
currentParts.Add(textContent);
}
else
{
// Rolle wechselt → vorherige Message abschließen
if (lastRole != null && currentParts.Count > 0)
{
messages.Add(new
{
role = lastRole,
content = string.Join("\n\n", currentParts)
});
}
lastRole = roleStr;
currentParts = new List<string> { textContent };
}
}
// Letzte Message abschließen
if (lastRole != null && currentParts.Count > 0)
{
messages.Add(new
{
role = lastRole,
content = string.Join("\n\n", currentParts)
});
}
// Sicherstellen, dass die Messages mit "user" beginnen
if (messages.Count > 0)
{
var firstJson = JsonSerializer.Serialize(messages[0]);
var firstEl = JsonSerializer.Deserialize<JsonElement>(firstJson);
if (firstEl.GetProperty("role").GetString() != "user")
{
messages.Insert(0, new { role = "user", content = "Bitte überprüfe die folgende Konversation und die Arbeit des Agents." });
}
}
else
{
messages.Add(new { role = "user", content = "Es gibt keine Konversation zum Überprüfen." });
}
return messages;
}
/// <summary>
/// Baut einen ChatContext für Sub-Agents (Preparer/Review/Finalizer) auf,
/// wenn diese einen OpenAI-kompatiblen Provider nutzen.
/// </summary>
private ChatContext BuildSubAgentChatContext(SubAgentContext sub, string chainId, string sessionId, string apiToken)
{
// SystemPrompt als Text extrahieren (SubAgent hat List<object> mit { type, text } Blöcken)
var systemText = "";
if (sub.SystemPrompt != null)
{
var parts = new List<string>();
foreach (var block in sub.SystemPrompt)
{
var json = JsonSerializer.Serialize(block);
var el = JsonSerializer.Deserialize<JsonElement>(json);
if (el.TryGetProperty("text", out var t))
parts.Add(t.GetString() ?? "");
}
systemText = string.Join("\n\n", parts);
}
return new ChatContext
{
ChainId = chainId,
SessionId = sessionId,
ModelId = sub.ModelId,
Provider = sub.Provider,
Workspace = sub.Workspace,
AgentId = sub.AgentId,
SystemPromptText = systemText,
UseMemory = sub.UseMemory,
WebSearchVersion = sub.WebSearchVersion,
MaxIterations = 20,
ApiToken = apiToken
};
}
// ── Shared Non-Streaming Tool Loop ──
/// <summary>
/// Optionale Callbacks für RunNonStreamingToolLoop.
/// </summary>
private class NonStreamingLoopCallbacks
{
/// <summary>Wird zu Beginn jeder Iteration aufgerufen (z.B. Typing-Indicator).</summary>
public Func<int, Task> OnIteration { get; set; }
/// <summary>SSE-Response für Client-Events. Null = keine SSE-Events.</summary>
public HttpResponse SseResponse { get; set; }
public CancellationToken SseCt { get; set; }
/// <summary>SSE-Event-Prefix (z.B. "review" → "review_text", "review_tool").</summary>
public string SsePrefix { get; set; }
public string AgentName { get; set; }
/// <summary>Tag für DB-Records (z.B. "review", "finalizer").</summary>
public string Tag { get; set; }
/// <summary>Ob Text-Embeddings in die Queue geschrieben werden.</summary>
public bool EnqueueEmbeddings { get; set; }
/// <summary>
/// Optionaler Tool-Executor. Parameter: (toolName, input, ct) → result.
/// Bei Fehler Exception werfen. Wenn null, wird Standard-Executor verwendet.
/// </summary>
public Func<string, JsonElement, CancellationToken, Task<string>> ExecuteTool { get; set; }
}
/// <summary>
/// Gemeinsamer Non-Streaming Tool-Loop für RunAgentChat und ChatNonStreaming.
/// Baut Payload, sendet API-Request, verarbeitet Content-Blöcke, führt Tools aus.
/// systemPrompt kann List&lt;object&gt; (Agent) oder string (ChatContext) sein.
/// </summary>
private async Task<(string text, int iterations)> RunNonStreamingToolLoop(
string modelId,
object systemPrompt,
ToolPayloadOptions toolOpts,
List<object> messages,
string chainId,
string sessionId,
string apiToken,
TokenUsage usage,
int baseIteration,
int maxIterations,
CancellationToken ct,
NonStreamingLoopCallbacks cb = null)
{
cb ??= new NonStreamingLoopCallbacks();
int iterations = 0;
var finalText = new StringBuilder();
while (iterations < maxIterations)
{
ct.ThrowIfCancellationRequested();
iterations++;
if (cb.OnIteration != null) await cb.OnIteration(iterations);
var payload = new Dictionary<string, object>
{
["model"] = modelId,
["max_tokens"] = 16384,
["messages"] = messages
};
if (systemPrompt is List<object> spBlocks && spBlocks.Count > 0)
payload["system"] = spBlocks;
else if (systemPrompt is string spText && !string.IsNullOrEmpty(spText))
payload["system"] = spText;
AddToolsToPayload(payload, toolOpts with { Iteration = iterations });
var client = CreateClient(apiToken);
var json = JsonSerializer.Serialize(payload);
LogPayload(json, sessionId, chainId, baseIteration + iterations);
var httpContent = new StringContent(json, Encoding.UTF8, "application/json");
using var httpResp = await client.SendAsync(
new HttpRequestMessage(HttpMethod.Post, "https://api.anthropic.com/v1/messages") { Content = httpContent },
HttpCompletionOption.ResponseHeadersRead,
ct
);
TryUpdateKeyMeta(httpResp, apiToken);
if (!httpResp.IsSuccessStatusCode)
{
var errBody = await httpResp.Content.ReadAsStringAsync(ct);
var errMsg = $"API {(int)httpResp.StatusCode}: {errBody}";
await _db.SaveMessage(new MessageRecord
{
ChainId = chainId, Iteration = baseIteration + iterations,
Role = "system", Type = "error",
Content = errMsg, IsError = true, Tag = cb.Tag
});
if (cb.SseResponse != null)
await SseHelper.TryWriteSse(cb.SseResponse, $"{cb.SsePrefix}_error", new { error = errMsg }, cb.SseCt);
break;
}
var respJson = await httpResp.Content.ReadAsStringAsync(ct);
var respDoc = JsonSerializer.Deserialize<JsonElement>(respJson);
if (respDoc.TryGetProperty("usage", out var u))
TrackUsage(usage, u);
var stopReason = respDoc.TryGetProperty("stop_reason", out var sr) ? sr.GetString() ?? "" : "";
var contentArr = respDoc.TryGetProperty("content", out var c) ? c : default;
if (contentArr.ValueKind != JsonValueKind.Array) break;
var assistantContent = new List<object>();
var toolResults = new List<object>();
foreach (var block in contentArr.EnumerateArray())
{
var blockType = block.TryGetProperty("type", out var bt) ? bt.GetString() ?? "" : "";
if (blockType == "text")
{
var text = block.GetProperty("text").GetString() ?? "";
finalText.Append(text);
assistantContent.Add(new { type = "text", text });
var msgId = await _db.SaveMessage(new MessageRecord
{
ChainId = chainId, Iteration = baseIteration + iterations,
Role = "assistant", Type = "text",
Content = text, Tag = cb.Tag
});
if (cb.EnqueueEmbeddings)
_embeddingBg?.Enqueue(msgId, chainId, sessionId, toolOpts.AgentId, "assistant", text);
if (cb.SseResponse != null)
await SseHelper.TryWriteSse(cb.SseResponse, $"{cb.SsePrefix}_text", new { text, agentName = cb.AgentName }, cb.SseCt);
}
else if (blockType == "tool_use")
{
var toolId = block.GetProperty("id").GetString() ?? "";
var toolName = block.GetProperty("name").GetString() ?? "";
JsonElement inputEl;
try { inputEl = block.GetProperty("input"); }
catch { inputEl = JsonSerializer.Deserialize<JsonElement>("{}"); }
assistantContent.Add(new { type = "tool_use", id = toolId, name = toolName, input = inputEl });
await _db.SaveMessage(new MessageRecord
{
ChainId = chainId, Iteration = baseIteration + iterations,
Role = "assistant", Type = "tool_use",
ToolName = toolName, ToolCallId = toolId,
Content = inputEl.ToString(), Tag = cb.Tag
});
ct.ThrowIfCancellationRequested();
string toolResult;
bool isError = false;
try
{
toolResult = cb.ExecuteTool != null
? await cb.ExecuteTool(toolName, inputEl, ct)
: await _toolService.ExecuteTool(toolName, inputEl, toolOpts.Workspace, _memoryService, toolOpts.AgentId, ct, vectorSearch: _vectorSearch, embeddingBg: _embeddingBg);
}
catch (OperationCanceledException) { throw; }
catch (Exception ex) { toolResult = $"Fehler: {ex.Message}"; isError = true; }
ct.ThrowIfCancellationRequested();
toolResults.Add(new
{
type = "tool_result",
tool_use_id = toolId,
content = TruncateToolResult(toolResult),
is_error = isError
});
await _db.SaveMessage(new MessageRecord
{
ChainId = chainId, Iteration = baseIteration + iterations,
Role = "user", Type = "tool_result",
ToolCallId = toolId,
Content = toolResult, IsError = isError, Tag = cb.Tag
});
if (cb.SseResponse != null)
await SseHelper.TryWriteSse(cb.SseResponse, $"{cb.SsePrefix}_tool", new { name = toolName, result = TruncateToolResult(toolResult), isError, agentName = cb.AgentName }, cb.SseCt);
}
}
messages.Add(new { role = "assistant", content = assistantContent });
if (stopReason != "tool_use" || toolResults.Count == 0)
break;
messages.Add(new { role = "user", content = toolResults });
}
return (finalText.ToString(), iterations);
}
/// <summary>
/// Agent-Chat (non-streaming API, SSE-Events für Client).
/// Wrapper um RunNonStreamingToolLoop für Review-/Finalizer-Agents.
/// </summary>
private Task<(string text, int iterations)> RunAgentChat(
HttpResponse response,
string modelId,
List<object> systemPrompt,
string agentId,
string agentWorkspace,
bool useMemory,
string webSearchVersion,
string agentToolChoice,
List<object> messages,
string chainId,
string sessionId,
string apiToken,
TokenUsage usage,
int baseIterations,
string agentType,
string agentName,
CancellationToken workCt,
CancellationToken sseCt)
{
return RunNonStreamingToolLoop(
modelId, systemPrompt,
new ToolPayloadOptions
{
Workspace = agentWorkspace, AgentId = agentId,
UseMemory = useMemory, WebSearchVersion = webSearchVersion,
ToolChoice = agentToolChoice
},
messages, chainId, sessionId, apiToken, usage,
baseIterations, 20, workCt,
new NonStreamingLoopCallbacks
{
SseResponse = response, SseCt = sseCt,
SsePrefix = agentType, AgentName = agentName,
Tag = agentType
});
}
/// <summary>
/// Non-Streaming Chat für Telegram. Gibt den finalen Antworttext zurück.
/// Wrapper um RunNonStreamingToolLoop mit Chain-Management und Embedding-Enqueue.
/// onIteration wird nach jeder Iteration aufgerufen (z.B. für Typing-Indicator).
/// </summary>
public async Task<string> ChatNonStreaming(
ChatContext ctx,
List<object> messages,
Func<int, Task> onIteration = null)
{
var sw = System.Diagnostics.Stopwatch.StartNew();
var usage = new TokenUsage();
var chain = new ChainRecord
{
ChainId = ctx.ChainId,
SessionId = ctx.SessionId,
Model = ctx.ModelId
};
try
{
var (text, iterations) = await RunNonStreamingToolLoop(
ctx.ModelId, ctx.SystemPromptText,
new ToolPayloadOptions
{
Workspace = ctx.Workspace, AgentId = ctx.AgentId,
UseMemory = ctx.UseMemory, WebSearchVersion = ctx.WebSearchVersion,
ToolChoice = "auto"
},
messages, ctx.ChainId, ctx.SessionId, ctx.ApiToken, usage,
0, ctx.MaxIterations, CancellationToken.None,
new NonStreamingLoopCallbacks
{
OnIteration = onIteration,
EnqueueEmbeddings = true,
ExecuteTool = (name, input, ct) =>
_toolService.ExecuteTool(name, input, ctx.Workspace, _memoryService, ctx.AgentId, ct,
vectorSearch: _vectorSearch, embeddingBg: _embeddingBg, db: _db,
chatService: _chatService, anthropicService: this, openAiService: _openAiService)
});
var finalStopReason = text.Length > 0 ? "end_turn" : "error";
await FinalizeChain(chain, usage, iterations, finalStopReason, sw);
return text;
}
catch (Exception ex)
{
await _db.SaveMessage(new MessageRecord
{
ChainId = ctx.ChainId, Iteration = 0,
Role = "system", Type = "error",
Content = $"Fehler: {ex.Message}", IsError = true
});
await FinalizeChain(chain, usage, 0, "error", sw);
return "";
}
}
}
+530
View File
@@ -0,0 +1,530 @@
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
using Gott.Endpoints;
using Gott.Models;
using Microsoft.AspNetCore.Http;
namespace Gott.Services;
public partial class AnthropicService
{
/// <summary>
/// Streaming-Chat über die Claude CLI.
/// Die CLI verwendet das lokale Token aus ~/.claude/.credentials.json.
/// </summary>
public async Task StreamChatViaCli(
HttpResponse response,
string modelId,
List<object> systemPrompt,
List<object> messages,
string chainId,
string workspace = null,
bool allowUserChoice = false,
bool planMode = false,
bool webSearch = false,
string sessionId = null,
string agentId = null,
RemoteConnectionManager remoteMgr = null,
CancellationToken workCt = default,
CancellationToken sseCt = default)
{
if (_cliService == null || !_cliService.IsAvailable)
{
await SseHelper.TryWriteSse(response, "error", new { error = "Claude CLI nicht verfügbar" }, sseCt);
await SseHelper.TryWriteSse(response, "done", new { chainId }, sseCt);
return;
}
var sw = System.Diagnostics.Stopwatch.StartNew();
var usage = new TokenUsage();
var fullText = new StringBuilder();
var tempImagePaths = new List<string>();
string mcpConfigPath = null;
bool isUserChoice = false;
var chain = new ChainRecord
{
ChainId = chainId,
SessionId = sessionId,
Model = modelId
};
try
{
// MCP-Config generieren (immer – für ask_user_choice, optional Memory-Tools + remote_execute)
var hasRemote = remoteMgr != null && !string.IsNullOrEmpty(sessionId) && remoteMgr.GetBySessionId(sessionId) != null;
mcpConfigPath = GenerateMcpConfig(sessionId, agentId, hasRemote);
// System-Prompt als Text extrahieren
var systemText = new StringBuilder();
if (systemPrompt != null)
{
foreach (var block in systemPrompt)
{
var json = JsonSerializer.Serialize(block);
var el = JsonSerializer.Deserialize<JsonElement>(json);
if (el.TryGetProperty("text", out var t))
systemText.AppendLine(t.GetString());
}
}
// Letzte User-Message als Prompt extrahieren
var prompt = ExtractLastUserText(messages) ?? "";
// Bild-Attachments aus letzter User-Message als Temp-Dateien speichern
// (CLI kann Bilder nicht inline empfangen, aber per Read-Tool lesen)
var imageRefs = ExtractLastUserImageRefs(messages, tempImagePaths);
if (!string.IsNullOrEmpty(imageRefs))
{
prompt = imageRefs + "\n" + prompt;
}
// Konversations-History als Kontext zum Prompt hinzufügen
var historyContext = BuildHistoryContext(messages);
if (!string.IsNullOrEmpty(historyContext))
{
prompt = $"<conversation_history>\n{historyContext}\n</conversation_history>\n\n<current_request>\n{prompt}\n</current_request>";
}
string choiceToolCallId = null;
var result = await _cliService.ChatStreamingAsync(
prompt,
async evt =>
{
switch (evt.Type)
{
case CliEventType.TextStart:
await SseHelper.TryWriteSse(response, "text_start", new { index = 0 }, sseCt);
break;
case CliEventType.TextDelta:
fullText.Append(evt.Text);
await SseHelper.TryWriteSse(response, "text_delta", new { text = evt.Text }, sseCt);
break;
case CliEventType.TextEnd:
await SseHelper.TryWriteSse(response, "text_end", new { }, sseCt);
break;
case CliEventType.ToolStart:
{
// Text-Puffer flushen bevor Tool-Use gespeichert wird (korrekte Reihenfolge in DB)
if (fullText.Length > 0)
{
await _db.SaveMessage(new MessageRecord
{
ChainId = chainId,
Iteration = 1,
Role = "assistant",
Type = "text",
Content = fullText.ToString()
});
fullText.Clear();
}
var mappedName = MapCliToolName(evt.ToolName);
await SseHelper.TryWriteSse(response, "tool_start", new { name = mappedName, id = evt.ToolId }, sseCt);
break;
}
case CliEventType.ToolEnd:
{
var mappedName = MapCliToolName(evt.ToolName);
var mappedInput = MapCliToolInput(evt.ToolName, evt.ToolInput);
// Tool-Use in DB speichern (damit Poll die Tools anzeigt)
await _db.SaveMessage(new MessageRecord
{
ChainId = chainId,
Iteration = 1,
Role = "assistant",
Type = "tool_use",
ToolName = mappedName,
ToolCallId = evt.ToolId,
Content = mappedInput
});
await SseHelper.TryWriteSse(response, "tool_end", new { name = mappedName, id = evt.ToolId, input = mappedInput }, sseCt);
// ask_user_choice via MCP: TCS erstellen + SSE user_choice Event
if (mappedName == "ask_user_choice")
{
try
{
var choiceInput = JsonSerializer.Deserialize<JsonElement>(mappedInput);
var cQuestion = choiceInput.TryGetProperty("question", out var cq) ? cq.GetString() ?? "" : "";
var cChoices = new List<string>();
if (choiceInput.TryGetProperty("choices", out var cc) && cc.ValueKind == JsonValueKind.Array)
foreach (var cOpt in cc.EnumerateArray())
cChoices.Add(cOpt.GetString() ?? "");
// Alte Antwort entfernen (falls vorhanden)
InternalEndpoint.ChoiceAnswers.TryRemove(sessionId, out _);
// SSE Event mit viaCli-Flag → Frontend zeigt Buttons + nutzt choice-respond Endpoint
choiceToolCallId = evt.ToolId;
await SseHelper.TryWriteSse(response, "user_choice", new
{
toolCallId = evt.ToolId,
question = cQuestion,
choices = cChoices,
viaCli = true
}, sseCt);
isUserChoice = true;
}
catch (JsonException) { /* Input nicht parsbar – MCP-Server gibt Fehler */ }
}
break;
}
case CliEventType.ToolResult:
{
// Tool-Result in DB speichern (damit es beim Neuladen sichtbar ist)
var resultContent = SseHelper.TruncateToolResult(evt.ToolResultContent ?? "");
await _db.SaveMessage(new MessageRecord
{
ChainId = chainId,
Iteration = 1,
Role = "user",
Type = "tool_result",
ToolCallId = evt.ToolId,
Content = resultContent,
IsError = evt.ToolResultIsError
});
// SSE Event an Client senden
await SseHelper.TryWriteSse(response, "tool_result", new
{
id = evt.ToolId,
result = resultContent,
isError = evt.ToolResultIsError
}, sseCt);
break;
}
}
},
modelId: modelId,
systemPrompt: systemText.Length > 0 ? systemText.ToString() : null,
workspace: workspace,
hasWorkspace: !string.IsNullOrEmpty(workspace),
planMode: planMode,
webSearch: webSearch,
mcpConfigPath: mcpConfigPath,
cancellationToken: workCt
);
// Bei CLI-Fehler → Fehlermeldung an Client senden und in DB speichern
if (!result.Success && !isUserChoice)
{
var errorMsg = !string.IsNullOrEmpty(result.Error) ? result.Error : "CLI hat keine Antwort geliefert";
// Error in DB speichern (damit er beim Neuladen sichtbar ist)
await _db.SaveMessage(new MessageRecord
{
ChainId = chainId,
Iteration = 1,
Role = "system",
Type = "error",
Content = $"CLI-Fehler: {errorMsg}"
});
await SseHelper.TryWriteSse(response, "error", new { error = $"CLI-Fehler: {errorMsg}" }, sseCt);
}
// Antwort in DB speichern
if (fullText.Length > 0)
{
await _db.SaveMessage(new MessageRecord
{
ChainId = chainId,
Iteration = 1,
Role = "assistant",
Type = "text",
Content = fullText.ToString()
});
}
// Token-Usage vom CLI-Result übernehmen
usage.InputTokens = result.InputTokens;
usage.OutputTokens = result.OutputTokens;
usage.CacheReadTokens = result.CacheReadTokens;
usage.CacheCreationTokens = result.CacheCreationTokens;
var stopReason = isUserChoice ? "awaiting_user_choice" : (result.Success ? "end_turn" : "error");
await FinalizeChain(chain, usage, result.NumTurns > 0 ? result.NumTurns : 1, stopReason, sw);
await SseHelper.TryWriteSse(response, "done", new
{
model = modelId,
inputTokens = result.InputTokens,
outputTokens = result.OutputTokens,
cacheCreationTokens = result.CacheCreationTokens,
cacheReadTokens = result.CacheReadTokens,
costUsd = result.CostUsd,
iterations = result.NumTurns > 0 ? result.NumTurns : 1,
chainId,
durationMs = sw.ElapsedMilliseconds,
stopReason,
viaCli = true
}, sseCt);
}
catch (OperationCanceledException) when (workCt.IsCancellationRequested)
{
// Stop-Button gedrückt → CLI-Prozess wurde gestoppt
// Wenn ask_user_choice aktiv war: als awaiting_user_choice markieren,
// damit die nächste Chain den Kontext (Frage + Optionen) behält.
var stoppedReason = isUserChoice ? "awaiting_user_choice" : "stopped";
await FinalizeChain(chain, usage, 1, stoppedReason, sw);
await SseHelper.TryWriteSse(response, "done", new
{
model = modelId,
chainId,
durationMs = sw.ElapsedMilliseconds,
stopReason = stoppedReason,
viaCli = true
}, CancellationToken.None);
}
catch (OperationCanceledException)
{
// Client-Disconnect (sseCt gecancelled, aber NICHT workCt)
// CLI-Prozess läuft weiter im Hintergrund – hier nur Chain finalisieren
// (sollte nicht mehr passieren, da CLI jetzt workCt nutzt)
await FinalizeChain(chain, usage, 1, "client_disconnected", sw);
}
catch (Exception ex)
{
var errorMsg = $"CLI-Fehler: {ex.Message}";
await _db.SaveMessage(new MessageRecord
{
ChainId = chainId,
Iteration = 1,
Role = "system",
Type = "error",
Content = errorMsg
});
await SseHelper.TryWriteSse(response, "error", new { error = errorMsg }, sseCt);
await FinalizeChain(chain, usage, 1, "error", sw);
await SseHelper.TryWriteSse(response, "done", new
{
model = modelId,
chainId,
durationMs = sw.ElapsedMilliseconds,
stopReason = "error",
viaCli = true
}, sseCt);
}
finally
{
// Temp-Bilddateien aufräumen
CleanupTempImages(tempImagePaths);
// MCP-Config Temp-Datei aufräumen
if (mcpConfigPath != null)
{
try { File.Delete(mcpConfigPath); } catch { }
}
}
}
/// <summary>
/// Mappt CLI-Tool-Namen (Edit, Write, Bash, etc.) auf gott-Tool-Namen (edit_file, write_file, etc.).
/// </summary>
private static string MapCliToolName(string cliName)
{
return cliName switch
{
"Edit" => "edit_file",
"Write" => "write_file",
"Read" => "read_file",
"Bash" => "execute",
"Glob" => "list_files",
"Grep" => "search_files",
"WebSearch" => "web_search",
"WebFetch" => "web_fetch",
"NotebookEdit" => "notebook_edit",
"mcp__gott-remote__remote_execute" => "remote_execute",
_ => cliName?.StartsWith("mcp__") == true ? cliName.Split("__").LastOrDefault() ?? cliName : cliName
};
}
/// <summary>
/// Mappt CLI-Tool-Input-Felder auf gott-Konventionen (z.B. file_path→path, old_string→old_text).
/// </summary>
private static string MapCliToolInput(string cliToolName, string inputJson)
{
if (string.IsNullOrEmpty(inputJson)) return inputJson;
try
{
var input = JsonSerializer.Deserialize<JsonElement>(inputJson);
if (input.ValueKind != JsonValueKind.Object) return inputJson;
switch (cliToolName)
{
case "Edit":
{
// file_path → path, old_string → old_text, new_string → new_text
var mapped = new Dictionary<string, object>();
foreach (var prop in input.EnumerateObject())
{
var key = prop.Name switch
{
"file_path" => "path",
"old_string" => "old_text",
"new_string" => "new_text",
_ => prop.Name
};
mapped[key] = prop.Value;
}
return JsonSerializer.Serialize(mapped);
}
case "Write":
{
// file_path → path
var mapped = new Dictionary<string, object>();
foreach (var prop in input.EnumerateObject())
{
var key = prop.Name == "file_path" ? "path" : prop.Name;
mapped[key] = prop.Value;
}
return JsonSerializer.Serialize(mapped);
}
case "Read":
{
// file_path → path
var mapped = new Dictionary<string, object>();
foreach (var prop in input.EnumerateObject())
{
var key = prop.Name == "file_path" ? "path" : prop.Name;
mapped[key] = prop.Value;
}
return JsonSerializer.Serialize(mapped);
}
case "Bash":
{
// command bleibt gleich
return inputJson;
}
default:
return inputJson;
}
}
catch
{
return inputJson;
}
}
/// <summary>
/// Baut einen Kontext-String aus der Konversations-History.
/// </summary>
private static string BuildHistoryContext(List<object> messages)
{
var sb = new StringBuilder();
// Letzte Message überspringen wenn es ein Text-Prompt ist (wird separat als current_request gesendet).
// Wenn die letzte Message ein tool_result ist (z.B. Choice-Antwort), NICHT überspringen.
int takeCount = messages.Count;
if (messages.Count > 0)
{
var lastJson = JsonSerializer.Serialize(messages[messages.Count - 1]);
var lastEl = JsonSerializer.Deserialize<JsonElement>(lastJson);
bool lastIsToolResult = false;
if (lastEl.TryGetProperty("content", out var lc) && lc.ValueKind == JsonValueKind.Array)
{
foreach (var b in lc.EnumerateArray())
{
var bType = b.TryGetProperty("type", out var bt) ? bt.GetString() ?? "" : "";
if (bType == "tool_result") { lastIsToolResult = true; break; }
}
}
if (!lastIsToolResult)
takeCount = messages.Count - 1;
}
foreach (var msg in messages.Take(takeCount))
{
var json = JsonSerializer.Serialize(msg);
var el = JsonSerializer.Deserialize<JsonElement>(json);
if (!el.TryGetProperty("role", out var role)) continue;
var roleStr = role.GetString() ?? "";
if (el.TryGetProperty("content", out var content))
{
if (content.ValueKind == JsonValueKind.String)
{
sb.AppendLine($"[{roleStr}]: {content.GetString()}");
}
else if (content.ValueKind == JsonValueKind.Array)
{
foreach (var block in content.EnumerateArray())
{
var blockType = block.TryGetProperty("type", out var bt) ? bt.GetString() ?? "" : "";
if (blockType == "text" && block.TryGetProperty("text", out var txt))
{
sb.AppendLine($"[{roleStr}]: {txt.GetString()}");
}
else if (blockType == "tool_use" && block.TryGetProperty("name", out var tn) && tn.GetString() == "ask_user_choice")
{
if (block.TryGetProperty("input", out var inp))
{
var q = inp.TryGetProperty("question", out var qq) ? qq.GetString() ?? "" : "";
var opts = new List<string>();
if (inp.TryGetProperty("choices", out var cc) && cc.ValueKind == JsonValueKind.Array)
foreach (var o in cc.EnumerateArray()) opts.Add(o.GetString() ?? "");
sb.AppendLine($"[assistant]: Ich habe gefragt: \"{q}\" mit Optionen: {string.Join(", ", opts)}");
}
}
else if (blockType == "tool_result" && block.TryGetProperty("content", out var trContent))
{
sb.AppendLine($"[user]: Du hast gewählt: \"{trContent.GetString()}\"");
}
}
}
}
}
return sb.ToString().Trim();
}
/// <summary>
/// Generiert eine temporäre MCP-Config-Datei für die Claude CLI.
/// Enthält den MCP-Server für remote_execute.
/// </summary>
private string GenerateMcpConfig(string sessionId, string agentId = null, bool hasRemote = false)
{
// Port aus Konfiguration oder Default
var port = 19000;
var scriptPath = Path.Combine(AppContext.BaseDirectory, "scripts", "mcp-remote-execute.py");
// Fallback: Quell-Verzeichnis prüfen (für Dev-Modus)
if (!File.Exists(scriptPath))
{
var projectRoot = Path.GetDirectoryName(AppPaths.WwwRoot) ?? AppContext.BaseDirectory;
var srcPath = Path.Combine(projectRoot, "scripts", "mcp-remote-execute.py");
if (File.Exists(srcPath)) scriptPath = srcPath;
}
var env = new Dictionary<string, string>
{
["GOTT_API_URL"] = $"http://localhost:{port}",
["GOTT_SESSION_ID"] = sessionId ?? "",
["GOTT_AUTH_TOKEN"] = Endpoints.InternalEndpoint.InternalToken,
["GOTT_AGENT_ID"] = agentId ?? "",
["GOTT_HAS_REMOTE"] = hasRemote ? "true" : "false"
};
var config = new
{
mcpServers = new Dictionary<string, object>
{
["gott-remote"] = new
{
command = "python3",
args = new[] { scriptPath },
env
}
}
};
var json = System.Text.Json.JsonSerializer.Serialize(config, new System.Text.Json.JsonSerializerOptions { WriteIndented = true });
var tempPath = Path.Combine(Path.GetTempPath(), $"gott-mcp-{sessionId}.json");
File.WriteAllText(tempPath, json);
return tempPath;
}
}
+750
View File
@@ -0,0 +1,750 @@
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Net.Http;
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
{
public async Task StreamChat(
HttpResponse response,
string modelId,
List<object> systemPrompt,
List<object> messages,
string chainId,
string workspace,
string apiToken,
string sessionId = null,
string agentId = null,
string userId = null,
bool useMemory = true,
string webSearchVersion = null,
int maxIterations = 100,
string toolChoice = "any",
SubAgentContext review = null,
SubAgentContext finalizer = null,
bool allowUserChoice = true,
bool planMode = false,
bool useAutoContext = true,
bool useGitHubSearch = false,
bool useSerperSearch = false,
CancellationToken workCt = default,
CancellationToken hardCt = default,
CancellationToken sseCt = default,
RemoteConnectionManager remoteMgr = null)
{
var sw = System.Diagnostics.Stopwatch.StartNew();
var usage = new TokenUsage();
int iterations = 0;
int reviewRound = 0;
string lastStopReason = "";
int promptTrimRetries = 0;
var chain = new ChainRecord
{
ChainId = chainId,
SessionId = sessionId,
Model = modelId
};
try
{
// ── Automatische Kontext-Injektion via Embedding-Suche ──
if (useAutoContext && _vectorSearch != null && systemPrompt != null && messages.Count > 0)
{
try
{
// Letzte User-Message als Query extrahieren
var lastUserText = ExtractLastUserText(messages);
if (!string.IsNullOrWhiteSpace(lastUserText))
{
var contextResults = await _vectorSearch.SearchAsync(lastUserText, agentId, topN: 5, minScore: 0.45f);
// Nur Ergebnisse aus anderen Sessions (nicht aktuelle Chain) und nicht zu kurze
contextResults.RemoveAll(r => r.ChainId == chainId || (r.TextPreview?.Length ?? 0) < 20);
// Memory-Dateien durchsuchen
var memoryResults = await _vectorSearch.SearchMemoryFilesAsync(lastUserText, agentId, topN: 3, minScore: 0.45f);
if (contextResults.Count > 0 || memoryResults.Count > 0)
{
var ctxSb = new StringBuilder();
ctxSb.AppendLine("\n<recalled_context>");
ctxSb.AppendLine("Die folgende Information wurde automatisch abgerufen und könnte relevant sein:");
foreach (var r in contextResults)
{
var date = DateTime.TryParse(r.CreatedAt, out var dt) ? dt.ToString("yyyy-MM-dd HH:mm") : r.CreatedAt;
ctxSb.AppendLine($"- [Gespräch, {r.Role}, {date}]: {r.TextPreview}");
}
foreach (var r in memoryResults)
{
ctxSb.AppendLine($"- [Memory, {r.FilePath}]: {r.TextPreview}");
}
ctxSb.AppendLine("</recalled_context>");
systemPrompt.Add(new { type = "text", text = ctxSb.ToString() });
}
}
}
catch (Exception ex)
{
Console.WriteLine($"[AutoContext] Error: {ex.Message}");
}
}
while (iterations < maxIterations)
{
// Hard-Stop: Sofortiger Abbruch aus while-Loop (Doppel-Klick Stop)
hardCt.ThrowIfCancellationRequested();
workCt.ThrowIfCancellationRequested();
iterations++;
var payload = new Dictionary<string, object>
{
["model"] = modelId,
["max_tokens"] = 16384,
["cache_control"] = new { type = "ephemeral" },
["messages"] = messages,
["stream"] = true
};
if (systemPrompt != null && systemPrompt.Count > 0)
payload["system"] = systemPrompt;
var hasRemote = remoteMgr?.GetBySessionId(sessionId) != null;
AddToolsToPayload(payload, new ToolPayloadOptions
{
Workspace = workspace,
AgentId = agentId,
UseMemory = useMemory,
WebSearchVersion = webSearchVersion,
ToolChoice = toolChoice,
Iteration = iterations,
AllowUserChoice = allowUserChoice,
PlanMode = planMode,
IncludeRemote = hasRemote,
IncludeGitHubSearch = useGitHubSearch && _toolService.HasGitHubToken,
IncludeSerperSearch = useSerperSearch && _toolService.HasSerperKey
});
var client = CreateClient(apiToken);
var json = JsonSerializer.Serialize(payload, new JsonSerializerOptions { WriteIndented = true });
LogPayload(json, sessionId, chainId, iterations);
var httpContent = new StringContent(json, Encoding.UTF8, "application/json");
using var httpResp = await client.SendAsync(
new HttpRequestMessage(HttpMethod.Post, "https://api.anthropic.com/v1/messages") { Content = httpContent },
HttpCompletionOption.ResponseHeadersRead,
workCt
);
TryUpdateKeyMeta(httpResp, apiToken);
if (!httpResp.IsSuccessStatusCode)
{
var errBody = await httpResp.Content.ReadAsStringAsync(workCt);
var errMsg = $"API {(int)httpResp.StatusCode}: {errBody}";
// Auto-Retry bei "prompt too long" / "exceed context limit": History kürzen und erneut versuchen
if ((int)httpResp.StatusCode == 400 &&
(errBody.Contains("prompt is too long") || errBody.Contains("exceed context limit")) &&
messages.Count > 4 && promptTrimRetries < 5)
{
promptTrimRetries++;
int trimCount = Math.Max(2, messages.Count / 5); // 20% der Messages entfernen
trimCount = TrimMessagesCleanly(messages, trimCount);
if (trimCount > 0)
{
await SseHelper.TryWriteSse(response, "info", new { message = $"History zu lang – {trimCount} ältere Messages entfernt, Retry ({promptTrimRetries}/5)..." }, sseCt);
iterations--; // Iteration nicht zählen, war ja kein echter Versuch
continue; // → nächste while-Iteration mit kürzerer History
}
}
await _db.SaveMessage(new MessageRecord
{
ChainId = chainId, Iteration = iterations,
Role = "system", Type = "error",
Content = errMsg, IsError = true
});
await SseHelper.TryWriteSse(response, "error", new { error = errMsg }, sseCt);
await FinalizeChain(chain, usage, iterations, "error", sw);
await SseHelper.TryWriteSse(response, "done", new { chainId = chain.ChainId }, sseCt);
return;
}
var (stopReason, assistantContent, executedTools, streamErrorInfo) = await ProcessStream(response, httpResp, usage, workspace, agentId, workCt, sseCt, remoteMgr, sessionId, userId, apiToken);
lastStopReason = stopReason;
promptTrimRetries = 0; // Erfolgreicher API-Call → Retry-Counter resetten
messages.Add(new { role = "assistant", content = assistantContent });
// Assistant-Blöcke persistieren
foreach (var block in assistantContent)
{
var blockJson = JsonSerializer.Serialize(block);
var el = JsonSerializer.Deserialize<JsonElement>(blockJson);
var blockType = el.GetProperty("type").GetString() ?? "";
if (blockType == "text")
{
var textContent = el.GetProperty("text").GetString();
var msgId = await _db.SaveMessage(new MessageRecord
{
ChainId = chainId, Iteration = iterations,
Role = "assistant", Type = "text",
Content = textContent
});
_embeddingBg?.Enqueue(msgId, chainId, sessionId, agentId, "assistant", textContent ?? "");
}
else if (blockType == "tool_use")
{
await _db.SaveMessage(new MessageRecord
{
ChainId = chainId, Iteration = iterations,
Role = "assistant", Type = "tool_use",
ToolName = el.GetProperty("name").GetString(),
ToolCallId = el.GetProperty("id").GetString(),
Content = el.GetProperty("input").ToString()
});
}
else if (blockType == "server_tool_use")
{
await _db.SaveMessage(new MessageRecord
{
ChainId = chainId, Iteration = iterations,
Role = "assistant", Type = "server_tool_use",
Content = el.GetRawText()
});
}
else if (blockType == "web_search_tool_result")
{
// Raw-JSON speichern für History-Rebuild (API braucht den vollständigen Block)
await _db.SaveMessage(new MessageRecord
{
ChainId = chainId, Iteration = iterations,
Role = "assistant", Type = "web_search_result",
Content = el.GetRawText()
});
}
else if (blockType.EndsWith("_tool_result"))
{
// Generische Server-Tool-Results (bash_code_execution_tool_result, code_execution_tool_result etc.)
await _db.SaveMessage(new MessageRecord
{
ChainId = chainId, Iteration = iterations,
Role = "assistant", Type = "server_tool_result",
Content = el.GetRawText()
});
}
}
// Tool Results persistieren
foreach (var tr in executedTools)
{
await _db.SaveMessage(new MessageRecord
{
ChainId = chainId, Iteration = iterations,
Role = "user", Type = "tool_result",
ToolCallId = tr.Id,
Content = tr.Result,
IsError = tr.IsError
});
}
if (stopReason != "tool_use")
{
// ── Review-Agent Phase ──
if (review != null && stopReason == "end_turn" && reviewRound < review.MaxReviewIterations)
{
var reviewResult = await RunReview(response, review, messages, chainId, workspace, sessionId, apiToken, usage, iterations, reviewRound, workCt, sseCt);
iterations += reviewResult.ReviewIterationCount;
reviewRound++;
if (reviewResult.HasFeedback)
{
// Feedback als User-Message an Worker zurückgeben → nächste Iteration
messages.Add(new Dictionary<string, object>
{
["role"] = "user",
["content"] = new List<object> { new { type = "text", text = reviewResult.Feedback } }
});
await _db.SaveMessage(new MessageRecord
{
ChainId = chainId, Iteration = iterations,
Role = "user", Type = "text",
Content = reviewResult.Feedback
});
continue; // → nächste Worker-Iteration
}
}
// ── Finalizer-Agent Phase ──
if (finalizer != null && stopReason == "end_turn")
{
var finalizerIterations = await RunFinalizer(response, finalizer, messages, chainId, workspace, sessionId, apiToken, usage, iterations, workCt, sseCt);
iterations += finalizerIterations;
}
var finalStopReason = string.IsNullOrEmpty(stopReason) ? "unexpected_end" : stopReason;
await FinalizeChain(chain, usage, iterations, finalStopReason, sw, streamErrorInfo);
await SseHelper.TryWriteSse(response, "done", new
{
model = modelId,
inputTokens = usage.InputTokens,
outputTokens = usage.OutputTokens,
cacheCreationTokens = usage.CacheCreationTokens,
cacheReadTokens = usage.CacheReadTokens,
iterations,
chainId,
durationMs = sw.ElapsedMilliseconds,
stopReason = finalStopReason,
streamErrorInfo,
planMode
}, sseCt);
return;
}
var toolResults = new List<object>();
foreach (var tr in executedTools)
{
toolResults.Add(new
{
type = "tool_result",
tool_use_id = tr.Id,
content = TruncateToolResult(tr.Result),
is_error = tr.IsError
});
}
messages.Add(new { role = "user", content = toolResults });
// Hard-Stop Check nach Tool-Ausführung: Sofort raus statt nächste Iteration
hardCt.ThrowIfCancellationRequested();
}
await FinalizeChain(chain, usage, iterations, "max_iterations", sw);
await SseHelper.TryWriteSse(response, "error", new { error = "Max iterations erreicht" }, sseCt);
await SseHelper.TryWriteSse(response, "done", new { chainId = chain.ChainId }, sseCt);
}
catch (OperationCanceledException) when (workCt.IsCancellationRequested || hardCt.IsCancellationRequested)
{
// Stop-Button gedrückt (workCt oder hardCt wurde gecancelled)
await FinalizeChain(chain, usage, iterations, "stopped", sw);
// sseCt ist hier bereits cancelled (gelinkt mit workCt), daher CancellationToken.None
// Der SSE-Stream zum Client ist noch offen – nur der Stop-Button hat ausgelöst
await SseHelper.TryWriteSse(response, "done", new
{
model = modelId,
inputTokens = usage.InputTokens,
outputTokens = usage.OutputTokens,
cacheCreationTokens = usage.CacheCreationTokens,
cacheReadTokens = usage.CacheReadTokens,
iterations,
chainId,
durationMs = sw.ElapsedMilliseconds,
stopReason = "stopped",
planMode
}, CancellationToken.None);
}
catch (Exception) when (workCt.IsCancellationRequested || hardCt.IsCancellationRequested)
{
// Stream-Dispose durch CancellationToken-Registration löst
// ObjectDisposedException/IOException aus statt OperationCanceledException.
// Wenn workCt/hardCt gecancelled ist, ist das ein regulärer Stop.
await FinalizeChain(chain, usage, iterations, "stopped", sw);
// sseCt ist hier bereits cancelled (gelinkt mit workCt), daher CancellationToken.None
await SseHelper.TryWriteSse(response, "done", new
{
model = modelId,
inputTokens = usage.InputTokens,
outputTokens = usage.OutputTokens,
cacheCreationTokens = usage.CacheCreationTokens,
cacheReadTokens = usage.CacheReadTokens,
iterations,
chainId,
durationMs = sw.ElapsedMilliseconds,
stopReason = "stopped",
planMode
}, CancellationToken.None);
}
catch (OperationCanceledException)
{
// Client-Disconnect (cancellationToken gecancelled, aber NICHT workCt/hardCt)
// Agent-Arbeit ist leider schon abgebrochen – Chain als "client_disconnected" finalisieren
// (sollte durch TryWriteSse eigentlich nicht mehr passieren)
await FinalizeChain(chain, usage, iterations, "client_disconnected", sw);
}
catch (Exception ex)
{
// Unerwarteter Fehler – in DB speichern und versuchen an Client zu senden
var errMsg = $"Stream-Fehler: {ex.Message}";
await _db.SaveMessage(new MessageRecord
{
ChainId = chainId, Iteration = iterations,
Role = "system", Type = "error",
Content = errMsg, IsError = true
});
await FinalizeChain(chain, usage, iterations, "error", sw);
await SseHelper.TryWriteSse(response, "error", new { error = errMsg }, sseCt);
await SseHelper.TryWriteSse(response, "done", new { chainId = chain.ChainId }, sseCt);
}
}
private async Task<(string StopReason, List<object> Content, List<ToolResult> Tools, string StreamErrorInfo)> ProcessStream(
HttpResponse response,
HttpResponseMessage httpResp,
TokenUsage usage,
string workspace,
string agentId = null,
CancellationToken workCt = default,
CancellationToken sseCt = default,
RemoteConnectionManager remoteMgr = null,
string sessionId = null,
string userId = null,
string apiToken = null)
{
var stream = await httpResp.Content.ReadAsStreamAsync(workCt);
// Bei Cancellation den Stream disposen, damit ReadLineAsync sofort abbricht.
// StreamReader.ReadLineAsync propagiert den CancellationToken nicht zuverlässig
// an den underlying NetworkStream – der Read blockiert sonst bis Anthropic
// das nächste Chunk sendet (kann bei extended thinking Minuten dauern).
using var ctReg = workCt.Register(() => { try { stream.Dispose(); } catch { } });
using var reader = new StreamReader(stream);
var stopReason = "";
var assistantContent = new List<object>();
var currentToolId = "";
var currentToolName = "";
var currentToolInputJson = new StringBuilder();
var currentTextBlock = new StringBuilder();
var blockTypes = new Dictionary<int, string>();
var webSearchResultBlocks = new Dictionary<int, object>();
var webSearchSources = new Dictionary<int, List<object>>();
var serverToolResultBlocks = new Dictionary<int, object>(); // bash_code_execution_tool_result, code_execution_tool_result etc.
var executedTools = new List<ToolResult>();
var streamErrors = new List<object>();
string line;
while ((line = await reader.ReadLineAsync(workCt)) != null)
{
if (!line.StartsWith("data: ")) continue;
var data = line["data: ".Length..];
if (string.IsNullOrWhiteSpace(data)) continue;
JsonElement evt;
try { evt = JsonSerializer.Deserialize<JsonElement>(data); }
catch { continue; }
var evtType = evt.TryGetProperty("type", out var t) ? t.GetString() ?? "" : "";
switch (evtType)
{
case "message_start":
{
if (evt.TryGetProperty("message", out var msg) &&
msg.TryGetProperty("usage", out var u))
TrackUsage(usage, u);
break;
}
case "content_block_start":
{
var idx = evt.GetProperty("index").GetInt32();
var cb = evt.GetProperty("content_block");
var cbType = cb.GetProperty("type").GetString() ?? "";
blockTypes[idx] = cbType;
if (cbType == "text")
{
currentTextBlock.Clear();
await SseHelper.TryWriteSse(response, "text_start", new { index = idx }, sseCt);
}
else if (cbType == "tool_use")
{
if (currentTextBlock.Length > 0)
{
assistantContent.Add(new { type = "text", text = currentTextBlock.ToString() });
currentTextBlock.Clear();
}
currentToolId = cb.GetProperty("id").GetString() ?? "";
currentToolName = cb.GetProperty("name").GetString() ?? "";
currentToolInputJson.Clear();
await SseHelper.TryWriteSse(response, "tool_start", new { name = currentToolName, id = currentToolId }, sseCt);
}
else if (cbType == "server_tool_use")
{
if (currentTextBlock.Length > 0)
{
assistantContent.Add(new { type = "text", text = currentTextBlock.ToString() });
currentTextBlock.Clear();
}
currentToolId = cb.GetProperty("id").GetString() ?? "";
currentToolName = cb.GetProperty("name").GetString() ?? "";
currentToolInputJson.Clear();
await SseHelper.TryWriteSse(response, "web_search_start", new { name = currentToolName, id = currentToolId }, sseCt);
}
else if (cbType == "web_search_tool_result")
{
var rawBlock = JsonSerializer.Deserialize<JsonElement>(cb.GetRawText());
webSearchResultBlocks[idx] = rawBlock;
webSearchSources[idx] = ExtractWebSearchSources(cb);
}
else if (cbType.EndsWith("_tool_result"))
{
// Generische Server-Tool-Results (bash_code_execution_tool_result, code_execution_tool_result etc.)
serverToolResultBlocks[idx] = JsonSerializer.Deserialize<JsonElement>(cb.GetRawText());
}
break;
}
case "content_block_delta":
{
var delta = evt.GetProperty("delta");
var deltaType = delta.GetProperty("type").GetString() ?? "";
if (deltaType == "text_delta")
{
var text = delta.GetProperty("text").GetString() ?? "";
currentTextBlock.Append(text);
await SseHelper.TryWriteSse(response, "text_delta", new { text }, sseCt);
}
else if (deltaType == "input_json_delta")
{
var partial = delta.GetProperty("partial_json").GetString() ?? "";
currentToolInputJson.Append(partial);
}
break;
}
case "content_block_stop":
{
var idx = evt.GetProperty("index").GetInt32();
if (blockTypes.TryGetValue(idx, out var bt))
{
if (bt == "text")
{
var text = currentTextBlock.ToString();
if (text.Length > 0)
{
assistantContent.Add(new { type = "text", text });
await SseHelper.TryWriteSse(response, "text_end", new { }, sseCt);
}
currentTextBlock.Clear();
}
else if (bt == "tool_use")
{
JsonElement inputEl;
try { inputEl = JsonSerializer.Deserialize<JsonElement>(currentToolInputJson.ToString()); }
catch { inputEl = JsonSerializer.Deserialize<JsonElement>("{}"); }
assistantContent.Add(new
{
type = "tool_use",
id = currentToolId,
name = currentToolName,
input = inputEl
});
// ask_user_choice: NICHT ausführen, sondern als SSE-Event an Client senden
if (currentToolName == ToolService.AskUserChoiceToolName)
{
var question = inputEl.TryGetProperty("question", out var q) ? q.GetString() ?? "" : "";
var choices = new List<string>();
if (inputEl.TryGetProperty("choices", out var c))
{
if (c.ValueKind == JsonValueKind.Array)
{
foreach (var ch in c.EnumerateArray())
choices.Add(ch.GetString() ?? "");
}
else if (c.ValueKind == JsonValueKind.String)
{
// Fallback: Claude hat einen einzelnen String statt Array gesendet
choices.Add(c.GetString() ?? "");
}
}
await SseHelper.TryWriteSse(response, "user_choice", new
{
toolCallId = currentToolId,
question,
choices
}, sseCt);
// Stream restlich konsumieren ohne weitere Verarbeitung
// (Claude sendet nach dem tool_use noch message_delta mit stop_reason)
stopReason = "awaiting_user_choice";
continue;
}
// Cancellation prüfen bevor Tool ausgeführt wird
workCt.ThrowIfCancellationRequested();
string toolResult;
bool isError = false;
try { toolResult = await _toolService.ExecuteTool(currentToolName, inputEl, workspace, _memoryService, agentId, workCt, remoteMgr: remoteMgr, sessionId: sessionId, vectorSearch: _vectorSearch, embeddingBg: _embeddingBg, db: _db, chatService: _chatService, anthropicService: this, openAiService: _openAiService, userId: userId, telegramService: _telegramService, apiToken: apiToken); }
catch (OperationCanceledException) { throw; }
catch (Exception ex) { toolResult = $"Fehler: {ex.Message}"; isError = true; }
workCt.ThrowIfCancellationRequested();
executedTools.Add(new ToolResult(currentToolId, toolResult, isError));
await SseHelper.TryWriteSse(response, "tool_end", new
{
name = currentToolName,
input = currentToolInputJson.ToString(),
result = toolResult,
isError
}, sseCt);
}
else if (bt == "server_tool_use")
{
// Server-side tool (Web Search) – Anthropic führt es aus, wir speichern nur die Anfrage
JsonElement inputEl;
try { inputEl = JsonSerializer.Deserialize<JsonElement>(currentToolInputJson.ToString()); }
catch { inputEl = JsonSerializer.Deserialize<JsonElement>("{}"); }
assistantContent.Add(new
{
type = "server_tool_use",
id = currentToolId,
name = currentToolName,
input = inputEl
});
}
else if (bt == "web_search_tool_result")
{
// Web-Search-Ergebnis von Anthropic – als Raw-JSON in assistantContent aufnehmen
assistantContent.Add(webSearchResultBlocks[idx]);
var sources = webSearchSources.TryGetValue(idx, out var s) ? s : new List<object>();
await SseHelper.TryWriteSse(response, "web_search_end", new { id = currentToolId, sources }, sseCt);
}
else if (bt.EndsWith("_tool_result") && serverToolResultBlocks.TryGetValue(idx, out var srvResultBlock))
{
// Generische Server-Tool-Results (bash_code_execution_tool_result, code_execution_tool_result etc.)
assistantContent.Add(srvResultBlock);
}
}
break;
}
case "message_delta":
{
if (evt.TryGetProperty("delta", out var d))
{
if (d.TryGetProperty("stop_reason", out var sr))
{
// Nicht überschreiben wenn bereits auf awaiting_user_choice gesetzt
if (stopReason != "awaiting_user_choice")
stopReason = sr.GetString() ?? "";
// Bei problematischen stop_reasons Debug-Infos sammeln
if (stopReason is "unexpected_end" or "max_tokens")
{
var debugInfo = new Dictionary<string, object>
{
["event"] = "message_delta",
["stop_reason"] = stopReason,
["output_tokens"] = usage.OutputTokens,
["text_length"] = currentTextBlock.Length,
["timestamp"] = DateTime.UtcNow.ToString("o")
};
if (d.TryGetProperty("stop_sequence", out var ss) && ss.ValueKind != JsonValueKind.Null)
debugInfo["stop_sequence"] = ss.GetString();
streamErrors.Add(debugInfo);
}
}
}
if (evt.TryGetProperty("usage", out var u2) &&
u2.TryGetProperty("output_tokens", out var ot))
usage.OutputTokens += ot.GetInt64();
break;
}
case "error":
{
// API Error-Event (z.B. overloaded_error)
var errorInfo = new Dictionary<string, object>
{
["event"] = "error",
["timestamp"] = DateTime.UtcNow.ToString("o"),
["raw"] = data
};
if (evt.TryGetProperty("error", out var err))
{
if (err.TryGetProperty("type", out var errType))
errorInfo["error_type"] = errType.GetString();
if (err.TryGetProperty("message", out var errMsg))
errorInfo["error_message"] = errMsg.GetString();
}
streamErrors.Add(errorInfo);
break;
}
}
}
if (currentTextBlock.Length > 0)
assistantContent.Add(new { type = "text", text = currentTextBlock.ToString() });
// Stream-Fehler als JSON serialisieren (null wenn keine Fehler)
string streamErrorInfo = streamErrors.Count > 0
? JsonSerializer.Serialize(streamErrors)
: null;
return (stopReason, assistantContent, executedTools, streamErrorInfo);
}
/// <summary>
/// Extrahiert Web-Search-Sources aus einem web_search_tool_result-Block.
/// </summary>
private static List<object> ExtractWebSearchSources(JsonElement block)
{
var sources = new List<object>();
if (block.TryGetProperty("content", out var searchContent) && searchContent.ValueKind == JsonValueKind.Array)
{
foreach (var item in searchContent.EnumerateArray())
{
var itemType = item.TryGetProperty("type", out var it) ? it.GetString() : "";
if (itemType == "web_search_result")
{
var url = item.TryGetProperty("url", out var u) ? u.GetString() ?? "" : "";
var title = item.TryGetProperty("title", out var tt) ? tt.GetString() ?? "" : "";
if (!string.IsNullOrEmpty(url))
sources.Add(new { url, title });
}
}
}
return sources;
}
/// <summary>
/// Parst einen server_tool_use-Block und gibt (id, name, input) zurück.
/// </summary>
private static (string id, string name, JsonElement input) ParseServerToolUse(JsonElement block)
{
var toolId = block.GetProperty("id").GetString() ?? "";
var toolName = block.GetProperty("name").GetString() ?? "";
JsonElement inputEl;
try { inputEl = block.GetProperty("input"); }
catch { inputEl = JsonSerializer.Deserialize<JsonElement>("{}"); }
return (toolId, toolName, inputEl);
}
/// <summary>
/// Aktualisiert den TokenUsage-Tracker aus einem usage-JSON-Element.
/// </summary>
private static void TrackUsage(TokenUsage usage, JsonElement u)
{
if (u.TryGetProperty("input_tokens", out var it)) usage.InputTokens += it.GetInt64();
if (u.TryGetProperty("output_tokens", out var ot)) usage.OutputTokens += ot.GetInt64();
if (u.TryGetProperty("cache_creation_input_tokens", out var cc)) usage.CacheCreationTokens += cc.GetInt64();
if (u.TryGetProperty("cache_read_input_tokens", out var cr)) usage.CacheReadTokens += cr.GetInt64();
}
}
File diff suppressed because it is too large. Load diff