From f8a985aad886d873860ca20cb2786739cd51dbb2 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Wed, 13 May 2026 16:20:54 +0800 Subject: [PATCH] =?UTF-8?q?feat(channels):=20=E6=96=B0=E5=A2=9E=E4=BC=9A?= =?UTF-8?q?=E8=AF=9D=E8=B7=AF=E7=94=B1=E3=80=81=E9=94=99=E8=AF=AF=E5=88=86?= =?UTF-8?q?=E7=B1=BB=E4=B8=8E=E8=81=8A=E5=A4=A9=E8=83=BD=E5=8A=9B=E6=94=AF?= =?UTF-8?q?=E6=8C=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. 新增ChannelStreamingProtocol协议的打字指示器、消息编辑支持和 fallback 发送方法 2. 重构SDK导入顺序，调整normalizer和retry的导入位置 3. 新增错误分类模块，实现异常类型归类逻辑 4. 新增SessionRouter抽象基类，实现基础会话路由逻辑 5. 增强重试SDK，添加总超时、熔断和指标回调支持 --- .../yuxi/channels/protocols/session_router.py | 64 +++++++++++++++++++ .../yuxi/channels/protocols/streaming.py | 12 ++++ backend/package/yuxi/channels/sdk/__init__.py | 16 ++--- .../yuxi/channels/sdk/error_classifier.py | 52 +++++++++++++++ backend/package/yuxi/channels/sdk/retry.py | 37 ++++++++++- 5 files changed, 172 insertions(+), 9 deletions(-) create mode 100644 backend/package/yuxi/channels/protocols/session_router.py create mode 100644 backend/package/yuxi/channels/sdk/error_classifier.py diff --git a/backend/package/yuxi/channels/protocols/session_router.py b/backend/package/yuxi/channels/protocols/session_router.py new file mode 100644 index 00000000..433388a2 --- /dev/null +++ b/backend/package/yuxi/channels/protocols/session_router.py @@ -0,0 +1,64 @@ +from __future__ import annotations + +from abc import ABC, abstractmethod +from typing import TYPE_CHECKING + +from yuxi.channels.models import SessionScope + +if TYPE_CHECKING: + from yuxi.channels.models import ChannelMessage + + +class SessionRouter(ABC): + def __init__(self, channel_type: str, account_id: str = "default"): + self.channel_type = channel_type + self.account_id = account_id + + @abstractmethod + def resolve_thread_key(self, message: ChannelMessage) -> str: + """解析消息对应的线程键""" + + @abstractmethod + def resolve_chat_id(self, message: ChannelMessage) -> str: + """解析聊天ID""" + + @abstractmethod + def resolve_session_scope(self, message: ChannelMessage) -> SessionScope: + """解析会话范围级别""" + + @abstractmethod + def normalize_target(self, target: str) -> str: + """标准化目标地址""" + + def resolve_agent_route(self, message: ChannelMessage) -> str: + thread_key = self.resolve_thread_key(message) + scope = self.resolve_session_scope(message) + return f"agent:main:{thread_key}:{scope.value}" + + def _build_key(self, scope: str, type_str: str, id_str: str) -> str: + return f"{self.channel_type}:{self.account_id}:{scope}:{type_str}:{id_str}" + + +class BaseSessionRouter(SessionRouter): + def normalize_target(self, target: str) -> str: + if target.startswith(f"{self.channel_type}:"): + return target + return f"{self.channel_type}:{target}" + + def resolve_thread_key(self, message: ChannelMessage) -> str: + identity = message.identity + chat_type = message.chat_type.value if hasattr(message.chat_type, "value") else str(message.chat_type) + return self._build_key("chat", chat_type, identity.channel_chat_id) + + def resolve_chat_id(self, message: ChannelMessage) -> str: + return message.identity.channel_chat_id + + def resolve_session_scope(self, message: ChannelMessage) -> SessionScope: + from yuxi.channels.models import ChatType + + chat_type = message.chat_type + if chat_type == ChatType.DIRECT: + return SessionScope.DIRECT + if chat_type == ChatType.THREAD: + return SessionScope.THREAD + return SessionScope.GROUP diff --git a/backend/package/yuxi/channels/protocols/streaming.py b/backend/package/yuxi/channels/protocols/streaming.py index a0caac8a..9928476a 100644 --- a/backend/package/yuxi/channels/protocols/streaming.py +++ b/backend/package/yuxi/channels/protocols/streaming.py @@ -14,4 +14,16 @@ class ChannelStreamingProtocol(Protocol): @property def block_streaming(self) -> bool: ... + @property + def supports_typing_indicator(self) -> bool: + return getattr(self, "capabilities", None) is not None and getattr(self.capabilities, "typing", False) + + @property + def supports_message_edit(self) -> bool: + return getattr(self, "capabilities", None) is not None and getattr(self.capabilities, "edit", False) + async def send_stream_chunk(self, chat_id: str, msg_id: str, chunk: str, finished: bool) -> DeliveryResult: ... + + async def send_typing_indicator(self, chat_id: str, active: bool) -> DeliveryResult: ... + + async def fallback_send(self, chat_id: str, text: str) -> DeliveryResult: ... diff --git a/backend/package/yuxi/channels/sdk/__init__.py b/backend/package/yuxi/channels/sdk/__init__.py index f60ac076..4d6d031e 100644 --- a/backend/package/yuxi/channels/sdk/__init__.py +++ b/backend/package/yuxi/channels/sdk/__init__.py @@ -1,12 +1,5 @@ -from yuxi.channels.sdk.retry import RetryConfig, with_retry -from yuxi.channels.sdk.dedupe import DedupeConfig, MessageDeduplicator from yuxi.channels.sdk.chunking import ChunkConfig, chunk_message -from yuxi.channels.sdk.normalizer import ( - extract_mentions, - extract_urls, - normalize_timestamp, - normalize_user_id, -) +from yuxi.channels.sdk.dedupe import DedupeConfig, MessageDeduplicator from yuxi.channels.sdk.format import ( TRUNCATION_MARKER, build_basic_outbound, @@ -14,6 +7,13 @@ from yuxi.channels.sdk.format import ( strip_html_tags, truncate_text, ) +from yuxi.channels.sdk.normalizer import ( + extract_mentions, + extract_urls, + normalize_timestamp, + normalize_user_id, +) +from yuxi.channels.sdk.retry import RetryConfig, with_retry __all__ = [ "RetryConfig", diff --git a/backend/package/yuxi/channels/sdk/error_classifier.py b/backend/package/yuxi/channels/sdk/error_classifier.py new file mode 100644 index 00000000..9d5f103a --- /dev/null +++ b/backend/package/yuxi/channels/sdk/error_classifier.py @@ -0,0 +1,52 @@ +from __future__ import annotations + +from enum import StrEnum + +from yuxi.channels.exceptions import ( + ChannelAuthenticationError, + ChannelException, + ChannelNotConnectedError, + ChannelRateLimitError, + ChannelTimeoutError, + MessageTooLargeError, + TokenExpiredError, +) + + +class ErrorCategory(StrEnum): + TRANSIENT = "transient" + PERMANENT = "permanent" + RATE_LIMIT = "rate_limit" + AUTH_EXPIRED = "auth_expired" + CIRCUIT_OPEN = "circuit_open" + TIMEOUT = "timeout" + + +_ERROR_CATEGORY_MAP: dict[type[Exception], ErrorCategory] = { + ChannelNotConnectedError: ErrorCategory.TRANSIENT, + ChannelRateLimitError: ErrorCategory.RATE_LIMIT, + TokenExpiredError: ErrorCategory.AUTH_EXPIRED, + ChannelAuthenticationError: ErrorCategory.PERMANENT, + ChannelTimeoutError: ErrorCategory.TIMEOUT, + MessageTooLargeError: ErrorCategory.PERMANENT, + TimeoutError: ErrorCategory.TIMEOUT, +} + + +def classify_error(error: Exception) -> ErrorCategory: + error_type = type(error) + if error_type in _ERROR_CATEGORY_MAP: + return _ERROR_CATEGORY_MAP[error_type] + + try: + from yuxi.channels.infra.circuit_breaker import CircuitBreakerOpenError + + if isinstance(error, CircuitBreakerOpenError): + return ErrorCategory.CIRCUIT_OPEN + except ImportError: + pass + + if isinstance(error, ChannelException): + return ErrorCategory.TRANSIENT if error.retryable else ErrorCategory.PERMANENT + + return ErrorCategory.TRANSIENT diff --git a/backend/package/yuxi/channels/sdk/retry.py b/backend/package/yuxi/channels/sdk/retry.py index 415b3eb5..295fbd58 100644 --- a/backend/package/yuxi/channels/sdk/retry.py +++ b/backend/package/yuxi/channels/sdk/retry.py @@ -2,13 +2,22 @@ from __future__ import annotations import asyncio import random +import time from collections.abc import Awaitable, Callable from dataclasses import dataclass +from enum import StrEnum from typing import Any from yuxi.utils.logging_config import logger +class RetryDecision(StrEnum): + RETRY = "retry" + CIRCUIT_BREAK = "circuit_break" + FALLBACK = "fallback" + FAIL = "fail" + + @dataclass class RetryConfig: max_retries: int = 3 @@ -17,6 +26,9 @@ class RetryConfig: backoff_factor: float = 2.0 jitter: bool = True retryable_exceptions: tuple[type[Exception], ...] = (Exception,) + total_timeout: float | None = None + circuit_breaker: Any = None + metrics_callback: Callable[[str, dict], None] | None = None async def with_retry( @@ -25,18 +37,41 @@ async def with_retry( ) -> Any: cfg = config or RetryConfig() last_exception: Exception | None = None + start_time = time.monotonic() for attempt in range(cfg.max_retries + 1): + if cfg.total_timeout is not None and (time.monotonic() - start_time) >= cfg.total_timeout: + raise last_exception or Exception("Retry total_timeout exceeded") + + if cfg.circuit_breaker is not None: + cb = cfg.circuit_breaker + if hasattr(cb, "state") and getattr(cb, "state", None) is not None: + from yuxi.channels.infra.circuit_breaker import CircuitState + + if cb.state == CircuitState.OPEN: + if cfg.metrics_callback: + cfg.metrics_callback("circuit_breaker_skip", {"attempt": attempt}) + raise last_exception or Exception("Circuit breaker is OPEN, skipping retry") + try: - return await func() + result = await func() + if cfg.metrics_callback: + cfg.metrics_callback("retry_success", {"attempt": attempt}) + return result except cfg.retryable_exceptions as e: last_exception = e if attempt == cfg.max_retries: break + if cfg.total_timeout is not None and (time.monotonic() - start_time) >= cfg.total_timeout: + break delay = min(cfg.base_delay * (cfg.backoff_factor**attempt), cfg.max_delay) if cfg.jitter: delay *= random.uniform(0.5, 1.5) + if cfg.metrics_callback: + cfg.metrics_callback("retry_attempt", {"attempt": attempt + 1, "error": str(e), "delay": delay}) logger.debug(f"[SDK/Retry] Attempt {attempt + 1}/{cfg.max_retries} failed: {e}, retrying in {delay:.2f}s") await asyncio.sleep(delay) + if cfg.metrics_callback: + cfg.metrics_callback("retry_exhausted", {"attempts": cfg.max_retries, "error": str(last_exception)}) raise last_exception # type: ignore[misc]