修复
This commit is contained in:
@@ -6,13 +6,16 @@ namespace app\common\service;
|
||||
/** Request-local numeric telemetry. Raw events, prompts, titles, URLs and IDs never leave it. */
|
||||
final class DifyStreamTiming
|
||||
{
|
||||
private const EVENTS = ['workflow_started', 'workflow_finished', 'node_started', 'node_finished',
|
||||
private const MAX_MILLISECONDS = 3600000;
|
||||
private const MAX_COUNT = 1000000000;
|
||||
private const EVENTS = ['workflow_started', 'workflow_finished', 'node_started', 'node_finished', 'node_retry',
|
||||
'message', 'agent_message', 'agent_thought', 'message_end', 'error', 'ping', 'choices'];
|
||||
private const TYPES = ['start', 'llm', 'knowledge-retrieval', 'tool', 'code', 'template-transform',
|
||||
'if-else', 'answer', 'end', 'agent', 'http-request', 'iteration', 'question-classifier',
|
||||
'variable-aggregator', 'parameter-extractor', 'list-operator', 'document-extractor'];
|
||||
private \Closure $clock;
|
||||
private float $started;
|
||||
private int $lastElapsed = 0;
|
||||
private array $metrics = [];
|
||||
private array $events = [];
|
||||
private array $nodes = [];
|
||||
@@ -24,7 +27,77 @@ final class DifyStreamTiming
|
||||
$this->started = ($this->clock)();
|
||||
}
|
||||
|
||||
private function elapsed(): int { return max(0, (int) floor(($this->clock)() - $this->started)); }
|
||||
private function elapsed(): int
|
||||
{
|
||||
$elapsed = ($this->clock)() - $this->started;
|
||||
if (!is_finite($elapsed)) return $this->lastElapsed;
|
||||
return $this->lastElapsed = max($this->lastElapsed,
|
||||
(int) min(self::MAX_MILLISECONDS, max(0, floor($elapsed))));
|
||||
}
|
||||
|
||||
private function increment(string $key): void
|
||||
{
|
||||
$this->metrics[$key] = min(self::MAX_COUNT, (int) ($this->metrics[$key] ?? 0) + 1);
|
||||
}
|
||||
|
||||
/** Abort signals also include validation rejection; they do not identify user intent. */
|
||||
public function interrupted(string $trigger): void
|
||||
{
|
||||
if (isset($this->metrics['interruption_detected_ms'])) return;
|
||||
$this->metrics['interruption_detected_ms'] = $this->elapsed();
|
||||
$this->metrics['interruption_trigger'] = in_array($trigger, ['abort_signal', 'callback_error', 'timeout'], true)
|
||||
? $trigger : 'other';
|
||||
}
|
||||
|
||||
/** Call only inside the callable that actually performs stop HTTP, never on lease polls. */
|
||||
public function stopIoBegin(): void
|
||||
{
|
||||
$now = $this->elapsed();
|
||||
$this->metrics['stop_io_begin_ms'] ??= $now;
|
||||
$this->metrics['last_stop_io_begin_ms'] = $now;
|
||||
$this->increment('stop_io_attempt_count');
|
||||
}
|
||||
|
||||
public function stopIoEnd(bool $accepted): void
|
||||
{
|
||||
$now = $this->elapsed();
|
||||
$this->metrics['stop_io_end_ms'] ??= $now;
|
||||
$this->metrics['last_stop_io_end_ms'] = $now;
|
||||
$this->increment('stop_io_completed_count');
|
||||
if ($accepted) {
|
||||
$this->increment('stop_io_accepted_count');
|
||||
$this->stopAccepted('local_io');
|
||||
}
|
||||
}
|
||||
|
||||
/** A peer can confirm acceptance without this transport performing stop I/O. */
|
||||
public function stopAccepted(string $source): void
|
||||
{
|
||||
if (isset($this->metrics['stop_accepted_ms'])) return;
|
||||
$this->metrics['stop_accepted_ms'] = $this->elapsed();
|
||||
$this->metrics['stop_accepted_source'] = in_array($source, ['local_io', 'coordinator'], true) ? $source : 'other';
|
||||
}
|
||||
|
||||
/** The parser calls this only after its bound-task and allowed-status terminal gate. */
|
||||
public function workflowTerminal(string $status): void
|
||||
{
|
||||
if (isset($this->metrics['workflow_terminal_ms']) || !in_array($status, ['stopped', 'succeeded', 'failed'], true)) return;
|
||||
$this->metrics['workflow_terminal_ms'] = $this->elapsed();
|
||||
$this->metrics['workflow_terminal_status'] = $status;
|
||||
}
|
||||
|
||||
/** Natural EOF means the HTTP response body completed; it is not provider/GPU proof. */
|
||||
public function transportReturned(string $outcome): void
|
||||
{
|
||||
if (isset($this->metrics['transport_return_ms'])) return;
|
||||
$this->metrics['transport_return_ms'] = $this->elapsed();
|
||||
$outcome = in_array($outcome, ['natural_eof', 'local_abort', 'timeout', 'transport_error', 'exception'], true)
|
||||
? $outcome : 'other';
|
||||
$this->metrics['transport_exit'] = $outcome;
|
||||
if ($outcome === 'natural_eof') $this->metrics['natural_eof_ms'] = $this->metrics['transport_return_ms'];
|
||||
}
|
||||
|
||||
public function transportClosed(): void { $this->metrics['transport_closed_ms'] ??= $this->elapsed(); }
|
||||
|
||||
public function headers(int $status): void
|
||||
{
|
||||
@@ -46,6 +119,20 @@ final class DifyStreamTiming
|
||||
$now = $this->elapsed();
|
||||
$this->events[$name] ??= ['first_ms' => $now, 'last_ms' => $now, 'count' => 0];
|
||||
++$this->events[$name]['count']; $this->events[$name]['last_ms'] = $now;
|
||||
if ($name === 'node_retry') {
|
||||
$retry = $decoded['data']['retry_index'] ?? null;
|
||||
if (is_numeric($retry) && (float) $retry >= 0 && (float) $retry <= 1000) {
|
||||
$this->metrics['max_node_retry_index'] = max((int) ($this->metrics['max_node_retry_index'] ?? 0), (int) $retry);
|
||||
}
|
||||
}
|
||||
$reasoning = $protocol === 'dify' ? ($decoded['thought'] ?? null)
|
||||
: ($decoded['choices'][0]['delta']['reasoning_content'] ?? null);
|
||||
if (is_string($reasoning) && $reasoning !== '') {
|
||||
$this->metrics['first_reasoning_frame_ms'] ??= $now;
|
||||
$this->metrics['reasoning_frame_count'] = (int) ($this->metrics['reasoning_frame_count'] ?? 0) + 1;
|
||||
$this->metrics['reasoning_frame_bytes'] = min(1000000000,
|
||||
(int) ($this->metrics['reasoning_frame_bytes'] ?? 0) + strlen($reasoning));
|
||||
}
|
||||
if (in_array($name, ['message', 'agent_message', 'choices'], true)) {
|
||||
$answer = $protocol === 'dify' ? ($decoded['answer'] ?? '') : ($decoded['choices'][0]['delta']['content'] ?? '');
|
||||
$visible = is_string($answer) && $answer !== '';
|
||||
@@ -83,6 +170,10 @@ final class DifyStreamTiming
|
||||
}
|
||||
}
|
||||
$usage = $name === 'message_end' ? ($decoded['metadata']['usage'] ?? []) : ($decoded['usage'] ?? []);
|
||||
if (is_array($usage) && !isset($usage['reasoning_tokens'])
|
||||
&& isset($usage['completion_tokens_details']['reasoning_tokens'])) {
|
||||
$usage['reasoning_tokens'] = $usage['completion_tokens_details']['reasoning_tokens'];
|
||||
}
|
||||
foreach (['prompt_tokens', 'completion_tokens', 'total_tokens', 'reasoning_tokens'] as $key) {
|
||||
$value = is_array($usage) ? ($usage[$key] ?? null) : null;
|
||||
if (is_numeric($value) && is_finite((float) $value) && (float) $value >= 0) {
|
||||
|
||||
Reference in New Issue
Block a user