Архитектура AI-лаборатории: Docker Swarm, FastAPI и 27 агентов — путь от одного скрипта

# agent_zero.py — тот самый первый скрипт, который сломался на второй день
import pandas as pd
import openpyxl

def match_samples(csv_path, excel_path):
    # Читаем данные из анализатора (CSV)
    df_lab = pd.read_csv(csv_path, encoding='windows-1251') # потому что наш древний анализатор так выдает
    # Читаем журнал регистрации (Excel)
    df_reg = pd.read_excel(excel_path, sheet_name='Журнал')

    # Сопоставляем по номеру пробы
    merged = pd.merge(df_lab, df_reg, on='Номер пробы', how='left')

    # Сохраняем результат
    merged.to_excel('result.xlsx', index=False)

if __name__ == "__main__":
    match_samples('data/2023-10-01_analyses.csv', 'data/registration.xlsx')

# agent_sanitizer.py — первая попытка внедрить «интеллект»
from rapidfuzz import fuzz, process
import pandas as pd

def fuzzy_match_samples(df_lab, df_reg):
    results = []

    for idx, lab_row in df_lab.iterrows():
        sample_id_lab = str(lab_row['Номер пробы']).strip()

        # Получаем список всех номеров пробы из журнала
        reg_ids = [str(id).strip() for id in df_reg['Номер пробы'].tolist()]

        # Нечёткое сравнение: находим лучшее соответствие
        best_match, score, index = process.extractOne(sample_id_lab, reg_ids, scorer=fuzz.ratio)

        if score >= 85: # Если совпадение больше 85% — считаем, что это та же проба
            matched_row = df_reg.iloc[index]
            # Объединяем данные
            merged_row = {**lab_row.to_dict(), **matched_row.to_dict()}
            results.append(merged_row)
        else:
            # Если соответствие не найдено — помечаем для ручной проверки
            lab_row['Статус'] = 'НЕ СОПОСТАВЛЕН'
            results.append(lab_row.to_dict())

    return pd.DataFrame(results)

# docker-compose.yml — версия 1.0, которая упала при первом запуске
version: '3.8'

services:
  agent-sanitizer:
    build: .
    volumes:
      - ./data:/app/data  # Плохая идея, но я тогда не знал
      - ./output:/app/output
    command: python agent_sanitizer.py

# initial_hell.py
import pandas as pd

# Читаем файл регистрации (CSV, который медсестра заполняет руками)
df = pd.read_csv('registration.csv', encoding='windows-1251')
print(df.head())

UnicodeDecodeError: 'utf-8' codec can't decode byte 0xff in position 0: invalid start byte

   Номер пробы  Фамилия Имя  Дата рождения  Назначенный анализ
0        12345   Иванов И.И.        01.01.1980            Глюкоза

# naive_matcher.py
import pandas as pd

registration_df = pd.read_csv('registration.csv', encoding='windows-1251')
results_df = pd.read_csv('results_biochemistry.csv')

# Пытаемся сопоставить по номеру пробы
merged_df = pd.merge(registration_df, results_df, on='Номер пробы', how='left')
print(f"Сопоставлено: {merged_df['Результат'].notna().sum()} из {len(merged_df)}")

def clean_sample_id(raw_id):
    try:
        # Убираем буквы, лишние пробелы
        cleaned = str(raw_id).strip().upper()
        if cleaned.endswith('А'):
            cleaned = cleaned.replace('А', '')
        return int(cleaned)
    except:
        return None

# candidate_matcher_v1.py
import pandas as pd
import re

def match_candidates(reg_row, results_df):
    candidates = []
    sample_id_raw = reg_row['Номер пробы']

    # Стратегия 1: Чистый номер (int)
    try:
        clean_id = int(re.sub(r'[^0-9]', '', str(sample_id_raw)))
        match = results_df[results_df['SampleID'] == clean_id]
        if len(match) == 1:
            candidates.append({'strategy': 'clean_int', 'data': match.iloc[0], 'confidence': 0.9})
    except:
        pass

    # Стратегия 2: Поиск по части строки (если есть буква)
    if any(char.isalpha() for char in str(sample_id_raw)):
        for _, result_row in results_df.iterrows():
            if str(sample_id_raw) in str(result_row['SampleID']) or str(result_row['SampleID']) in str(sample_id_raw):
                candidates.append({'strategy': 'substring', 'data': result_row, 'confidence': 0.6})

    # Стратегия 3: По имени и дате (если аппарат их передал)
    # ... (этот код был длинным и жалким)

    return candidates

# Применяем к каждой пробе
registration_df = pd.read_csv('registration.csv', encoding='windows-1251')
results_df = pd.read_csv('results_biochemistry.csv')

matched_count = 0
for index, row in registration_df.iterrows():
    candidates = match_candidates(row, results_df)
    if candidates:
        # Выбираем кандидата с максимальной confidence
        best_match = max(candidates, key=lambda x: x['confidence'])
        # ... записываем результат
        matched_count += 1

print(f"Сопоставлено кандидатами: {matched_count} из {len(registration_df)}")

# agent_v0.1_hell.py - НЕ ДЕЛАЙТЕ ТАК
import os
import shutil
from datetime import datetime

def process_folder(base_path):
    for root, dirs, files in os.walk(base_path):
        for file in files:
            if file.endswith('.pdf'):
                # Это PDF-заявка от отделения?
                if 'заявка' in file.lower():
                    if 'хирургия' in root.lower():
                        dest = '/archive/хирургия/заявки/'
                    elif 'терапия' in root.lower():
                        dest = '/archive/терапия/заявки/'
                    else:
                        dest = '/archive/разное/'
                    # А если файл старше 30 дней?
                    filepath = os.path.join(root, file)
                    file_time = os.path.getmtime(filepath)
                    if datetime.now().timestamp() - file_time > 2592000:
                        dest = dest.replace('заявки', 'старые_заявки')
                    # Но если это срочный анализ?
                    if 'cito' in file.lower():
                        dest = '/priority/cito/' + dest
                # А может, это результат анализа?
                elif 'результат' in file.lower():
                    # Проверить, есть ли ID пациента в имени?
                    # ... и так ещё 200 строк if-else

# agent_v0.2_skeleton.py - уже теплее
import os
import json
from datetime import datetime
from typing import Dict, Any, Optional
from dataclasses import dataclass, asdict

@dataclass
class AgentState:
    name: str
    current_task: Optional[str] = None
    processed_items: int = 0
    last_error: Optional[str] = None
    context: Dict[str, Any] = None

    def to_dict(self):
        return asdict(self)

    def update_context(self, key: str, value: Any):
        if self.context is None:
            self.context = {}
        self.context[key] = value

class BaseAgent:
    def __init__(self, name: str, work_dir: str):
        self.state = AgentState(name=name)
        self.work_dir = work_dir
        self._load_state()  # Пытаемся восстановиться после падения

    def _load_state(self):
        state_file = os.path.join(self.work_dir, f"{self.state.name}_state.json")
        if os.path.exists(state_file):
            with open(state_file, 'r', encoding='utf-8') as f:
                data = json.load(f)
                self.state = AgentState(**data)
                print(f"[{self.state.name}] State loaded. Processed: {self.state.processed_items}")

    def _save_state(self):
        state_file = os.path.join(self.work_dir, f"{self.state.name}_state.json")
        with open(state_file, 'w', encoding='utf-8') as f:
            json.dump(self.state.to_dict(), f, ensure_ascii=False, indent=2)

    def run(self):
        """Основной цикл агента. Шаблонный метод."""
        try:
            self._pre_run()
            self._process()
            self._post_run()
        except Exception as e:
            self.state.last_error = str(e)
            self._save_state()
            raise

    def _pre_run(self):
        self.state.current_task = "starting"
        print(f"[{self.state.name}] Starting at {datetime.now()}")

    def _process(self):
        """Эту логику будут переопределять наследники."""
        raise NotImplementedError

    def _post_run(self):
        self.state.current_task = None
        self.state.last_error = None
        self._save_state()
        print(f"[{self.state.name}] Finished. Total processed: {self.state.processed_items}")

# sanitary_agent_v1.py - первый боевой агент
import os
import shutil
import hashlib
from pathlib import Path
from .base_agent import BaseAgent

class SanitaryAgent(BaseAgent):
    def __init__(self, work_dir: str, inbox_path: str):
        super().__init__(name="SanitaryAgent_v1", work_dir=work_dir)
        self.inbox_path = Path(inbox_path)
        self.processed_base = Path(work_dir) / "processed"
        self.processed_base.mkdir(parents=True, exist_ok=True)
        # Простая, но расширяемая классификация по ключевым словам
        self.rules = {
            'application': ['заявка', 'направление', 'request', 'app'],
            'result': ['результат', 'анализ', 'отчет', 'result', 'lab'],
            'scan': ['скан', 'паспорт', 'страховка', 'scan', 'id'],
            'image': ['.jpg', '.jpeg', '.png', '.tiff']  # для микроскопии
        }

    def _classify_file(self, file_path: Path) -> str:
        """Определяем тип файла. Пока по имени, потом добавим ML."""
        name = file_path.name.lower()
        suffix = file_path.suffix.lower()

        for file_type, keywords in self.rules.items():
            if any(keyword in name for keyword in keywords):
                return file_type
        # Если суффикс изображения, но не опознан
        if suffix in self.rules['image']:
            return 'image'
        return 'unknown'

    def _process(self):
        self.state.current_task = "scanning_inbox"
        files = list(self.inbox_path.iterdir())
        print(f"[{self.state.name}] Found {len(files)} items in inbox.")

        for item in files:
            if item.is_file():
                self._process_file(item)

    def _process_file(self, file_path: Path):
        file_type = self._classify_file(file_path)
        dest_dir = self.processed_base / file_type / datetime.now().strftime("%Y-%m-%d")
        dest_dir.mkdir(parents=True, exist_ok=True)

        # Генерируем уникальное имя, чтобы избежать перезаписи
        file_hash = hashlib.md5(file_path.read_bytes()).hexdigest()[:8]
        new_name = f"{datetime.now().strftime('%H%M%S')}_{file_hash}{file_path.suffix}"
        dest_path = dest_dir / new_name

        try:
            shutil.move(str(file_path), str(dest_path))
            # Логируем действие в контекст агента и в отдельный файл
            log_entry = {
                "timestamp": datetime.now().isoformat(),
                "original_name": file_path.name,
                "new_path": str(dest_path),
                "file_type": file_type
            }
            self.state.update_context(f"file_{self.state.processed_items}", log_entry)
            self.state.processed_items += 1
            print(f"  -> Moved '{file_path.name}' to '{dest_path}' as '{file_type}'")

        except Exception as e:
            error_msg = f"Failed to process {file_path.name}: {e}"
            print(f"  [ERROR] {error_msg}")
            self.state.last_error = error_msg
            # Перемещаем проблемный файл в карантин
            quarantine_dir = self.processed_base / "quarantine"
            quarantine_dir.mkdir(exist_ok=True)
            shutil.move(str(file_path), str(quarantine_dir / file_path.name))

cd /opt/lab_ai
python -m agents.sanitary_agent_v1 --work-dir ./data --inbox ./hot_inbox

# /etc/cron.d/lab_ai_sanitary
*/5 * * * * root cd /opt/lab_ai && /usr/bin/python3 -m agents.sanitary_agent_v1 --work-dir ./data --inbox ./hot_inbox >> /var/log/lab_ai/sanitary.log 2>&1

# orchestrator/main.py
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
from typing import List, Optional
import httpx
import logging
import asyncio

# Конфигурация агентов
AGENT_CONFIG = {
    "classifier": {"url": "http://classifier:8001", "health_endpoint": "/health"},
    "validator": {"url": "http://validator:8002", "health_endpoint": "/health"},
    "sanitizer": {"url": "http://sanitizer:8003", "health_endpoint": "/health"},
    "report_generator": {"url": "http://report_generator:8004", "health_endpoint": "/health"},
}

app = FastAPI(title="Lab AI Orchestrator", version="1.0.0")
logger = logging.getLogger("orchestrator")

class ProcessingRequest(BaseModel):
    task_type: str  # "classify", "validate", "sanitize_and_report"
    data: List[dict]  # Список образцов с метаданными
    priority: Optional[str] = "normal"

@app.get("/health")
async def health():
    """Эндпоинт здоровья оркестратора."""
    return {"status": "healthy", "service": "orchestrator"}

@app.post("/process")
async def process_task(request: ProcessingRequest, background_tasks: BackgroundTasks):
    """Основной эндпоинт для обработки задач."""
    logger.info(f"Received task: {request.task_type} with {len(request.data)} samples.")

    # Определяем целевого агента
    target_agent = None
    if request.task_type.startswith("classify"):
        target_agent = "classifier"
    elif request.task_type.startswith("validate"):
        target_agent = "validator"
    elif request.task_type.startswith("sanitize"):
        # Комплексная задача: сначала санитар, потом классификатор, потом генератор отчетов
        background_tasks.add_task(complex_sanitize_task, request.data)
        return {"message": "Complex sanitization task started in background", "task_id": "async_123"}

    if not target_agent:
        raise HTTPException(status_code=400, detail=f"Unknown task type: {request.task_type}")

    # Проверяем здоровье агента перед отправкой
    health_url = AGENT_CONFIG[target_agent]["url"] + AGENT_CONFIG[target_agent]["health_endpoint"]
    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            health_response = await client.get(health_url)
            if health_response.status_code != 200:
                raise HTTPException(status_code=503, detail=f"Agent {target_agent} is unhealthy.")
    except httpx.RequestError as e:
        logger.error(f"Health check failed for {target_agent}: {e}")
        raise HTTPException(status_code=503, detail=f"Agent {target_agent} is unreachable.")

    # Отправляем задачу агент-сервису
    agent_url = AGENT_CONFIG[target_agent]["url"] + "/process"
    try:
        async with httpx.AsyncClient(timeout=30.0) as client:
            response = await client.post(agent_url, json={"samples": request.data})
            result = response.json()
    except httpx.RequestError as e:
        logger.error(f"Task failed for {target_agent}: {e}")
        raise HTTPException(status_code=500, detail=f"Agent {target_agent} processing error.")

    logger.info(f"Task completed by {target_agent}. Results: {len(result.get('classified_samples', []))}")
    return {"processed_by": target_agent, "results": result}

async def complex_sanitize_task(samples):
    """Пример сложной задачи, выполняемой фоном."""
    # Это упрощённый пример. В реальности тут цепочка вызовов.
    logger.info("Starting complex sanitize task in background...")
    # ... вызовы к sanitizer, затем classifier, затем report_generator
    await asyncio.sleep(1)
    logger.info("Complex sanitize task completed.")

# docker-compose.yml для Docker Swarm
version: '3.8'

services:
  orchestrator:
    image: lab-ai-orchestrator:1.0.0
    build:
      context: ./orchestrator
      dockerfile: Dockerfile
    ports:
      - "8000:8000"
    environment:
      - LOG_LEVEL=INFO
      - AGENT_TIMEOUT=30
    networks:
      - lab-ai-net
    deploy:
      replicas: 2
      restart_policy:
        condition: on-failure
        delay: 10s
        max_attempts: 3
      placement:
        constraints:
          - node.role == manager  # Оркестратор только на manager ноде

  classifier:
    image: lab-ai-classifier:1.0.2
    build:
      context: ./agents/classifier
      dockerfile: Dockerfile
    ports:
      - "8001:8001"
    environment:
      - MODEL_PATH=/app/models/blood_classifier_v3.onnx
      - THRESHOLD_URGENT=0.85
    volumes:
      - classifier_models:/app/models
    networks:
      - lab-ai-net
    deploy:
      replicas: 3
      restart_policy:
        condition: any
      resources:
        limits:
          memory: 1G
        reservations:
          memory: 512M

  validator:
    image: lab-ai-validator:1.0.1
    build:
      context: ./agents/validator
      dockerfile: Dockerfile
    ports:
      - "8002:8002"
    environment:
      - ALLOWED_TESTS=WBC,RBC,HGB,PLT
      - MAX_SAMPLE_AGE_HOURS=72
    networks:
      - lab-ai-net
    deploy:
      replicas: 2

  sanitizer:
    image: lab-ai-sanitizer:1.0.3
    build:
      context: ./agents/sanitizer
      dockerfile: Dockerfile
    ports:
      - "8003:8003"
    environment:
      - SCAN_INTERVAL_SECONDS=300
      - QUARANTINE_PATH=/app/quarantine
    volumes:
      - lab_data:/data/raw
      - sanitizer_quarantine:/app/quarantine
    networks:
      - lab-ai-net
    deploy:
      replicas: 2

  report_generator:
    image: lab-ai-report-generator:1.0.0
    build:
      context: ./agents/report_generator
      dockerfile: Dockerfile
    ports:
      - "8004:8004"
    environment:
      - TEMPLATE_DIR=/app/templates
      - OUTPUT_DIR=/app/generated_reports
    volumes:
      - report_templates:/app/templates
      - generated_reports:/app/generated_reports
    networks:
      - lab-ai-net
    deploy:
      replicas: 2

  # Мониторинг (простейший)
  monitor:
    image: lab-ai-monitor:0.1.0
    build:
      context: ./monitor
      dockerfile: Dockerfile
    ports:
      - "8080:8080"
    environment:
      - ORCHESTRATOR_URL=http://orchestrator:8000/health
      - AGENTS_URLS=http://classifier:8001/health,http://validator:8002/health
    networks:
      - lab-ai-net
    deploy:
      replicas: 1
      placement:
        constraints:
          - node.role == manager

networks:
  lab-ai-net:
    driver: overlay  # Ключевое для Swarm: overlay сеть для связи между нодами

volumes:
  lab_data:
    driver: local
  classifier_models:
    driver: local
  sanitizer_quarantine:
    driver: local
  report_templates:
    driver: local
  generated_reports:
    driver: local

# Сборка всех образов (долго, но необходимо)
docker-compose build --parallel

# Инициализация Swarm на основной ноде (мой ноутбук стал manager)
docker swarm init

# Деплой стека в Swarm
docker stack deploy -c docker-compose.yml lab-ai-stack

# Проверка статуса
docker stack services lab-ai-stack

# Логи конкретного сервиса (когда что-то падало)
docker service logs lab-ai-stack_classifier --tail 50 --follow

# docker-compose.v2.yml - фрагмент с лимитами
services:
  report_agent:
    image: lab/report-agent:1.2
    deploy:
      replicas: 2
      resources:
        limits:
          cpus: '1.0'
          memory: 2G
        reservations:
          cpus: '0.5'
          memory: 1G
    environment:
      - AGENT_SLEEP=0.1 # Агрессивность, от 0.1 до 1.0

  validator_agent:
    image: lab/validator:1.1
    deploy:
      replicas: 5
      resources:
        limits:
          cpus: '0.5'
          memory: 768M
        reservations:
          cpus: '0.1'
          memory: 256M
    environment:
      - VALIDATION_BATCH_SIZE=10

# health_check.py в каждом агенте (FastAPI)
from fastapi import APIRouter, HTTPException
import psutil, redis, asyncpg
from datetime import datetime, timedelta

router = APIRouter()

@router.get("/health")
async def deep_health():
    errors = []
    # 1. Redis
    try:
        r = redis.Redis(host="redis", port=6379, socket_connect_timeout=2)
        r.ping()
    except Exception as e:
        errors.append(f"Redis: {e}")

    # 2. Postgres
    try:
        conn = await asyncpg.connect("postgresql://user:pass@postgres:5432/lab")
        await conn.close()
    except Exception as e:
        errors.append(f"Postgres: {e}")

    # 3. CPU sanity check
    cpu_percent = psutil.cpu_percent(interval=1)
    if cpu_percent > 95.0:
        errors.append(f"CPU overload: {cpu_percent}%")

    # 4. Memory check (более 90% от лимита Docker)
    mem = psutil.virtual_memory()
    if mem.percent > 90:
        errors.append(f"Memory high: {mem.percent}%")

    if errors:
        raise HTTPException(status_code=503, detail={"status": "unhealthy", "errors": errors})

    return {"status": "healthy", "timestamp": datetime.utcnow().isoformat()}

# docker-compose.monitor.yml
version: '3.8'
services:
  prometheus:
    image: prom/prometheus:v2.45.0
    volumes:
      - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml
      - prom_data:/prometheus
    command:
      - '--config.file=/etc/prometheus/prometheus.yml'
      - '--storage.tsdb.retention.time=30d'
    ports:
      - "9090:9090"
    deploy:
      placement:
        constraints:
          - node.role == manager

  grafana:
    image: grafana/grafana:10.0.0
    volumes:
      - grafana_data:/var/lib/grafana
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin123 # Да, я знаю. Не в продакшен.
    ports:
      - "3000:3000"
    deploy:
      placement:
        constraints:
          - node.role == manager

volumes:
  prom_data:
  grafana_data:

# monitoring/prometheus.yml
global:
  scrape_interval: 15s
  evaluation_interval: 15s

scrape_configs:
  - job_name: 'ai-agents'
    static_configs:
      - targets:
        - 'report_agent:8000'
        - 'validator_agent:8000'
        - 'classifier_agent:8000'
        - 'lis_connector:8000'
        # ... и ещё 23 цели
    metrics_path: /metrics
    relabel_configs:
      - source_labels: [__address__]
        target_label: __param_target
      - source_labels: [__param_target]
        target_label: instance
      - target_label: __address__
        replacement: prometheus:9090  # Blackbox exporter, если бы он был

  - job_name: 'agent-health'
    metrics_path: /health
    static_configs:
      - targets:
        - 'report_agent:8000'
        - 'validator_agent:8000'
        # ... все агены
    relabel_configs:
      - source_labels: [__address__]
        regex: '(.*):8000'
        target_label: agent
        replacement: '$1'

  loki:
    image: grafana/loki:2.9.0
    command: -config.file=/etc/loki/local-config.yaml
    ports:
      - "3100:3100"

  promtail:
    image: grafana/promtail:2.9.0
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
      - ./monitoring/promtail-config.yml:/etc/promtail/config.yml
    command: -config.file=/etc/promtail/config.yml

# monitoring/promtail-config.yml
server:
  http_listen_port: 9080
  grpc_listen_port: 0

positions:
  filename: /tmp/positions.yaml

clients:
  - url: http://loki:3100/loki/api/v1/push

scrape_configs:
  - job_name: docker
    docker_sd_configs:
      - host: unix:///var/run/docker.sock
        refresh_interval: 5s
    relabel_configs:
      - source_labels: ['__meta_docker_container_name']
        regex: '/(.*)'
        target_label: 'container'
      - source_labels: ['__meta_docker_service_name']
        target_label: 'service'
      - source_labels: ['__meta_docker_service_task_slot']
        target_label: 'task_slot'

# В самом начале проекта. core_interface.py
import pandas as pd
from pathlib import Path
import json

class LabDataInterface:
    def __init__(self, incoming_root: str):
        self.root = Path(incoming_root)
        # ВСЕ агенты потом будут работать через этот класс
        # Он знает ВСЮ специфику моих файлов

    def list_new_samples(self) -> list[dict]:
        # Не просто список файлов, а сразу dict с метаданными
        # Например: {'path': '...', 'timestamp': '...', 'type': 'blood'}
        pass

    def load_sample_data(self, sample_path: str) -> pd.DataFrame:
        # Здесь все костыли для моих кривых CSV
        # pd.read_csv(... encoding='cp1251', sep=';', on_bad_lines='warn')
        pass

    def write_result(self, sample_id: str, result: dict):
        # Единый метод записи результата для всех агентов
        # В стандартный формат, в стандартную папку
        pass

# docker-compose.v1.yml — для одного агента!
version: '3.8'
services:
  agent-sanitizer:
    build:
      context: ./agents/sanitizer
      dockerfile: Dockerfile
    volumes:
      - /mnt/lab/incoming:/app/incoming:ro
      - ./logs:/app/logs
    environment:
      - LOG_LEVEL=INFO
    restart: unless-stopped

# agent_base.py
from prometheus_client import Counter, Histogram
import time

PROCESSED_FILES = Counter('agent_processed_files_total', 'Total processed files')
PROCESS_TIME = Histogram('agent_process_time_seconds', 'Time per file processing')

def process_file(self, file_path):
    start_time = time.time()
    try:
        # ... ваша логика ...
        PROCESSED_FILES.inc()
    finally:
        PROCESS_TIME.observe(time.time() - start_time)

@app.get("/health")
def health():
    return {"status": "healthy", "timestamp": datetime.utcnow()}

agents/
├── core/                 # Общий код: интерфейс к данным, логирование, метрики
├── agent_types/
│   ├── classifier/      # Тип 1: классификатор (общий код)
│   ├── validator/       # Тип 2: валидатор (общий код)
│   ├── reporter/        # Тип 3: генератор отчетов (общий код)
├── instances/
│   ├── classifier_blood/ # Конкретный агент: классификатор крови
│   │   ├── config.json  # Конфигурация
│   │   ├── model.pth    # Его конкретная модель
│   │   └── Dockerfile   # Тонкий Dockerfile, который использует общий
│   ├── classifier_urine/
│   ├── validator_main/

# orchestrator_simple.py
import redis  # Используем Redis как очередь
import json

r = redis.Redis(host='redis', port=6379)

def submit_task(task_type: str, file_meta: dict):
    task_id = str(uuid.uuid4())
    task = {
        'id': task_id,
        'type': task_type,
        'data': file_meta,
        'status': 'pending',
        'created_at': datetime.utcnow().isoformat()
    }
    r.set(f'task:{task_id}', json.dumps(task))
    r.rpush(f'queue:{task_type}', task_id)