This commit is contained in:
Your Name
2026-10-09 14:13:57 +08:00
parent 51d8fd409e
commit 5706e9a6a5
15 changed files with 732 additions and 162 deletions
@@ -19,7 +19,6 @@ from doctor_workstation.core.errors import (
ApiBusinessError, ApiBusinessError,
ApiHttpError, ApiHttpError,
ApiProtocolError, ApiProtocolError,
ApiTimeoutError,
ApiTransportError, ApiTransportError,
AuthenticationExpiredError, AuthenticationExpiredError,
) )
@@ -1413,11 +1412,11 @@ class RemoteDoctorRepository:
task: str = "custom", task: str = "custom",
cancelled: Callable[[], bool] | None = None, cancelled: Callable[[], bool] | None = None,
) -> Iterator[dict[str, Any]]: ) -> Iterator[dict[str, Any]]:
"""Stream a diagnosis answer, with one legacy fallback before first content.""" """Stream a diagnosis answer; use the legacy route only when streaming is absent."""
clean_prompt, clean_task = _diagnosis_ai_request(diagnosis_id, prompt, task) clean_prompt, clean_task = _diagnosis_ai_request(diagnosis_id, prompt, task)
body = {"id": diagnosis_id, "prompt": clean_prompt, "task": clean_task} body = {"id": diagnosis_id, "prompt": clean_prompt, "task": clean_task}
received_delta = False received_event = False
received_done = False received_done = False
try: try:
for raw_event in self.client.post_event_stream( for raw_event in self.client.post_event_stream(
@@ -1428,13 +1427,12 @@ class RemoteDoctorRepository:
): ):
if cancelled is not None and cancelled(): if cancelled is not None and cancelled():
return return
received_event = True
event = _normalise_diagnosis_ai_event(raw_event) event = _normalise_diagnosis_ai_event(raw_event)
if event is None: if event is None:
continue continue
kind = event["event"] kind = event["event"]
if kind == "delta": if kind == "done":
received_delta = True
elif kind == "done":
received_done = True received_done = True
yield event yield event
if kind == "done": if kind == "done":
@@ -1443,12 +1441,16 @@ class RemoteDoctorRepository:
return return
if not received_done: if not received_done:
raise ApiProtocolError("AI assistant stream ended before done") raise ApiProtocolError("AI assistant stream ended before done")
except (ApiHttpError, ApiProtocolError, ApiTimeoutError, ApiTransportError): except ApiHttpError as error:
if received_delta or (cancelled is not None and cancelled()): if (
error.status_code not in {404, 405}
or received_event
or (cancelled is not None and cancelled())
):
raise raise
# Older deployments do not expose the stream route. Submit exactly one # Only a missing stream route on an older deployment justifies another
# request through the confirmed non-streaming endpoint in that case. # full request. Timeouts, transport failures and malformed streams do not.
result = self.analyze_diagnosis_ai( result = self.analyze_diagnosis_ai(
diagnosis_id, diagnosis_id,
clean_prompt, clean_prompt,
@@ -3146,6 +3146,8 @@ class _AiStreamWorker(QRunnable):
@Slot() @Slot()
def run(self) -> None: def run(self) -> None:
try: try:
if self.is_cancelled():
return
stream = getattr(self.repository, "stream_diagnosis_ai", None) stream = getattr(self.repository, "stream_diagnosis_ai", None)
if callable(stream): if callable(stream):
events = stream( events = stream(
@@ -3169,6 +3171,8 @@ class _AiStreamWorker(QRunnable):
{"event": "delta", "text": answer, "fallback": True}, {"event": "delta", "text": answer, "fallback": True},
{**payload, "event": "done", "fallback": True}, {**payload, "event": "done", "fallback": True},
) )
if self.is_cancelled():
return
for event in events: for event in events:
if self.is_cancelled(): if self.is_cancelled():
return return
@@ -3207,13 +3211,18 @@ class _ClickCard(QFrame):
AI_CONTEXT_MAX_CHARS = 320 AI_CONTEXT_MAX_CHARS = 320
AI_PROMPT_LIMIT = 500 AI_PROMPT_LIMIT = 500
#: 首个增量到达前显示的占位文案,避免留下一块没有任何说明的空白气泡。 #: 等待服务端整理诊单资料时显示的占位文案。
AI_STREAM_PENDING_TEXT = "正在生成…" AI_STREAM_PENDING_TEXT = "正在整理诊单资料…"
#: 服务端确认资料已准备好、尚未收到回答时显示的占位文案。
AI_STREAM_WAITING_TEXT = "资料已整理,正在等待 AI 回复…"
#: 连接结束但一个字都没收到时的兜底文案。 #: 连接结束但一个字都没收到时的兜底文案。
AI_STREAM_SILENT_TEXT = ( AI_STREAM_SILENT_TEXT = (
"AI 助手没有返回内容,可能是服务端未响应或连接中断,请稍后重试。" "AI 助手没有返回内容,可能是服务端未响应或连接中断,请稍后重试。"
) )
AI_STREAM_TOTAL_TIMEOUT_MS = 180_000
AI_STREAM_TIMEOUT_TEXT = "等待 AI 回复超时,已停止本次等待,请稍后重试。"
AI_CONTEXT_SEPARATOR = "\n\n— 医生提问 —\n" AI_CONTEXT_SEPARATOR = "\n\n— 医生提问 —\n"
@@ -3534,6 +3543,9 @@ class AiConsultDialog(QDialog):
self._flush_timer.setSingleShot(True) self._flush_timer.setSingleShot(True)
self._flush_timer.setInterval(40) self._flush_timer.setInterval(40)
self._flush_timer.timeout.connect(self._flush_stream_chunks) self._flush_timer.timeout.connect(self._flush_stream_chunks)
self._stream_timeout_timer = QTimer(self)
self._stream_timeout_timer.setSingleShot(True)
self._stream_timeout_timer.timeout.connect(self._stream_timed_out)
self.setObjectName("AiConsultDialog") self.setObjectName("AiConsultDialog")
self.setWindowTitle("问诊详情") self.setWindowTitle("问诊详情")
# QDialog 默认只带关闭按钮,医生无法把这个信息密度很高的窗口放大到整屏。 # QDialog 默认只带关闭按钮,医生无法把这个信息密度很高的窗口放大到整屏。
@@ -6028,6 +6040,7 @@ class AiConsultDialog(QDialog):
worker.signals.finished.connect( worker.signals.finished.connect(
lambda: self._stream_finished(generation, stream_generation, worker) lambda: self._stream_finished(generation, stream_generation, worker)
) )
self._stream_timeout_timer.start(AI_STREAM_TOTAL_TIMEOUT_MS)
QThreadPool.globalInstance().start(worker) QThreadPool.globalInstance().start(worker)
QTimer.singleShot(0, self._scroll_chat_to_bottom) QTimer.singleShot(0, self._scroll_chat_to_bottom)
@@ -6046,6 +6059,8 @@ class AiConsultDialog(QDialog):
kind = str(payload.get("event") or "").lower() kind = str(payload.get("event") or "").lower()
if kind == "start": if kind == "start":
self._stream_meta.update(payload) self._stream_meta.update(payload)
if self._stream_bubble is not None and not self._stream_text and not self._pending_chunks:
self._stream_bubble.set_payload(AI_STREAM_WAITING_TEXT)
return return
if kind == "delta": if kind == "delta":
chunk = payload.get("text") chunk = payload.get("text")
@@ -6058,6 +6073,7 @@ class AiConsultDialog(QDialog):
return return
self._stream_meta.update(payload) self._stream_meta.update(payload)
self._stream_completed = True self._stream_completed = True
self._stream_timeout_timer.stop()
if not self._stream_text and not self._pending_chunks: if not self._stream_text and not self._pending_chunks:
answer = first_value(payload, "answer", "content", default="") answer = first_value(payload, "answer", "content", default="")
if answer not in (None, ""): if answer not in (None, ""):
@@ -6094,6 +6110,7 @@ class AiConsultDialog(QDialog):
) -> None: ) -> None:
if not self._stream_is_current(generation, stream_generation): if not self._stream_is_current(generation, stream_generation):
return return
self._stream_timeout_timer.stop()
self._flush_stream_chunks() self._flush_stream_chunks()
message = friendly_error(error) message = friendly_error(error)
if self._stream_text: if self._stream_text:
@@ -6112,10 +6129,11 @@ class AiConsultDialog(QDialog):
) -> None: ) -> None:
if not self._stream_is_current(generation, stream_generation): if not self._stream_is_current(generation, stream_generation):
return return
self._stream_timeout_timer.stop()
if self._stream_worker is worker: if self._stream_worker is worker:
self._stream_worker = None self._stream_worker = None
# 连接结束却既没有 done 也没有报错时(服务端静默断开),占位气泡会永远 # 连接结束却既没有 done 也没有报错时(服务端静默断开),
# 停在“正在生成…”。这里补一条明确说明,而不是留一块空白卡片。 # 将阶段占位文案替换成明确说明。
if ( if (
not self._stream_completed not self._stream_completed
and not self._stream_text and not self._stream_text
@@ -6127,9 +6145,23 @@ class AiConsultDialog(QDialog):
self._asking = False self._asking = False
self.send_button.setEnabled(True) self.send_button.setEnabled(True)
def _stream_timed_out(self) -> None:
if not self._asking or self._stream_completed:
return
self._flush_stream_chunks()
bubble = self._stream_bubble
text = self._stream_text
self._cancel_stream()
if bubble is not None:
bubble.set_payload(
f"{text}\n\n> {AI_STREAM_TIMEOUT_TEXT}" if text else AI_STREAM_TIMEOUT_TEXT
)
bubble.set_time_text(datetime.now().strftime("%H:%M"))
def _cancel_stream(self) -> None: def _cancel_stream(self) -> None:
self._stream_generation += 1 self._stream_generation += 1
self._flush_timer.stop() self._flush_timer.stop()
self._stream_timeout_timer.stop()
if self._stream_worker is not None: if self._stream_worker is not None:
self._stream_worker.cancel() self._stream_worker.cancel()
self._stream_worker = None self._stream_worker = None
+63
View File
@@ -1491,6 +1491,55 @@ def test_pending_and_silently_closed_streams_never_show_a_blank_bubble(
dialog.close() dialog.close()
def test_stalled_stream_exits_pending_state_and_ignores_late_events(
application: QApplication,
monkeypatch: pytest.MonkeyPatch,
) -> None:
dialog = _silent_dialog(monkeypatch)
dialog.show()
application.processEvents()
dialog._ask("请总结当前病情")
generation, stream_generation = dialog._generation, dialog._stream_generation
bubble = dialog._stream_bubble
worker = dialog._stream_worker
assert bubble is not None and worker is not None
assert dialog._stream_timeout_timer.isActive()
dialog._stream_timed_out()
assert worker.is_cancelled()
assert not dialog._stream_timeout_timer.isActive()
assert not dialog._asking
assert dialog.send_button.isEnabled()
assert bubble._raw_payload == ai_consult_module.AI_STREAM_TIMEOUT_TEXT
calls: list[int] = []
monkeypatch.setattr(
dialog.repository,
"stream_diagnosis_ai",
lambda *args, **kwargs: calls.append(1),
raising=False,
)
worker.run()
assert calls == []
dialog._stream_event(generation, stream_generation, {"event": "delta", "text": "迟到回复"})
dialog._stream_finished(generation, stream_generation, worker)
assert bubble._raw_payload == ai_consult_module.AI_STREAM_TIMEOUT_TEXT
dialog._ask("请补充说明")
second_bubble = dialog._stream_bubble
assert second_bubble is not None
dialog._stream_event(
dialog._generation,
dialog._stream_generation,
{"event": "delta", "text": "已收到部分分析"},
)
dialog._stream_timed_out()
assert second_bubble._raw_payload.startswith("已收到部分分析\n\n>")
assert ai_consult_module.AI_STREAM_TIMEOUT_TEXT in second_bubble._raw_payload
dialog.close()
def test_answered_stream_replaces_the_pending_placeholder( def test_answered_stream_replaces_the_pending_placeholder(
application: QApplication, application: QApplication,
monkeypatch: pytest.MonkeyPatch, monkeypatch: pytest.MonkeyPatch,
@@ -1501,15 +1550,29 @@ def test_answered_stream_replaces_the_pending_placeholder(
dialog._ask("请总结当前病情") dialog._ask("请总结当前病情")
generation, stream_generation = dialog._generation, dialog._stream_generation generation, stream_generation = dialog._generation, dialog._stream_generation
assert dialog._stream_bubble is not None
assert dialog._stream_bubble._raw_payload == ai_consult_module.AI_STREAM_PENDING_TEXT
dialog._stream_event(
generation,
stream_generation,
{"event": "start", "source_summary": {"patient_name": "敏感姓名"}},
)
assert dialog._stream_bubble._raw_payload == ai_consult_module.AI_STREAM_WAITING_TEXT
assert "敏感姓名" not in dialog._stream_bubble._raw_payload
dialog._stream_event(generation, stream_generation, {"event": "delta", "text": "证候:"}) dialog._stream_event(generation, stream_generation, {"event": "delta", "text": "证候:"})
dialog._flush_timer.stop()
dialog._flush_stream_chunks()
assert dialog._stream_bubble._raw_payload == "证候:"
dialog._stream_event(generation, stream_generation, {"event": "delta", "text": "脾肾两虚"}) dialog._stream_event(generation, stream_generation, {"event": "delta", "text": "脾肾两虚"})
dialog._stream_event(generation, stream_generation, {"event": "done", "model_label": "千问"}) dialog._stream_event(generation, stream_generation, {"event": "done", "model_label": "千问"})
dialog._stream_finished(generation, stream_generation, dialog._stream_worker) dialog._stream_finished(generation, stream_generation, dialog._stream_worker)
application.processEvents() application.processEvents()
assert dialog._stream_text == "证候:脾肾两虚" assert dialog._stream_text == "证候:脾肾两虚"
assert not dialog._stream_timeout_timer.isActive()
texts = _bubble_texts(dialog) texts = _bubble_texts(dialog)
assert not any(ai_consult_module.AI_STREAM_PENDING_TEXT in text for text in texts) assert not any(ai_consult_module.AI_STREAM_PENDING_TEXT in text for text in texts)
assert not any(ai_consult_module.AI_STREAM_WAITING_TEXT in text for text in texts)
assert not any(ai_consult_module.AI_STREAM_SILENT_TEXT in text for text in texts) assert not any(ai_consult_module.AI_STREAM_SILENT_TEXT in text for text in texts)
dialog.close() dialog.close()
+61 -3
View File
@@ -8,7 +8,13 @@ from typing import Any
import pytest import pytest
from doctor_workstation.core.errors import ApiBusinessError, ApiHttpError, ApiProtocolError from doctor_workstation.core.errors import (
ApiBusinessError,
ApiHttpError,
ApiProtocolError,
ApiTimeoutError,
ApiTransportError,
)
from doctor_workstation.core.models import Appointment, Consultation, PageResult, Prescription from doctor_workstation.core.models import Appointment, Consultation, PageResult, Prescription
from doctor_workstation.services.mock_repository import DemoDoctorRepository from doctor_workstation.services.mock_repository import DemoDoctorRepository
from doctor_workstation.services.repository import ( from doctor_workstation.services.repository import (
@@ -774,10 +780,11 @@ def test_remote_diagnosis_ai_stream_normalises_chunks_in_order() -> None:
assert client.post_calls == [] assert client.post_calls == []
def test_remote_diagnosis_ai_stream_falls_back_once_but_not_for_error_event() -> None: @pytest.mark.parametrize("status_code", [404, 405])
def test_remote_diagnosis_ai_stream_falls_back_for_missing_route(status_code: int) -> None:
class MissingStreamClient(RecordingClient): class MissingStreamClient(RecordingClient):
def post_event_stream(self, *args: Any, **kwargs: Any): def post_event_stream(self, *args: Any, **kwargs: Any):
raise ApiHttpError("missing", status_code=404) raise ApiHttpError("missing", status_code=status_code)
missing_client = MissingStreamClient() missing_client = MissingStreamClient()
events = list( events = list(
@@ -789,6 +796,57 @@ def test_remote_diagnosis_ai_stream_falls_back_once_but_not_for_error_event() ->
"tcm.diagnosis/aiAssistant" "tcm.diagnosis/aiAssistant"
] ]
@pytest.mark.parametrize(
"stream_error",
[
ApiHttpError("server error", status_code=500),
ApiTimeoutError("timed out"),
ApiTransportError("disconnected"),
ApiProtocolError("invalid event stream"),
],
)
def test_remote_diagnosis_ai_stream_failures_do_not_issue_full_request(
stream_error: Exception,
) -> None:
class FailedStreamClient(RecordingClient):
def post_event_stream(self, *args: Any, **kwargs: Any):
raise stream_error
client = FailedStreamClient()
with pytest.raises(type(stream_error)):
list(RemoteDoctorRepository(client).stream_diagnosis_ai(501, "请分析"))
assert client.post_calls == []
@pytest.mark.parametrize("after_start", [False, True])
def test_remote_diagnosis_ai_stream_incomplete_response_does_not_fall_back(
after_start: bool,
) -> None:
class IncompleteStreamClient(RecordingClient):
def post_event_stream(self, *args: Any, **kwargs: Any):
if after_start:
yield {"event": "start", "data": {"model_key": "qwen"}}
client = IncompleteStreamClient()
with pytest.raises(ApiProtocolError, match="ended before done"):
list(RemoteDoctorRepository(client).stream_diagnosis_ai(501, "请分析"))
assert client.post_calls == []
def test_remote_diagnosis_ai_stream_does_not_fall_back_after_start() -> None:
class FailedAfterStartClient(RecordingClient):
def post_event_stream(self, *args: Any, **kwargs: Any):
yield {"event": "start", "data": {"model_key": "qwen"}}
raise ApiHttpError("missing", status_code=404)
client = FailedAfterStartClient()
with pytest.raises(ApiHttpError):
list(RemoteDoctorRepository(client).stream_diagnosis_ai(501, "请分析"))
assert client.post_calls == []
def test_remote_diagnosis_ai_stream_error_event_does_not_fall_back() -> None:
class ErrorStreamClient(RecordingClient): class ErrorStreamClient(RecordingClient):
def post_event_stream(self, *args: Any, **kwargs: Any): def post_event_stream(self, *args: Any, **kwargs: Any):
yield {"event": "error", "data": {"message": "模型繁忙"}} yield {"event": "error", "data": {"message": "模型繁忙"}}
@@ -22,6 +22,7 @@ use app\adminapi\logic\tcm\DiagnosisLogic;
use app\adminapi\logic\tcm\PatientAiReportLogic; use app\adminapi\logic\tcm\PatientAiReportLogic;
use app\adminapi\logic\tcm\TrackingNoteLogic; use app\adminapi\logic\tcm\TrackingNoteLogic;
use app\adminapi\service\AssistantSseProtocol; use app\adminapi\service\AssistantSseProtocol;
use app\adminapi\service\AssistantSseTiming;
use app\adminapi\validate\tcm\DiagnosisValidate; use app\adminapi\validate\tcm\DiagnosisValidate;
use app\common\model\Order; use app\common\model\Order;
use app\common\model\WechatChatRecord; use app\common\model\WechatChatRecord;
@@ -930,6 +931,7 @@ class DiagnosisController extends BaseAdminController
*/ */
public function aiAssistantStream() public function aiAssistantStream()
{ {
$timing = new AssistantSseTiming();
// 登录由全局中间件完成;请求校验、旧助手权限与 DataScope 必须全部 // 登录由全局中间件完成;请求校验、旧助手权限与 DataScope 必须全部
// 在任何 SSE header / start 事件之前完成,失败时仍返回标准 JSON。 // 在任何 SSE header / start 事件之前完成,失败时仍返回标准 JSON。
$params = (new DiagnosisValidate())->post()->goCheck('aiAssistant'); $params = (new DiagnosisValidate())->post()->goCheck('aiAssistant');
@@ -938,17 +940,18 @@ class DiagnosisController extends BaseAdminController
(string) $params['task'], (string) $params['task'],
(string) ($params['prompt'] ?? ''), (string) ($params['prompt'] ?? ''),
$this->adminId, $this->adminId,
$this->adminInfo $this->adminInfo,
$timing
); );
if ($prepared === null) { if ($prepared === null) {
return $this->fail(DiagnosisAiLogic::getError()); return $this->fail(DiagnosisAiLogic::getError());
} }
$this->runAssistantSse($prepared); $this->runAssistantSse($prepared, $timing);
} }
/** @param array<string,mixed> $prepared */ /** @param array<string,mixed> $prepared */
private function runAssistantSse(array $prepared): void private function runAssistantSse(array $prepared, AssistantSseTiming $timing): void
{ {
while (ob_get_level() > 0) { while (ob_get_level() > 0) {
ob_end_clean(); ob_end_clean();
@@ -984,7 +987,7 @@ class DiagnosisController extends BaseAdminController
return !connection_aborted(); return !connection_aborted();
}; };
$emit('start', [ $startEmitted = $emit('start', [
'task' => (string) ($prepared['task'] ?? ''), 'task' => (string) ($prepared['task'] ?? ''),
'model_key' => (string) ($prepared['profile'] ?? ''), 'model_key' => (string) ($prepared['profile'] ?? ''),
'diagnosis_id' => (int) ($prepared['diagnosis_id'] ?? 0), 'diagnosis_id' => (int) ($prepared['diagnosis_id'] ?? 0),
@@ -995,14 +998,37 @@ class DiagnosisController extends BaseAdminController
: [], : [],
'message' => '已连接,正在生成…', 'message' => '已连接,正在生成…',
]); ]);
$sseStartedAt = hrtime(true);
$firstDeltaMs = null;
$timing->log('sse_start', ['status' => $startEmitted ? 'ok' : 'disconnected']);
try { try {
$result = DiagnosisAiLogic::streamPreparedAssistant( $result = DiagnosisAiLogic::streamPreparedAssistant(
$prepared, $prepared,
static fn (string $delta): bool => $emit('delta', ['text' => $delta]), static function (string $delta) use ($emit, $timing, $sseStartedAt, &$firstDeltaMs): bool {
static fn (): bool => connection_aborted() === 1 $observedAtMs = $firstDeltaMs === null && $delta !== ''
? AssistantSseTiming::elapsedMs($sseStartedAt)
: null;
$accepted = $emit('delta', ['text' => $delta]);
if ($accepted && $observedAtMs !== null) {
$firstDeltaMs = $observedAtMs;
$timing->log('first_delta', [
'duration_ms' => $firstDeltaMs,
'delta_bytes' => strlen($delta),
]);
}
return $accepted;
},
static fn (): bool => connection_aborted() === 1,
$timing
); );
if (connection_aborted()) { if (connection_aborted()) {
$timing->log('sse_done', [
'duration_ms' => AssistantSseTiming::elapsedMs($sseStartedAt),
'first_delta_observed' => $firstDeltaMs !== null,
'first_delta_ms' => $firstDeltaMs,
'status' => 'disconnected',
]);
exit; exit;
} }
if ($result === null) { if ($result === null) {
@@ -1013,14 +1039,22 @@ class DiagnosisController extends BaseAdminController
} else { } else {
$emit('done', $result); $emit('done', $result);
} }
$timing->log('sse_done', [
'duration_ms' => AssistantSseTiming::elapsedMs($sseStartedAt),
'first_delta_observed' => $firstDeltaMs !== null,
'first_delta_ms' => $firstDeltaMs,
'status' => $result === null ? 'error' : 'ok',
]);
} catch (\Throwable $e) { } catch (\Throwable $e) {
\think\facade\Log::warning('diagnosis ai assistant sse failed ' . json_encode([ \think\facade\Log::warning('diagnosis ai assistant sse failed ' . json_encode([
'diagnosis_id' => (int) ($prepared['diagnosis_id'] ?? 0), 'trace_id' => $timing->id(),
'profile' => (string) ($prepared['profile'] ?? ''),
'task' => (string) ($prepared['task'] ?? ''),
'admin_id' => (int) ($prepared['admin_id'] ?? 0),
'exception_class' => get_class($e),
], JSON_UNESCAPED_SLASHES | JSON_INVALID_UTF8_SUBSTITUTE)); ], JSON_UNESCAPED_SLASHES | JSON_INVALID_UTF8_SUBSTITUTE));
$timing->log('sse_done', [
'duration_ms' => AssistantSseTiming::elapsedMs($sseStartedAt),
'first_delta_observed' => $firstDeltaMs !== null,
'first_delta_ms' => $firstDeltaMs,
'status' => 'exception',
]);
$emit('error', [ $emit('error', [
'code' => 'AI_ASSISTANT_FAILED', 'code' => 'AI_ASSISTANT_FAILED',
'message' => 'AI 助手暂时不可用,请稍后重试', 'message' => 'AI 助手暂时不可用,请稍后重试',
@@ -12,6 +12,7 @@ use app\common\model\tcm\DiagnosisAiReport;
use app\common\service\DifyChatService; use app\common\service\DifyChatService;
use app\common\service\NihaixiaClinicalSkill; use app\common\service\NihaixiaClinicalSkill;
use app\common\service\TcmAncientBooksReference; use app\common\service\TcmAncientBooksReference;
use app\adminapi\service\AssistantSseTiming;
use think\facade\Db; use think\facade\Db;
use think\facade\Log; use think\facade\Log;
@@ -359,8 +360,18 @@ class DiagnosisAiLogic extends BaseLogic
string $task, string $task,
string $prompt, string $prompt,
int $adminId, int $adminId,
array $adminInfo array $adminInfo,
?AssistantSseTiming $timing = null
): ?array { ): ?array {
$prepareStartedAt = hrtime(true);
$aggregateMs = 0;
$compactionMs = 0;
$sourceBytes = 0;
$queryBytes = 0;
$attachmentCount = 0;
$compactionAttempted = false;
$status = 'error';
try {
self::$assistantErrorCode = 'AI_ASSISTANT_FAILED'; self::$assistantErrorCode = 'AI_ASSISTANT_FAILED';
$task = strtolower(trim($task)); $task = strtolower(trim($task));
if (!isset(self::ASSISTANT_TASKS[$task])) { if (!isset(self::ASSISTANT_TASKS[$task])) {
@@ -390,7 +401,11 @@ class DiagnosisAiLogic extends BaseLogic
return null; return null;
} }
$aggregateStartedAt = hrtime(true);
$context = self::buildCaseContext($diagnosis, $adminId, $adminInfo); $context = self::buildCaseContext($diagnosis, $adminId, $adminInfo);
$aggregateMs = AssistantSseTiming::elapsedMs($aggregateStartedAt);
$sourceBytes = strlen((string) ($context['case_text'] ?? ''));
$attachmentCount = count(is_array($context['files'] ?? null) ? $context['files'] : []);
if ($context['case_lines'] === []) { if ($context['case_lines'] === []) {
self::setError('患者纵向资料为空或聚合失败,无法使用 AI 助手'); self::setError('患者纵向资料为空或聚合失败,无法使用 AI 助手');
return null; return null;
@@ -405,9 +420,23 @@ class DiagnosisAiLogic extends BaseLogic
return null; return null;
} }
if (!self::fitContextForPrompt($context, $profile)) { $compactionAttempted = $sourceBytes > self::MAX_PROMPT_SOURCE_BYTES;
$compactionStartedAt = hrtime(true);
if (!self::fitContextForPrompt($context, $profile, $timing)) {
$compactionMs = $compactionAttempted ? AssistantSseTiming::elapsedMs($compactionStartedAt) : 0;
return null; return null;
} }
$compactionMs = $compactionAttempted ? AssistantSseTiming::elapsedMs($compactionStartedAt) : 0;
$query = self::buildAssistantPrompt($context, $task, $prompt);
$inputs = self::buildUpstreamInputs(
$context,
'病例问诊助手',
self::ASSISTANT_PROMPT_VERSION
);
$knowledgeSource = NihaixiaClinicalSkill::knowledgeSource();
$queryBytes = strlen($query);
$status = 'ok';
return [ return [
'diagnosis_id' => $diagnosisId, 'diagnosis_id' => $diagnosisId,
@@ -415,13 +444,9 @@ class DiagnosisAiLogic extends BaseLogic
'model_name' => $model, 'model_name' => $model,
'model_label' => $modelLabel, 'model_label' => $modelLabel,
'task' => $task, 'task' => $task,
'inputs' => self::buildUpstreamInputs( 'inputs' => $inputs,
$context, 'query' => $query,
'病例问诊助手', 'knowledge_source' => $knowledgeSource,
self::ASSISTANT_PROMPT_VERSION
),
'query' => self::buildAssistantPrompt($context, $task, $prompt),
'knowledge_source' => NihaixiaClinicalSkill::knowledgeSource(),
'user' => 'admin-diagnosis-assistant-' . $adminId, 'user' => 'admin-diagnosis-assistant-' . $adminId,
'admin_id' => $adminId, 'admin_id' => $adminId,
'files' => is_array($context['files'] ?? null) ? $context['files'] : [], 'files' => is_array($context['files'] ?? null) ? $context['files'] : [],
@@ -432,6 +457,20 @@ class DiagnosisAiLogic extends BaseLogic
? $context['source_diagnosis_ids'] ? $context['source_diagnosis_ids']
: [], : [],
]; ];
} finally {
if ($timing !== null) {
$timing->log('prepare', [
'duration_ms' => AssistantSseTiming::elapsedMs($prepareStartedAt),
'aggregate_ms' => $aggregateMs,
'compaction_ms' => $compactionMs,
'source_bytes' => $sourceBytes,
'query_bytes' => $queryBytes,
'attachment_count' => $attachmentCount,
'compaction_attempted' => $compactionAttempted,
'status' => $status,
]);
}
}
} }
/** /**
@@ -443,20 +482,30 @@ class DiagnosisAiLogic extends BaseLogic
public static function streamPreparedAssistant( public static function streamPreparedAssistant(
array $prepared, array $prepared,
callable $onDelta, callable $onDelta,
?callable $shouldAbort = null ?callable $shouldAbort = null,
?AssistantSseTiming $timing = null
): ?array { ): ?array {
$diagnosisId = (int) ($prepared['diagnosis_id'] ?? 0); $diagnosisId = (int) ($prepared['diagnosis_id'] ?? 0);
$profile = (string) ($prepared['profile'] ?? ''); $profile = (string) ($prepared['profile'] ?? '');
$adminId = (int) ($prepared['admin_id'] ?? 0); $adminId = (int) ($prepared['admin_id'] ?? 0);
$deliveredDelta = false; $deliveredDelta = false;
$forwardDelta = static function (string $delta) use (&$deliveredDelta, $onDelta) { $deltaCount = 0;
$deltaBytes = 0;
$retryAttempted = false;
$retryMs = 0;
$status = 'error';
$upstreamStartedAt = hrtime(true);
$forwardDelta = static function (string $delta) use (&$deliveredDelta, &$deltaCount, &$deltaBytes, $onDelta) {
$accepted = $onDelta($delta); $accepted = $onDelta($delta);
if ($accepted !== false) { if ($accepted !== false) {
$deliveredDelta = true; $deliveredDelta = true;
++$deltaCount;
$deltaBytes += strlen($delta);
} }
return $accepted; return $accepted;
}; };
try {
try { try {
$result = DifyChatService::streamChat( $result = DifyChatService::streamChat(
$profile, $profile,
@@ -468,6 +517,7 @@ class DiagnosisAiLogic extends BaseLogic
is_array($prepared['files'] ?? null) ? $prepared['files'] : [] is_array($prepared['files'] ?? null) ? $prepared['files'] : []
); );
} catch (\Throwable $e) { } catch (\Throwable $e) {
if ($timing === null) {
self::logAssistantFailure( self::logAssistantFailure(
$diagnosisId, $diagnosisId,
$profile, $profile,
@@ -475,12 +525,15 @@ class DiagnosisAiLogic extends BaseLogic
$e, $e,
(string) ($prepared['task'] ?? '') (string) ($prepared['task'] ?? '')
); );
}
$status = 'exception';
self::$assistantErrorCode = 'UPSTREAM_UNAVAILABLE'; self::$assistantErrorCode = 'UPSTREAM_UNAVAILABLE';
self::setError('AI 助手暂时不可用,请稍后重试'); self::setError('AI 助手暂时不可用,请稍后重试');
return null; return null;
} }
if (empty($result['ok'])) { if (empty($result['ok'])) {
if ($timing === null) {
self::logAssistantUpstreamError( self::logAssistantUpstreamError(
$diagnosisId, $diagnosisId,
$profile, $profile,
@@ -488,6 +541,7 @@ class DiagnosisAiLogic extends BaseLogic
(string) ($prepared['task'] ?? ''), (string) ($prepared['task'] ?? ''),
is_array($result) ? $result : [] is_array($result) ? $result : []
); );
}
// Some Dify-compatible gateways accept blocking chat but reject or // Some Dify-compatible gateways accept blocking chat but reject or
// incompletely terminate streaming responses. Before any delta has // incompletely terminate streaming responses. Before any delta has
@@ -502,6 +556,8 @@ class DiagnosisAiLogic extends BaseLogic
true true
) )
) { ) {
$retryAttempted = true;
$retryStartedAt = hrtime(true);
try { try {
$result = DifyChatService::chat( $result = DifyChatService::chat(
$profile, $profile,
@@ -511,6 +567,7 @@ class DiagnosisAiLogic extends BaseLogic
is_array($prepared['files'] ?? null) ? $prepared['files'] : [] is_array($prepared['files'] ?? null) ? $prepared['files'] : []
); );
} catch (\Throwable $e) { } catch (\Throwable $e) {
if ($timing === null) {
self::logAssistantFailure( self::logAssistantFailure(
$diagnosisId, $diagnosisId,
$profile, $profile,
@@ -519,7 +576,10 @@ class DiagnosisAiLogic extends BaseLogic
(string) ($prepared['task'] ?? '') (string) ($prepared['task'] ?? '')
); );
} }
}
$retryMs = AssistantSseTiming::elapsedMs($retryStartedAt);
if (empty($result['ok'])) { if (empty($result['ok'])) {
if ($timing === null) {
self::logAssistantUpstreamError( self::logAssistantUpstreamError(
$diagnosisId, $diagnosisId,
$profile, $profile,
@@ -530,8 +590,24 @@ class DiagnosisAiLogic extends BaseLogic
} }
} }
} }
}
return self::formatAssistantResult($prepared, $result); $formatted = self::formatAssistantResult($prepared, $result);
$status = $formatted === null ? 'error' : 'ok';
return $formatted;
} finally {
if ($timing !== null) {
$timing->log('upstream_done', [
'duration_ms' => AssistantSseTiming::elapsedMs($upstreamStartedAt),
'retry_ms' => $retryMs,
'retry_attempted' => $retryAttempted,
'delta_count' => $deltaCount,
'delta_bytes' => $deltaBytes,
'status' => $status,
'error_code' => $status !== 'ok' ? self::getAssistantErrorCode() : '',
]);
}
}
} }
/** /**
@@ -1477,7 +1553,11 @@ PROMPT;
* *
* @param array<string,mixed> $context * @param array<string,mixed> $context
*/ */
private static function fitContextForPrompt(array &$context, string $profile): bool private static function fitContextForPrompt(
array &$context,
string $profile,
?AssistantSseTiming $timing = null
): bool
{ {
$caseText = (string) ($context['case_text'] ?? ''); $caseText = (string) ($context['case_text'] ?? '');
if ($caseText === '' || strlen($caseText) <= self::MAX_PROMPT_SOURCE_BYTES) { if ($caseText === '' || strlen($caseText) <= self::MAX_PROMPT_SOURCE_BYTES) {
@@ -1487,12 +1567,12 @@ PROMPT;
try { try {
$compacted = PatientAiReportLogic::compactSourceForPrompt($profile, $caseText); $compacted = PatientAiReportLogic::compactSourceForPrompt($profile, $caseText);
} catch (\Throwable $e) { } catch (\Throwable $e) {
if ($timing === null) {
Log::warning('diagnosis ai context compaction failed', [ Log::warning('diagnosis ai context compaction failed', [
'diagnosis_id' => (int) ($context['diagnosis_id'] ?? 0),
'profile' => $profile,
'source_bytes' => strlen($caseText), 'source_bytes' => strlen($caseText),
'exception_class' => get_class($e), 'exception_class' => get_class($e),
]); ]);
}
self::setError('患者纵向资料过大,AI 分片读取失败,请稍后重试'); self::setError('患者纵向资料过大,AI 分片读取失败,请稍后重试');
return false; return false;
} }
@@ -0,0 +1,89 @@
<?php
declare(strict_types=1);
namespace app\adminapi\service;
use think\facade\Log;
/** A request-local, patient-free timing record for the diagnosis assistant SSE path. */
final class AssistantSseTiming
{
private const PHASES = ['prepare', 'sse_start', 'first_delta', 'upstream_done', 'sse_done'];
private const NUMBER_FIELDS = [
'duration_ms', 'aggregate_ms', 'compaction_ms', 'retry_ms',
'source_bytes', 'query_bytes', 'attachment_count',
'delta_bytes', 'delta_count', 'first_delta_ms',
];
private const BOOLEAN_FIELDS = ['compaction_attempted', 'retry_attempted', 'first_delta_observed'];
private const ERROR_CODES = [
'AI_ASSISTANT_FAILED', 'UPSTREAM_UNAVAILABLE', 'UPSTREAM_REJECTED',
'INCOMPLETE_RESPONSE', 'EMPTY_RESPONSE', 'UPSTREAM_TIMEOUT',
'CONFIG_DISABLED', 'INVALID_PROFILE', 'CONFIG_MISSING', 'CONFIG_INVALID',
'CURL_UNAVAILABLE', 'SKILL_UNAVAILABLE', 'CLIENT_DISCONNECTED',
'STREAM_DELIVERY_FAILED', 'REQUEST_BUILD_FAILED', 'CURL_INIT_FAILED',
'UPSTREAM_BUSY', 'INVALID_RESPONSE',
];
private string $traceId;
public function __construct()
{
$this->traceId = bin2hex(random_bytes(16));
}
public function id(): string
{
return $this->traceId;
}
public static function elapsedMs(int $startedAt): int
{
return max(0, (int) floor((hrtime(true) - $startedAt) / 1000000));
}
/**
* Only a fixed set of numeric and boolean metadata reaches the log. In particular,
* arbitrary prompt text, patient identifiers, URLs and credentials are discarded.
*
* @param array<string,mixed> $metrics
* @return array<string,int|bool|string>
*/
public function event(string $phase, array $metrics = []): array
{
$event = [
'trace_id' => $this->traceId,
'phase' => in_array($phase, self::PHASES, true) ? $phase : 'upstream_done',
];
foreach (self::NUMBER_FIELDS as $field) {
if (isset($metrics[$field]) && is_int($metrics[$field])) {
$event[$field] = max(0, $metrics[$field]);
}
}
foreach (self::BOOLEAN_FIELDS as $field) {
if (isset($metrics[$field]) && is_bool($metrics[$field])) {
$event[$field] = $metrics[$field];
}
}
if (isset($metrics['status']) && in_array($metrics['status'], ['ok', 'error', 'exception', 'disconnected'], true)) {
$event['status'] = $metrics['status'];
}
if (isset($metrics['error_code']) && in_array($metrics['error_code'], self::ERROR_CODES, true)) {
$event['error_code'] = $metrics['error_code'];
}
return $event;
}
/** @param array<string,mixed> $metrics */
public function log(string $phase, array $metrics = []): void
{
try {
Log::info('diagnosis ai assistant timing ' . json_encode($this->event($phase, $metrics)));
} catch (\Throwable $ignored) {
// Telemetry must never interrupt the clinical request.
}
}
}
@@ -12,6 +12,15 @@ final class ClinicalKnowledgeReference
return TcmAncientBooksReference::augmentQuery(NihaixiaClinicalSkill::augmentQuery($query)); return TcmAncientBooksReference::augmentQuery(NihaixiaClinicalSkill::augmentQuery($query));
} }
/** @param list<array<string,mixed>> $ancientBookSources */
public static function augmentQueryWithAncientBookSources(string $query, array $ancientBookSources): string
{
return TcmAncientBooksReference::augmentQueryWithSources(
NihaixiaClinicalSkill::augmentQuery($query),
$ancientBookSources
);
}
/** @param array<int,array<string,mixed>> $messages /** @param array<int,array<string,mixed>> $messages
* @return array<int,array<string,mixed>> * @return array<int,array<string,mixed>>
*/ */
@@ -79,7 +79,7 @@ class DifyChatService
$ancientBookSources = ClinicalKnowledgeReference::ancientBookSources($query); $ancientBookSources = ClinicalKnowledgeReference::ancientBookSources($query);
try { try {
$query = ClinicalKnowledgeReference::augmentQuery($query); $query = ClinicalKnowledgeReference::augmentQueryWithAncientBookSources($query, $ancientBookSources);
} catch (\RuntimeException $error) { } catch (\RuntimeException $error) {
return self::error('SKILL_UNAVAILABLE', '临床 AI 参考资料不可用'); return self::error('SKILL_UNAVAILABLE', '临床 AI 参考资料不可用');
} }
@@ -239,7 +239,7 @@ class DifyChatService
$ancientBookSources = ClinicalKnowledgeReference::ancientBookSources($query); $ancientBookSources = ClinicalKnowledgeReference::ancientBookSources($query);
try { try {
$query = ClinicalKnowledgeReference::augmentQuery($query); $query = ClinicalKnowledgeReference::augmentQueryWithAncientBookSources($query, $ancientBookSources);
} catch (\RuntimeException $error) { } catch (\RuntimeException $error) {
return self::error('SKILL_UNAVAILABLE', '临床 AI 参考资料不可用'); return self::error('SKILL_UNAVAILABLE', '临床 AI 参考资料不可用');
} }
@@ -163,7 +163,16 @@ final class TcmAncientBooksReference
if (str_contains($query, self::MARKER)) { if (str_contains($query, self::MARKER)) {
return $query; return $query;
} }
$reference = self::referenceForText($query); return self::augmentQueryWithSources($query, self::sourcesForText($query));
}
/** @param list<array<string,mixed>> $sources */
public static function augmentQueryWithSources(string $query, array $sources): string
{
if (str_contains($query, self::MARKER)) {
return $query;
}
$reference = self::referenceForSources($sources);
return $reference === '' ? $query : $reference . "\n\n" . $query; return $reference === '' ? $query : $reference . "\n\n" . $query;
} }
@@ -193,7 +202,12 @@ final class TcmAncientBooksReference
public static function referenceForText(string $text): string public static function referenceForText(string $text): string
{ {
$sources = self::sourcesForText($text); return self::referenceForSources(self::sourcesForText($text));
}
/** @param list<array<string,mixed>> $sources */
public static function referenceForSources(array $sources): string
{
if ($sources === []) { if ($sources === []) {
return ''; return '';
} }
+1 -1
View File
@@ -15,7 +15,7 @@ php server/scripts/install_tcm_ancient_books.php
```bash ```bash
mv server/knowledge/tcm-ancient-books server/knowledge/tcm-ancient-books.submodule-backup mv server/knowledge/tcm-ancient-books server/knowledge/tcm-ancient-books.submodule-backup
git pull git pull
php server/scripts/install_tcm_ancient_books.php
``` ```
确认新目录已有 701 个 TXT 且索引成功后,旧备份目录即可清理。 确认新目录已有 701 个 TXT 且索引成功后,旧备份目录即可清理。
@@ -71,7 +71,7 @@ assistantStreamExpect(
$actionStart = strpos($controller, 'public function aiAssistantStream()'); $actionStart = strpos($controller, 'public function aiAssistantStream()');
$checkAt = strpos($controller, "goCheck('aiAssistant')", $actionStart); $checkAt = strpos($controller, "goCheck('aiAssistant')", $actionStart);
$prepareAt = strpos($controller, 'DiagnosisAiLogic::prepareAssistant(', $actionStart); $prepareAt = strpos($controller, 'DiagnosisAiLogic::prepareAssistant(', $actionStart);
$runAt = strpos($controller, '$this->runAssistantSse($prepared)', $actionStart); $runAt = strpos($controller, '$this->runAssistantSse($prepared, $timing)', $actionStart);
$headerAt = strpos($controller, "header('Content-Type: text/event-stream; charset=utf-8')", $actionStart); $headerAt = strpos($controller, "header('Content-Type: text/event-stream; charset=utf-8')", $actionStart);
assistantStreamExpect( assistantStreamExpect(
$actionStart !== false && $checkAt > $actionStart && $prepareAt > $checkAt && $runAt > $prepareAt && $headerAt > $runAt, $actionStart !== false && $checkAt > $actionStart && $prepareAt > $checkAt && $runAt > $prepareAt && $headerAt > $runAt,
@@ -0,0 +1,73 @@
<?php
declare(strict_types=1);
require dirname(__DIR__) . '/vendor/autoload.php';
use app\adminapi\service\AssistantSseTiming;
function timingExpect(bool $condition, string $message): void
{
if (!$condition) {
fwrite(STDERR, "FAIL: {$message}\n");
exit(1);
}
}
$one = new AssistantSseTiming();
$two = new AssistantSseTiming();
$first = $one->event('prepare', [
'duration_ms' => 120,
'aggregate_ms' => 75,
'compaction_ms' => 0,
'source_bytes' => 1000,
'query_bytes' => 1100,
'attachment_count' => 2,
'compaction_attempted' => false,
'status' => 'ok',
'patient_name' => '患者姓名',
'query' => '患者正文',
'url' => 'https://private.example/path',
'api_key' => 'secret',
]);
$second = $one->event('upstream_done', [
'duration_ms' => 4300,
'retry_ms' => 1200,
'retry_attempted' => true,
'delta_count' => 4,
'delta_bytes' => 24,
'status' => 'error',
'error_code' => 'UPSTREAM_TIMEOUT',
]);
timingExpect(preg_match('/^[a-f0-9]{32}$/', $first['trace_id']) === 1, 'trace id is random hex only');
timingExpect($first['trace_id'] !== $two->event('prepare')['trace_id'], 'requests use distinct trace ids');
timingExpect($first['trace_id'] === $second['trace_id'], 'all phases share one trace id');
timingExpect(
$first['aggregate_ms'] === 75 && $first['attachment_count'] === 2 && $first['compaction_attempted'] === false,
'prepare phase retains useful numeric metadata'
);
timingExpect(
$second['retry_attempted'] === true && $second['retry_ms'] === 1200
&& $second['error_code'] === 'UPSTREAM_TIMEOUT',
'upstream phase retains retry, duration and safe status'
);
$encoded = json_encode($first, JSON_UNESCAPED_UNICODE);
foreach (['患者姓名', '患者正文', 'private.example', 'secret'] as $sensitive) {
timingExpect(!str_contains($encoded, $sensitive), 'timing log excludes sensitive payload');
}
timingExpect(!isset($one->event('prepare', ['duration_ms' => '患者正文'])['duration_ms']), 'numeric fields reject text');
timingExpect(!isset($one->event('upstream_done', ['error_code' => 'KEY=secret'])['error_code']), 'error code is validated');
timingExpect(!isset($one->event('upstream_done', ['error_code' => 'API_KEY_SECRET'])['error_code']), 'unknown machine codes are not logged');
$startedAt = hrtime(true);
timingExpect(AssistantSseTiming::elapsedMs($startedAt) >= 0, 'monotonic elapsed time is non-negative');
$controller = file_get_contents(dirname(__DIR__) . '/app/adminapi/controller/tcm/DiagnosisController.php');
$logic = file_get_contents(dirname(__DIR__) . '/app/adminapi/logic/tcm/DiagnosisAiLogic.php');
timingExpect(is_string($controller) && is_string($logic), 'instrumented source is readable');
timingExpect(str_contains($controller, "log('first_delta'") && str_contains($controller, "log('sse_done'"), 'SSE first delta and completion are timed');
timingExpect(str_contains($logic, "log('prepare'") && str_contains($logic, "log('upstream_done'"), 'prepare and upstream are timed');
timingExpect(str_contains($logic, 'compaction_attempted') && str_contains($logic, 'retry_attempted'), 'compression and compatibility retry are recorded');
echo "Diagnosis AI assistant timing: OK\n";
@@ -0,0 +1,102 @@
<?php
declare(strict_types=1);
namespace app\common\service {
function config(string $name): array
{
return $GLOBALS['ancientReuseConfig'];
}
function curl_init(): \stdClass
{
return new \stdClass();
}
function curl_setopt_array(\stdClass $handle, array $options): bool
{
$handle->options = $options;
$GLOBALS['ancientReuseRequests'][] = json_decode($options[CURLOPT_POSTFIELDS], true);
return true;
}
function curl_exec(\stdClass $handle): string
{
$options = $handle->options;
if (isset($options[CURLOPT_WRITEFUNCTION])) {
$options[CURLOPT_HEADERFUNCTION]($handle, "HTTP/1.1 200 OK\r\n");
$options[CURLOPT_WRITEFUNCTION]($handle,
"data: {\"event\":\"message\",\"answer\":\"offline\"}\n\n"
. "data: {\"event\":\"message_end\"}\n\n");
return '';
}
return '{"answer":"offline"}';
}
function curl_errno(\stdClass $handle): int { return 0; }
function curl_getinfo(\stdClass $handle, int $option): int { return 200; }
function curl_close(\stdClass $handle): void {}
}
namespace {
require dirname(__DIR__) . '/vendor/autoload.php';
use app\common\service\ClinicalKnowledgeReference;
use app\common\service\DifyChatService;
use app\common\service\TcmAncientBooksReference;
function ancientReuseExpect(bool $condition, string $message): void
{
if (!$condition) {
fwrite(STDERR, "FAIL: {$message}\n");
exit(1);
}
}
$indexProperty = new ReflectionProperty(TcmAncientBooksReference::class, 'index');
$indexProperty->setValue(null, [
'commit' => TcmAncientBooksReference::COMMIT,
'index_version' => TcmAncientBooksReference::INDEX_VERSION,
'terms' => [
'消渴' => [[
'title' => '甲书', 'chapter' => '消渴篇', 'path' => '甲书.txt',
'line' => 42, 'excerpt' => '消渴原文。', 'score' => 15,
]],
],
'chapter_terms' => [
'痛风' => [[
'title' => '乙书', 'chapter' => '痛风篇', 'path' => '乙书.txt',
'line' => 88, 'excerpt' => '痛风原文。', 'score' => 12,
]],
],
]);
$ancientReuseConfig = [
'enable' => true, 'base_url' => 'https://ai.example.test/v1/chat-messages',
'timeout' => 30, 'models' => ['qwen' => ['name' => 'offline', 'api_key' => 'test-key']],
];
foreach (['患者糖尿病,请分析。', '患者痛风,请分析。', '患者有未收录的症状。'] as $case) {
$expectedSources = ClinicalKnowledgeReference::ancientBookSources($case);
$expectedPrompt = ClinicalKnowledgeReference::augmentQuery($case);
foreach (['blocking', 'streaming'] as $mode) {
$ancientReuseRequests = [];
if ($mode === 'blocking') {
$result = DifyChatService::chat('qwen', [], $case, 'offline-user');
} else {
$deltas = [];
$result = DifyChatService::streamChat('qwen', [], $case, 'offline-user',
static function (string $delta) use (&$deltas): void { $deltas[] = $delta; });
ancientReuseExpect($deltas === ['offline'], 'streaming forwards the upstream delta');
}
ancientReuseExpect($result['ok'] === true, "$mode request succeeds");
ancientReuseExpect(count($ancientReuseRequests) === 1, "$mode sends one request");
ancientReuseExpect($ancientReuseRequests[0]['query'] === $expectedPrompt,
"$mode prompt remains byte-for-byte identical");
ancientReuseExpect($result['ancient_book_sources'] === $expectedSources,
"$mode returns exactly the selected citations");
}
}
$indexProperty->setValue(null, null);
echo "Dify ancient-books reuse contract passed.\n";
}
@@ -52,6 +52,20 @@ ancientExpect(str_contains($augmented, NihaixiaClinicalSkill::VERSION), 'existin
ancientExpect(str_ends_with($augmented, $query), 'original patient question remains intact'); ancientExpect(str_ends_with($augmented, $query), 'original patient question remains intact');
ancientExpect(ClinicalKnowledgeReference::augmentQuery($augmented) === $augmented, 'retry does not duplicate references'); ancientExpect(ClinicalKnowledgeReference::augmentQuery($augmented) === $augmented, 'retry does not duplicate references');
foreach ([$query, '患者出现痛风症状。', '患者有未收录的症状。', '阶段=text 患者糖尿病', '阶段=final 患者糖尿病'] as $case) {
$selected = ClinicalKnowledgeReference::ancientBookSources($case);
ancientExpect(
ClinicalKnowledgeReference::augmentQueryWithAncientBookSources($case, $selected)
=== ClinicalKnowledgeReference::augmentQuery($case),
'selected sources preserve the exact legacy prompt for each clinical case'
);
}
ancientExpect(
ClinicalKnowledgeReference::augmentQueryWithAncientBookSources('患者有未收录的症状。', [])
=== ClinicalKnowledgeReference::augmentQuery('患者有未收录的症状。'),
'an unrelated request receives no preceding patient citations'
);
$messages = [['role' => 'user', 'content' => $query]]; $messages = [['role' => 'user', 'content' => $query]];
$wire = ClinicalKnowledgeReference::augmentMessages($messages); $wire = ClinicalKnowledgeReference::augmentMessages($messages);
ancientExpect(count($wire) === 2 && str_contains($wire[0]['content'], '测试古籍'), 'chat receives reference in system message'); ancientExpect(count($wire) === 2 && str_contains($wire[0]['content'], '测试古籍'), 'chat receives reference in system message');