Merge branch 'master' of http://git.zhenyangtang.com.cn/ZYT/zyt
This commit is contained in:
@@ -23,10 +23,12 @@ use app\adminapi\logic\tcm\PatientAiReportLogic;
|
||||
use app\adminapi\logic\tcm\TrackingNoteLogic;
|
||||
use app\adminapi\service\AssistantSseProtocol;
|
||||
use app\adminapi\service\AssistantSseTiming;
|
||||
use app\adminapi\service\AssistantClinicalRejectDiagnostic;
|
||||
use app\adminapi\service\AssistantRequestControl;
|
||||
use app\adminapi\validate\tcm\DiagnosisValidate;
|
||||
use app\common\model\Order;
|
||||
use app\common\model\WechatChatRecord;
|
||||
use app\common\service\DifyWorkflowRunStatus;
|
||||
|
||||
/**
|
||||
* 中医辨房病因诊单控制器
|
||||
@@ -951,12 +953,20 @@ class DiagnosisController extends BaseAdminController
|
||||
$profile = '';
|
||||
$sseStartedAt = hrtime(true);
|
||||
$disconnected = false;
|
||||
$shouldAbort = static function () use (&$control, &$disconnected): bool {
|
||||
$shouldAbort = static function () use (&$control, &$disconnected, &$emit): bool {
|
||||
if (connection_aborted() === 1 && !$disconnected) {
|
||||
$disconnected = true;
|
||||
if ($control !== null) $control->signalCancel();
|
||||
}
|
||||
return $disconnected || ($control !== null && $control->shouldAbort());
|
||||
if ($disconnected || ($control !== null && $control->shouldAbort())) return true;
|
||||
// The cURL wait loop polls here even when the model sends no bytes.
|
||||
// A throttled comment prevents the reverse proxy's idle timeout.
|
||||
if ($emit !== null) $emit('heartbeat', []);
|
||||
if (connection_aborted() === 1 && !$disconnected) {
|
||||
$disconnected = true;
|
||||
if ($control !== null) $control->signalCancel();
|
||||
}
|
||||
return $disconnected;
|
||||
};
|
||||
$options = [
|
||||
'should_abort' => $shouldAbort,
|
||||
@@ -977,24 +987,50 @@ class DiagnosisController extends BaseAdminController
|
||||
'on_progress' => static function (array $progress) use (&$emit): void {
|
||||
if ($emit !== null) $emit('progress', $progress);
|
||||
},
|
||||
'on_clinical_rejection' => static function (string $unit, array $issues) use ($timing): void {
|
||||
try {
|
||||
AssistantClinicalRejectDiagnostic::log($timing, $unit, $issues);
|
||||
} catch (\Throwable $ignored) {
|
||||
// Diagnostic failure must never affect the clinical decision or delivery.
|
||||
}
|
||||
},
|
||||
'can_start_attempt' => static function () use (&$control): bool {
|
||||
return $control !== null && $control->canStartAttempt();
|
||||
},
|
||||
'on_stop' => static function (bool $confirmed, string $taskId = '') use (&$control): void {
|
||||
if ($control !== null && $taskId !== '') $control->markStop($confirmed, $taskId);
|
||||
},
|
||||
'on_transfer_end' => static function (string $taskId = '') use (&$control): void {
|
||||
if ($control !== null) $control->clearTask($taskId !== '' ? $taskId : null);
|
||||
'on_timeout' => static function () use (&$control): void {
|
||||
if ($control !== null) $control->signalTimeout();
|
||||
},
|
||||
'on_cancel_task' => static function (string $taskId, callable $performStop) use (&$control): array {
|
||||
'on_transfer_end' => static function (string $taskId = '', string $submissionId = '', string $disposition = 'unconfirmed') use (&$control): void {
|
||||
if ($control !== null && $submissionId !== '') {
|
||||
$control->closeSubmittedGeneration($submissionId, $taskId, $disposition, [DifyWorkflowRunStatus::class, 'lookup']);
|
||||
} elseif ($control !== null && $taskId !== '') $control->closeGeneration($taskId, [DifyWorkflowRunStatus::class, 'lookup']);
|
||||
},
|
||||
'on_cancel_task' => static function (string $taskId, callable $performStop, string $reason = 'cancelled') use (&$control): array {
|
||||
if ($control === null) return ['accepted' => false, 'pending' => false];
|
||||
$control->signalCancel();
|
||||
if ($control->hasHandoffTask($taskId)) {
|
||||
// bindTask already retained this unexpected new task and
|
||||
// requested cleanup. Do not turn the race into user cancel.
|
||||
return $control->stopBoundTask($performStop, $taskId);
|
||||
}
|
||||
if ($reason === 'timeout') $control->signalTimeout();
|
||||
elseif ($reason === 'validation') $control->signalValidationFailure();
|
||||
else $control->signalCancel();
|
||||
return $control->stopBoundTask($performStop, $taskId);
|
||||
},
|
||||
'on_terminal' => static function (string $taskId, string $status) use (&$control): void {
|
||||
if ($control !== null) $control->markTerminal($taskId, $status);
|
||||
'on_terminal' => static function (string $taskId, string $status, string $workflowRunId = '') use (&$control): void {
|
||||
if ($control !== null) $control->markTerminal($taskId, $status, $workflowRunId);
|
||||
},
|
||||
];
|
||||
$upstreamUser = $options['upstream_user'];
|
||||
$options['on_task'] = static function (string $taskId) use (&$control, &$profile, $upstreamUser): void {
|
||||
if ($control !== null) $control->bindTask($profile, $upstreamUser, $taskId);
|
||||
$options['on_submission'] = static function () use (&$control, &$profile, $upstreamUser): string {
|
||||
if ($control === null) throw new \RuntimeException('Assistant submission control unavailable');
|
||||
return $control->beginSubmission($profile, $upstreamUser);
|
||||
};
|
||||
$options['on_task'] = static function (string $taskId, string $workflowRunId = '', string $submissionId = '') use (&$control, &$profile, $upstreamUser): void {
|
||||
if ($control !== null) $control->bindTask($profile, $upstreamUser, $taskId, $workflowRunId, $submissionId);
|
||||
};
|
||||
try {
|
||||
$prepared = DiagnosisAiLogic::prepareAssistant(
|
||||
@@ -1005,16 +1041,28 @@ class DiagnosisController extends BaseAdminController
|
||||
return $this->fail(DiagnosisAiLogic::getError());
|
||||
}
|
||||
if ($prepared === null) {
|
||||
$this->finishAssistantError($emit, $control, $shouldAbort());
|
||||
$cancelled = $shouldAbort();
|
||||
$this->logAssistantGenerationDrain($timing, AssistantSseTiming::elapsedMs($sseStartedAt));
|
||||
$this->finishAssistantError($emit, $control, $cancelled);
|
||||
$timing->log('sse_done', ['duration_ms' => AssistantSseTiming::elapsedMs($sseStartedAt),
|
||||
'status' => $cancelled ? 'cancelled' : 'error']);
|
||||
exit;
|
||||
}
|
||||
$this->runAssistantSse($prepared, $timing, $emit, $shouldAbort, $options, $control, $sseStartedAt);
|
||||
} catch (\Throwable $e) {
|
||||
\think\facade\Log::warning('diagnosis ai assistant sse failed ' . json_encode(['trace_id' => $timing->id()]));
|
||||
$this->logAssistantGenerationDrain($timing, AssistantSseTiming::elapsedMs($sseStartedAt));
|
||||
if ($emit === null) {
|
||||
$timing->log('sse_done', ['duration_ms' => AssistantSseTiming::elapsedMs($sseStartedAt), 'status' => 'exception']);
|
||||
return $this->fail('AI 助手暂时不可用,请稍后重试');
|
||||
}
|
||||
try {
|
||||
if ($control !== null) $control->finish('error');
|
||||
$this->reconcileClosedAssistant($control);
|
||||
} catch (\Throwable $ignored) {}
|
||||
$timing->log('sse_done', ['duration_ms' => AssistantSseTiming::elapsedMs($sseStartedAt), 'status' => 'exception']);
|
||||
if ($emit === null) return $this->fail('AI 助手暂时不可用,请稍后重试');
|
||||
try { if ($control !== null) $control->finish('error'); } catch (\Throwable $ignored) {}
|
||||
$emit('error', ['code' => 'AI_ASSISTANT_FAILED', 'message' => 'AI 助手暂时不可用,请稍后重试']);
|
||||
$emit('error', ['code' => 'AI_ASSISTANT_FAILED', 'message' => 'AI 助手暂时不可用,请稍后重试']
|
||||
+ $this->safeAssistantState($control));
|
||||
exit;
|
||||
}
|
||||
}
|
||||
@@ -1043,24 +1091,57 @@ class DiagnosisController extends BaseAdminController
|
||||
$protocol = new AssistantSseProtocol();
|
||||
return function (string $event, array $payload) use ($protocol): bool {
|
||||
if ($protocol->isTerminal() || connection_aborted()) return false;
|
||||
$encoded = $protocol->encode($event, $payload);
|
||||
if ($encoded === null) return false;
|
||||
$encoded = $event === 'heartbeat' ? $protocol->heartbeat() : $protocol->encode($event, $payload);
|
||||
if ($encoded === null) return $event === 'heartbeat';
|
||||
echo $encoded;
|
||||
$this->flushSseOutput();
|
||||
return !connection_aborted();
|
||||
};
|
||||
}
|
||||
|
||||
private function finishAssistantError(callable $emit, ?AssistantRequestControl $control, bool $cancelled): void
|
||||
private function finishAssistantError(callable $emit, ?AssistantRequestControl $control, bool $cancelled, bool $recover = true): void
|
||||
{
|
||||
if ($control !== null) $control->finish($cancelled ? 'cancelled' : 'error');
|
||||
if ($recover) $this->reconcileClosedAssistant($control);
|
||||
$state = $this->safeAssistantState($control);
|
||||
if ($cancelled) {
|
||||
$emit('progress', ['phase' => 'cancelled', 'status' => 'cancelled', 'message' => '本次请求已取消']
|
||||
+ ($control !== null ? $control->publicState() : []));
|
||||
+ $state);
|
||||
}
|
||||
$emit('error', ['code' => $cancelled ? 'AI_ASSISTANT_CANCELLED' : DiagnosisAiLogic::getAssistantErrorCode(),
|
||||
'message' => $cancelled ? '本次请求已取消' : DiagnosisAiLogic::getError()]
|
||||
+ ($cancelled ? [] : DiagnosisAiLogic::getAssistantErrorMeta()));
|
||||
+ ($cancelled ? [] : DiagnosisAiLogic::getAssistantErrorMeta()) + $state);
|
||||
}
|
||||
|
||||
/** Invoked only after finish; never inside cURL callbacks or the cancel endpoint. */
|
||||
private function reconcileClosedAssistant(?AssistantRequestControl $control): void
|
||||
{
|
||||
if ($control === null) return;
|
||||
try {
|
||||
$recovery = $control->reconcileClosedTasks([DifyWorkflowRunStatus::class, 'lookup']);
|
||||
// Separate from generation/metadata-drain timing; aggregates contain no IDs or bodies.
|
||||
\think\facade\Log::write('diagnosis ai assistant terminal recovery ' . json_encode($recovery), 'info');
|
||||
} catch (\Throwable $ignored) {
|
||||
// Recovery failure must not replace the original report/error outcome.
|
||||
}
|
||||
}
|
||||
|
||||
private function safeAssistantState(?AssistantRequestControl $control): array
|
||||
{
|
||||
try {
|
||||
return $control !== null ? $control->publicState() : [];
|
||||
} catch (\Throwable $ignored) {
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
private function logAssistantGenerationDrain(AssistantSseTiming $timing, int $durationMs): void
|
||||
{
|
||||
try {
|
||||
\think\facade\Log::write('diagnosis ai assistant generation and drain ' . json_encode([
|
||||
'trace_id' => $timing->id(), 'duration_ms' => $durationMs,
|
||||
]), 'info');
|
||||
} catch (\Throwable $ignored) {}
|
||||
}
|
||||
|
||||
/** @param array<string,mixed> $prepared */
|
||||
@@ -1083,14 +1164,18 @@ class DiagnosisController extends BaseAdminController
|
||||
$shouldAbort, $timing, $options
|
||||
);
|
||||
$cancelled = $shouldAbort();
|
||||
// Preserve generation plus its existing metadata drain before independent recovery.
|
||||
$this->logAssistantGenerationDrain($timing, AssistantSseTiming::elapsedMs($sseStartedAt));
|
||||
if ($result === null || $cancelled) {
|
||||
$this->finishAssistantError($emit, $control, $cancelled);
|
||||
} else {
|
||||
$control->finish('done');
|
||||
if ($control->publicState()['status'] === 'cancelled') {
|
||||
$this->finishAssistantError($emit, $control, true);
|
||||
$this->reconcileClosedAssistant($control);
|
||||
$state = $this->safeAssistantState($control);
|
||||
if (($state['status'] ?? '') === 'cancelled') {
|
||||
$this->finishAssistantError($emit, $control, true, false);
|
||||
} else {
|
||||
$emit('done', $result);
|
||||
$emit('done', $result + $state);
|
||||
}
|
||||
}
|
||||
$timing->log('sse_done', ['duration_ms' => AssistantSseTiming::elapsedMs($sseStartedAt),
|
||||
|
||||
Reference in New Issue
Block a user