OptiFood AI 🥗🤖

An AI-powered Telegram bot for intelligent food logging — text, voice, photo, and barcode — with a five-layer recognition pipeline, anti-hallucination validation, and an async-first architecture.

---

📋 Table of Contents

• Overview
• Features
• Architecture
• AI Pipeline
• Tech Stack
• Project Structure
• Quick Start
• Configuration
• Testing
• Contributing
• License

---

🧭 Overview

OptiFood AI is a production-grade Telegram bot that lets users log meals by simply sending a text description, voice message, photo, or barcode scan. The bot automatically identifies food items, calculates macros (calories, protein, fat, carbs, fiber, alcohol), and tracks daily nutrition goals.

The backend runs as two independent Docker services: a FastAPI + aiogram webhook API and a TaskIQ async worker. Communication between them uses Redis Pub/Sub — no polling, no blocking the event loop.

---

---

---

✨ Features

Voice input	Photo recognition	Barcode scan

• Five-layer AI recognition pipeline (L1–L5) with explicit fallback conditions, from Redis cache to LLM
• Anti-hallucination validator — dual-layer check via physical nutrient limits and the Atwater formula
• Intelligent AI Mentor — daily macro analysis with contextual recommendations powered by GPT
• 
• Premium subscription — paywall for analytics and photo-log, managed via premium_until field
• 
• Async-first — PostgreSQL via asyncpg, Redis async client, aiogram 3.x, TaskIQ worker
• Strict layered architecture — Port/Adapter (DIP), Unit of Work, Repository, Orchestrator Facade
• Full observability — Sentry initialized independently in both api and worker processes; structured logging

---

🏗 Architecture

Layer Rules

Layer	Location	Can import	Forbidden
Presentation	app/bot/handlers/, middlewares/	core/services/, core/ports/, schemas/	infrastructure/database/models directly
Application	app/core/services/, orchestrator/	core/skills/, core/ports/, schemas/	Any infrastructure/ import
Domain Skills	app/core/skills/	core/services/ai_service.py, schemas/	infrastructure/
Infrastructure	app/infrastructure/, app/worker/	schemas/	Business logic

---

🧠 AI Pipeline (L1→L5)

Each level is activated only if the previous one returned no result:

Level	Method	Latency	Condition
L1	Redis Cache (MD5 key, TTL 24h)	~0 ms	Simple queries only
L2	Fuzzy Local DB (Rapidfuzz, in-memory)	~2–5 ms	Cache miss
L3	Vector Semantic Search (Pinecone + OpenAI Embeddings, cached 30d)	~50–100 ms	Fuzzy miss
L4	LLM Fallback — GPT Structured JSON Output	~1–3 s	Vector miss or complex_query
L5	Background indexing (asyncio.gather)	async	After L4 hit — indexes result into Redis + Pinecone

Pre-Parser (FoodParserPipeline): normalizes input, extracts quantity + units, detects complex_query (multiple numbers or conjunctions). Rules live in core/constants.py.

Anti-hallucination (NutritionCalculator.sanitize_dto):

1. Physical limits — protein + fat + carbs ≤ 100 g/100 g; calories ≤ 950 kcal/100 g
2. Atwater formula — if deviation from protein×4 + fat×9 + carbs×4 exceeds 30%, recalculates automatically

---

🛠 Tech Stack

Category	Technology
Bot framework	aiogram 3.x
Web framework	FastAPI + Uvicorn
Database	PostgreSQL 15 via asyncpg + SQLAlchemy 2.0 (async)
Migrations	Alembic
Cache / FSM	Redis 7 (FSM storage, Pub/Sub, LLM cache, embeddings)
Task queue	TaskIQ + taskiq-redis
AI / LLM	OpenAI (GPT-4o, Whisper, Embeddings) via litellm
Vector search	Pinecone via httpx (async)
Fuzzy search	Rapidfuzz (in-memory index)
Barcode	zxing-cpp (asyncio.to_thread)
External food data	Open Food Facts via aiohttp
NLP / Lemmatization	pymorphy3 + pymorphy3-dicts-uk, NLTK
Localization	aiogram-i18n + Fluent (.ftl files)
Config	pydantic-settings + SecretStr
Monitoring	Sentry SDK (independent in api + worker)
Image processing	Pillow
Testing	pytest, pytest-asyncio, pytest-mock, pytest-cov

---

📁 Project Structure

OptiFood AI/
├── .github/workflows/           # CI/CD — GitHub Actions
├── alembic/                     # Database migrations
│   └── versions/
├── app/
│   ├── main.py                  # Entry point: FastAPI + aiogram webhook
│   ├── config.py                # Pydantic Settings, SecretStr for all API keys
│   ├── bot/
│   │   ├── pubsub.py            # Redis Pub/Sub: worker result listener
│   │   ├── handlers/            # common, food, admin, errors
│   │   ├── middlewares/         # user_context, throttling, state_clear, i18n, onboarding
│   │   ├── keyboards/           # inline, reply
│   │   └── states/              # FoodLogState, EditWeightState, SettingsState
│   ├── core/                    # Domain layer — never imports infrastructure directly
│   │   ├── constants.py         # LLM prompts, regex patterns, Atwater coefficients
│   │   ├── exceptions.py        # DomainError, RecognitionFailedError, BarcodeNotFoundError
│   │   ├── orchestrator/        # FoodOrchestrator — single stable entry point for worker tasks
│   │   ├── ports/               # ABC interfaces: ITaskQueue, IFoodCache, IVectorStore, IFuzzyIndex
│   │   ├── services/            # ai, analytics, media, nutrition_calculator, nutrition, recognition, user
│   │   └── skills/              # Atomic pipeline components: pre_parser, text_normalizer, barcode,
│   │                            # dto_builder, nutrition_calc, ocr, voice, mentor
│   ├── infrastructure/          # Concrete port implementations + external integrations
│   │   ├── cache/               # FuzzyProductIndex (Rapidfuzz), redis_client (FoodCache)
│   │   ├── database/            # SQLAlchemy models, UnitOfWork, repositories, VectorStore (Pinecone)
│   │   ├── external_api/        # openfoodfacts_client (aiohttp)
│   │   ├── storage/             # local_media.py (S3-ready interface)
│   │   └── task_queue/          # taskiq_queue.py — ITaskQueue implementation
│   ├── locales/                 # Fluent .ftl files (uk, en)
│   ├── schemas/                 # Pydantic DTOs: FoodItemDTO, MealCreateDTO, OrchestratorResult, UserDTO
│   ├── utils/                   # TelegramFormatter, UIBuilder
│   └── worker/
│       ├── broker.py            # TaskIQ broker + SentryMiddleware
│       └── tasks.py             # Task entry points, TaskNotifier, FoodTaskCoordinator
├── scripts/                     # i18n audit and auto-fix utilities
├── tests/
│   ├── bot/                     # [INSERT TESTS — middleware, handlers]
│   ├── core/                    # [INSERT TESTS — services, dto_builder, nutrition_calc, pre_parser, recognition pipeline]
│   └── infrastructure/          # [INSERT TESTS — repositories, fuzzy index, openfoodfacts client]
├── docker-compose.yml
├── Dockerfile                   # Multi-stage build (builder + runtime), non-root user
├── requirements.txt
├── pytest.ini
└── .pre-commit-config.yaml      # ruff (lint + format), trailing whitespace, secret detection

---

🚀 Quick Start

Prerequisites

• Docker & Docker Compose
• A Telegram Bot Token (@BotFather)
• OpenAI API Key
• Pinecone API Key + Index name

1. Clone and configure

git clone https://github.com/PyDevDeep/optifood-ai.git
cd optifood-ai
cp .env.example .env   # fill in all required values

2. Start services

docker compose up --build

This starts three containers:

Service	Role
redis	FSM storage, Pub/Sub, cache
optifood_api	FastAPI + aiogram webhook; runs alembic upgrade head on startup
optifood_worker	TaskIQ worker; shares /tmp/optifood_media volume with api

3. Register webhook

curl "https://api.telegram.org/bot<TOKEN>/setWebhook?url=https://<YOUR_DOMAIN>/webhook"

---

⚙️ Configuration

All sensitive values are managed through app/config.py (Pydantic SecretStr). Set them via .env:

# Telegram
TELEGRAM_BOT_TOKEN=...
WEBHOOK_URL=https://yourdomain.com/webhook
WEBHOOK_SECRET=...

# OpenAI
OPENAI_API_KEY=...

# Pinecone
PINECONE_API_KEY=...
PINECONE_INDEX_NAME=...

# PostgreSQL
DATABASE_URL=postgresql+asyncpg://user:pass@db:5432/optifood

# Redis
REDIS_HOST=redis
REDIS_PORT=6379

# Sentry
SENTRY_DSN_API=...
SENTRY_DSN_WORKER=...

# Premium
ADMIN_IDS=[INSERT_TELEGRAM_ID]

⚠️ Never commit .env to the repository. It is listed in .gitignore.

---

🧪 Testing

# Run all tests
pytest

# With coverage report
pytest --cov=app --cov-report=term-missing

# Run a specific layer
pytest tests/core/
pytest tests/infrastructure/
pytest tests/bot/

Test structure mirrors the application layers:

The project is thoroughly tested with 248 asynchronous tests covering all architectural layers:

Directory	Focus
tests/bot/	Presentation layer: Middlewares (AntiSpam, Quota, UserContext) and FSM state routing.
tests/core/	Domain logic: NutritionCalculator (Atwater limits), NLP parsing, AI Service fallbacks, and isolated business services.
tests/infrastructure/	Data access: FuzzyProductIndex, SQLAlchemy repositories, and integrations (OpenFoodFacts).

---

🤝 Contributing

1. Fork the repository
2. Create a feature branch: git checkout -b feat/your-feature
3. Follow the layer rules — no cross-layer imports
4. All UI strings go to .ftl files under app/locales/; hardcoding in handlers fails pre-commit
5. Run pre-commit run --all-files before pushing (ruff lint + format, secret detection)
6. Open a Pull Request with a clear description of the change

---

📄 License

🛡 License

This project is licensed under the MIT License. See the LICENSE file for details.