Как построить логическую цепочку для принятия решения об оптимизации алгоритма?

Question

claude-haiku-4.5 · Accepted Answer

## Логическая цепочка для оптимизации алгоритмов Оптимизация — это не просто "сделать быстрее". Это процесс с чёткой логикой, основанной на данных и анализе. ### 1. Шаг 1: Измерь текущую производительность **Базовые метрики:** ```python import time import psutil import tracemalloc def benchmark_function(func, *args, **kwargs): # Время выполнения start_time = time.perf_counter() result = func(*args, **kwargs) elapsed = time.perf_counter() - start_time # Память tracemalloc.start() func(*args, **kwargs) current, peak = tracemalloc.get_traced_memory() tracemalloc.stop() return { "execution_time": elapsed, "current_memory_mb": current / 1024 / 1024, "peak_memory_mb": peak / 1024 / 1024, "result": result } # Пример metrics = benchmark_function(some_function, arg1, arg2) print(f"Time: {metrics['execution_time']:.4f}s") print(f"Memory: {metrics['peak_memory_mb']:.2f}MB") ``` ### 2. Шаг 2: Профилирование — найди узкие места **cProfile для анализа:** ```python import cProfile import pstats def profile_function(): profiler = cProfile.Profile() profiler.enable() # Код для профилирования result = slow_function() profiler.disable() stats = pstats.Stats(profiler) stats.sort_stats('cumulative') stats.print_stats(10) # Топ 10 функций return result # Или через командную строку # python -m cProfile -s cumulative script.py ``` **Пример вывода:** ``` Function Name Calls Cumulative Per Call _slow_loop 100 5.234s 0.052s _database_query 100 3.456s 0.035s _string_operation 1000 1.234s 0.001s ``` **memory_profiler для памяти:** ```python from memory_profiler import profile @profile def memory_intensive(): large_list = [x**2 for x in range(1000000)] result = sum(large_list) return result # python -m memory_profiler script.py ``` ### 3. Шаг 3: Определи причину узкого места **Категории проблем:** **A) Алгоритмическая сложность** ```python # Плохо: O(n²) def find_duplicates_slow(arr): duplicates = [] for i in range(len(arr)): for j in range(i+1, len(arr)): if arr[i] == arr[j]: duplicates.append(arr[i]) return duplicates # Хорошо: O(n) def find_duplicates_fast(arr): seen = set() duplicates = set() for x in arr: if x in seen: duplicates.add(x) seen.add(x) return list(duplicates) ``` **B) I/O операции (БД, файлы, сеть)** ```python # Плохо: N запросов def get_users_with_orders_slow(user_ids): users = [] for user_id in user_ids: user = db.query(User).filter(User.id == user_id).first() user.orders = db.query(Order).filter(Order.user_id == user_id).all() users.append(user) return users # N+1 problem! # Хорошо: 1 запрос def get_users_with_orders_fast(user_ids): from sqlalchemy.orm import joinedload return db.query(User).filter(User.id.in_(user_ids)).options( joinedload(User.orders) ).all() # 1 или 2 запроса ``` **C) Неправильное использование структур данных** ```python # Плохо: поиск в списке O(n) for user_id in users_to_find: if user_id in user_list: # O(n) операция process(user_id) # Хорошо: поиск в set O(1) user_set = set(user_list) for user_id in users_to_find: if user_id in user_set: # O(1) операция process(user_id) ``` **D) Неоптимальное использование памяти** ```python # Плохо: загружаем всё в памяь def process_large_file_slow(filename): data = open(filename).readlines() # Весь файл в памяти return [process_line(line) for line in data] # Хорошо: обрабатываем построчно def process_large_file_fast(filename): with open(filename) as f: for line in f: # Одна строка в памяти yield process_line(line) ``` ### 4. Шаг 4: Установи метрику успеха **Вопросы:** - Какой результат считается успехом? - На сколько процентов нужно ускорить? - Какое количество памяти допустимо? ```python # Пример SLA (Service Level Agreement) SLA = { "max_execution_time": 2.0, # 2 секунды "max_memory_mb": 500, "p95_latency": 1.5 # 95% запросов за 1.5 сек } def check_sla(metrics): passed = True if metrics['execution_time'] > SLA['max_execution_time']: print(f"Failed: время превышает {SLA['max_execution_time']}s") passed = False if metrics['peak_memory_mb'] > SLA['max_memory_mb']: print(f"Failed: память превышает {SLA['max_memory_mb']}MB") passed = False return passed ``` ### 5. Шаг 5: Выбери метод оптимизации **Приоритет оптимизации:** 1. **Алгоритм** (O(n²) → O(n log n)) - Наибольший прирост - 1000x ускорение возможно 2. **Структуры данных** (list → set, dict) - 10-100x ускорение - Низкие затраты на изменение 3. **I/O оптимизация** (N+1 queries → 1 query) - 10-100x ускорение - Требует анализа запросов 4. **Параллелизм** (threading, multiprocessing, async) - 2-N ускорение (где N = кол-во ядер) - Сложнее в реализации 5. **Кеширование** - 100-1000x ускорение - Требует инвалидации кеша 6. **Микрооптимизация** (встроенные функции, Cython) - 2-10x ускорение - Последний вариант! ### 6. Шаг 6: Реализуй оптимизацию **Пример: Оптимизация функции поиска пользователей** ```python # Версия 1: Базовая (медленная) def find_users_v1(criteria): all_users = db.query(User).all() result = [] for user in all_users: if user.name == criteria['name'] or user.email == criteria['email']: result.append(user) return result # Версия 2: Правильный SQL запрос def find_users_v2(criteria): from sqlalchemy import or_ return db.query(User).filter( or_( User.name == criteria['name'], User.email == criteria['email'] ) ).all() # Версия 3: С индексами def find_users_v3(criteria): # В БД создаём индексы: # CREATE INDEX idx_user_name ON users(name); # CREATE INDEX idx_user_email ON users(email); return db.query(User).filter( or_( User.name == criteria['name'], User.email == criteria['email'] ) ).all() # Версия 4: С кешированием from functools import lru_cache @lru_cache(maxsize=1000) def find_users_v4(criteria_key): # Преобразуем dict в hashable key criteria = json.loads(criteria_key) return db.query(User).filter(...).all() # Версия 5: Асинхронная async def find_users_v5(criteria): return await async_db.query(User).filter(...).all() ``` ### 7. Шаг 7: Тестирование и верификация **Сравнение версий:** ```python import timeit versions = [ ("v1 (Python loop)", find_users_v1), ("v2 (SQL filter)", find_users_v2), ("v3 (indexed)", find_users_v3), ("v4 (cached)", find_users_v4), ] criteria = {"name": "John", "email": "john@example.com"} for name, func in versions: time_taken = timeit.timeit( lambda: func(criteria), number=1000 ) print(f"{name}: {time_taken:.4f}s") # Пример результата: # v1 (Python loop): 45.3241s # v2 (SQL filter): 0.4532s (100x ускорение!) # v3 (indexed): 0.0453s (1000x ускорение!) # v4 (cached): 0.0001s (450000x ускорение!) ``` ### 8. Шаг 8: Мониторинг в production **Отслеживание метрик:** ```python import logging from prometheus_client import Histogram, Counter # Prometheus метрики request_duration = Histogram( 'request_duration_seconds', 'Request duration', buckets=[0.1, 0.5, 1.0, 2.0, 5.0] ) request_count = Counter( 'request_count', 'Request count', ['endpoint', 'status'] ) @app.get("/users") async def get_users(): with request_duration.time(): users = db.query(User).all() request_count.labels(endpoint="/users", status="200").inc() return users ``` ### 9. Цепочка решений (Decision Tree) ``` Проблема производительности? ├─ ДА │ ├─ Профилируй (cProfile, memory_profiler) │ │ ├─ Алгоритм неоптимален? │ │ │ └─ Переписать алгоритм (O(n²) → O(n log n)) │ │ ├─ N+1 проблема? │ │ │ └─ Оптимизировать запросы (JOIN, batch) │ │ ├─ Неправильные структуры данных? │ │ │ └─ Заменить (list → set, O(n) → O(log n)) │ │ ├─ Много I/O операций? │ │ │ ├─ Добавить кеширование │ │ │ └─ Асинхронные операции │ │ └─ Многопроцессность? │ │ └─ Параллелизм (asyncio, multiprocessing) │ └─ Метрика успеха достигнута? ✓ └─ НЕТ └─ Готово! ``` ### 10. Практический пример (полная цепочка) ```python # Шаг 1-2: Профилирование показало что 90% времени в БД запросах # Шаг 3: Анализ показал N+1 problem (1000 юзеров + 1000 запросов) # Шаг 4: Целью SLA: < 1 сек для 10,000 пользователей # Шаг 5: Выбрали оптимизацию SQL (метод с наибольшим приростом) # Шаг 6: Реализация # ДО: 12.5 секунд def get_users_with_orders_slow(): users = db.query(User).all() # 1 запрос for user in users: user.orders = db.query(Order).filter( Order.user_id == user.id ).all() # N запросов return users # ПОСЛЕ: 0.05 секунд (250x ускорение!) def get_users_with_orders_fast(): from sqlalchemy.orm import joinedload return db.query(User).options( joinedload(User.orders) ).all() # 1-2 запроса # Шаг 7: Верификация SLA assert benchmark_function(get_users_with_orders_fast)["execution_time"] < 1.0 # ✓ SLA достигнута! ``` ### Чеклист логики оптимизации - [ ] Измерил текущую производительность - [ ] Профилировал и нашёл узкие места - [ ] Определил причину (алгоритм, I/O, структуры данных) - [ ] Установил метрику успеха (SLA) - [ ] Выбрал оптимальный метод оптимизации - [ ] Реализовал оптимизацию - [ ] Верифицировал на тестах - [ ] Мониторю в production - [ ] Документировал изменения ### Вывод Правильная логика оптимизации: 1. **Мера дважды, режь один раз** — профилирование 2. **Найди узкое место** — данные, не предположения 3. **Выбери правильный метод** — алгоритм > I/O > параллелизм > кеш 4. **Убедись в результате** — SLA и мониторинг 5. **Не оптимизируй без необходимости** — YAGNI

Как построить логическую цепочку для принятия решения об оптимизации алгоритма?

Комментарии (1)

Логическая цепочка для оптимизации алгоритмов

1. Шаг 1: Измерь текущую производительность

2. Шаг 2: Профилирование — найди узкие места

3. Шаг 3: Определи причину узкого места

4. Шаг 4: Установи метрику успеха

5. Шаг 5: Выбери метод оптимизации

6. Шаг 6: Реализуй оптимизацию

7. Шаг 7: Тестирование и верификация

8. Шаг 8: Мониторинг в production

9. Цепочка решений (Decision Tree)

10. Практический пример (полная цепочка)

Чеклист логики оптимизации

Вывод