Roog (顾新培)
a4d2ad1e93
变更内容： - 将 `redis` 客户端替换为 `redis.asyncio` 实现。 - 系统中同步方法调整为异步方法，提升事件循环效率。 - 在 `MetricsManager` 中添加异步初始化及关闭逻辑，避免阻塞问题。 - 更新便捷函数以支持异步上下文，并添加同步模式的兼容方法。 - 调整 Worker、JobManager、API 路由等模块，适配异步指标操作。 - 扩展单元测试，覆盖新增的异步方法及 Redis 操作逻辑。 - 简化 Dockerfile，取消开发依赖安装命令。
78 lines
2.3 KiB
Python
78 lines
2.3 KiB
Python
"""算法基类"""
|
|
|
|
from abc import ABC, abstractmethod
|
|
from typing import Any, Dict
|
|
import time
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class BaseAlgorithm(ABC):
|
|
"""算法基类，所有算法必须继承此类"""
|
|
|
|
def __init__(self):
|
|
self.name = self.__class__.__name__
|
|
self.version = "1.0.0"
|
|
|
|
@abstractmethod
|
|
def process(self, *args, **kwargs) -> Dict[str, Any]:
|
|
"""
|
|
算法处理逻辑，子类必须实现此方法
|
|
|
|
Returns:
|
|
Dict[str, Any]: 算法处理结果
|
|
"""
|
|
pass
|
|
|
|
def execute(self, *args, **kwargs) -> Dict[str, Any]:
|
|
"""
|
|
执行算法，包含埋点和错误处理
|
|
|
|
Returns:
|
|
Dict[str, Any]: 包含结果和元数据的字典
|
|
"""
|
|
from ..core.metrics_unified import incr_sync, observe_sync
|
|
|
|
start_time = time.time()
|
|
status = "success"
|
|
|
|
try:
|
|
logger.info(f"Starting algorithm: {self.name}")
|
|
result = self.process(*args, **kwargs)
|
|
elapsed_time = time.time() - start_time
|
|
|
|
logger.info(f"Algorithm {self.name} completed successfully in {elapsed_time:.3f}s")
|
|
|
|
return {
|
|
"success": True,
|
|
"result": result,
|
|
"metadata": {
|
|
"algorithm": self.name,
|
|
"version": self.version,
|
|
"elapsed_time": elapsed_time,
|
|
},
|
|
}
|
|
|
|
except Exception as e:
|
|
status = "error"
|
|
elapsed_time = time.time() - start_time
|
|
logger.error(f"Algorithm {self.name} failed: {str(e)}", exc_info=True)
|
|
|
|
return {
|
|
"success": False,
|
|
"error": str(e),
|
|
"metadata": {
|
|
"algorithm": self.name,
|
|
"version": self.version,
|
|
"elapsed_time": elapsed_time,
|
|
},
|
|
}
|
|
finally:
|
|
# 记录算法执行指标
|
|
elapsed_time = time.time() - start_time
|
|
incr_sync("algorithm_executions_total", {"algorithm": self.name, "status": status})
|
|
observe_sync(
|
|
"algorithm_execution_duration_seconds", {"algorithm": self.name}, elapsed_time
|
|
)
|