Комментарии (1)
Ответ сгенерирован нейросетью и может содержать ошибки
Мой текущий проект: PrepBro — AI-платформа для подготовки к собеседованиям
Работаю на роли Python/Backend разработчика в стартапе PrepBro. Это интересный проект, который объединяет AI, микросервисную архитектуру и real-time систему. Расскажу о сути проекта и своём вкладе.
О проекте
PrepBro — платформа, которая помогает разработчикам подготовиться к собеседованиям через:
- Банк вопросов по разным специальностям (Python Developer, Frontend, DevOps и т.д.)
- AI-агентов, которые генерируют ответы на вопросы
- Систему оценивания, которая проверяет качество ответов
- Telegram-бота для удобного взаимодействия
Платформа решает реальную проблему: подготовка к интервью требует времени, а ответы нужны быстро и адаптированы к выбранной специальности.
Архитектура
┌─────────────────┐
│ Telegram Bot │ (aiogram 3.x)
│ (Handlers) │
└────────┬────────┘
│
▼
┌─────────────────────────┐
│ FastAPI Backend │
│ - /api/v1/questions │
│ - /api/v1/answers │
│ - /api/v1/dev/* │
└────────┬────────────────┘
│
┌────┼────┐
▼ ▼
┌────────┐ ┌───────────────────┐
│ DB │ │ AI Agents │
│(Goose) │ │ (LLM Service) │
└────────┘ └───────────────────┘
│ │
└──────┬───────┘
▼
┌──────────────┐
│ PostgreSQL │
│ TimescaleDB │
└──────────────┘
Технологический стек
Backend:
- Python 3.12 с FastAPI для высокопроизводительного API
- SQLAlchemy ORM для работы с БД + Pydantic для валидации
- Goose для миграций (raw SQL, а не Alembic)
- PostgreSQL как основная БД
- Telegram Bot API через aiogram 3.x с FSM
DevOps & Infrastructure:
- Docker Compose для локального развития
- Git + Dokku для CI/CD (git push dokku main)
- SSH на port 24822 для деплоймента
- make команды для управления (make lint, make test)
Testing:
- pytest с флагом
--reuse-dbдля быстрых тестов - VCR.py для записи HTTP responses в тестах (no real API calls)
- Playwright MCP для E2E тестирования фронта
- Coverage > 90% обязательное требование
Мои задачи и вклад
1. Разработка API endpoints
# Пример: получить вопрос для ответа
@router.get("/api/v1/agent/answering/next")
async def get_next_question(
profession_id: UUID,
current_user: User = Depends(get_current_user),
) -> QuestionResponse:
"""Получить следующий вопрос для которого ещё нет ответа"""
question = await QuestionRepository.find_unanswered(
profession_id=profession_id,
user_id=current_user.id
)
return QuestionResponse.from_entity(question)
# И отправить ответ
@router.post("/api/v1/agent/answering/{question_id}")
async def submit_answer(
question_id: UUID,
payload: AnswerPayload,
current_user: User = Depends(get_current_user),
) -> AnswerResponse:
answer = Answer(
question_id=question_id,
user_id=current_user.id,
content=payload.content
)
await AnswerRepository.save(answer)
return AnswerResponse.from_entity(answer)
2. Integration с AI service
# Вызов AI агента для генерации ответа
class AIAnsweringService:
async def generate_answer(self, question: Question) -> str:
"""Генерирует ответ используя LLM"""
prompt = self._build_prompt(question)
response = await self.llm_client.complete(
prompt=prompt,
temperature=0.7,
max_tokens=2048
)
return response.text
def _build_prompt(self, question: Question) -> str:
return f"""You are an expert {question.profession_name}.
Answer the following question:
{question.title}
{question.text or ''}
Provide detailed, practical answer in Russian.
"""
3. Telegram Bot handlers
# Простой и чистый handler для Telegram
@router.message(Command("start"))
async def start_command(message: Message, state: FSMContext) -> None:
user = await UserService.get_or_create(message.from_user.id)
await message.answer(
f"Привет, {user.name}! Выбери специальность для подготовки."
)
await state.set_state(AnsweringState.choosing_profession)
@router.callback_query(StateFilter(AnsweringState.choosing_profession))
async def profession_selected(
callback: CallbackQuery,
state: FSMContext
) -> None:
profession_id = UUID(callback.data)
next_question = await QuestionService.get_next(
profession_id=profession_id,
user_id=callback.from_user.id
)
await callback.message.edit_text(
f"Q: {next_question.title}\n{next_question.text or ''}"
)
await state.update_data(question_id=next_question.id)
await state.set_state(AnsweringState.waiting_answer)
4. Database миграции
-- migrations/0001_initial_schema.sql
CREATE TABLE questions (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
title TEXT NOT NULL,
text TEXT,
profession_id UUID NOT NULL REFERENCES professions(id),
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_questions_profession ON questions(profession_id);
CREATE TABLE answers (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
question_id UUID NOT NULL REFERENCES questions(id) ON DELETE CASCADE,
user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
content TEXT NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_answers_question_user ON answers(question_id, user_id);
5. Clean Architecture & SOLID
backend/
├── domain/ # Бизнес-логика (Entity, ValueObject)
│ ├── question.py # Question aggregate
│ └── answer.py # Answer aggregate
├── application/ # Use cases & Services
│ ├── answering_service.py
│ └── question_service.py
├── infrastructure/ # Repository, DB, External Services
│ ├── repositories/
│ ├── database/
│ └── ai_client/
└── presentation/ # API Routes & Telegram Handlers
├── api/
└── telegram/
Зависимости только внутрь (presentation → application → domain).
Challenges и как их решал
Challenge 1: Race conditions при выборе следующего вопроса
-- Решение: SELECT FOR UPDATE SKIP LOCKED
SELECT * FROM questions
WHERE profession_id = @profession_id
AND id NOT IN (SELECT question_id FROM answers WHERE user_id = @user_id)
FOR UPDATE SKIP LOCKED
LIMIT 1;
Challenge 2: Медленные LLM запросы (30+ сек)
- Использовал async/await, background tasks
- Добавил progress indicator в Telegram
- Кешировал частые ответы
Challenge 3: Тестирование Telegram Bot
# Используем dispatcher.feed_update() вместо реального API
async def test_start_command():
from aiogram import Dispatcher
dispatcher = Dispatcher()
# ... регистрируем handlers
message = Message(text="/start", from_user=User(id=123))
await dispatcher.feed_update(message)
# Проверяем, что ответ корректный
Результаты
- ✅ API полностью покрыт тестами (90%+ coverage)
- ✅ Telegram Bot обрабатывает 500+ запросов/день
- ✅ Zero downtime deployments через Dokku
- ✅ Response time < 200ms для 95% запросов
- ✅ Code quality: A grade (ruff linting)
Что интересного для карьеры
- Полный стек — от API до Telegram, от DB до DevOps
- AI integration — работа с LLM, prompt engineering
- Архитектура — DDD, clean architecture, microservices
- Production опыт — реальные пользователи, monitoring, performance
- Быстрая итерация — стартап = быстрые решения
Проект научил меня не только писать код, но и думать о масштабировании, надёжности и user experience с первого дня разработки.