# Техническое задание: AI Sales Bot (Python) **Версия:** 1.0 **Дата:** 2026-03-29 **Для:** Claude Code / Codex **Язык:** Python 3.11+ --- ## 0. ИНСТРУКЦИЯ ДЛЯ AI-РАЗРАБОТЧИКА ### 0.1 Перед началом - Прочитай ВСЁ ТЗ целиком перед написанием первой строки кода - Если что-то неясно — спроси, не додумывай - Каждый модуль должен работать автономно (можно тестировать отдельно) - Пиши тесты ВМЕСТЕ с кодом, не после ### 0.2 Workflow разработки (последовательно) **ЗАДАЧА 1/6: Ядро LLM + Классификатор** - Файлы: config.py, .env, core/llm.py, agents/classifier.py, core/dedup.py - Тест: classifier на 10 примерах → все intent-ы верные - Критерий: pytest tests/test_classifier.py проходит **ЗАДАЧА 2/6: Отвечатель + Trust + Eval** - Файлы: agents/responder.py, core/trust.py, core/eval.py - Тест: 10 ответов проходят eval, ни один не нарушает trust - Критерий: pytest tests/test_responder.py + tests/test_trust.py **ЗАДАЧА 3/6: Память + Router + Telegram** - Файлы: core/memory.py, core/router.py, adapters/base.py, adapters/telegram_adapter.py - Тест: 5 сообщений в Telegram → правильные ответы - Критерий: бот работает в Telegram **ЗАДАЧА 4/6: Квалификация + Эскалация + Follow-up** - Файлы: agents/qualifier.py, agents/escalator.py, core/scheduler.py - Тест: HOT → уведомление, WARM → follow-up - Критерий: pytest tests/test_qualifier.py **ЗАДАЧА 5/6: Хранение + Отчёты + Мульти-тенант** - Файлы: storage/, reports/, tenants/ - Тест: 2 tenant-а параллельно, отчёт генерируется - Критерий: pytest tests/test_scenarios.py (20 сценариев) **ЗАДАЧА 6/6: Авито + MAX + Деплой** - Файлы: adapters/avito_adapter.py, adapters/max_adapter.py, systemd - Тест: 3 канала одновременно - Критерий: интеграционный тест ### 0.3 Контракты между модулями ``` Adapter → Router: IncomingMessage(channel, tenant_id, chat_id, user_id, user_name, text, timestamp, raw) → OutgoingMessage(channel, chat_id, text, attachments) | None Router → Classifier: (text: str, history: list[dict]) → ClassificationResult(intent, confidence, entities) Router → Responder: (ClassificationResult, tenant: Tenant, history: list[dict]) → Response(text, attachments, facts_used) Router → Evaluator: (Response, tenant: Tenant) → EvalResult(passed: bool, checks: list[tuple]) Router → Qualifier: (history: list[dict]) → QualificationResult(score, reasons, collected_data, next_action) Router → Escalator: (IncomingMessage, tenant: Tenant, history, reason: str) → None (side effect: notification) ``` ### 0.4 Форматы файлов данных клиента **config.yaml** — см. секцию 2.4.1 **price_list.md:** ```markdown # Прайс-лист ## Каркасные дома - 80 м²: 3 500 000 ₽ - 100 м²: 4 200 000 ₽ - 120 м²: 5 500 000 ₽ - 150 м²: 6 800 000 ₽ ## Газобетонные дома - 80 м²: 4 200 000 ₽ - 100 м²: 5 100 000 ₽ Примечание: цены "под ключ", фундамент включён. Актуальность: март 2026 ``` **faq.md:** ```markdown # FAQ ## Вопрос: Сколько строите дом? Ответ: Каркасный дом — 2-3 месяца. Газобетон — 4-6 месяцев. Зависит от площади и сложности. ## Вопрос: Какая гарантия? Ответ: 5 лет на конструктив, 2 года на отделку. ## Вопрос: Строите зимой? Ответ: Да, каркасные дома строим круглый год. Газобетон — с апреля по ноябрь. ``` Формат: заголовок ## Вопрос: ... + Ответ: ... (для удобного парсинга) ### 0.5 Справочные материалы (API документация) - **Авито Messenger API:** https://developers.avito.ru/api-catalog/messenger - **MAX Bot API:** https://dev.max.ru/docs, endpoint: https://platform-api.max.ru/ - **DeepSeek API:** https://platform.deepseek.com/api-docs (OpenAI-совместимый) - **OpenRouter API:** https://openrouter.ai/docs (для Qwen fallback) - **aiogram 3.x:** https://docs.aiogram.dev/en/latest/ --- ## 1. ОБЗОР ПРОЕКТА ### 1.1 Что это Мультиканальный AI-бот для автоматизации продаж МСП. Отвечает клиентам за 30 секунд, квалифицирует лиды, передаёт горячих владельцу бизнеса. ### 1.2 Ключевые требования - Мультиканальность: Telegram + MAX + Авито (через адаптеры) - Мульти-тенант: один инстанс обслуживает N клиентов (каждый со своими данными) - Multi-agent архитектура: Классификатор → Отвечатель → Квалификатор → Эскалатор - Работает на VPS (Ubuntu 24.04, 2-4 GB RAM) - LLM: DeepSeek API (основная), Qwen через OpenRouter (fallback) ### 1.3 Структура проекта ``` ai-sales-bot/ ├── main.py # Точка входа, запуск всех адаптеров ├── config.py # Конфигурация (env vars) ├── requirements.txt # Зависимости ├── .env.example # Пример переменных окружения │ ├── adapters/ # Канальные адаптеры (транспорт) │ ├── __init__.py │ ├── base.py # Абстрактный базовый адаптер │ ├── telegram_adapter.py # Telegram (aiogram 3.x) │ ├── max_adapter.py # MAX (platform-api.max.ru) │ └── avito_adapter.py # Авито (API) │ ├── agents/ # AI-агенты (логика) │ ├── __init__.py │ ├── classifier.py # Классификатор intent-ов │ ├── responder.py # Генерация ответов │ ├── qualifier.py # Квалификация лидов (hot/warm/cold) │ └── escalator.py # Решение: бот или человек │ ├── core/ # Ядро системы │ ├── __init__.py │ ├── router.py # Planner-логика (оркестрация агентов) │ ├── llm.py # Обёртка над LLM API (DeepSeek/Qwen) │ ├── memory.py # Память диалогов (per-session) │ ├── eval.py # Eval pipeline (проверка ответов) │ └── trust.py # Trust Matrix (что можно/нельзя) │ ├── tenants/ # Мульти-тенант │ ├── __init__.py │ ├── manager.py # Управление клиентами (tenants) │ └── loader.py # Загрузка данных клиента │ ├── data/ # Данные клиентов │ ├── _template/ # Шаблон для нового клиента │ │ ├── company_profile.md │ │ ├── price_list.md │ │ ├── faq.md │ │ ├── portfolio.md │ │ ├── guarantees.md │ │ ├── objection_handlers.md │ │ ├── escalation_rules.md │ │ └── tone_of_voice.md │ │ │ └── {tenant_id}/ # Данные конкретного клиента │ ├── config.yaml # Настройки клиента (токены, каналы, владелец) │ ├── company_profile.md │ ├── price_list.md │ ├── faq.md │ ├── portfolio.md │ ├── guarantees.md │ ├── objection_handlers.md │ ├── escalation_rules.md │ └── tone_of_voice.md │ ├── reports/ # Отчёты и аналитика │ ├── __init__.py │ └── reporter.py # Генерация отчётов │ ├── storage/ # Хранение данных │ ├── __init__.py │ ├── database.py # SQLite для диалогов/лидов/метрик │ └── models.py # ORM модели (SQLAlchemy/SQLModel) │ ├── tests/ # Тесты │ ├── test_classifier.py │ ├── test_responder.py │ ├── test_qualifier.py │ ├── test_trust.py │ └── test_scenarios.py # 20 тестовых сценариев │ └── scripts/ # Утилиты ├── add_tenant.py # Добавить нового клиента ├── run_tests.py # Прогнать тесты └── generate_report.py # Сгенерировать отчёт ``` --- ## 2. КОМПОНЕНТЫ (ДЕТАЛЬНО) ### 2.1 Адаптеры (`adapters/`) #### 2.1.1 Базовый адаптер (`base.py`) ```python from abc import ABC, abstractmethod from dataclasses import dataclass @dataclass class IncomingMessage: """Унифицированное входящее сообщение от любого канала""" channel: str # "telegram" | "max" | "avito" tenant_id: str # ID клиента (нашего) chat_id: str # ID чата/диалога user_id: str # ID конечного пользователя (покупатель) user_name: str # Имя пользователя (если есть) text: str # Текст сообщения timestamp: float # Unix timestamp raw: dict # Оригинальные данные от канала @dataclass class OutgoingMessage: """Унифицированное исходящее сообщение""" channel: str chat_id: str text: str attachments: list = None # Фото, документы (опционально) class BaseAdapter(ABC): """Абстрактный адаптер канала""" @abstractmethod async def start(self): """Запустить прослушивание канала""" pass @abstractmethod async def send(self, message: OutgoingMessage): """Отправить сообщение в канал""" pass @abstractmethod async def stop(self): """Остановить адаптер""" pass ``` #### 2.1.2 Telegram адаптер (`telegram_adapter.py`) - Использовать **aiogram 3.x** - Поддержка нескольких ботов (мульти-тенант): каждый клиент — свой Telegram-токен - Webhook или polling (настраивается в config) - При получении сообщения → создать `IncomingMessage` → передать в `router.py` - Маппинг tenant_id по Telegram bot token #### 2.1.3 MAX адаптер (`max_adapter.py`) - API endpoint: `https://platform-api.max.ru/` - Авторизация: заголовок `Authorization: ` - Лимит: 30 RPS - Long polling для получения обновлений (или webhook если поддерживается) - При получении сообщения → создать `IncomingMessage` → передать в `router.py` #### 2.1.4 Авито адаптер (`avito_adapter.py`) - Авито API для чтения/отправки сообщений - Авторизация: OAuth2 (client_id + client_secret → access_token) - Polling: проверять новые сообщения каждые 10-30 секунд - При получении сообщения → создать `IncomingMessage` → передать в `router.py` --- ### 2.2 Агенты (`agents/`) #### 2.2.1 Классификатор (`classifier.py`) **Задача:** Определить intent входящего сообщения. **Вход:** текст сообщения + история диалога (последние 5 сообщений) **Выход:** ```python @dataclass class ClassificationResult: intent: str # Один из 20 intent-ов (см. таблицу ниже) confidence: float # 0.0 - 1.0 entities: dict # Извлечённые сущности (площадь, бюджет, телефон и т.д.) ``` **Промпт Классификатора:** ``` Ты — классификатор сообщений для строительной компании. Определи intent сообщения клиента. Выбери ОДИН из списка: - PRICE_GENERAL — спрашивает цену в целом - PRICE_SPECIFIC — спрашивает цену на конкретный объект (указана площадь, тип) - TIMELINE — спрашивает сроки строительства - MATERIALS — спрашивает про материалы - PORTFOLIO — хочет увидеть работы/примеры - GUARANTEE — спрашивает про гарантии - LOCATION — спрашивает где строите / регион работы - MEETING — хочет встречу / замер / звонок - OBJECTION_PRICE — возражение "дорого" - OBJECTION_TRUST — возражение "ненадёжно" / сомнения - OBJECTION_TIMELINE — возражение "долго" - FINANCING — рассрочка / ипотека / кредит - PROCESS — как проходит стройка / этапы - DOCUMENTS — какие документы нужны - LAND — вопрос про участок - COMPARISON — сравнение материалов / технологий - SEASONAL — строительство зимой / в межсезонье - GREETING — приветствие, начало диалога - COMPLAINT — жалоба, негатив - SPAM — спам, реклама, нерелевантное - UNKNOWN — не удалось определить Также извлеки сущности (если есть): - area: площадь в м² (число) - budget: бюджет (число) - phone: телефон - name: имя клиента - location: город/регион - house_type: тип дома (каркасный, газобетон, кирпич) Ответ строго в JSON: {"intent": "...", "confidence": 0.XX, "entities": {...}} ``` **Модель:** DeepSeek (дешёвая, задача простая) #### 2.2.2 Отвечатель (`responder.py`) **Задача:** Сгенерировать ответ на основе intent + данных клиента (tenant). **Вход:** - `ClassificationResult` (intent + entities) - Context: данные tenant-а (прайс, FAQ, портфолио и т.д.) - История диалога (последние 5-10 сообщений) **Выход:** ```python @dataclass class Response: text: str # Текст ответа attachments: list # Фото/документы (опционально) needs_eval: bool # Нужна ли проверка eval pipeline facts_used: list # Какие факты из базы использованы (для eval) ``` **Промпт Отвечателя:** ``` Ты — менеджер по продажам компании {company_name}. ПРАВИЛА (ОБЯЗАТЕЛЬНО): 1. Отвечай ТОЛЬКО на основе предоставленных данных компании 2. НИКОГДА не выдумывай цены, сроки или факты 3. Если информации нет в данных — скажи "Уточню у менеджера и вернусь с ответом" 4. ЗАПРЕЩЕНО: обещать скидки, обсуждать юридику, принимать оплату 5. При назывании цены ВСЕГДА добавляй: "Точную стоимость рассчитаем после замера" 6. При назывании сроков ВСЕГДА добавляй: "±2 недели в зависимости от условий" 7. Стиль общения: {tone_of_voice} ДАННЫЕ КОМПАНИИ: {company_profile} ПРАЙС: {price_list} FAQ: {faq} ГАРАНТИИ: {guarantees} ПОРТФОЛИО: {portfolio} ИСТОРИЯ ДИАЛОГА: {conversation_history} INTENT: {intent} СООБЩЕНИЕ КЛИЕНТА: {user_message} Ответь клиенту. В конце укажи какие факты из данных компании ты использовал (для проверки). Формат: {"response": "...", "facts_used": ["price_list: каркасный дом 120м² = 5.5 млн", ...]} ``` **Контекст загружается по intent:** - PRICE_* → price_list.md + company_profile.md - PORTFOLIO → portfolio.md - GUARANTEE → guarantees.md - OBJECTION_* → objection_handlers.md - FAQ-вопросы → faq.md - Всегда: company_profile.md + escalation_rules.md **Модель:** DeepSeek (основная), Qwen (fallback) #### 2.2.3 Квалификатор (`qualifier.py`) **Задача:** Оценить лида после 3+ сообщений. **Вход:** Полная история диалога **Выход:** ```python @dataclass class QualificationResult: score: str # "hot" | "warm" | "cold" reasons: list # Почему такой скор collected_data: dict # Собранные данные: имя, телефон, бюджет, площадь next_action: str # Рекомендуемое действие ``` **Логика скоринга:** ``` HOT (готов к сделке): - Назвал бюджет ИЛИ площадь - Оставил телефон или попросил перезвонить - Спрашивает про встречу/замер → Действие: НЕМЕДЛЕННО уведомить владельца WARM (интерес есть): - Задал 3+ вопросов по существу - Спрашивает детали (материалы, этапы, сроки) - Но не оставил контакт → Действие: Follow-up через 24 часа, предложить замер COLD (просто смотрит): - 1-2 общих вопроса - Не уточняет детали - Возможно спам → Действие: Ответить и завершить ``` **Модель:** DeepSeek #### 2.2.4 Эскалатор (`escalator.py`) **Задача:** Решить, нужно ли передать диалог человеку. **Триггеры эскалации:** ```python ESCALATION_TRIGGERS = { # По intent "intent_based": ["COMPLAINT", "UNKNOWN"], # По содержанию "keywords": ["суд", "жалоба", "прокуратура", "обман", "мошенники", "юрист"], # По поведению "no_answer_count": 3, # Бот не смог ответить 3 раза подряд "negative_sentiment": True, # Клиент негативно настроен # По квалификации "hot_lead": True, # HOT лид → уведомить владельца } ``` **Действия при эскалации:** 1. Отправить владельцу уведомление в Telegram: ``` 🔥 Горячий лид! Клиент: {name} ({phone}) Канал: {channel} Интерес: {summary} Бюджет: {budget} Площадь: {area} Последние сообщения: {last_3_messages} ``` 2. Клиенту сказать: «Передаю ваш запрос менеджеру, он свяжется с вами в ближайшее время» --- ### 2.3 Ядро (`core/`) #### 2.3.1 Router / Planner (`router.py`) **Задача:** Оркестрация агентов. Главный контроллер. **Логика обработки сообщения:** ```python async def process_message(message: IncomingMessage) -> OutgoingMessage: # 1. Загрузить данные tenant-а tenant = tenant_manager.get(message.tenant_id) # 2. Загрузить/создать историю диалога history = memory.get_history(message.tenant_id, message.chat_id) # 3. Классифицировать classification = await classifier.classify(message.text, history) # 4. Проверить эскалацию if escalator.should_escalate(classification, history): await escalator.escalate(message, tenant, history) return OutgoingMessage( channel=message.channel, chat_id=message.chat_id, text="Передаю ваш запрос менеджеру. Он свяжется с вами в ближайшее время!" ) # 5. Сгенерировать ответ response = await responder.respond(classification, tenant, history) # 6. Проверить через eval pipeline eval_result = await evaluator.check(response, tenant) if not eval_result.passed: # Ответ не прошёл проверку → fallback response.text = "Уточню информацию у менеджера и вернусь с ответом!" await escalator.escalate(message, tenant, history, reason="eval_failed") # 7. Сохранить в историю memory.add(message.tenant_id, message.chat_id, message.text, response.text) # 8. Квалифицировать (если 3+ сообщений) if len(history) >= 3: qualification = await qualifier.qualify(history) if qualification.score == "hot": await escalator.notify_owner(message, tenant, qualification) # Сохранить квалификацию в БД storage.save_lead(message, qualification) # 9. Сохранить метрики storage.save_metric(message, classification, response) return OutgoingMessage( channel=message.channel, chat_id=message.chat_id, text=response.text, attachments=response.attachments ) ``` #### 2.3.2 LLM обёртка (`llm.py`) ```python class LLMClient: """Обёртка над LLM API с fallback""" def __init__(self): self.primary = DeepSeekClient(api_key=config.DEEPSEEK_API_KEY) self.fallback = OpenRouterClient( api_key=config.OPENROUTER_API_KEY, model="qwen/qwen-2.5-72b-instruct" ) async def complete(self, messages: list, temperature: float = 0.3) -> str: """Отправить запрос к LLM с fallback""" try: return await self.primary.complete(messages, temperature) except Exception as e: logger.warning(f"DeepSeek failed: {e}, falling back to Qwen") return await self.fallback.complete(messages, temperature) ``` **Параметры LLM:** - Temperature: 0.3 (стабильные ответы, минимум креатива) - Max tokens: 500 (ответы клиентам короткие) - Для классификатора: temperature 0.1, max_tokens 200 #### 2.3.3 Память диалогов (`memory.py`) ```python class ConversationMemory: """Хранение истории диалогов per-session""" MAX_HISTORY = 10 # Максимум сообщений в контексте def get_history(self, tenant_id: str, chat_id: str) -> list: """Получить историю диалога (последние MAX_HISTORY сообщений)""" pass def add(self, tenant_id: str, chat_id: str, user_msg: str, bot_msg: str): """Добавить пару сообщений в историю""" pass def summarize_if_needed(self, tenant_id: str, chat_id: str): """Если история > MAX_HISTORY → суммаризировать старые сообщения""" pass def clear(self, tenant_id: str, chat_id: str): """Очистить историю (конец диалога)""" pass ``` **Хранение:** SQLite (таблица `conversations`) **Защита от context degradation:** - Хранить максимум 10 сообщений - При превышении → суммаризировать первые 5 в одно предложение #### 2.3.4 Eval Pipeline (`eval.py`) ```python class Evaluator: """Проверка ответов бота перед отправкой""" async def check(self, response: Response, tenant: Tenant) -> EvalResult: """Проверить ответ на корректность""" checks = [] # 1. Проверка цен if self._mentions_price(response.text): price_ok = self._verify_price(response.text, tenant.price_list) checks.append(("price_check", price_ok)) # 2. Проверка сроков if self._mentions_timeline(response.text): timeline_ok = self._verify_timeline(response.text, tenant.company_profile) checks.append(("timeline_check", timeline_ok)) # 3. Проверка запрещённых обещаний forbidden_ok = self._check_forbidden(response.text) checks.append(("forbidden_check", forbidden_ok)) # 4. Проверка Trust Matrix trust_ok = self._check_trust_matrix(response.text) checks.append(("trust_check", trust_ok)) passed = all(ok for _, ok in checks) return EvalResult(passed=passed, checks=checks) def _check_forbidden(self, text: str) -> bool: """Проверить что бот не обещает запрещённое""" FORBIDDEN_PATTERNS = [ r"скидк[аиуе]", r"бесплатно", r"гарантирую", r"100%", r"точная (цена|стоимость)", r"(возврат|вернём) деньг", ] for pattern in FORBIDDEN_PATTERNS: if re.search(pattern, text, re.IGNORECASE): return False return True ``` #### 2.3.5 Trust Matrix (`trust.py`) ```python TRUST_MATRIX = { "can_do": [ "answer_faq", # Отвечать на FAQ "quote_price_range", # Называть диапазон цен (из прайса) "show_portfolio", # Показывать портфолио "schedule_meeting", # Назначать встречу/замер "collect_contacts", # Собирать контакты "describe_process", # Описывать процесс стройки "compare_materials", # Сравнивать материалы ], "can_do_with_disclaimer": [ "quote_exact_price", # Цена + "точную рассчитаем после замера" "quote_timeline", # Сроки + "±2 недели" "describe_guarantee", # Гарантия (только стандартный текст из базы) ], "cannot_do": [ "promise_discount", # Обещать скидки "discuss_legal", # Обсуждать юридику "accept_payment", # Принимать оплату "confirm_unverified", # Подтверждать непроверенные факты "make_commitments", # Давать обязательства от имени компании "share_competitor_info",# Обсуждать конкурентов негативно ] } ``` --- ### 2.4 Мульти-тенант (`tenants/`) #### 2.4.1 Конфиг клиента (`data/{tenant_id}/config.yaml`) ```yaml tenant_id: "stroydream" company_name: "СтройДрим" active: true # Каналы channels: telegram: enabled: true bot_token: "BOT_TOKEN_HERE" max: enabled: false bot_token: "" avito: enabled: true client_id: "AVITO_CLIENT_ID" client_secret: "AVITO_CLIENT_SECRET" # Владелец (куда слать уведомления) owner: telegram_chat_id: "123456789" name: "Иван Петрович" # Настройки бота settings: language: "ru" tone: "formal" # formal | friendly | casual response_max_length: 500 # Макс. длина ответа в символах working_hours: "08:00-20:00" # Рабочие часы (вне них: "Ответим утром") timezone: "Europe/Kaliningrad" # LLM llm: model: "deepseek" # deepseek | qwen | auto temperature: 0.3 ``` #### 2.4.2 Менеджер клиентов (`manager.py`) ```python class TenantManager: """Управление клиентами""" def load_all(self) -> dict: """Загрузить всех активных клиентов из data/""" pass def get(self, tenant_id: str) -> Tenant: """Получить данные клиента""" pass def reload(self, tenant_id: str): """Перезагрузить данные клиента (при обновлении прайса и т.д.)""" pass def add(self, tenant_id: str, config: dict): """Добавить нового клиента""" pass ``` --- ### 2.5 Хранение (`storage/`) #### 2.5.1 База данных: SQLite **Таблицы:** ```sql -- Диалоги CREATE TABLE conversations ( id INTEGER PRIMARY KEY, tenant_id TEXT NOT NULL, channel TEXT NOT NULL, -- telegram/max/avito chat_id TEXT NOT NULL, user_id TEXT, user_name TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, status TEXT DEFAULT 'active' -- active/closed/escalated ); -- Сообщения CREATE TABLE messages ( id INTEGER PRIMARY KEY, conversation_id INTEGER REFERENCES conversations(id), role TEXT NOT NULL, -- user/bot text TEXT NOT NULL, intent TEXT, -- Определённый intent confidence REAL, timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); -- Лиды CREATE TABLE leads ( id INTEGER PRIMARY KEY, conversation_id INTEGER REFERENCES conversations(id), tenant_id TEXT NOT NULL, score TEXT NOT NULL, -- hot/warm/cold name TEXT, phone TEXT, budget TEXT, area TEXT, house_type TEXT, notes TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); -- Метрики CREATE TABLE metrics ( id INTEGER PRIMARY KEY, tenant_id TEXT NOT NULL, date DATE NOT NULL, total_messages INTEGER DEFAULT 0, total_conversations INTEGER DEFAULT 0, hot_leads INTEGER DEFAULT 0, warm_leads INTEGER DEFAULT 0, cold_leads INTEGER DEFAULT 0, escalations INTEGER DEFAULT 0, eval_failures INTEGER DEFAULT 0, avg_response_time_ms INTEGER DEFAULT 0 ); ``` --- ### 2.6 Отчёты (`reports/`) #### 2.6.1 Еженедельный отчёт ```python class Reporter: """Генерация отчётов для клиентов""" async def weekly_report(self, tenant_id: str) -> str: """Сгенерировать еженедельный отчёт""" # Собрать метрики за неделю # Форматировать в читаемый текст # Отправить владельцу в Telegram pass ``` **Формат отчёта:** ``` 📊 Еженедельный отчёт: {company_name} Период: {date_from} — {date_to} 💬 Диалогов: {total_conversations} 📩 Сообщений: {total_messages} 🔥 Горячих лидов: {hot_leads} 🟡 Тёплых: {warm_leads} 🔵 Холодных: {cold_leads} ⚡ Среднее время ответа: {avg_response_time} сек 📈 Конверсия в лида: {conversion}% ❓ Топ-5 вопросов клиентов: 1. {top_question_1} ({count_1} раз) 2. {top_question_2} ({count_2} раз) ... ⚠️ Эскалаций: {escalations} ❌ Ошибок eval: {eval_failures} ``` --- ## 3. КОНФИГУРАЦИЯ ### 3.1 Переменные окружения (`.env`) ```env # LLM DEEPSEEK_API_KEY=sk-... OPENROUTER_API_KEY=sk-or-... # База данных DATABASE_URL=sqlite:///data/bot.db # Логирование LOG_LEVEL=INFO LOG_FILE=logs/bot.log # Уведомления (admin) ADMIN_TELEGRAM_TOKEN=... ADMIN_TELEGRAM_CHAT_ID=... # Авито (общие) AVITO_CLIENT_ID=... AVITO_CLIENT_SECRET=... ``` ### 3.2 Зависимости (`requirements.txt`) ``` aiogram==3.x aiohttp>=3.9 sqlalchemy>=2.0 pydantic>=2.0 pyyaml>=6.0 python-dotenv>=1.0 httpx>=0.27 structlog>=24.0 ``` --- ## 4. ЗАЩИТА ОТ СБОЕВ ### 4.1 Specification drift protection В `router.py` каждые 5 сообщений принудительно вставлять system prompt: ```python if len(history) % 5 == 0: messages.insert(1, { "role": "system", "content": f"НАПОМИНАНИЕ: Ты менеджер {tenant.company_name}. " f"Отвечай ТОЛЬКО на основе данных компании. " f"НЕ выдумывай. НЕ обещай скидки." }) ``` ### 4.2 Sycophantic confirmation protection В промпте Отвечателя жёсткое правило: ``` КРИТИЧЕСКОЕ ПРАВИЛО: Если клиент утверждает факт (цену, срок, гарантию), а этого факта НЕТ в данных компании — НЕ ПОДТВЕРЖДАЙ. Скажи: "Позвольте уточню эту информацию у менеджера." ``` ### 4.3 Context degradation protection В `memory.py`: ```python MAX_HISTORY = 10 # При превышении → суммаризация через LLM # "Суммаризируй предыдущий диалог в 2 предложения" ``` ### 4.4 Silent failure detection В `eval.py`: - Проверка всех цен по прайсу - Проверка сроков по профилю - Проверка на запрещённые паттерны - Логирование ВСЕХ ответов в БД для последующего human review ### 4.5 Cascading failure protection В `router.py`: ```python try: classification = await classifier.classify(...) except Exception: # Классификатор упал → fallback: отвечаем общим ответом return generic_response("Спасибо за обращение! Менеджер свяжется с вами.") ``` ### 4.6 Rate limiting ```python # Защита от спама MAX_MESSAGES_PER_MINUTE = 10 # От одного пользователя MAX_MESSAGES_PER_HOUR = 100 # От одного пользователя ``` --- ## 5. ТЕСТЫ ### 5.1 Тестовые сценарии (`tests/test_scenarios.py`) 20 сценариев для автоматического тестирования: ```python TEST_SCENARIOS = [ # 1. Простой вопрос о цене {"input": "Сколько стоит дом?", "expected_intent": "PRICE_GENERAL"}, # 2. Конкретный вопрос о цене {"input": "Каркасный дом 120 м², сколько выйдет?", "expected_intent": "PRICE_SPECIFIC"}, # 3. Сроки {"input": "За сколько построите?", "expected_intent": "TIMELINE"}, # 4. Портфолио {"input": "Покажите ваши работы", "expected_intent": "PORTFOLIO"}, # 5. Гарантия {"input": "Какая гарантия на дом?", "expected_intent": "GUARANTEE"}, # 6. Встреча {"input": "Хочу приехать посмотреть, когда можно?", "expected_intent": "MEETING"}, # 7. Возражение "дорого" {"input": "Что-то дороговато у вас", "expected_intent": "OBJECTION_PRICE"}, # 8. Возражение "доверие" {"input": "А откуда мне знать что вы нормально строите?", "expected_intent": "OBJECTION_TRUST"}, # 9. Жалоба {"input": "Вы обманщики! Позвоню в прокуратуру!", "expected_intent": "COMPLAINT"}, # 10. Спам {"input": "Продам кирпич оптом, дёшево", "expected_intent": "SPAM"}, # 11. Trust Matrix: не обещает скидку {"input": "Дайте скидку 20%", "check": "response_not_contains", "pattern": "скидк"}, # 12. Trust Matrix: не подтверждает ложный факт {"input": "Вы же строите за 2 недели, верно?", "check": "response_not_confirms"}, # 13. Eval: цена из прайса {"input": "Сколько стоит каркасный 100м²?", "check": "price_from_pricelist"}, # 14. Квалификация HOT {"history": ["Хочу дом 150м²", "Бюджет 8 млн", "Позвоните мне: 89001234567"], "expected_score": "hot"}, # 15. Квалификация COLD {"history": ["Сколько стоит?"], "expected_score": "cold"}, # И т.д. — 20 сценариев ] ``` --- ## 6. ЗАПУСК ### 6.1 Установка ```bash # 1. Клонировать репозиторий git clone ai-sales-bot cd ai-sales-bot # 2. Создать виртуальное окружение python3 -m venv venv source venv/bin/activate # 3. Установить зависимости pip install -r requirements.txt # 4. Настроить переменные окружения cp .env.example .env # Заполнить .env (API ключи) # 5. Создать БД python scripts/init_db.py # 6. Добавить первого клиента python scripts/add_tenant.py --id demo --name "Демо Стройка" # Заполнить data/demo/*.md # 7. Запустить python main.py ``` ### 6.2 Деплой (systemd) ```ini [Unit] Description=AI Sales Bot After=network.target [Service] Type=simple User=bot WorkingDirectory=/opt/ai-sales-bot ExecStart=/opt/ai-sales-bot/venv/bin/python main.py Restart=always RestartSec=10 EnvironmentFile=/opt/ai-sales-bot/.env [Install] WantedBy=multi-user.target ``` ### 6.3 Мониторинг - Логи: `logs/bot.log` (structlog, JSON формат) - Метрики: таблица `metrics` в SQLite - Алерты: если бот не ответил > 60 сек → уведомление админу - Healthcheck: `GET /health` → 200 OK (если добавить aiohttp сервер) --- ## 7. ПОРЯДОК РАЗРАБОТКИ ### Фаза 1: Ядро (приоритет) 1. `config.py` + `.env` 2. `core/llm.py` (обёртка DeepSeek + fallback) 3. `agents/classifier.py` (классификатор intent-ов) 4. `agents/responder.py` (генерация ответов) 5. `core/trust.py` (Trust Matrix) 6. `core/eval.py` (проверка ответов) 7. `core/memory.py` (память диалогов) 8. `core/router.py` (оркестрация) ### Фаза 2: Telegram 9. `adapters/telegram_adapter.py` 10. `tenants/manager.py` + `loader.py` 11. `storage/database.py` + `models.py` 12. `main.py` (запуск) 13. Тесты 20 сценариев ### Фаза 3: Квалификация + отчёты 14. `agents/qualifier.py` 15. `agents/escalator.py` 16. `reports/reporter.py` ### Фаза 4: Доп. каналы 17. `adapters/avito_adapter.py` 18. `adapters/max_adapter.py` ### Фаза 5: Утилиты 19. `scripts/add_tenant.py` 20. `scripts/generate_report.py` 21. systemd + деплой --- ## 8. ДОПОЛНЕНИЯ (v1.1) ### 8.1 Обработка медиа (`adapters/`) Добавить в каждый адаптер: ```python async def handle_media(message: IncomingMessage) -> IncomingMessage: """Обработка нетекстовых сообщений""" if message.type == "photo": message.text = "[ФОТО] Клиент отправил фото" # Ответ: "Спасибо за фото! Подскажите, какой у вас вопрос?" elif message.type == "voice": # Вариант 1: транскрипция через Whisper API message.text = await whisper_transcribe(message.file) # Вариант 2: "Пришлите текстом, так смогу точнее ответить" elif message.type == "document": message.text = "[ДОКУМЕНТ] Клиент отправил документ" # Ответ: "Получил документ, передам менеджеру" await escalator.escalate(message, reason="document_received") elif message.type == "sticker": # Игнорировать стикеры return None return message ``` ### 8.2 Follow-up / Автонапоминания (`core/scheduler.py`) ```python class FollowUpScheduler: """Автоматические напоминания для warm-лидов""" RULES = { "warm": [ {"delay_hours": 24, "message": "Добрый день! Вы интересовались {topic}. Остались вопросы?"}, {"delay_hours": 72, "message": "Напоминаю о нашем предложении по {topic}. Готовы обсудить детали?"}, ], "hot_no_response": [ {"delay_hours": 2, "message": "Менеджер скоро свяжется с вами! Если срочно — позвоните {phone}"}, ] } async def check_and_send(self): """Проверять каждые 30 мин, кому пора отправить follow-up""" pass # Запускать через asyncio.create_task в main.py # Останавливать follow-up если клиент ответил ``` ### 8.3 Дедупликация сообщений (`core/dedup.py`) ```python class MessageDeduplicator: """Защита от дубликатов (Авито/Telegram могут слать дважды)""" TTL = 300 # 5 минут def __init__(self): self._seen = {} # {message_hash: timestamp} def is_duplicate(self, message: IncomingMessage) -> bool: key = f"{message.channel}:{message.chat_id}:{message.text}" h = hashlib.md5(key.encode()).hexdigest() if h in self._seen and time.time() - self._seen[h] < self.TTL: return True self._seen[h] = time.time() return False ``` ### 8.4 Статус диалога и возврат после эскалации Добавить в таблицу `conversations`: ```sql status TEXT DEFAULT 'active' -- active: бот обрабатывает -- escalated: бот молчит, ждёт менеджера -- returned: менеджер вернул боту -- closed: диалог завершён ``` Логика в `router.py`: ```python async def process_message(message): conv = storage.get_conversation(message.tenant_id, message.chat_id) if conv.status == "escalated": # Бот молчит, пересылает менеджеру await forward_to_owner(message, tenant) return None # Не отвечаем # ... обычная обработка ``` Команда для менеджера (в боте-уведомителе): ``` /return {chat_id} — вернуть диалог боту /close {chat_id} — закрыть диалог /takeover {chat_id} — забрать диалог себе ``` ### 8.5 Тройной fallback LLM ```python class LLMClient: async def complete(self, messages, temperature=0.3): # Уровень 1: DeepSeek try: return await self.deepseek.complete(messages, temperature) except Exception: pass # Уровень 2: Qwen через OpenRouter try: return await self.qwen.complete(messages, temperature) except Exception: pass # Уровень 3: Шаблонный ответ (без LLM) logger.critical("All LLMs failed!") return FALLBACK_RESPONSE FALLBACK_RESPONSE = "Спасибо за обращение! Менеджер свяжется с вами в ближайшее время." ``` ### 8.6 Рабочие часы В `router.py`: ```python def is_working_hours(tenant) -> bool: tz = pytz.timezone(tenant.settings.timezone) now = datetime.now(tz) start, end = tenant.settings.working_hours.split("-") start_h, start_m = map(int, start.split(":")) end_h, end_m = map(int, end.split(":")) current = now.hour * 60 + now.minute return start_h * 60 + start_m <= current <= end_h * 60 + end_m async def process_message(message): if not is_working_hours(tenant): return OutgoingMessage( text=f"Спасибо за обращение! Мы работаем с {tenant.settings.working_hours}. " f"Ответим вам первым делом утром!" ) # ... обычная обработка ``` ### 8.7 Обновление данных клиента через бота Команды для владельца (через бот-уведомитель): ``` /update_price — обновить прайс (прислать текст или файл) /update_faq — добавить вопрос в FAQ /stats — текущая статистика /leads — список горячих лидов за сегодня /pause — приостановить бота /resume — возобновить бота ``` ### 8.8 Бэкап БД Добавить `scripts/backup.sh`: ```bash #!/bin/bash DATE=$(date +%Y-%m-%d) cp /opt/ai-sales-bot/data/bot.db /opt/ai-sales-bot/backups/bot_${DATE}.db # Хранить последние 30 бэкапов ls -t /opt/ai-sales-bot/backups/ | tail -n +31 | xargs rm -f ``` Cron: `0 3 * * * /opt/ai-sales-bot/scripts/backup.sh` ### 8.9 Защита персданных ```python # В storage/models.py from cryptography.fernet import Fernet class PersonalDataEncryptor: """Шифрование телефонов и имён в БД""" def __init__(self, key: str): self.cipher = Fernet(key.encode()) def encrypt(self, value: str) -> str: return self.cipher.encrypt(value.encode()).decode() def decrypt(self, value: str) -> str: return self.cipher.decrypt(value.encode()).decode() # Автоудаление старых данных async def cleanup_old_data(): """Удалить персданные старше 90 дней""" storage.execute( "DELETE FROM leads WHERE created_at < datetime('now', '-90 days')" ) ``` Добавить в `requirements.txt`: `cryptography>=42.0` ### 8.10 Webhook для Telegram (production) В `adapters/telegram_adapter.py`: ```python class TelegramAdapter(BaseAdapter): async def start(self): if config.TELEGRAM_USE_WEBHOOK: # Webhook (production) — быстрее, надёжнее await bot.set_webhook( url=f"https://{config.DOMAIN}/webhook/tg/{self.tenant_id}" ) # Запустить aiohttp сервер для приёма webhook else: # Polling (development) — проще для отладки await dp.start_polling(bot) ``` --- ## 9. ОБНОВЛЁННЫЙ ПОРЯДОК РАЗРАБОТКИ ### Фаза 1: Ядро (5-7 дней) 1. config.py + .env 2. core/llm.py (DeepSeek + Qwen + fallback) 3. core/dedup.py (дедупликация) 4. agents/classifier.py 5. agents/responder.py 6. core/trust.py 7. core/eval.py 8. core/memory.py 9. core/router.py (+ рабочие часы) ### Фаза 2: Telegram + хранение (3-5 дней) 10. adapters/base.py (+ обработка медиа) 11. adapters/telegram_adapter.py (polling) 12. tenants/manager.py + loader.py 13. storage/database.py + models.py (+ шифрование) 14. main.py ### Фаза 3: Квалификация + эскалация (2-3 дня) 15. agents/qualifier.py 16. agents/escalator.py (+ статус диалога + возврат) 17. core/scheduler.py (follow-up) ### Фаза 4: Отчёты + команды (2-3 дня) 18. reports/reporter.py 19. Команды владельца (/update_price, /stats, /leads) 20. scripts/backup.sh ### Фаза 5: Тесты + деплой (2-3 дня) 21. tests/ (20 сценариев) 22. systemd + webhook 23. Мониторинг + алерты ### Фаза 6: Доп. каналы (3-5 дней) 24. adapters/avito_adapter.py 25. adapters/max_adapter.py **Общий срок: 15-25 дней** (для Claude Code — быстрее) --- *Файл обновлён: 2026-03-29 (v1.1)* *Путь: /root/.openclaw/workspace/projects/ai-agency/TECHNICAL-SPEC.md*