Meleki
5fe3f8ac87
- Session hat optionalen ReviewAgentId + MaxReviewIterations - Nach end_turn des Worker-Agents wird der Review-Agent aufgerufen - Review-Agent bekommt Worker-Konversation als Kontext + kann Tools nutzen - [APPROVED] in Review-Antwort → fertig, sonst Feedback an Worker - Worker korrigiert und wird erneut reviewed (max N Runden) - SSE-Events: review_start, review_text, review_tool_use, review_feedback, review_approved - Chat-UI: Review-Phase als eigene Bubble mit Header, Text, Status - Session-Page: Review-Agent-Picker (gleicher Agent-Card-Helper) - Dashboard: Review-Agent-Badge auf Session-Cards - DB-Migration: review_agent_id, max_review_iterations auf sessions
1115 lines
50 KiB
C#
1115 lines
50 KiB
C#
using System;
|
||
using System.Collections.Generic;
|
||
using System.IO;
|
||
using System.IO.Compression;
|
||
using System.Net.Http;
|
||
using System.Net.Http.Headers;
|
||
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 class AnthropicService
|
||
{
|
||
private readonly IHttpClientFactory _httpFactory;
|
||
private readonly string _token;
|
||
private readonly ToolService _toolService;
|
||
private readonly DatabaseService _db;
|
||
private readonly MemoryService _memoryService;
|
||
private readonly bool _payloadLogging;
|
||
private readonly bool _payloadLoggingGzip;
|
||
|
||
public AnthropicService(IHttpClientFactory httpFactory, string token, ToolService toolService, DatabaseService db, MemoryService memoryService = null, bool payloadLogging = false, bool payloadLoggingGzip = false)
|
||
{
|
||
_httpFactory = httpFactory;
|
||
_token = token;
|
||
_toolService = toolService;
|
||
_db = db;
|
||
_memoryService = memoryService;
|
||
_payloadLogging = payloadLogging;
|
||
_payloadLoggingGzip = payloadLoggingGzip;
|
||
}
|
||
|
||
public async Task StreamChat(
|
||
HttpResponse response,
|
||
string modelId,
|
||
List<object> systemPrompt,
|
||
List<object> messages,
|
||
string chainId,
|
||
string workspace,
|
||
string sessionId = null,
|
||
string agentId = null,
|
||
bool useMemory = true,
|
||
string webSearchVersion = null,
|
||
int maxIterations = 100,
|
||
string toolChoice = "any",
|
||
ReviewContext review = null,
|
||
CancellationToken cancellationToken = default)
|
||
{
|
||
var sw = System.Diagnostics.Stopwatch.StartNew();
|
||
var usage = new TokenUsage();
|
||
int iterations = 0;
|
||
int reviewRound = 0;
|
||
string lastStopReason = "";
|
||
|
||
var chain = new ChainRecord
|
||
{
|
||
ChainId = chainId,
|
||
SessionId = sessionId,
|
||
Model = modelId
|
||
};
|
||
|
||
try
|
||
{
|
||
while (iterations < maxIterations)
|
||
{
|
||
cancellationToken.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 hasWorkspace = !string.IsNullOrEmpty(workspace);
|
||
var includeMemory = useMemory && _memoryService != null && !string.IsNullOrEmpty(agentId);
|
||
var hasTools = hasWorkspace || includeMemory;
|
||
var hasWebSearch = !string.IsNullOrEmpty(webSearchVersion);
|
||
|
||
if (hasTools || hasWebSearch)
|
||
{
|
||
var tools = new List<object>();
|
||
if (hasTools)
|
||
tools.AddRange(_toolService.GetToolDefinitions(includeFileTools: hasWorkspace, includeMemory: includeMemory));
|
||
if (hasWebSearch)
|
||
tools.Add(new { type = webSearchVersion, name = "web_search", max_uses = 5 });
|
||
payload["tools"] = tools;
|
||
// "any" erzwingt Tool-Nutzung, aber nur in der ersten Iteration,
|
||
// damit Claude nicht endlos Tools aufruft wenn die Aufgabe erledigt ist.
|
||
var effectiveChoice = (toolChoice == "any" && iterations == 1) ? "any" : "auto";
|
||
payload["tool_choice"] = new { type = effectiveChoice };
|
||
}
|
||
|
||
var client = CreateClient();
|
||
var json = JsonSerializer.Serialize(payload, new JsonSerializerOptions { WriteIndented = true });
|
||
LogPayload(json, sessionId, chainId, iterations);
|
||
var httpContent = new StringContent(json, Encoding.UTF8, "application/json");
|
||
|
||
var httpResp = await client.SendAsync(
|
||
new HttpRequestMessage(HttpMethod.Post, "https://api.anthropic.com/v1/messages") { Content = httpContent },
|
||
HttpCompletionOption.ResponseHeadersRead,
|
||
cancellationToken
|
||
);
|
||
|
||
if (!httpResp.IsSuccessStatusCode)
|
||
{
|
||
var errBody = await httpResp.Content.ReadAsStringAsync(cancellationToken);
|
||
var errMsg = $"API {(int)httpResp.StatusCode}: {errBody}";
|
||
await _db.SaveMessage(new MessageRecord
|
||
{
|
||
ChainId = chainId, Iteration = iterations,
|
||
Role = "system", Type = "error",
|
||
Content = errMsg, IsError = true
|
||
});
|
||
await TryWriteSse(response, "error", new { error = errMsg }, cancellationToken);
|
||
await FinalizeChain(chain, usage, iterations, "error", sw);
|
||
await TryWriteSse(response, "done", new { chainId = chain.ChainId }, cancellationToken);
|
||
return;
|
||
}
|
||
|
||
var (stopReason, assistantContent, executedTools, streamErrorInfo) = await ProcessStream(response, httpResp, usage, workspace, agentId, cancellationToken);
|
||
lastStopReason = stopReason;
|
||
|
||
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")
|
||
{
|
||
await _db.SaveMessage(new MessageRecord
|
||
{
|
||
ChainId = chainId, Iteration = iterations,
|
||
Role = "assistant", Type = "text",
|
||
Content = el.GetProperty("text").GetString()
|
||
});
|
||
}
|
||
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()
|
||
});
|
||
}
|
||
}
|
||
|
||
// 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, usage, iterations, reviewRound, cancellationToken);
|
||
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
|
||
}
|
||
}
|
||
|
||
var finalStopReason = string.IsNullOrEmpty(stopReason) ? "unexpected_end" : stopReason;
|
||
await FinalizeChain(chain, usage, iterations, finalStopReason, sw, streamErrorInfo);
|
||
await 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
|
||
}, cancellationToken);
|
||
return;
|
||
}
|
||
|
||
var toolResults = new List<object>();
|
||
foreach (var tr in executedTools)
|
||
{
|
||
// Tool-Results in der History auf 20.000 Zeichen kürzen.
|
||
// Das vollständige Ergebnis wurde bereits in der DB gespeichert.
|
||
// 20k Zeichen ≈ ~5k Tokens – verhindert Token-Explosion bei großen
|
||
// read_file / execute-Outputs die sich über alle weiteren Iterationen schleppen.
|
||
const int MaxToolResultChars = 20_000;
|
||
var content = tr.Result.Length > MaxToolResultChars
|
||
? tr.Result[..MaxToolResultChars] + $"\n\n[... gekürzt, {tr.Result.Length - MaxToolResultChars} weitere Zeichen ...]"
|
||
: tr.Result;
|
||
|
||
toolResults.Add(new
|
||
{
|
||
type = "tool_result",
|
||
tool_use_id = tr.Id,
|
||
content,
|
||
is_error = tr.IsError
|
||
});
|
||
}
|
||
|
||
messages.Add(new { role = "user", content = toolResults });
|
||
}
|
||
|
||
await FinalizeChain(chain, usage, iterations, "max_iterations", sw);
|
||
await TryWriteSse(response, "error", new { error = "Max iterations erreicht" }, cancellationToken);
|
||
await TryWriteSse(response, "done", new { chainId = chain.ChainId }, cancellationToken);
|
||
}
|
||
catch (OperationCanceledException)
|
||
{
|
||
// Client hat die Verbindung getrennt (z.B. Browser-Refresh)
|
||
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 TryWriteSse(response, "error", new { error = errMsg }, cancellationToken);
|
||
await TryWriteSse(response, "done", new { chainId = chain.ChainId }, cancellationToken);
|
||
}
|
||
}
|
||
|
||
// ── 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 gesamte Konversation
|
||
/// und kann Tools nutzen (z.B. read_file um die Arbeit zu prüfen).
|
||
/// Antwortet er mit [APPROVED] → fertig. Sonst → Feedback wird an den Worker zurückgegeben.
|
||
/// reviewRound = aktuelle Runde (0-basiert), wird vom Caller hochgezählt.
|
||
/// </summary>
|
||
private async Task<ReviewResult> RunReview(
|
||
HttpResponse response,
|
||
ReviewContext review,
|
||
List<object> workerMessages,
|
||
string chainId,
|
||
string workspace,
|
||
string sessionId,
|
||
TokenUsage usage,
|
||
int baseIterations,
|
||
int reviewRound,
|
||
CancellationToken cancellationToken)
|
||
{
|
||
cancellationToken.ThrowIfCancellationRequested();
|
||
|
||
// SSE: Review-Phase Start
|
||
await TryWriteSse(response, "review_start", new
|
||
{
|
||
round = reviewRound + 1,
|
||
maxRounds = review.MaxReviewIterations,
|
||
agentId = review.AgentId,
|
||
agentName = review.Agent?.Name ?? "Review-Agent"
|
||
}, cancellationToken);
|
||
|
||
// Review-Agent bekommt eigene Message-History
|
||
var reviewMessages = BuildReviewMessages(workerMessages);
|
||
|
||
// Review-Agent iteriert mit eigenen Tools
|
||
var (reviewText, reviewIterations) = await RunReviewAgentChat(
|
||
response, review, reviewMessages, chainId, workspace, sessionId,
|
||
usage, baseIterations, cancellationToken);
|
||
|
||
var result = new ReviewResult { ReviewIterationCount = reviewIterations };
|
||
|
||
// Prüfe ob [APPROVED]
|
||
if (!string.IsNullOrEmpty(reviewText) && reviewText.Contains("[APPROVED]"))
|
||
{
|
||
await TryWriteSse(response, "review_approved", new
|
||
{
|
||
round = reviewRound + 1,
|
||
agentName = review.Agent?.Name ?? "Review-Agent"
|
||
}, cancellationToken);
|
||
result.HasFeedback = false;
|
||
return result;
|
||
}
|
||
|
||
// Feedback – an Worker zurückspielen
|
||
result.HasFeedback = true;
|
||
result.Feedback = $"[Review-Feedback von {review.Agent?.Name ?? "Review-Agent"} (Runde {reviewRound + 1}/{review.MaxReviewIterations})]:\n{reviewText}";
|
||
|
||
await TryWriteSse(response, "review_feedback", new
|
||
{
|
||
round = reviewRound + 1,
|
||
maxRounds = review.MaxReviewIterations,
|
||
feedback = reviewText,
|
||
agentName = review.Agent?.Name ?? "Review-Agent"
|
||
}, cancellationToken);
|
||
|
||
return result;
|
||
}
|
||
|
||
/// <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)
|
||
{
|
||
// Worker-Konversation zusammenfassen
|
||
var summary = new StringBuilder();
|
||
summary.AppendLine("Hier ist die bisherige Konversation zwischen User und Worker-Agent. Bitte überprüfe ob die Aufgabe korrekt erledigt wurde:\n");
|
||
summary.AppendLine("---");
|
||
|
||
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))
|
||
{
|
||
if (content.ValueKind == JsonValueKind.String)
|
||
{
|
||
summary.AppendLine($"\n**{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")
|
||
{
|
||
var text = block.TryGetProperty("text", out var t) ? t.GetString() ?? "" : "";
|
||
summary.AppendLine($"\n**{roleStr}:** {text}");
|
||
}
|
||
else if (blockType == "tool_use")
|
||
{
|
||
var toolName = block.TryGetProperty("name", out var tn) ? tn.GetString() ?? "" : "";
|
||
var toolInput = block.TryGetProperty("input", out var ti) ? ti.ToString() : "{}";
|
||
summary.AppendLine($"\n**{roleStr} [Tool: {toolName}]:** {toolInput}");
|
||
}
|
||
else if (blockType == "tool_result")
|
||
{
|
||
var resultContent = block.TryGetProperty("content", out var rc) ? rc.ToString() : "";
|
||
// Tool-Results kürzen für Review-Kontext
|
||
if (resultContent.Length > 2000)
|
||
resultContent = resultContent[..2000] + "\n[... gekürzt]";
|
||
summary.AppendLine($"\n**{roleStr} [Tool-Result]:** {resultContent}");
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
summary.AppendLine("\n---");
|
||
summary.AppendLine("\nBitte überprüfe ob die Aufgabe korrekt erledigt wurde. Du kannst dazu auch Tools verwenden (z.B. Dateien lesen). Antworte mit exakt [APPROVED] wenn alles korrekt ist, oder beschreibe was falsch ist und korrigiert werden muss.");
|
||
|
||
return new List<object>
|
||
{
|
||
new Dictionary<string, object>
|
||
{
|
||
["role"] = "user",
|
||
["content"] = new List<object> { new { type = "text", text = summary.ToString() } }
|
||
}
|
||
};
|
||
}
|
||
|
||
/// <summary>
|
||
/// Führt den Review-Agent Chat aus (non-streaming für den Agent, aber SSE-Events für den Client).
|
||
/// Gibt den finalen Text und die Anzahl der Iterationen zurück.
|
||
/// </summary>
|
||
private async Task<(string text, int iterations)> RunReviewAgentChat(
|
||
HttpResponse response,
|
||
ReviewContext review,
|
||
List<object> messages,
|
||
string chainId,
|
||
string workspace,
|
||
string sessionId,
|
||
TokenUsage usage,
|
||
int baseIterations,
|
||
CancellationToken cancellationToken)
|
||
{
|
||
int iterations = 0;
|
||
var finalText = new StringBuilder();
|
||
int maxToolIterations = 20; // Review-Agent bekommt max 20 Tool-Iterationen
|
||
|
||
while (iterations < maxToolIterations)
|
||
{
|
||
cancellationToken.ThrowIfCancellationRequested();
|
||
iterations++;
|
||
|
||
var payload = new Dictionary<string, object>
|
||
{
|
||
["model"] = review.ModelId,
|
||
["max_tokens"] = 8192,
|
||
["messages"] = messages,
|
||
["stream"] = false
|
||
};
|
||
|
||
if (review.SystemPrompt != null && review.SystemPrompt.Count > 0)
|
||
payload["system"] = review.SystemPrompt;
|
||
|
||
var hasWorkspace = !string.IsNullOrEmpty(workspace);
|
||
var includeMemory = review.UseMemory && _memoryService != null && !string.IsNullOrEmpty(review.AgentId);
|
||
var hasTools = hasWorkspace || includeMemory;
|
||
var hasWebSearch = !string.IsNullOrEmpty(review.WebSearchVersion);
|
||
|
||
if (hasTools || hasWebSearch)
|
||
{
|
||
var tools = new List<object>();
|
||
if (hasTools)
|
||
tools.AddRange(_toolService.GetToolDefinitions(includeFileTools: hasWorkspace, includeMemory: includeMemory));
|
||
if (hasWebSearch)
|
||
tools.Add(new { type = review.WebSearchVersion, name = "web_search", max_uses = 5 });
|
||
payload["tools"] = tools;
|
||
var effectiveChoice = (review.ToolChoice == "any" && iterations == 1) ? "any" : "auto";
|
||
payload["tool_choice"] = new { type = effectiveChoice };
|
||
}
|
||
|
||
var client = CreateClient();
|
||
var json = JsonSerializer.Serialize(payload, new JsonSerializerOptions { WriteIndented = true });
|
||
LogPayload(json, sessionId, chainId, baseIterations + iterations);
|
||
var httpContent = new StringContent(json, Encoding.UTF8, "application/json");
|
||
|
||
var httpResp = await client.SendAsync(
|
||
new HttpRequestMessage(HttpMethod.Post, "https://api.anthropic.com/v1/messages") { Content = httpContent },
|
||
cancellationToken: cancellationToken);
|
||
|
||
var respBody = await httpResp.Content.ReadAsStringAsync(cancellationToken);
|
||
|
||
if (!httpResp.IsSuccessStatusCode)
|
||
{
|
||
return ($"Review-Agent API Fehler: {(int)httpResp.StatusCode}: {respBody}", iterations);
|
||
}
|
||
|
||
var resp = JsonSerializer.Deserialize<JsonElement>(respBody);
|
||
|
||
// Usage tracken (auf den gemeinsamen Zähler)
|
||
if (resp.TryGetProperty("usage", out var 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();
|
||
}
|
||
|
||
var stopReason = resp.TryGetProperty("stop_reason", out var sr) ? sr.GetString() ?? "" : "";
|
||
var content = resp.GetProperty("content");
|
||
|
||
var assistantContent = new List<object>();
|
||
var toolResults = new List<object>();
|
||
|
||
foreach (var block in content.EnumerateArray())
|
||
{
|
||
var blockType = block.GetProperty("type").GetString() ?? "";
|
||
|
||
if (blockType == "text")
|
||
{
|
||
var text = block.GetProperty("text").GetString() ?? "";
|
||
assistantContent.Add(new { type = "text", text });
|
||
finalText.Append(text);
|
||
|
||
// Review-Text per SSE an den Client streamen
|
||
await TryWriteSse(response, "review_text", new
|
||
{
|
||
text,
|
||
agentName = review.Agent?.Name ?? "Review-Agent",
|
||
agentId = review.AgentId
|
||
}, cancellationToken);
|
||
|
||
// In DB speichern (als Review-Message markiert)
|
||
await _db.SaveMessage(new MessageRecord
|
||
{
|
||
ChainId = chainId,
|
||
Iteration = baseIterations + iterations,
|
||
Role = "assistant",
|
||
Type = "text",
|
||
Content = $"[Review: {review.Agent?.Name ?? "Review-Agent"}] {text}"
|
||
});
|
||
}
|
||
else if (blockType == "tool_use")
|
||
{
|
||
var toolId = block.GetProperty("id").GetString() ?? "";
|
||
var toolName = block.GetProperty("name").GetString() ?? "";
|
||
var toolInput = block.GetProperty("input");
|
||
|
||
assistantContent.Add(new { type = "tool_use", id = toolId, name = toolName, input = toolInput });
|
||
|
||
// Tool-Use per SSE
|
||
await TryWriteSse(response, "review_tool_use", new
|
||
{
|
||
toolName,
|
||
toolId,
|
||
input = toolInput.ToString(),
|
||
agentName = review.Agent?.Name ?? "Review-Agent"
|
||
}, cancellationToken);
|
||
|
||
await _db.SaveMessage(new MessageRecord
|
||
{
|
||
ChainId = chainId,
|
||
Iteration = baseIterations + iterations,
|
||
Role = "assistant",
|
||
Type = "tool_use",
|
||
Content = toolInput.GetRawText(),
|
||
ToolName = toolName,
|
||
ToolCallId = toolId
|
||
});
|
||
|
||
string toolResult;
|
||
bool isError = false;
|
||
try { toolResult = await _toolService.ExecuteTool(toolName, toolInput, workspace, _memoryService, review.AgentId, cancellationToken); }
|
||
catch (Exception ex) { toolResult = $"Fehler: {ex.Message}"; isError = true; }
|
||
|
||
const int MaxToolResultChars = 20_000;
|
||
var truncatedResult = toolResult.Length > MaxToolResultChars
|
||
? toolResult[..MaxToolResultChars] + $"\n\n[... gekürzt, {toolResult.Length - MaxToolResultChars} weitere Zeichen ...]"
|
||
: toolResult;
|
||
|
||
toolResults.Add(new
|
||
{
|
||
type = "tool_result",
|
||
tool_use_id = toolId,
|
||
content = truncatedResult,
|
||
is_error = isError
|
||
});
|
||
|
||
await _db.SaveMessage(new MessageRecord
|
||
{
|
||
ChainId = chainId,
|
||
Iteration = baseIterations + iterations,
|
||
Role = "user",
|
||
Type = "tool_result",
|
||
Content = toolResult,
|
||
ToolName = toolName,
|
||
ToolCallId = toolId,
|
||
IsError = isError
|
||
});
|
||
}
|
||
}
|
||
|
||
messages.Add(new Dictionary<string, object> { ["role"] = "assistant", ["content"] = assistantContent });
|
||
|
||
if (stopReason == "end_turn" || stopReason == "stop_sequence" || toolResults.Count == 0)
|
||
break;
|
||
|
||
messages.Add(new Dictionary<string, object> { ["role"] = "user", ["content"] = toolResults });
|
||
}
|
||
|
||
return (finalText.ToString(), iterations);
|
||
}
|
||
|
||
/// <summary>
|
||
/// Non-Streaming Chat für Telegram. Gibt den finalen Antworttext zurück.
|
||
/// 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();
|
||
int iterations = 0;
|
||
var finalText = new StringBuilder();
|
||
|
||
var chain = new ChainRecord
|
||
{
|
||
ChainId = ctx.ChainId,
|
||
SessionId = ctx.SessionId,
|
||
Model = ctx.ModelId
|
||
};
|
||
|
||
try
|
||
{
|
||
while (iterations < ctx.MaxIterations)
|
||
{
|
||
iterations++;
|
||
onIteration?.Invoke(iterations);
|
||
|
||
var payload = new Dictionary<string, object>
|
||
{
|
||
["model"] = ctx.ModelId,
|
||
["max_tokens"] = 16384,
|
||
["messages"] = messages,
|
||
["stream"] = false
|
||
};
|
||
|
||
if (ctx.SystemPrompt != null && ctx.SystemPrompt.Count > 0)
|
||
payload["system"] = ctx.SystemPrompt;
|
||
|
||
var hasWorkspace = !string.IsNullOrEmpty(ctx.Workspace);
|
||
var includeMemory = ctx.UseMemory && _memoryService != null && !string.IsNullOrEmpty(ctx.AgentId);
|
||
var hasTools = hasWorkspace || includeMemory;
|
||
var hasWebSearch = !string.IsNullOrEmpty(ctx.WebSearchVersion);
|
||
|
||
if (hasTools || hasWebSearch)
|
||
{
|
||
var tools = new List<object>();
|
||
if (hasTools)
|
||
tools.AddRange(_toolService.GetToolDefinitions(includeFileTools: hasWorkspace, includeMemory: includeMemory));
|
||
if (hasWebSearch)
|
||
tools.Add(new { type = ctx.WebSearchVersion, name = "web_search", max_uses = 5 });
|
||
payload["tools"] = tools;
|
||
var effectiveChoice = (ctx.ToolChoice == "any" && iterations == 1) ? "any" : "auto";
|
||
payload["tool_choice"] = new { type = effectiveChoice };
|
||
}
|
||
|
||
var client = CreateClient();
|
||
var json = JsonSerializer.Serialize(payload, new JsonSerializerOptions { WriteIndented = true });
|
||
LogPayload(json, ctx.SessionId, ctx.ChainId, iterations);
|
||
var httpContent = new StringContent(json, Encoding.UTF8, "application/json");
|
||
|
||
var httpResp = await client.SendAsync(
|
||
new HttpRequestMessage(HttpMethod.Post, "https://api.anthropic.com/v1/messages") { Content = httpContent });
|
||
|
||
var respBody = await httpResp.Content.ReadAsStringAsync();
|
||
|
||
if (!httpResp.IsSuccessStatusCode)
|
||
{
|
||
var errMsg = $"API {(int)httpResp.StatusCode}: {respBody}";
|
||
await _db.SaveMessage(new MessageRecord
|
||
{
|
||
ChainId = ctx.ChainId, Iteration = iterations,
|
||
Role = "system", Type = "error",
|
||
Content = errMsg, IsError = true
|
||
});
|
||
await FinalizeChain(chain, usage, iterations, "error", sw);
|
||
throw new Exception($"Claude API Fehler: {(int)httpResp.StatusCode}");
|
||
}
|
||
|
||
var resp = JsonSerializer.Deserialize<JsonElement>(respBody);
|
||
|
||
// Usage tracken
|
||
if (resp.TryGetProperty("usage", out var 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();
|
||
}
|
||
|
||
var stopReason = resp.TryGetProperty("stop_reason", out var sr) ? sr.GetString() ?? "" : "";
|
||
var content = resp.GetProperty("content");
|
||
|
||
var assistantContent = new List<object>();
|
||
var toolResults = new List<object>();
|
||
|
||
foreach (var block in content.EnumerateArray())
|
||
{
|
||
var blockType = block.GetProperty("type").GetString() ?? "";
|
||
|
||
if (blockType == "text")
|
||
{
|
||
var text = block.GetProperty("text").GetString() ?? "";
|
||
assistantContent.Add(new { type = "text", text });
|
||
finalText.Append(text);
|
||
|
||
await _db.SaveMessage(new MessageRecord
|
||
{
|
||
ChainId = ctx.ChainId, Iteration = iterations,
|
||
Role = "assistant", Type = "text",
|
||
Content = text
|
||
});
|
||
}
|
||
else if (blockType == "tool_use")
|
||
{
|
||
var toolId = block.GetProperty("id").GetString() ?? "";
|
||
var toolName = block.GetProperty("name").GetString() ?? "";
|
||
var toolInput = block.GetProperty("input");
|
||
|
||
assistantContent.Add(new { type = "tool_use", id = toolId, name = toolName, input = toolInput });
|
||
|
||
await _db.SaveMessage(new MessageRecord
|
||
{
|
||
ChainId = ctx.ChainId, Iteration = iterations,
|
||
Role = "assistant", Type = "tool_use",
|
||
Content = toolInput.GetRawText(),
|
||
ToolName = toolName, ToolCallId = toolId
|
||
});
|
||
|
||
string toolResult;
|
||
bool isError = false;
|
||
try { toolResult = await _toolService.ExecuteTool(toolName, toolInput, ctx.Workspace, _memoryService, ctx.AgentId); }
|
||
catch (Exception ex) { toolResult = $"Fehler: {ex.Message}"; isError = true; }
|
||
|
||
// Tool-Result in der History kürzen (wie bei StreamChat)
|
||
const int MaxToolResultChars = 20_000;
|
||
var truncatedResult = toolResult.Length > MaxToolResultChars
|
||
? toolResult[..MaxToolResultChars] + $"\n\n[... gekürzt, {toolResult.Length - MaxToolResultChars} weitere Zeichen ...]"
|
||
: toolResult;
|
||
|
||
toolResults.Add(new
|
||
{
|
||
type = "tool_result",
|
||
tool_use_id = toolId,
|
||
content = truncatedResult,
|
||
is_error = isError
|
||
});
|
||
|
||
await _db.SaveMessage(new MessageRecord
|
||
{
|
||
ChainId = ctx.ChainId, Iteration = iterations,
|
||
Role = "user", Type = "tool_result",
|
||
Content = toolResult,
|
||
ToolName = toolName, ToolCallId = toolId,
|
||
IsError = isError
|
||
});
|
||
}
|
||
}
|
||
|
||
messages.Add(new Dictionary<string, object> { ["role"] = "assistant", ["content"] = assistantContent });
|
||
|
||
if (stopReason == "end_turn" || stopReason == "stop_sequence" || toolResults.Count == 0)
|
||
{
|
||
await FinalizeChain(chain, usage, iterations, stopReason, sw);
|
||
break;
|
||
}
|
||
|
||
messages.Add(new Dictionary<string, object> { ["role"] = "user", ["content"] = toolResults });
|
||
}
|
||
}
|
||
catch
|
||
{
|
||
if (iterations > 0)
|
||
await FinalizeChain(chain, usage, iterations, "error", sw);
|
||
throw;
|
||
}
|
||
|
||
return finalText.Length > 0 ? finalText.ToString() : "(Keine Textantwort)";
|
||
}
|
||
|
||
private void LogPayload(string json, string sessionId, string chainId, int iteration)
|
||
{
|
||
if (!_payloadLogging) return;
|
||
try
|
||
{
|
||
var logsDir = Path.Combine(AppContext.BaseDirectory, "data", "logs");
|
||
Directory.CreateDirectory(logsDir);
|
||
var timestamp = DateTimeOffset.UtcNow.ToUnixTimeSeconds();
|
||
var baseName = $"{timestamp}-{sessionId}-{chainId}-iter{iteration}-payload";
|
||
|
||
if (_payloadLoggingGzip)
|
||
{
|
||
var filePath = Path.Combine(logsDir, baseName + ".json.gz");
|
||
using var fs = File.Create(filePath);
|
||
using var gz = new GZipStream(fs, CompressionLevel.Fastest);
|
||
using var writer = new StreamWriter(gz, Encoding.UTF8);
|
||
writer.Write(json);
|
||
}
|
||
else
|
||
{
|
||
File.WriteAllText(Path.Combine(logsDir, baseName + ".json"), json);
|
||
}
|
||
}
|
||
catch { /* Logging darf nie die App crashen */ }
|
||
}
|
||
|
||
private async Task FinalizeChain(ChainRecord chain, TokenUsage usage, int iterations, string stopReason, System.Diagnostics.Stopwatch sw, string streamErrorInfo = null)
|
||
{
|
||
chain.InputTokens = usage.InputTokens;
|
||
chain.OutputTokens = usage.OutputTokens;
|
||
chain.CacheCreationTokens = usage.CacheCreationTokens;
|
||
chain.CacheReadTokens = usage.CacheReadTokens;
|
||
chain.Iterations = iterations;
|
||
chain.StopReason = stopReason;
|
||
chain.StreamErrorInfo = streamErrorInfo;
|
||
chain.DurationMs = sw.ElapsedMilliseconds;
|
||
await _db.UpdateChain(chain);
|
||
}
|
||
|
||
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 cancellationToken = default)
|
||
{
|
||
var stream = await httpResp.Content.ReadAsStreamAsync(cancellationToken);
|
||
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 executedTools = new List<ToolResult>();
|
||
var streamErrors = new List<object>();
|
||
|
||
string line;
|
||
while ((line = await reader.ReadLineAsync(cancellationToken)) != 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))
|
||
{
|
||
if (u.TryGetProperty("input_tokens", out var it))
|
||
usage.InputTokens += it.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();
|
||
}
|
||
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 WriteSseEvent(response, "text_start", new { index = idx }, cancellationToken);
|
||
}
|
||
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 WriteSseEvent(response, "tool_start", new { name = currentToolName, id = currentToolId }, cancellationToken);
|
||
}
|
||
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 WriteSseEvent(response, "web_search_start", new { name = currentToolName, id = currentToolId }, cancellationToken);
|
||
}
|
||
else if (cbType == "web_search_tool_result")
|
||
{
|
||
// Gesamten Block als JSON speichern für die Conversation-History
|
||
var rawBlock = JsonSerializer.Deserialize<JsonElement>(cb.GetRawText());
|
||
webSearchResultBlocks[idx] = rawBlock;
|
||
}
|
||
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 WriteSseEvent(response, "text_delta", new { text }, cancellationToken);
|
||
}
|
||
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 WriteSseEvent(response, "text_end", new { }, cancellationToken);
|
||
}
|
||
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
|
||
});
|
||
|
||
// Cancellation prüfen bevor Tool ausgeführt wird
|
||
cancellationToken.ThrowIfCancellationRequested();
|
||
|
||
string toolResult;
|
||
bool isError = false;
|
||
try { toolResult = await _toolService.ExecuteTool(currentToolName, inputEl, workspace, _memoryService, agentId, cancellationToken); }
|
||
catch (OperationCanceledException) { throw; }
|
||
catch (Exception ex) { toolResult = $"Fehler: {ex.Message}"; isError = true; }
|
||
|
||
cancellationToken.ThrowIfCancellationRequested();
|
||
executedTools.Add(new ToolResult(currentToolId, toolResult, isError));
|
||
|
||
await WriteSseEvent(response, "tool_end", new
|
||
{
|
||
name = currentToolName,
|
||
input = currentToolInputJson.ToString(),
|
||
result = toolResult.Length > 500 ? toolResult[..500] + "..." : toolResult,
|
||
isError
|
||
}, cancellationToken);
|
||
}
|
||
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]);
|
||
await WriteSseEvent(response, "web_search_end", new { id = currentToolId }, cancellationToken);
|
||
}
|
||
}
|
||
break;
|
||
}
|
||
|
||
case "message_delta":
|
||
{
|
||
if (evt.TryGetProperty("delta", out var d))
|
||
{
|
||
if (d.TryGetProperty("stop_reason", out var sr))
|
||
{
|
||
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);
|
||
}
|
||
|
||
private HttpClient CreateClient()
|
||
{
|
||
var client = _httpFactory.CreateClient();
|
||
client.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", _token);
|
||
client.DefaultRequestHeaders.Add("anthropic-version", "2023-06-01");
|
||
client.DefaultRequestHeaders.Add("anthropic-beta", "claude-code-20250219,oauth-2025-04-20");
|
||
client.DefaultRequestHeaders.Add("User-Agent", "claude-cli/2.1.2 (external, cli)");
|
||
client.DefaultRequestHeaders.Add("x-app", "cli");
|
||
client.DefaultRequestHeaders.Accept.Add(new MediaTypeWithQualityHeaderValue("application/json"));
|
||
client.DefaultRequestHeaders.Add("anthropic-dangerous-direct-browser-access", "true");
|
||
return client;
|
||
}
|
||
|
||
public static async Task WriteSseEvent(HttpResponse response, string eventType, object data, CancellationToken ct = default)
|
||
{
|
||
ct.ThrowIfCancellationRequested();
|
||
var json = JsonSerializer.Serialize(data);
|
||
await response.WriteAsync($"event: {eventType}\ndata: {json}\n\n", ct);
|
||
await response.Body.FlushAsync(ct);
|
||
}
|
||
|
||
private static async Task TryWriteSse(HttpResponse response, string eventType, object data, CancellationToken ct)
|
||
{
|
||
try
|
||
{
|
||
if (!ct.IsCancellationRequested)
|
||
await WriteSseEvent(response, eventType, data, ct);
|
||
}
|
||
catch
|
||
{
|
||
// Client bereits disconnected – ignorieren
|
||
}
|
||
}
|
||
}
|
||
|
||
public record ToolResult(string Id, string Result, bool IsError);
|
||
|
||
public class TokenUsage
|
||
{
|
||
public long InputTokens { get; set; }
|
||
public long OutputTokens { get; set; }
|
||
public long CacheCreationTokens { get; set; }
|
||
public long CacheReadTokens { get; set; }
|
||
}
|