From 8766e829f9f112788341733334660103db6019d0 Mon Sep 17 00:00:00 2001 From: eezd Date: Fri, 6 Nov 2026 16:25:16 +0800 Subject: [PATCH] Coordinate AI retries and recover shared provider outages Retry transient requests after one and three minutes and coordinate provider recovery across processes with bounded probe intervals. Preserve completed results when recovery bookkeeping fails, keep concurrent failure accounting accurate, and expose retry state in the UI. --- backend/src/alphapulse/ai/llm.py | 241 +++++++++++++----- backend/src/alphapulse/ai/recovery.py | 85 ++++++ backend/src/alphapulse/api/chat.py | 4 +- backend/src/alphapulse/api/settings.py | 12 + backend/src/alphapulse/news_research/api.py | 4 + backend/src/alphapulse/news_research/model.py | 9 +- .../alphapulse/news_research/scheduling.py | 3 +- backend/src/alphapulse/trading/analyst.py | 18 +- backend/tests/conftest.py | 9 + backend/tests/test_ai_recovery.py | 160 ++++++++++++ backend/tests/test_failure_paths.py | 6 +- backend/tests/test_news_transport.py | 29 +-- deploy/README.md | 4 +- scripts/ci.py | 2 + scripts/smoke_ai_recovery.py | 75 ++++++ web/src/common/apis/chat/type.ts | 3 +- web/src/common/apis/news-research/index.ts | 2 +- web/src/common/apis/news-research/type.ts | 2 +- web/src/common/apis/settings/index.ts | 7 +- .../components/AiAvailability/index.vue | 40 +++ web/src/pages/chat/components/MessageList.vue | 3 + web/src/pages/chat/index.vue | 7 + web/src/pages/chat/type.ts | 1 + web/src/pages/home/index.vue | 3 + .../settings/components/ProviderSettings.vue | 2 + web/src/pages/trading/index.vue | 2 + web/tests/ai-availability.test.ts | 45 ++++ 27 files changed, 670 insertions(+), 108 deletions(-) create mode 100644 backend/src/alphapulse/ai/recovery.py create mode 100644 backend/tests/test_ai_recovery.py create mode 100644 scripts/smoke_ai_recovery.py create mode 100644 web/src/common/components/AiAvailability/index.vue create mode 100644 web/tests/ai-availability.test.ts diff --git a/backend/src/alphapulse/ai/llm.py b/backend/src/alphapulse/ai/llm.py index 68f9048..ab0e790 100644 --- a/backend/src/alphapulse/ai/llm.py +++ b/backend/src/alphapulse/ai/llm.py @@ -7,12 +7,13 @@ """ import asyncio +import hashlib import json import logging +import uuid from collections.abc import AsyncIterator from contextvars import ContextVar from datetime import UTC, datetime -from email.utils import parsedate_to_datetime from dataclasses import dataclass, field from typing import Any, Literal @@ -21,6 +22,7 @@ import httpx import httpx2 # anthropic SDK 使用的 HTTP 客户端，异常类与 httpx 不通用 import openai +from alphapulse.ai.recovery import Gate from alphapulse.core.crypto import decrypt from alphapulse.models import AiProvider @@ -28,11 +30,8 @@ from alphapulse.models import AiProvider TIMEOUT_SECONDS = 600 # 思考强度：两种协议共有的取值。OpenAI 另有 none/minimal，但实测经 Sub2API 会卡住后 502，不提供 Effort = Literal["low", "medium", "high", "xhigh", "max"] -# SDK 自带重试只针对连接错误、429 和 5xx -MAX_RETRIES = 2 -# 只重试 SDK 不处理的流内错误和读取中断，且只在还没有输出内容时重试。 -STREAM_RETRIES = 3 -STREAM_RETRY_DELAY_SECONDS = 2 +# SDK 重试关闭，所有生成请求仅由公共流负责重试。 +MAX_RETRIES = 0 # 输出被截断时回给模型的工具结果：参数不完整，不能执行 TRUNCATED_RESULT = json.dumps({"error": "输出达到最大输出上限被截断，工具参数不完整，未执行。请精简内容后重新调用"}, ensure_ascii=False) @@ -40,6 +39,7 @@ logger = logging.getLogger("alphapulse.ai.llm") news_retry = ContextVar("news_model_retry", default=None) +fail_fast = ContextVar("ai_fail_fast", default=False) class LlmError(Exception): @@ -106,70 +106,178 @@ async def list_models(provider: AiProvider) -> list[str]: raise LlmError('无法读取模型列表；请检查接入配置，或手动填写模型名称') from error +@dataclass +class RetryStatus: + scope: str + attempt: int + retry_at: str + reason: str + + +def recovery_gate(provider): + identity = json.dumps([provider.protocol, api_root(provider.base_url), decrypt(provider.api_key_encrypted)]) + return Gate(hashlib.sha256(identity.encode()).hexdigest()) + + +async def availability(provider): + try: + state = await recovery_gate(provider).call('read') + except Exception as error: + raise LlmError('无法读取模型故障协调状态，请检查 Redis') from error + return {'state': 'probing' if state['probe'] else 'waiting' if state['level'] else 'normal', + 'retry_at': datetime.fromtimestamp(state['retry_at'], UTC).isoformat() if state['level'] else None, + 'level': state['level']} + + async def stream( provider: AiProvider, system: str, messages: list[dict], tools: list[ToolSpec], effort: Effort | None = None, -) -> AsyncIterator[TextDelta | ThinkingStarted | Completion]: - """先逐段产出 TextDelta / ThinkingStarted，最后产出一个 Completion。effort 为空表示不传，由模型决定""" - policy = news_retry.get() - retries = 2 if policy is not None else STREAM_RETRIES - for attempt in range(retries + 1): - output = False - if policy is not None: policy["output"] = False + *, before_retry=None, +) -> AsyncIterator[TextDelta | ThinkingStarted | Completion | RetryStatus]: + """仅重发未开始生成的当前请求；共享故障等待优先于局部重试。""" + gate = recovery_gate(provider) + identifier = uuid.uuid4().hex + retries = 0 + waited = False + policy = {'output': False} + + async def coordinate(operation, epoch=-1): try: - if provider.protocol == "openai": - events = _stream_openai(provider, system, messages, tools, effort) - else: - events = _stream_anthropic(provider, system, messages, tools, effort) - async for event in events: - output = True - yield event - return - # 流式读取过程中的网络错误 SDK 不包装，以原始的 httpx / httpx2 异常抛出 - except (openai.APIError, anthropic.APIError, httpx.TransportError, httpx2.TransportError) as e: - # 已经推送过内容再重试会重复输出 - delay = _news_retry_delay(e, attempt) if policy is not None else STREAM_RETRY_DELAY_SECONDS - retryable = delay is not None if policy is not None else _retryable(e) - if output or (policy is not None and policy.get("output")) or attempt == retries or not retryable: - raise _llm_error(e) from e - if policy is not None: + return await gate.call(operation, identifier, epoch) + except Exception as error: + raise LlmError('模型故障协调不可用，请检查 Redis；未绕过保护发送请求') from error + + async def record_outcome(operation, epoch): + # 已完成响应及取消语义不能被协调状态写入或租约清理失败覆盖。 + # 新请求仍必须通过严格的 enter 门禁，失败时不会绕过 Redis。 + try: + await coordinate(operation, epoch) + except LlmError: + logger.warning('模型响应状态登记失败：%s；保留原始结果', operation) + + async def notice(scope, count, timestamp, reason): + from alphapulse.news_research.progress import step, current + if current.get() is not None: + from alphapulse.ai.recovery import mark_waiting + await mark_waiting(identifier) + at = datetime.fromtimestamp(timestamp, UTC).isoformat() + step('等待模型重试', retry_scope=scope, retry_attempt=count, retry_at=at) + return RetryStatus(scope, count, at, reason) + + async def heartbeat(epoch): + while True: + await asyncio.sleep(30) + await coordinate('heartbeat', epoch) + + try: + while True: + admission = await coordinate('enter') + if not admission['admitted']: + status = await notice('upstream', admission['level'], + max(admission['retry_at'], admission['now']+1), '同一接入暂时不可用，等待单请求试探恢复') + if fail_fast.get(): + raise LlmError(f'{status.reason}；下次尝试：{status.retry_at}') + yield status + waited = True + await asyncio.sleep(1) + continue + epoch = admission['epoch'] + pulse = asyncio.create_task(heartbeat(epoch)) if admission['probe'] else None + output = False + try: + if waited and before_retry: + await before_retry() from alphapulse.news_research.progress import step - step(f"上游暂时拒绝：{_llm_error(e)}；{delay:g}秒后重试（{attempt + 1}/2）") - logger.warning("模型调用失败，%ss 后第 %s 次重试：%s", delay, attempt + 1, _llm_error(e)) - await asyncio.sleep(delay) - if policy is not None: - from alphapulse.news_research.progress import step - step(f"等待模型响应（重试 {attempt + 1}/2）") + from alphapulse.ai.recovery import clear_waiting + await clear_waiting(identifier) + step('等待模型响应', retry_at=None, retry_scope=None) + policy['output'] = False + output = False + adapter = _stream_openai if provider.protocol == 'openai' else _stream_anthropic + completed = False + async for event in _guarded_events(adapter(provider, system, messages, tools, effort), policy): + output |= not isinstance(event, Completion) + if isinstance(event, Completion): + completed = True + # 调用方可能取得 Completion 后直接关闭流；先尝试登记恢复，失败仍保留完整响应。 + await record_outcome('success', epoch) + yield event + if not completed: + raise LlmError('模型连接提前结束，未收到完整响应的结束标志') + return + except (openai.APIError, anthropic.APIError, httpx.TransportError, httpx2.TransportError, LlmError) as error: + temporary = _temporary(error) + state = await coordinate('failure', epoch) if temporary else await coordinate('read') + if output or policy['output'] or _timeout(error) or not temporary: + raise _llm_error(error) from error + if state['level']: + waited = True + continue + if retries >= 2: + raise _llm_error(error) from error + delay = (60, 180)[retries] + retries += 1 + yield await notice('request', retries, datetime.now(UTC).timestamp()+delay, '单请求临时故障，等待重试') + await _wait_local(delay, coordinate) + waited = True + finally: + if pulse: + pulse.cancel() + try: + await pulse + except asyncio.CancelledError: + pass + except LlmError: + logger.warning('模型试探租约续期失败；保留原始结果') + await record_outcome('release', epoch) + finally: + from alphapulse.ai.recovery import clear_waiting + await clear_waiting(identifier) -def _news_retry_delay(error, attempt): - """One bounded retry owner for news; never repeat a timed out request.""" - if isinstance(error, (openai.APITimeoutError, anthropic.APITimeoutError, httpx.TimeoutException, httpx2.TimeoutException)): - return None +async def _guarded_events(events, policy): + # ContextVar 仅覆盖模型读取，不跨 yield 泄漏到调用方或异步生成器清理任务。 + try: + while True: + token = news_retry.set(policy) + try: + event = await anext(events) + except StopAsyncIteration: + return + finally: + news_retry.reset(token) + yield event + finally: + await events.aclose() + + +async def _wait_local(delay, coordinate): + deadline = asyncio.get_running_loop().time()+delay + while asyncio.get_running_loop().time() < deadline: + await asyncio.sleep(max(0, min(1, deadline-asyncio.get_running_loop().time()))) + if (await coordinate('read'))['level']: + break + + +def _timeout(error): + return isinstance(error, (openai.APITimeoutError, anthropic.APITimeoutError, + httpx.TimeoutException, httpx2.TimeoutException)) + + +def _temporary(error): status = getattr(error, 'status_code', None) - body = getattr(error, 'body', None) - detail = body.get('error', body) if isinstance(body, dict) else {} - message = str((detail.get('message') if isinstance(detail, dict) else detail) or getattr(error, 'message', '')).lower() - concurrency = _in_stream_error(error) and 'concurrency limit exceeded' in message - connection = isinstance(error, (openai.APIConnectionError, anthropic.APIConnectionError, httpx.TransportError, httpx2.TransportError)) - if not (status == 429 or status is not None and 500 <= status <= 599 or concurrency or connection): - return None - delay = 30 * (attempt + 1) - response = getattr(error, 'response', None) - header = response.headers.get('retry-after') if response is not None else None - if header: - try: - try: seconds = float(header) - except ValueError: seconds = (parsedate_to_datetime(header) - datetime.now(UTC)).total_seconds() - if seconds > 120: return None - if seconds > 0: delay = max(delay, seconds) - except (ValueError, TypeError, OverflowError): - pass - return delay + message = str(error).lower() + return (_timeout(error) or status == 429 or status is not None and 500 <= status <= 599 + or isinstance(error, (openai.APIConnectionError, anthropic.APIConnectionError, + httpx.TransportError, httpx2.TransportError)) + or _in_stream_error(error) and any(term in message for term in + ('concurrency limit', 'overloaded', 'rate limit', 'temporarily unavailable')) + or isinstance(error, LlmError) and '未收到完整响应的结束标志' in message) def _news_output(): policy = news_retry.get() - if policy is not None: policy['output'] = True + if policy is not None: + policy['output'] = True def _in_stream_error(e: Exception) -> bool: @@ -180,16 +288,9 @@ def _in_stream_error(e: Exception) -> bool: return isinstance(e, anthropic.APIStatusError) and e.status_code == 200 -def _retryable(e: Exception) -> bool: - """建连重试归 SDK；这里只处理流内错误与读取中断，超时不重试。""" - if isinstance(e, (openai.APITimeoutError, anthropic.APITimeoutError, httpx.TimeoutException, httpx2.TimeoutException)): - return False - return _in_stream_error(e) or isinstance( - e, (httpx.TransportError, httpx2.TransportError), - ) - - def _llm_error(e: Exception) -> LlmError: + if isinstance(e, LlmError): + return e if _in_stream_error(e): error = e.body.get("error") if isinstance(e.body, dict) else None message = error.get("message") if isinstance(error, dict) else None @@ -231,7 +332,7 @@ def _openai_messages(system: str, messages: list[dict]) -> list[dict]: async def _stream_openai(provider: AiProvider, system: str, messages: list[dict], tools: list[ToolSpec], effort: Effort | None) -> AsyncIterator[TextDelta | Completion]: client = openai.AsyncOpenAI( api_key=decrypt(provider.api_key_encrypted), base_url=f"{api_root(provider.base_url)}/v1", - timeout=TIMEOUT_SECONDS, max_retries=0 if news_retry.get() is not None else MAX_RETRIES, + timeout=TIMEOUT_SECONDS, max_retries=0, ) kwargs: dict[str, Any] = {} if tools: @@ -310,7 +411,7 @@ def _anthropic_messages(messages: list[dict]) -> list[dict]: async def _stream_anthropic(provider: AiProvider, system: str, messages: list[dict], tools: list[ToolSpec], effort: Effort | None) -> AsyncIterator[TextDelta | ThinkingStarted | Completion]: client = anthropic.AsyncAnthropic( api_key=decrypt(provider.api_key_encrypted), base_url=api_root(provider.base_url), - timeout=TIMEOUT_SECONDS, max_retries=0 if news_retry.get() is not None else MAX_RETRIES, + timeout=TIMEOUT_SECONDS, max_retries=0, ) kwargs: dict[str, Any] = {} if tools: diff --git a/backend/src/alphapulse/ai/recovery.py b/backend/src/alphapulse/ai/recovery.py new file mode 100644 index 0000000..2f81e3f --- /dev/null +++ b/backend/src/alphapulse/ai/recovery.py @@ -0,0 +1,85 @@ +"""跨进程模型接入熔断；只有持有租约的请求可以试探恢复。""" +import json +import logging +from redis.exceptions import RedisError +from alphapulse.core.redis import KEY_PREFIX, redis + +# 时间以 Redis 为准。epoch 隔离旧请求；失败窗口只计不同逻辑请求。 +SCRIPT = r''' +local raw = redis.call('GET', KEYS[1]) +local s = raw and cjson.decode(raw) or {epoch=0, level=0, retry_at=0, failures={}} +local now = tonumber(redis.call('TIME')[1]) +local op, id, epoch = ARGV[1], ARGV[2], tonumber(ARGV[3]) +local function save() + redis.call('SET', KEYS[1], cjson.encode(s), 'EX', 86400) +end +local function open() + s.level = math.min(s.level+1, 3) + s.retry_at = now + ({180,600,900})[s.level] + s.epoch = s.epoch+1 + s.owner = nil; s.lease_until = nil; s.failures = {} +end +local admitted = false +if op == 'enter' then + if s.level == 0 then admitted = true + elseif now >= s.retry_at and (not s.owner or s.lease_until <= now) then + s.owner=id; s.lease_until=now+660; admitted=true; save() + end +elseif epoch == s.epoch then + if op == 'success' then + if s.level == 0 or s.owner == id then + -- 正常成功只清空失败窗口，不使并发中的其他请求失效。 + -- 只有从整体故障恢复才切换代次，隔离旧试探结果。 + local next_epoch = s.epoch + if s.level > 0 then next_epoch = next_epoch+1 end + s={epoch=next_epoch,level=0,retry_at=0,failures={}}; save() + end + elseif op == 'failure' then + if s.level > 0 then + if s.owner == id then open(); save() end + else + for key, at in pairs(s.failures) do + if at <= now-60 then s.failures[key]=nil end + end + s.failures[id]=now + local count=0 + for _ in pairs(s.failures) do count=count+1 end + if count >= 3 then open() end + save() + end + elseif s.owner == id then + if op == 'heartbeat' then s.lease_until=now+660 + elseif op == 'release' then s.owner=nil; s.lease_until=nil end + save() + end +end +return cjson.encode({epoch=s.epoch,level=s.level, retry_at=s.retry_at, + probe=s.owner ~= nil, admitted=admitted, now=now}) +''' + + +class Gate: + def __init__(self, scope): + self.key = KEY_PREFIX + 'ai:recovery:' + scope + + async def call(self, operation, request_id='', epoch=-1): + return json.loads(await redis.eval(SCRIPT, 1, self.key, operation, request_id, epoch)) + + +async def mark_waiting(identifier): + import time + await redis.zadd(KEY_PREFIX + 'ai:news-waiting', {identifier: time.time()+300}) + + +async def clear_waiting(identifier): + try: + await redis.zrem(KEY_PREFIX + 'ai:news-waiting', identifier) + except RedisError: + # 展示计数有过期兜底，清理失败不能掩盖原始错误或取消。 + logging.getLogger(__name__).warning('无法清理模型等待计数') + + +async def waiting_count(): + import time + await redis.zremrangebyscore(KEY_PREFIX + 'ai:news-waiting', '-inf', time.time()) + return await redis.zcard(KEY_PREFIX + 'ai:news-waiting') diff --git a/backend/src/alphapulse/api/chat.py b/backend/src/alphapulse/api/chat.py index 5ffbced..14b4ec9 100644 --- a/backend/src/alphapulse/api/chat.py +++ b/backend/src/alphapulse/api/chat.py @@ -170,7 +170,9 @@ async def _message_events(body: SendIn) -> AsyncIterator[ServerSentEvent]: yield event("delta", {"text": item.text}) elif isinstance(item, llm.ThinkingStarted): yield event("thinking", {}) - else: + elif isinstance(item, llm.RetryStatus): + yield event("retry", {"scope": item.scope, "attempt": item.attempt, "retry_at": item.retry_at, "reason": item.reason}) + elif isinstance(item, llm.Completion): completion = item except llm.LlmError as e: if text: diff --git a/backend/src/alphapulse/api/settings.py b/backend/src/alphapulse/api/settings.py index aebc246..cb68fe2 100644 --- a/backend/src/alphapulse/api/settings.py +++ b/backend/src/alphapulse/api/settings.py @@ -170,10 +170,20 @@ async def delete_provider(provider_id: int, session: Annotated[AsyncSession, Dep return success(None) +@router.get("/provider-availability") +async def provider_availability(session: Annotated[AsyncSession, Depends(get_session)]): + providers = list(await session.scalars(select(AiProvider).order_by(AiProvider.id))) + try: + return success([{'provider_id': p.id, 'name': p.name, **await llm.availability(p)} for p in providers]) + except llm.LlmError as error: + raise HTTPException(503, str(error)) from error + + @router.post("/providers/{provider_id}/test") async def test_provider(provider_id: int, session: Annotated[AsyncSession, Depends(get_session)]) -> ApiResponse[TestResult]: provider = await get_provider(session, provider_id) started = time.monotonic() + token = llm.fail_fast.set(True) try: completion = None async for event in llm.stream(provider, "只回复 OK。", [{"role": "user", "text": "ping"}], [], provider.reasoning_effort): @@ -181,6 +191,8 @@ async def test_provider(provider_id: int, session: Annotated[AsyncSession, Depen completion = event except llm.LlmError as e: return success(TestResult(ok=False, elapsed_ms=round((time.monotonic() - started) * 1000), error=str(e))) + finally: + llm.fail_fast.reset(token) return success(TestResult(ok=True, elapsed_ms=round((time.monotonic() - started) * 1000), reply=completion.text[:200])) diff --git a/backend/src/alphapulse/news_research/api.py b/backend/src/alphapulse/news_research/api.py index 48920d5..76647be 100644 --- a/backend/src/alphapulse/news_research/api.py +++ b/backend/src/alphapulse/news_research/api.py @@ -197,6 +197,8 @@ async def enable(session: DB, body: EnableInput | None = None): "summary": "", "content": "本条为虚构测试。示例公司宣布计划维护，具体时间未知。", } + from alphapulse.ai import llm + token = llm.fail_fast.set(True) original_config = row.config control_revision = row.control_revision try: @@ -261,6 +263,8 @@ async def enable(session: DB, body: EnableInput | None = None): raise HTTPException(422, str(error)) from error except Exception as error: raise HTTPException(502, f"两阶段验证失败：{error}") from error + finally: + llm.fail_fast.reset(token) row = await session.get( ResearchSettings, 1, populate_existing=True, with_for_update=True ) diff --git a/backend/src/alphapulse/news_research/model.py b/backend/src/alphapulse/news_research/model.py index 36ab07f..07584e2 100644 --- a/backend/src/alphapulse/news_research/model.py +++ b/backend/src/alphapulse/news_research/model.py @@ -23,12 +23,9 @@ class SubmissionFailure(ValueError): async def model_stream(*args): """Scope transport retry changes to news requests only.""" - token = llm.news_retry.set({"output": False}) - try: - async for item in llm.stream(*args): - yield item - finally: - llm.news_retry.reset(token) + async for item in llm.stream(*args): + yield item + async def validate_submission(provider, system, messages, schema, completion, usages, allow_repair=True): """Repair structural or time-contract errors once using the same evidence.""" diff --git a/backend/src/alphapulse/news_research/scheduling.py b/backend/src/alphapulse/news_research/scheduling.py index 5f143a6..c2ba278 100644 --- a/backend/src/alphapulse/news_research/scheduling.py +++ b/backend/src/alphapulse/news_research/scheduling.py @@ -166,5 +166,6 @@ async def queue_status(session, settings): reasons.append('no_work' if not any( q.get(s, 0) for q in queues.values() for s in ('missing', 'stale', 'index_pending', 'assign_pending') ) and not any(q.get('ready', 0) for q in events.values()) else 'dispatching') - return {'running': running, 'capacity': capacity, 'running_by_stage': stages, + from alphapulse.ai.recovery import waiting_count + return {'waiting_retry': await waiting_count(), 'running': running, 'capacity': capacity, 'running_by_stage': stages, 'draining': not settings.enabled and running > 0, 'events': events, 'idle_reasons': reasons, **queues} diff --git a/backend/src/alphapulse/trading/analyst.py b/backend/src/alphapulse/trading/analyst.py index 88d6a78..f672700 100644 --- a/backend/src/alphapulse/trading/analyst.py +++ b/backend/src/alphapulse/trading/analyst.py @@ -96,6 +96,7 @@ class AnalysisError(Exception): @dataclass class Prepared: exchange: str + symbol: str provider: AiProvider effort: str | None dry_run: bool @@ -248,13 +249,24 @@ async def prepare(symbol: str, reasons: list[str], analysis_id: int) -> Prepared "## 预取的行情（实盘数据，已取，不要用相同参数重复调用）", await prefetch(symbol), ] - return Prepared(exchange=exchange, provider=provider, effort=effort, dry_run=dry_run, system=system, user_text="\n".join(lines), context=context) + return Prepared(exchange=exchange, symbol=symbol, provider=provider, effort=effort, dry_run=dry_run, system=system, user_text="\n".join(lines), context=context) async def complete(prepared: Prepared, messages: list[dict], tools: list[llm.ToolSpec]) -> llm.Completion: - async for item in llm.stream(prepared.provider, prepared.system, messages, tools, prepared.effort): + async def refresh(): + # 只更新当前请求的行情，不重新执行此前的工具或交易操作。 + latest = "重试前最新行情（此前行情仅为历史快照）：\n" + await prefetch(prepared.symbol) + if messages[-1].get('retry_snapshot'): + messages.pop() + messages.append({'role': 'user', 'text': latest, 'retry_snapshot': True}) + if estimate_tokens(prepared.system) + estimate_tokens(dump(messages)) + prepared.provider.max_tokens > prepared.provider.context_window: + raise AnalysisError('刷新行情后超过上下文上限，未重发') + completion = None + async for item in llm.stream(prepared.provider, prepared.system, messages, tools, prepared.effort, before_retry=refresh): if isinstance(item, llm.Completion): - return item + completion = item + if completion is not None: + return completion raise AssertionError("llm.stream 没有产出 Completion") diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index a68720a..b6a955e 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -11,4 +11,13 @@ def trading(monkeypatch): }.items(): monkeypatch.setenv(key, value) from alphapulse.trading import executor + from unittest.mock import AsyncMock + from alphapulse.ai import llm, recovery + class HealthyGate: + async def call(self, operation, request_id='', epoch=-1): + return {'epoch': 0, 'level': 0, 'retry_at': 0, 'probe': False, 'admitted': True, 'now': 0} + monkeypatch.setattr(llm, 'recovery_gate', lambda _: HealthyGate()) + monkeypatch.setattr(recovery, 'mark_waiting', AsyncMock()) + monkeypatch.setattr(recovery, 'clear_waiting', AsyncMock()) + monkeypatch.setattr(recovery, 'waiting_count', AsyncMock(return_value=0)) return executor diff --git a/backend/tests/test_ai_recovery.py b/backend/tests/test_ai_recovery.py new file mode 100644 index 0000000..da22674 --- /dev/null +++ b/backend/tests/test_ai_recovery.py @@ -0,0 +1,160 @@ +"""隔离 Redis 验证实际 Lua 原子操作；不访问生产接入。""" +import asyncio +from types import SimpleNamespace as N +from unittest.mock import AsyncMock + +import httpx +import openai +import pytest + + + +def test_global_failure_takes_over_local_retry_and_recovers(trading, monkeypatch): + from alphapulse.ai import llm + states = iter([ + dict(level=0, epoch=0, admitted=True, probe=False, now=0, retry_at=0), + dict(level=1, epoch=1, admitted=False, probe=False, now=0, retry_at=180), + dict(level=1, epoch=1, admitted=False, probe=False, now=0, retry_at=180), + dict(level=1, epoch=1, admitted=True, probe=True, now=180, retry_at=180), + ]) + async def coordinate(operation, *args): + if operation in ('enter', 'failure'): + return next(states) + return dict(level=0) + gate = N(call=AsyncMock(side_effect=coordinate)) + monkeypatch.setattr(llm, 'recovery_gate', lambda _: gate) + monkeypatch.setattr(llm.asyncio, 'sleep', AsyncMock()) + local = AsyncMock() + monkeypatch.setattr(llm, '_wait_local', local) + calls = [] + async def adapter(*args): + calls.append(args) + if len(calls) == 1: + raise openai.RateLimitError('limited', response=httpx.Response(429, request=httpx.Request('POST', 'https://test.invalid')), body={}) + yield llm.Completion(text='done') + monkeypatch.setattr(llm, '_stream_openai', adapter) + async def run(): + return [x async for x in llm.stream(N(protocol='openai'), '', [], [])] + result = asyncio.run(run()) + assert result[-1].text == 'done' and len(calls) == 2 + assert any(isinstance(x, llm.RetryStatus) and x.scope == 'upstream' for x in result) + local.assert_not_awaited() + + +def test_coordinator_failure_does_not_send(trading, monkeypatch): + from alphapulse.ai import llm + adapter = AsyncMock() + monkeypatch.setattr(llm, '_stream_openai', adapter) + monkeypatch.setattr(llm, 'recovery_gate', lambda _: N(call=AsyncMock(side_effect=RuntimeError('offline')))) + async def run(): + with pytest.raises(llm.LlmError, match='协调不可用'): + async for _ in llm.stream(N(protocol='openai'), '', [], []): + pass + asyncio.run(run()) + adapter.assert_not_called() + + +def test_fail_fast_does_not_send_or_wait(trading, monkeypatch): + from alphapulse.ai import llm + gate = N(call=AsyncMock(return_value=dict(level=2, epoch=1, admitted=False, probe=False, now=100, retry_at=700))) + monkeypatch.setattr(llm, 'recovery_gate', lambda _: gate) + adapter = AsyncMock() + monkeypatch.setattr(llm, '_stream_openai', adapter) + async def run(): + token = llm.fail_fast.set(True) + try: + with pytest.raises(llm.LlmError, match='下次尝试'): + async for _ in llm.stream(N(protocol='openai'), '', [], []): + pass + finally: + llm.fail_fast.reset(token) + asyncio.run(run()) + adapter.assert_not_called() + + +def test_completed_probe_is_confirmed_before_caller_returns(trading, monkeypatch): + from alphapulse.ai import llm + gate = N(call=AsyncMock(return_value=dict(level=0, epoch=1, admitted=True, probe=True, now=1, retry_at=0))) + monkeypatch.setattr(llm, 'recovery_gate', lambda _: gate) + async def adapter(*args): + yield llm.Completion(text='done') + monkeypatch.setattr(llm, '_stream_openai', adapter) + async def run(): + stream = llm.stream(N(protocol='openai'), '', [], []) + assert (await anext(stream)).text == 'done' + assert any(call.args[0] == 'success' for call in gate.call.await_args_list) + await stream.aclose() + asyncio.run(run()) + + +def test_retry_refreshes_only_current_request(trading, monkeypatch): + from alphapulse.ai import llm + calls = [] + async def adapter(provider, system, messages, tools, effort): + calls.append(list(messages)) + if len(calls) == 1: + raise httpx.ConnectError('unavailable') + yield llm.Completion(text='ok') + messages = [{'role': 'tool', 'tool_call_id': 'already-done', 'result': 'retained'}] + async def refresh(): + messages.append({'role': 'user', 'text': 'fresh snapshot'}) + monkeypatch.setattr(llm, '_stream_openai', adapter) + monkeypatch.setattr(llm, '_wait_local', AsyncMock()) + async def run(): + return [x async for x in llm.stream(N(protocol='openai'), '', messages, [], before_retry=refresh)] + assert asyncio.run(run())[-1].text == 'ok' + assert len(calls) == 2 and calls[1][0] == calls[0][0] + assert calls[1][-1]['text'] == 'fresh snapshot' + + +@pytest.mark.parametrize('failed_operation', ['success', 'release', 'both']) +def test_completed_result_survives_coordination_write_failure(trading, monkeypatch, failed_operation): + from alphapulse.ai import llm + from alphapulse.news_research.model import complete + from alphapulse.news_research.contracts import IndexResult + calls = [] + + async def coordinate(operation, *args): + calls.append(operation) + if operation == failed_operation or failed_operation == 'both' and operation in ('success', 'release'): + raise RuntimeError('synthetic Redis outage') + return dict(level=0, epoch=0, admitted=True, probe=False, now=0, retry_at=0) + + async def adapter(*args): + yield llm.Completion(tool_calls=[{'name': 'submit', 'id': '1', 'arguments': { + 'keywords': [{'canonical': '新闻', 'aliases': ['news'], 'quote': 'news'}], + }}], usage={'input_tokens': 100, 'output_tokens': 10}) + + monkeypatch.setattr(llm, 'recovery_gate', lambda _: N(call=coordinate)) + monkeypatch.setattr(llm, '_stream_openai', adapter) + async def run(): + return await complete(N(protocol='openai', max_tokens=100, context_window=250000, + reasoning_effort=None), '', {}, IndexResult) + output, usage = asyncio.run(run()) + assert output.keywords[0].canonical == '新闻' + assert usage == {'input_tokens': 100, 'output_tokens': 10} + assert calls.count('enter') == 1 and 'release' in calls + + +def test_cancellation_survives_lease_release_failure(trading, monkeypatch): + from alphapulse.ai import llm + started = asyncio.Event() + async def coordinate(operation, *args): + if operation == 'release': + raise RuntimeError('synthetic Redis outage') + return dict(level=0, epoch=0, admitted=True, probe=False, now=0, retry_at=0) + async def adapter(*args): + started.set() + await asyncio.Event().wait() + yield + monkeypatch.setattr(llm, 'recovery_gate', lambda _: N(call=coordinate)) + monkeypatch.setattr(llm, '_stream_openai', adapter) + async def run(): + async def consume(): + return [x async for x in llm.stream(N(protocol='openai'), '', [], [])] + task = asyncio.create_task(consume()) + await started.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + asyncio.run(run()) diff --git a/backend/tests/test_failure_paths.py b/backend/tests/test_failure_paths.py index 3952c08..c39cc35 100644 --- a/backend/tests/test_failure_paths.py +++ b/backend/tests/test_failure_paths.py @@ -58,7 +58,7 @@ def test_model_requires_protocol_end_marker(trading, monkeypatch, protocol, comp asyncio.run(check()) -def test_connection_failure_uses_only_sdk_retry_budget(trading, monkeypatch): +def test_connection_failure_has_one_retry_owner(trading, monkeypatch): from alphapulse.ai import llm attempts = 0 def handle(request): @@ -66,12 +66,14 @@ def test_connection_failure_uses_only_sdk_retry_budget(trading, monkeypatch): attempts += 1 raise httpx.ConnectError("测试连接失败", request=request) provider = install_model_transport(llm, monkeypatch, "openai", handle) + from unittest.mock import AsyncMock + monkeypatch.setattr(llm, "_wait_local", AsyncMock()) async def check(): with pytest.raises(llm.LlmError, match="无法连接"): async for _ in llm.stream(provider, "system", [], []): pass asyncio.run(check()) - assert attempts == llm.MAX_RETRIES + 1 + assert attempts == 3 def test_pipeline_error_keeps_history_checkpoint_and_marks_run_failed(trading, monkeypatch): diff --git a/backend/tests/test_news_transport.py b/backend/tests/test_news_transport.py index af572b2..27120b7 100644 --- a/backend/tests/test_news_transport.py +++ b/backend/tests/test_news_transport.py @@ -17,7 +17,7 @@ def test_news_transport_retry_budget_and_output_guard(trading, monkeypatch, mode calls.append(args) if mode == 'partial': yield llm.TextDelta('started') if mode == 'tool_fragment': llm._news_output() - if mode == 'success' and len(calls) == 3: + if mode in ('success', 'long_wait') and len(calls) == 3: yield llm.Completion(usage={'input_tokens': 1}) return request = httpx.Request('POST', 'https://example.invalid/v1/chat/completions') @@ -27,31 +27,22 @@ def test_news_transport_retry_budget_and_output_guard(trading, monkeypatch, mode raise openai.APIError('Concurrency limit exceeded for account', request, body={'error': {'message': 'Concurrency limit exceeded for account'}}) monkeypatch.setattr(llm, '_stream_openai', adapter) sleep = AsyncMock() - monkeypatch.setattr(llm.asyncio, 'sleep', sleep) + monkeypatch.setattr(llm, '_wait_local', sleep) record = {} async def run(): token = current.set(record) try: return [item async for item in model_stream(N(protocol='openai'), '', [], [], 'max')] finally: current.reset(token) - if mode == 'success': assert asyncio.run(run())[-1].usage == {'input_tokens': 1} + if mode in ('success', 'long_wait'): assert asyncio.run(run())[-1].usage == {'input_tokens': 1} else: with pytest.raises(llm.LlmError): asyncio.run(run()) - retries = mode in ('success', 'exhausted') + retries = mode in ('success', 'exhausted', 'long_wait') assert len(calls) == (3 if retries else 1) - assert [call.args[0] for call in sleep.await_args_list] == ([30, 60] if retries else []) + assert [call.args[0] for call in sleep.await_args_list] == ([60, 180] if retries else []) assert llm.news_retry.get() is None - if retries: assert '2/2' in record['phase'] + if retries: assert record['retry_attempt'] == 2 -def test_retry_after_seconds_and_dates(trading): - from alphapulse.ai import llm - from datetime import UTC, datetime, timedelta - from email.utils import format_datetime - request = httpx.Request('POST', 'https://example.invalid') - for value in ('90', format_datetime(datetime.now(UTC) + timedelta(seconds=95))): - error = openai.RateLimitError('wait', response=httpx.Response(429, headers={'retry-after': value}, request=request), body={}) - assert 89 <= llm._news_retry_delay(error, 0) <= 96 - def test_cancelling_backoff_releases_context_and_progress(trading, monkeypatch): from alphapulse.ai import llm @@ -61,11 +52,11 @@ def test_cancelling_backoff_releases_context_and_progress(trading, monkeypatch): async def adapter(*args): raise openai.APIError('Concurrency limit exceeded', httpx.Request('POST', 'https://example.invalid'), body={}) yield - async def sleep(delay): + async def sleep(delay, coordinate): entered.set() await asyncio.Event().wait() monkeypatch.setattr(llm, '_stream_openai', adapter) - monkeypatch.setattr(llm.asyncio, 'sleep', sleep) + monkeypatch.setattr(llm, '_wait_local', sleep) async def run(): async def consume(): try: @@ -82,7 +73,7 @@ def test_cancelling_backoff_releases_context_and_progress(trading, monkeypatch): @pytest.mark.parametrize('protocol', ['openai', 'anthropic']) @pytest.mark.parametrize('news', [True, False]) -def test_sdk_retry_is_disabled_only_for_news(trading, monkeypatch, protocol, news): +def test_sdk_retry_is_disabled_for_every_request(trading, monkeypatch, protocol, news): from alphapulse.ai import llm options=[] class Client: @@ -107,4 +98,4 @@ def test_sdk_retry_is_disabled_only_for_news(trading, monkeypatch, protocol, new try: return [item async for item in llm.stream(provider,'rules',[],[],None)] finally: llm.news_retry.reset(token) assert asyncio.run(run())[-1] == llm.Completion() - assert options[0]['max_retries'] == (0 if news else llm.MAX_RETRIES) + assert options[0]['max_retries'] == 0 diff --git a/deploy/README.md b/deploy/README.md index 76c5ac3..73c3562 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -179,7 +179,9 @@ Use `scripts/smoke_news_research.py` only with disposable CI PostgreSQL credenti Real-news acceptance requires explicit permission for the bounded sample and necessary model access. Export original records in a read-only repeatable-read transaction, copy only necessary provider fields to a private isolated configuration, and never copy the production `.env`. After processing, review keyword coverage, event boundaries, reposts, evidence, dates and ratings, retain a credential-free report, and remove temporary keys and test containers. The initial November 3 round used thirty news items and additive tables only; the subsequent chronological supplement uses twelve frozen originals and a separate additional request budget. Legacy sample replay/reset commands target a different schema and are not part of this workflow; do not run them for this transition. Legacy reports and task counts do not prove current quality or savings. Missing usage is not zero cost. -News transport uses at most two retries after 30/60 seconds, honoring valid Retry-After up to 120 seconds while retaining the worker slot. Partial output and permanent failures are not replayed. Structural submission repair is bounded to one correction using already available evidence. Default Skill changes are not automatically published to live nodes. Research tools are restricted to local sources; evidence gaps are reported without attempting external access. +All AI generation requests share one transport retry owner; SDK retries are disabled. A single request retries after 60 and then 180 seconds while retaining its task slot. Partial text, thinking or tool arguments, ambiguous timeouts and permanent errors are not replayed. Three distinct transiently failing requests within sixty seconds without a completed success open a Redis-coordinated circuit scoped to protocol, normalized endpoint and credential fingerprint. It waits 180, 600 and then 900 seconds, repeating 900-second waits until recovery. Only one real request probes recovery across processes; successful completion closes the circuit. An expiring, renewed lease recovers crashed probes. Normal concurrent successes clear the failure window without invalidating peers that subsequently fail. Complete responses and known usage remain available when post-response bookkeeping or lease cleanup fails; admission for new requests still requires Redis coordination. These fixed intervals replace upstream Retry-After scheduling, including five-hour limit hints. Redis coordination failure blocks outgoing generation rather than bypassing protection. + +Chat emits additive `retry` SSE events and cancellation stops waiting. `GET /api/settings/provider-availability` reports normal/waiting/probing per configured provider without credentials. News queue status adds `waiting_retry`; waiting requests retain their accepted task positions. Trading retries refresh market input without replaying previous tools or trades. Synchronous connection tests and news enable validation fail promptly while the shared circuit is waiting; they do not enable processing after a failed validation. CI runs `scripts/smoke_ai_recovery.py` against its disposable Redis, using `AI_RECOVERY_TEST_REDIS_URL`. Structural submission repair is bounded to one correction using already available evidence. Default Skill changes are not automatically published to live nodes. Research tools are restricted to local sources; evidence gaps are reported without attempting external access. ## Global economic calendar diff --git a/scripts/ci.py b/scripts/ci.py index 53ec989..6125fd3 100644 --- a/scripts/ci.py +++ b/scripts/ci.py @@ -180,6 +180,8 @@ def smoke(build, label): run('docker', 'run', '-d', '--name', names['web'], '--network-alias', 'web', *common, build['images']['web']['local']) wait_command('docker', 'exec', names['api'], 'python', '-c', 'import urllib.request; urllib.request.urlopen("http://localhost:8000/api/health", timeout=3)') + print(run('docker', 'exec', '-i', '-e', 'AI_RECOVERY_TEST_REDIS_URL=' + env['REDIS_URL'], + names['api'], 'python', '-', input=(ROOT / 'scripts/smoke_ai_recovery.py').read_text()), flush=True) print(run('docker', 'exec', '-i', names['api'], 'python', '-', input=(ROOT / 'scripts/smoke_dynamic_queue.py').read_text()), flush=True) print(run('docker', 'exec', '-i', names['api'], 'python', '-', diff --git a/scripts/smoke_ai_recovery.py b/scripts/smoke_ai_recovery.py new file mode 100644 index 0000000..dee5c1d --- /dev/null +++ b/scripts/smoke_ai_recovery.py @@ -0,0 +1,75 @@ +"""在明确指定的隔离 Redis 上验证接入恢复与租约互斥。""" +import asyncio +import os +import uuid + +def main(): + url = os.environ['AI_RECOVERY_TEST_REDIS_URL'] + from redis.asyncio import Redis + from alphapulse.ai.recovery import SCRIPT + + async def run(): + redis = Redis.from_url(url, decode_responses=True) + import json + key = 'ap:ai:test:' + uuid.uuid4().hex + # 仅替换时钟，实际执行生产 Lua，检查跨客户端的租约互斥。 + script = SCRIPT.replace("tonumber(redis.call('TIME')[1])", 'tonumber(ARGV[4])') + async def call(op, identity='', epoch=-1, now=1000): + return json.loads(await redis.eval(script, 1, key, op, identity, epoch, now)) + try: + for _ in range(4): + assert (await call('failure', 'same', 0))['level'] == 0 + # 四个并发请求先成功一个，其后其他三个失败必须正常计数。 + admissions = [await call('enter', name) for name in ('good', 'a', 'b', 'c')] + epoch = admissions[0]['epoch'] + await call('success', 'good', epoch) + assert (await call('read'))['epoch'] == epoch + await call('failure', 'a', epoch, 1000) + await call('failure', 'b', epoch, 1001) + opened = await call('failure', 'c', epoch, 1002) + assert opened['level'] == 1 and opened['retry_at'] == 1182 + assert not (await call('enter', 'early', now=1181))['admitted'] + claims = await asyncio.gather(*(call('enter', str(i), now=1182) for i in range(20))) + assert sum(s['admitted'] for s in claims) == 1 + winner = str(next(i for i, s in enumerate(claims) if s['admitted'])) + epoch = claims[int(winner)]['epoch'] + # 旧失败不推进；失效的并发成功不解除当前故障。 + assert (await call('failure', 'late', epoch-1, 1183))['level'] == 1 + assert (await call('success', 'late', epoch-1, 1183))['level'] == 1 + second = await call('failure', winner, epoch, 1183) + assert second['level'] == 2 and second['retry_at'] == 1783 + third = await call('enter', 'probe2', now=1783) + state = await call('failure', 'probe2', third['epoch'], 1784) + assert state['level'] == 3 and state['retry_at'] == 2684 + third = await call('enter', 'probe3', now=2684) + state = await call('failure', 'probe3', third['epoch'], 2685) + assert state['retry_at'] == 3585 + probe = await call('enter', 'crashed', now=3585) + await call('heartbeat', 'crashed', probe['epoch'], 4000) + assert not (await call('enter', 'other', now=4200))['admitted'] + recovered = await call('enter', 'other', now=4661) + assert recovered['admitted'] + assert (await call('success', 'other', recovered['epoch'], 4662))['level'] == 0 + # 窗口边界：过期失败不积累，显式释放试探不冒充恢复成功。 + epoch = (await call('read'))['epoch'] + await call('failure', 'a', epoch, 5000) + await call('failure', 'b', epoch, 5061) + assert (await call('failure', 'c', epoch, 5062))['level'] == 0 + await call('failure', 'd', epoch, 5063) + probe = await call('enter', 'cancelled', now=5243) + assert probe['admitted'] + await call('release', 'cancelled', probe['epoch'], 5244) + assert (await call('enter', 'replacement', now=5244))['admitted'] + # 另一个接入不受此故障影响。 + independent = await redis.eval(script, 1, key+'-other', 'enter', 'fresh', -1, 5244) + assert json.loads(independent)['admitted'] + finally: + await redis.delete(key) + await redis.aclose() + asyncio.run(run()) + + + +if __name__ == "__main__": + main() + print("AI recovery atomic acceptance passed") diff --git a/web/src/common/apis/chat/type.ts b/web/src/common/apis/chat/type.ts index 7f15b82..b098dc8 100644 --- a/web/src/common/apis/chat/type.ts +++ b/web/src/common/apis/chat/type.ts @@ -51,7 +51,8 @@ export interface SendChatRequestData { } export type ChatStreamEvent - = | { name: "prompt", data: { estimated_tokens: number, context_window: number, provider: string } } + = | { name: "retry", data: { scope: string, attempt: number, retry_at: string, reason: string } } + | { name: "prompt", data: { estimated_tokens: number, context_window: number, provider: string } } | { name: "thinking", data: Record } | { name: "done", data: Record } | { name: "context_updated", data: { id: number } } diff --git a/web/src/common/apis/news-research/index.ts b/web/src/common/apis/news-research/index.ts index 91c15b7..9eb5b19 100644 --- a/web/src/common/apis/news-research/index.ts +++ b/web/src/common/apis/news-research/index.ts @@ -3,7 +3,7 @@ import { request } from "@/http/axios" export const getSettingsApi = () => request({ url: "/risk/settings", method: "get" }) export const saveSettingsApi = (data: Research.Config & { start_at: string, base_revision: number }) => request({ url: "/risk/settings", method: "put", data }) -export const enableApi = () => request({ url: "/risk/enable", method: "post" }) +export const enableApi = () => request({ url: "/risk/enable", method: "post", timeout: 6200000 }) export const getCoverageApi = (start_at: string) => request({ url: "/risk/coverage", method: "get", params: { start_at } }) export const pauseApi = () => request({ url: "/risk/pause", method: "post" }) export const createSkillsApi = () => request<{ index: number, research: number }>({ url: "/risk/skill-templates", method: "post" }) diff --git a/web/src/common/apis/news-research/type.ts b/web/src/common/apis/news-research/type.ts index 9ae6a3e..ee4dc6a 100644 --- a/web/src/common/apis/news-research/type.ts +++ b/web/src/common/apis/news-research/type.ts @@ -1,7 +1,7 @@ export interface Keyword { canonical: string, aliases: string[], quote: string } export interface ModelConfig { provider_id: number | null, model: string, reasoning_effort: string | null, skill_id: number | null } export interface Config { index: ModelConfig, research: ModelConfig, concurrency: number } -export interface QueueStatus { running: number, capacity: number, running_by_stage: { index: number, assign: number, research: number }, draining: boolean, realtime: Record, backfill: Record, events: { realtime: Record, backfill: Record }, idle_reasons: string[] } +export interface QueueStatus { waiting_retry?: number, running: number, capacity: number, running_by_stage: { index: number, assign: number, research: number }, draining: boolean, realtime: Record, backfill: Record, events: { realtime: Record, backfill: Record }, idle_reasons: string[] } export interface Settings { control_revision: number, realtime_from: string | null, queue: QueueStatus, batch?: null, config: Partial, enabled: boolean, start_at: string } export interface Coverage { start_at: string, checked_at: string, total: number, unprocessed: number, missing: number, stale: number, pending: number, processing: number, failed: number, completed: number, outside: { unprocessed: number, stale: number, pending: number, processing: number, failed: number } } export interface Package { id: number, news_id: number, title: string, url: string, state: string, error: string | null, keywords: Keyword[], events: { id: number, title: string, reason: string, quote: string }[] } diff --git a/web/src/common/apis/settings/index.ts b/web/src/common/apis/settings/index.ts index 361a936..1e8ec2f 100644 --- a/web/src/common/apis/settings/index.ts +++ b/web/src/common/apis/settings/index.ts @@ -45,13 +45,13 @@ export function deleteProviderApi(id: number) { }) } -/** 测试模型连接；按后端 SDK 重试上限等待 */ +/** 测试模型连接；按单请求重试上限等待 */ export function testProviderApi(id: number) { return request({ url: `/settings/providers/${id}/test`, method: "post", data: null, - timeout: 1900000 + timeout: 2100000 }) } @@ -89,3 +89,6 @@ export function testExchangeApi() { timeout: 30000 }) } + +export interface ProviderAvailability { provider_id: number, name: string, state: string, retry_at: string | null, level: number } +export const listProviderAvailabilityApi = () => request({ url: "/settings/provider-availability", method: "get", notify: false }) diff --git a/web/src/common/components/AiAvailability/index.vue b/web/src/common/components/AiAvailability/index.vue new file mode 100644 index 0000000..c0601e1 --- /dev/null +++ b/web/src/common/components/AiAvailability/index.vue @@ -0,0 +1,40 @@ + + + diff --git a/web/src/pages/chat/components/MessageList.vue b/web/src/pages/chat/components/MessageList.vue index 014a90b..e495efd 100644 --- a/web/src/pages/chat/components/MessageList.vue +++ b/web/src/pages/chat/components/MessageList.vue @@ -51,6 +51,9 @@ function prettyResult(result: string) { AlphaPulse AI {{ item.time ? formatTime(item.time) : "" }} +
+ {{ item.retryNotice }} +
思考中… diff --git a/web/src/pages/chat/index.vue b/web/src/pages/chat/index.vue index 6ac2e80..6bf14d0 100644 --- a/web/src/pages/chat/index.vue +++ b/web/src/pages/chat/index.vue @@ -91,15 +91,21 @@ const onEvent: ChatStreamHandler = (event) => { estimatedTokens.value = data.estimated_tokens // OpenAI 格式推理期间没有任何推送，先显示等待状态直到出字 currentAssistant().thinking = true + } else if (name === "retry") { + const item = currentAssistant() + item.thinking = false + item.retryNotice = `${data.reason}；下次尝试 ${new Date(data.retry_at).toLocaleString()}` } else if (name === "thinking") { currentAssistant().thinking = true } else if (name === "delta") { const item = currentAssistant() item.thinking = false + item.retryNotice = undefined item.text += data.text } else if (name === "usage") { // 一次模型调用结束；之后的工具调用挂在这条回复上，新的文字另起一条 const item = currentAssistant() + item.retryNotice = undefined item.closed = true item.thinking = false } else if (name === "tool_call") { @@ -124,6 +130,7 @@ const onEvent: ChatStreamHandler = (event) => { } else if (name === "error") { const item = currentAssistant() item.thinking = false + item.retryNotice = undefined item.error = data.message } scrollToBottom() diff --git a/web/src/pages/chat/type.ts b/web/src/pages/chat/type.ts index 9731391..41e1383 100644 --- a/web/src/pages/chat/type.ts +++ b/web/src/pages/chat/type.ts @@ -17,6 +17,7 @@ export interface DisplayItem { closed?: boolean /** 生成中：模型正在思考，还没有输出文字 */ thinking?: boolean + retryNotice?: string /** 已落库的消息时间；生成中的没有 */ time?: string } diff --git a/web/src/pages/home/index.vue b/web/src/pages/home/index.vue index 004b5cd..478d610 100644 --- a/web/src/pages/home/index.vue +++ b/web/src/pages/home/index.vue @@ -5,6 +5,7 @@ import type { SkillNode } from "@@/apis/skills/type" import * as api from "@@/apis/news-research" import { listProviderModelsApi, listProvidersApi } from "@@/apis/settings" import { listSkillTreeApi } from "@@/apis/skills" +import AiAvailability from "@@/components/AiAvailability/index.vue" import { EFFORTS } from "@@/constants/ai" import dayjs from "dayjs" @@ -372,6 +373,7 @@ onBeforeUnmount(() => {