feat: add trace context and resilience utilities for LLM calls

src/analysis/utils/llm_utils.py

@@ -7,6 +7,15 @@ import asyncio
 from typing import Any
 
 from astrbot.api import logger
+from ...utils.resilience import CircuitBreaker, global_llm_rate_limiter
+
+_circuit_breakers = {}
+
+
+def _get_circuit_breaker(provider_id: str) -> CircuitBreaker:
+    if provider_id not in _circuit_breakers:
+        _circuit_breakers[provider_id] = CircuitBreaker(name=f"provider_{provider_id}")
+    return _circuit_breakers[provider_id]
 
 
 async def _try_get_provider_id_by_id(
@@ -192,7 +201,8 @@ async def call_provider_with_retry(
     Returns:
         LLM生成的结果，失败时返回None
     """
-    timeout = config_manager.get_llm_timeout()
+    # 注意: 超时由 AstrBot Provider 内部配置控制，不再使用插件层 asyncio.wait_for
+    # 用户可在 AstrBot WebUI 中为每个 Provider 配置 timeout 参数
     retries = config_manager.get_llm_retries()
     backoff = config_manager.get_llm_backoff()
 
@@ -225,26 +235,35 @@ async def call_provider_with_retry(
                 )
                 return None
 
-            # 使用新的 llm_generate API
-            # 注意：llm_generate 可能不直接支持 max_tokens 和 temperature 参数，
-            # 取决于 AstrBot 版本和具体实现。如果支持 kwargs，可以传递。
-            # 这里假设支持 kwargs 传递给底层 provider。
-            # 使用 asyncio.wait_for 包裹，继续遵守 timeout 参数并在超时时抛出 TimeoutError。
-            llm_resp = await asyncio.wait_for(
-                context.llm_generate(
-                    chat_provider_id=provider_id,
-                    prompt=prompt,
-                    max_tokens=max_tokens,
-                    temperature=temperature,
-                ),
-                timeout=timeout,
-            )
+            # 获取熔断器
+            cb = _get_circuit_breaker(provider_id)
+            if not cb.allow_request():
+                logger.warning(f"Provider {provider_id} 熔断器已打开，跳过本次请求")
+                return None
 
-            return llm_resp
+            # 使用全局限流器 + 熔断器记录
+            # 超时由 Provider 内部控制，无需外层 wait_for
+            try:
+                async with global_llm_rate_limiter:
+                    llm_resp = await context.llm_generate(
+                        chat_provider_id=provider_id,
+                        prompt=prompt,
+                        max_tokens=max_tokens,
+                        temperature=temperature,
+                    )
+
+                # 成功记录
+                cb.record_success()
+                return llm_resp
+
+            except Exception as e:
+                # 失败记录
+                cb.record_failure()
+                raise e
 
         except asyncio.TimeoutError as e:
             last_exc = e
-            logger.warning(f"LLM请求超时: 第{attempt}次, timeout={timeout}s")
+            logger.warning(f"LLM请求超时: 第{attempt}次 (Provider 内部超时)")
         except Exception as e:
             last_exc = e
             logger.warning(f"LLM请求失败: 第{attempt}次, 错误: {last_exc}")

src/utils/resilience.py

@@ -0,0 +1,101 @@
+import time
+import asyncio
+from typing import Dict
+from astrbot.api import logger
+
+
+class CircuitBreaker:
+    """
+    简单的熔断器实现 (Simple Circuit Breaker)
+    """
+
+    STATE_CLOSED = "CLOSED"
+    STATE_OPEN = "OPEN"
+    STATE_HALF_OPEN = "HALF_OPEN"
+
+    def __init__(
+        self,
+        failure_threshold: int = 5,
+        recovery_timeout: int = 60,
+        name: str = "default",
+    ):
+        self.name = name
+        self.failure_threshold = failure_threshold
+        self.recovery_timeout = recovery_timeout
+
+        self.failure_count = 0
+        self.state = self.STATE_CLOSED
+        self.last_failure_time = 0
+
+    def record_failure(self):
+        """记录一次失败"""
+        self.failure_count += 1
+        if (
+            self.state == self.STATE_CLOSED
+            and self.failure_count >= self.failure_threshold
+        ):
+            self._open_circuit()
+        elif self.state == self.STATE_HALF_OPEN:
+            # 在半开状态下，一次失败直接重新打开熔断器
+            self._open_circuit()
+
+    def record_success(self):
+        """记录一次成功"""
+        if self.state == self.STATE_HALF_OPEN:
+            self._close_circuit()
+        elif self.state == self.STATE_CLOSED:
+            # 成功则重置失败计数 (可选，这里选择连续失败才熔断)
+            self.failure_count = 0
+
+    def allow_request(self) -> bool:
+        """是否允许请求"""
+        if self.state == self.STATE_OPEN:
+            if time.time() - self.last_failure_time > self.recovery_timeout:
+                self._half_open_circuit()
+                return True
+            return False
+        return True
+
+    def _open_circuit(self):
+        self.state = self.STATE_OPEN
+        self.last_failure_time = time.time()
+        logger.warning(
+            f"CircuitBreaker[{self.name}] 熔断器已打开! 暂停请求 {self.recovery_timeout} 秒。"
+        )
+
+    def _close_circuit(self):
+        self.state = self.STATE_CLOSED
+        self.failure_count = 0
+        logger.info(f"CircuitBreaker[{self.name}] 熔断器已关闭，服务恢复。")
+
+    def _half_open_circuit(self):
+        self.state = self.STATE_HALF_OPEN
+        logger.info(f"CircuitBreaker[{self.name}] 进入半开状态，尝试恢复...")
+
+
+class GlobalRateLimiter:
+    """
+    全局限流器 (Global Rate Limiter)
+    使用 asyncio.Semaphore 控制并发数
+    """
+
+    _instance = None
+    _semaphore = None
+
+    @classmethod
+    def get_instance(cls, max_concurrency: int = 3):
+        if cls._instance is None:
+            cls._instance = cls()
+            cls._semaphore = asyncio.Semaphore(max_concurrency)
+        return cls._instance
+
+    @property
+    def semaphore(self):
+        if self._semaphore is None:
+            # Fallback if accessed before get_instance called with arg
+            self._semaphore = asyncio.Semaphore(3)
+        return self._semaphore
+
+
+# 默认全局限流实例
+global_llm_rate_limiter = GlobalRateLimiter.get_instance(max_concurrency=3).semaphore

src/utils/trace_context.py

@@ -0,0 +1,50 @@
+import contextvars
+import logging
+import uuid
+import time
+
+# 定义 ContextVar
+_trace_id_ctx = contextvars.ContextVar("trace_id", default="")
+
+class TraceContext:
+    """
+    链路追踪上下文管理器
+    """
+    
+    @staticmethod
+    def set(trace_id: str):
+        """设置当前上下文的 TraceID"""
+        return _trace_id_ctx.set(trace_id)
+
+    @staticmethod
+    def get() -> str:
+        """获取当前上下文的 TraceID"""
+        return _trace_id_ctx.get()
+        
+    @staticmethod
+    def generate(prefix: str = "") -> str:
+        """生成一个新的 TraceID (Prefix + Timestamp + UUID前8位)"""
+        timestamp = int(time.time())
+        unique_id = str(uuid.uuid4())[:8]
+        if prefix:
+            return f"{prefix}-{timestamp}-{unique_id}"
+        return f"{timestamp}-{unique_id}"
+
+    @staticmethod
+    def clear():
+        """清除当前上下文的 TraceID"""
+        _trace_id_ctx.set("")
+
+class TraceLogFilter(logging.Filter):
+    """
+    日志过滤器，自动注入 TraceID
+    """
+    def filter(self, record):
+        trace_id = _trace_id_ctx.get()
+        if trace_id:
+            # 将 trace_id 注入到 record 中，同时也修改 msg 以便在不支持自定义 format 的 logger 中也能看到
+            record.trace_id = trace_id
+            record.msg = f"[{trace_id}] {record.msg}"
+        else:
+            record.trace_id = ""
+        return True