diff --git a/src/analysis/utils/llm_utils.py b/src/analysis/utils/llm_utils.py index e527dee..5274907 100644 --- a/src/analysis/utils/llm_utils.py +++ b/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}") diff --git a/src/utils/resilience.py b/src/utils/resilience.py new file mode 100644 index 0000000..fabb030 --- /dev/null +++ b/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 diff --git a/src/utils/trace_context.py b/src/utils/trace_context.py new file mode 100644 index 0000000..c01be87 --- /dev/null +++ b/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