ZYT/zyt
2
0
Fork 0
Code Issues Actions Packages Projects Releases Wiki Activity

更新

Browse Source
This commit is contained in:
Your Name
2026-08-27 15:41:09 +08:00
parent 74ff568ba4
commit 75e214dc08
6 changed files with 303 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/app/common/service/DifyChatService.php
+201 -104
View File
@@ -79,58 +79,78 @@ class DifyChatService
$formatted = null;
foreach (self::buildAttemptPlan($normalized['kept'], $normalized['dropped']) as $attempt) {
$requestSpecs = self::buildRequestSpecs(
$baseUrl,
$model,
$inputs,
$query,
$user,
false,
$attempt['files'],
$attempt['omitted']
);
$lastResponse = null;
$lastSpec = [];
// 协议回退会把最初的“附件被拒”换成另一协议的状态码，因此降级判断
// 必须记住本轮出现过的附件拒绝信号，而不能只看最后一次响应。
$fileRejected = false;
$inputRejected = false;
foreach ($requestSpecs as $index => $requestSpec) {
$elapsedSeconds = (int) floor(microtime(true) - $startedAt);
$remainingTimeout = $timeout - $elapsedSeconds;
if ($remainingTimeout < self::MIN_TIMEOUT) {
return self::error(
'UPSTREAM_TIMEOUT',
'模型响应超时，请稍后重试',
self::elapsedMilliseconds($startedAt)
foreach (self::buildInputAttemptPlan($inputs) as $inputIndex => $attemptInputs) {
if ($inputIndex > 0 && !$inputRejected) {
break;
}
$requestSpecs = self::buildRequestSpecs(
$baseUrl,
$model,
$attemptInputs,
$query,
$user,
false,
$attempt['files'],
$attempt['omitted']
);
$lastResponse = null;
$lastSpec = [];
foreach ($requestSpecs as $index => $requestSpec) {
$elapsedSeconds = (int) floor(microtime(true) - $startedAt);
$remainingTimeout = $timeout - $elapsedSeconds;
if ($remainingTimeout < self::MIN_TIMEOUT) {
return self::error(
'UPSTREAM_TIMEOUT',
'模型响应超时，请稍后重试',
self::elapsedMilliseconds($startedAt)
);
}
$response = self::sendRequest(
$requestSpec['url'],
$requestSpec['payload'],
$apiKey,
$remainingTimeout
);
$lastResponse = $response;
$lastSpec = $requestSpec;
$fileRejected = $fileRejected || self::isFileRejection($response, $attempt['files']);
$inputRejected = $inputRejected || self::isInputRejection(
$response,
$requestSpec,
$attemptInputs
);
// /v1 在两种协议中都是合法基址。仅在明确表示路径不存在时尝试另一协议，
// 避免因业务参数错误而重复提交同一份临床数据。
$hasFallback = isset($requestSpecs[$index + 1]);
if (
$hasFallback
&& !$inputRejected
&& self::shouldTryNextProtocol($response, false)
) {
continue;
}
break;
}
$response = self::sendRequest(
$requestSpec['url'],
$requestSpec['payload'],
$apiKey,
$remainingTimeout
);
$lastResponse = $response;
$lastSpec = $requestSpec;
$fileRejected = $fileRejected || self::isFileRejection($response, $attempt['files']);
// /v1 在两种协议中都是合法基址。仅在明确表示路径不存在时尝试另一协议，
// 避免因业务参数错误而重复提交同一份临床数据。
$hasFallback = isset($requestSpecs[$index + 1]);
if ($hasFallback && self::shouldTryNextProtocol($response, false)) {
$lastResponse = $lastResponse ?? ['body' => '', 'errno' => 0, 'http_code' => 0];
$formatted = self::formatResponse($lastResponse, $startedAt);
self::logUpstreamFailure($lastSpec, $lastResponse, $query, $attempt['files'], $formatted);
if (!empty($formatted['ok'])) {
return $formatted;
}
// Dify 只接受应用中已声明且满足长度约束的 inputs。病例正文已经完整
// 放在 query 中，因此 invalid_param 时可安全地用空 inputs 重试一次。
if ($inputIndex === 0 && $inputRejected) {
continue;
}
break;
}
$lastResponse = $lastResponse ?? ['body' => '', 'errno' => 0, 'http_code' => 0];
$formatted = self::formatResponse($lastResponse, $startedAt);
self::logUpstreamFailure($lastSpec, $lastResponse, $query, $attempt['files'], $formatted);
if (!empty($formatted['ok'])) {
return $formatted;
}
// 附件整体被拒时退回纯文本重试，附件清单已在下一轮尝试中补齐。
if (!$fileRejected) {
return $formatted;
@@ -195,63 +215,82 @@ class DifyChatService
$formatted = null;
foreach (self::buildAttemptPlan($normalized['kept'], $normalized['dropped']) as $attempt) {
$requestSpecs = self::buildRequestSpecs(
$baseUrl,
$model,
$inputs,
$query,
$user,
true,
$attempt['files'],
$attempt['omitted']
);
$lastResponse = null;
$lastSpec = [];
// 协议回退会把最初的“附件被拒”换成另一协议的状态码，因此降级判断
// 必须记住本轮出现过的附件拒绝信号，而不能只看最后一次响应。
$fileRejected = false;
$inputRejected = false;
foreach ($requestSpecs as $index => $requestSpec) {
$elapsedSeconds = (int) floor(microtime(true) - $startedAt);
$remainingTimeout = $timeout - $elapsedSeconds;
if ($remainingTimeout < self::MIN_TIMEOUT) {
return self::error(
'UPSTREAM_TIMEOUT',
'模型响应超时，请稍后重试',
self::elapsedMilliseconds($startedAt)
foreach (self::buildInputAttemptPlan($inputs) as $inputIndex => $attemptInputs) {
if ($inputIndex > 0 && !$inputRejected) {
break;
}
$requestSpecs = self::buildRequestSpecs(
$baseUrl,
$model,
$attemptInputs,
$query,
$user,
true,
$attempt['files'],
$attempt['omitted']
);
$lastResponse = null;
$lastSpec = [];
foreach ($requestSpecs as $index => $requestSpec) {
$elapsedSeconds = (int) floor(microtime(true) - $startedAt);
$remainingTimeout = $timeout - $elapsedSeconds;
if ($remainingTimeout < self::MIN_TIMEOUT) {
return self::error(
'UPSTREAM_TIMEOUT',
'模型响应超时，请稍后重试',
self::elapsedMilliseconds($startedAt)
);
}
$response = self::sendStreamRequest(
$requestSpec['protocol'],
$requestSpec['url'],
$requestSpec['payload'],
$apiKey,
$remainingTimeout,
$onDelta,
$shouldAbort
);
$lastResponse = $response;
$lastSpec = $requestSpec;
$fileRejected = $fileRejected || self::isFileRejection($response, $attempt['files']);
$inputRejected = $inputRejected || self::isInputRejection(
$response,
$requestSpec,
$attemptInputs
);
// 只在尚未向下游发送任何文本、且明确为路径不支持时尝试另一协议。
$hasFallback = isset($requestSpecs[$index + 1]);
if (
$hasFallback
&& !$inputRejected
&& self::shouldTryNextProtocol($response, true)
) {
continue;
}
break;
}
$response = self::sendStreamRequest(
$requestSpec['protocol'],
$requestSpec['url'],
$requestSpec['payload'],
$apiKey,
$remainingTimeout,
$onDelta,
$shouldAbort
);
$lastResponse = $response;
$lastSpec = $requestSpec;
$fileRejected = $fileRejected || self::isFileRejection($response, $attempt['files']);
// 只在尚未向下游发送任何文本、且明确为路径不支持时尝试另一协议。
$hasFallback = isset($requestSpecs[$index + 1]);
$lastResponse = $lastResponse ?? self::emptyStreamResponse(0);
$formatted = self::formatStreamResponse($lastResponse, $startedAt);
self::logUpstreamFailure($lastSpec, $lastResponse, $query, $attempt['files'], $formatted);
if (!empty($formatted['ok'])) {
return $formatted;
}
if (
$hasFallback
&& self::shouldTryNextProtocol($response, true)
$inputIndex === 0
&& $inputRejected
&& empty($lastResponse['emitted'])
) {
continue;
}
break;
}
$lastResponse = $lastResponse ?? self::emptyStreamResponse(0);
$formatted = self::formatStreamResponse($lastResponse, $startedAt);
self::logUpstreamFailure($lastSpec, $lastResponse, $query, $attempt['files'], $formatted);
if (!empty($formatted['ok'])) {
return $formatted;
}
// 已经推给医生的文本不能重复输出，因此只在一个字都没发出去时才降级重试。
if (!empty($lastResponse['emitted']) || !$fileRejected) {
return $formatted;
@@ -454,6 +493,66 @@ class DifyChatService
return $attempts;
}
/**
* Dify 应用输入变量由发布时的表单定义决定。先保留结构化输入；若上游明确
* 拒绝输入，再使用空对象兼容旧应用。病例正文始终在 query 中，不会丢失。
*
* @param array<string,mixed> $inputs
* @return array<int,array<string,mixed>>
*/
private static function buildInputAttemptPlan(array $inputs): array
{
return $inputs === [] ? [[]] : [$inputs, []];
}
/**
* @param array<string,mixed> $response
* @param array<string,mixed> $requestSpec
* @param array<string,mixed> $inputs
*/
private static function isInputRejection(array $response, array $requestSpec, array $inputs): bool
{
if (
$inputs === []
|| ($requestSpec['protocol'] ?? '') !== 'dify'
|| (int) ($response['errno'] ?? 0) !== 0
|| !empty($response['emitted'])
) {
return false;
}
$httpCode = (int) ($response['http_code'] ?? 0);
if (in_array($httpCode, [413, 422], true)) {
return true;
}
$upstreamCode = strtolower(self::responseUpstreamCode($response));
if ($httpCode === 400) {
// 标准 Dify 会给出 invalid_param；部分兼容网关只保留 400，因此空码
// 也允许一次无 inputs 重试。额度、模型或应用状态错误不能重复提交。
return $upstreamCode === ''
|| in_array($upstreamCode, ['invalid_param', 'payload_too_large', 'request_too_large'], true);
}
return !empty($response['upstream_error'])
&& in_array(
$upstreamCode,
['invalid_param', 'payload_too_large', 'request_too_large'],
true
);
}
/** @param array<string,mixed> $response */
private static function responseUpstreamCode(array $response): string
{
$upstreamCode = self::cleanUpstreamCode($response['upstream_code'] ?? '');
if ($upstreamCode !== '' || !isset($response['body'])) {
return $upstreamCode;
}
$decoded = json_decode((string) $response['body'], true);
return is_array($decoded) ? self::cleanUpstreamCode($decoded['code'] ?? '') : '';
}
/**
* @param array<int,array<string,string>> $files
*/
@@ -535,10 +634,10 @@ class DifyChatService
/**
* Decide whether an ambiguous base URL should be tried with the other wire
* protocol. A 400/415/422 response cannot have started generation, and a
* 2xx stream with no delivered delta but no valid terminal frame is also
* safe to retry. Authentication, rate-limit and server failures retain
* their original diagnosis instead of being hidden by a second request.
* protocol. Only a missing/unsupported endpoint is a blocking-mode protocol
* signal. A 2xx stream with no delivered delta but no valid terminal frame
* is also safe to retry. Business validation, authentication, rate-limit and
* server failures retain their original diagnosis instead of being hidden.
*
* @param array<string,mixed> $response
*/
@@ -548,7 +647,7 @@ class DifyChatService
return false;
}
$httpCode = (int) ($response['http_code'] ?? 0);
if (in_array($httpCode, [400, 404, 405, 415, 422, 501], true)) {
if (in_array($httpCode, [404, 405, 501], true)) {
return true;
}
if (!$streaming || $httpCode < 200 || $httpCode >= 300 || !empty($response['emitted'])) {
@@ -1092,25 +1191,23 @@ class DifyChatService
return;
}
$url = (string) ($requestSpec['url'] ?? '');
$upstreamCode = (string) ($response['upstream_code'] ?? '');
if ($upstreamCode === '' && isset($response['body'])) {
$decoded = json_decode((string) $response['body'], true);
$upstreamCode = is_array($decoded)
? self::cleanUpstreamCode($decoded['code'] ?? '')
: '';
}
Log::warning('prescription ai upstream request failed', [
$context = [
'protocol' => (string) ($requestSpec['protocol'] ?? ''),
'endpoint_path' => (string) (parse_url($url, PHP_URL_PATH) ?? ''),
'http_code' => (int) ($response['http_code'] ?? 0),
'curl_errno' => (int) ($response['errno'] ?? 0),
// 上游自有错误码（如 invalid_param），用于区分附件超限、鉴权、模型故障。
'upstream_code' => $upstreamCode,
'upstream_code' => self::responseUpstreamCode($response),
'query_bytes' => strlen($query),
'file_count' => count($files),
'error_code' => (string) ($formatted['error_code'] ?? 'UNKNOWN'),
'latency_ms' => (int) ($formatted['latency_ms'] ?? 0),
]);
];
// ThinkPHP 文件日志不会自动输出未参与占位符替换的 context；显式序列化
// 这组不含凭据、主机名、患者正文的诊断字段，确保线上日志真正可用。
Log::warning(
'prescription ai upstream request failed ' . json_encode($context, JSON_UNESCAPED_SLASHES)
);
}
/**