Как создать умных ботов на Python: пошаговое руководство – от идеи к коду

import re
import random

# Определяем паттерны ответов
patterns = {
r'привет|здравствуй|хай': ['Привет!', 'Здравствуйте!', 'Рад вас видеть!'],
r'как дела|как ты': ['Отлично, спасибо!', 'Всё хорошо, а у вас?'],
r'прощай|пока|до свидания': ['До свидания!', 'Пока!', 'До новых встреч!'],
r'помощь|что ты умеешь': ['Я могу отвечать на простые вопросы.', 'Я чат-бот. Задайте мне вопрос!']
}

def get_response(user_input):
user_input = user_input.lower()

for pattern, responses in patterns.items():
if re.search(pattern, user_input):
return random.choice(responses)

return "Извините, я не понимаю вас. Можете перефразировать?"

# Простой интерфейс для общения с ботом
def chat():
print("Бот: Привет! Я простой чат-бот. Напишите 'выход', чтобы завершить.")

while True:
user_input = input("Вы: ")

if user_input.lower() == 'выход':
print("Бот: До свидания!")
break

response = get_response(user_input)
print(f"Бот: {response}")

if __name__ == "__main__":
chat()

from chatterbot import ChatBot
from chatterbot.trainers import ChatterBotCorpusTrainer

# Создаем экземпляр бота
bot = ChatBot('MyBot', 
storage_adapter='chatterbot.storage.SQLStorageAdapter',
logic_adapters=[
'chatterbot.logic.MathematicalEvaluation',
'chatterbot.logic.BestMatch'
]
)

# Создаем тренера и обучаем бота
trainer = ChatterBotCorpusTrainer(bot)
trainer.train("chatterbot.corpus.russian") # Для англоязычного бота используйте "english"

# Функция для получения ответа от бота
def get_bot_response(user_input):
return bot.get_response(user_input)

# Интерфейс чата
def run_chatbot():
print("Бот: Привет! Я бот на базе ChatterBot. Напишите 'выход', чтобы завершить.")

while True:
user_input = input("Вы: ")

if user_input.lower() == 'выход':
print("Бот: До свидания!")
break

response = get_bot_response(user_input)
print(f"Бот: {response}")

if __name__ == "__main__":
run_chatbot()

import logging
from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler, MessageHandler, filters, ContextTypes

# Настраиваем логирование
logging.basicConfig(
format='%(asctime)s – %(name)s – %(levelname)s – %(message)s',
level=logging.INFO
)

# Функция для команды /start
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
await update.message.reply_text('Привет! Я ваш бот. Чем могу помочь?')

# Обработка текстовых сообщений
async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
user_text = update.message.text
# Здесь можно использовать функцию get_bot_response из предыдущего примера
response = f"Вы сказали: {user_text}" # Заглушка
await update.message.reply_text(response)

if __name__ == '__main__':
# Создаем приложение
application = ApplicationBuilder().token('YOUR_TELEGRAM_BOT_TOKEN').build()

# Добавляем обработчики
application.add_handler(CommandHandler("start", start))
application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))

# Запускаем бота
application.run_polling()

import requests
from datetime import datetime

def get_weather(city):
api_key = "YOUR_OPENWEATHER_API_KEY"
url = f"http://api.openweathermap.org/data/2.5/weather?q={city}&appid={api_key}&units=metric"

response = requests.get(url)
data = response.json()

if response.status_code == 200:
temp = data['main']['temp']
weather_desc = data['weather'][0]['description']
humidity = data['main']['humidity']

return f"Погода в {city}: {weather_desc}, температура: {temp}°C, влажность: {humidity}%"
else:
return "Не удалось получить данные о погоде. Проверьте название города."

# Интеграция функции с ботом
def process_weather_request(user_input):
# Извлекаем название города из запроса пользователя
# Простой пример – предполагаем, что город указан после слова "погода"
if "погода" in user_input.lower():
parts = user_input.lower().split("погода")
if len(parts) > 1 and parts[1].strip():
city = parts[1].strip()
return get_weather(city)

return "Укажите город после слова 'погода', например: 'погода Москва'"

pip install SpeechRecognition PyAudio gTTS pyttsx3 playsound

import speech_recognition as sr
import pyttsx3
import time

# Инициализация движка преобразования текста в речь
def init_tts_engine():
engine = pyttsx3.init()
voices = engine.getProperty('voices')
# Выбор голоса (может отличаться в зависимости от ОС)
engine.setProperty('voice', voices[0].id) # Обычно индекс 0 – мужской голос, 1 – женский
return engine

# Функция для преобразования текста в речь
def speak(engine, text):
engine.say(text)
engine.runAndWait()

# Функция распознавания речи
def recognize_speech():
r = sr.Recognizer()

with sr.Microphone() as source:
print("Слушаю...")
audio = r.listen(source)

try:
text = r.recognize_google(audio, language="ru-RU") # Для русского языка
print(f"Вы сказали: {text}")
return text.lower()
except sr.UnknownValueError:
return "Не удалось распознать речь"
except sr.RequestError:
return "Ошибка сервиса распознавания речи"

# Обработка команд (аналогично текстовому боту)
def process_command(command):
if "привет" in command:
return "Здравствуйте! Чем могу помочь?"
elif "время" in command:
current_time = time.strftime("%H:%M")
return f"Текущее время {current_time}"
elif "пока" in command or "до свидания" in command:
return "До свидания! Хорошего дня!"
else:
return "Извините, я не понимаю эту команду"

# Основной цикл голосового помощника
def main():
engine = init_tts_engine()
speak(engine, "Голосовой помощник запущен. Чем могу помочь?")

while True:
command = recognize_speech()

if "выход" in command:
speak(engine, "Выключаюсь. До свидания!")
break

response = process_command(command)
speak(engine, response)
print(response)

if __name__ == "__main__":
main()

from gtts import gTTS
from playsound import playsound
import os
import uuid

# Функция преобразования текста в речь с использованием gTTS
def speak_gtts(text, language='ru'):
filename = f"voice_{str(uuid.uuid4())}.mp3"
tts = gTTS(text=text, lang=language, slow=False)
tts.save(filename)
playsound(filename)
os.remove(filename)

# Модифицированная основная функция
def main_gtts():
speak_gtts("Голосовой помощник запущен. Чем могу помочь?")

while True:
command = recognize_speech()

if "выход" in command:
speak_gtts("Выключаюсь. До свидания!")
break

response = process_command(command)
speak_gtts(response)
print(response)

import requests

def get_weather(city):
api_key = "YOUR_OPENWEATHER_API_KEY"
url = f"http://api.openweathermap.org/data/2.5/weather?q={city}&appid={api_key}&units=metric"

response = requests.get(url)
data = response.json()

if response.status_code == 200:
temp = data['main']['temp']
weather_desc = data['weather'][0]['description']

return f"Погода в городе {city}: {weather_desc}, температура: {temp} градусов Цельсия"
else:
return "Не удалось получить данные о погоде. Проверьте название города."

# Дополненная функция обработки команд
def process_command_extended(command):
if "привет" in command:
return "Здравствуйте! Чем могу помочь?"
elif "время" in command:
current_time = time.strftime("%H:%M")
return f"Текущее время {current_time}"
elif "погода" in command:
command_parts = command.split("погода")
if len(command_parts) > 1 and command_parts[1].strip():
city = command_parts[1].strip()
return get_weather(city)
else:
return "Укажите город. Например, скажите 'погода Москва'"
elif "пока" in command or "до свидания" in command:
return "До свидания! Хорошего дня!"
else:
return "Извините, я не понимаю эту команду"

def enhanced_speech_recognition():
r = sr.Recognizer()

with sr.Microphone() as source:
print("Настройка окружающего шума...")
r.adjust_for_ambient_noise(source, duration=1)
print("Слушаю...")

r.pause_threshold = 0.8 # Пауза перед окончанием фразы
r.energy_threshold = 300 # Порог громкости речи

audio = r.listen(source, timeout=5, phrase_time_limit=10)

try:
text = r.recognize_google(audio, language="ru-RU")
print(f"Вы сказали: {text}")
return text.lower()
except sr.UnknownValueError:
return "Не удалось распознать речь"
except sr.RequestError:
return "Ошибка сервиса распознавания речи"
except Exception as e:
return f"Произошла ошибка: {str(e)}"

import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.naive_bayes import MultinomialNB
from sklearn.pipeline import Pipeline

# Обучающие данные: примеры фраз и их классификация
training_data = [
("как дела", "greeting"),
("привет", "greeting"),
("здравствуйте", "greeting"),
("какая погода в москве", "weather"),
("погода в санкт-петербурге", "weather"),
("прогноз погоды", "weather"),
("который час", "time"),
("текущее время", "time"),
("сколько времени", "time"),
("до свидания", "goodbye"),
("пока", "goodbye"),
("прощай", "goodbye")
]

# Разделение данных на фразы и метки
phrases, labels = zip(*training_data)

# Создание и обучение классификатора
def train_intent_classifier():
# Создаем конвейер: векторизация TF-IDF + наивный байесовский классификатор
pipeline = Pipeline([
('tfidf', TfidfVectorizer()),
('classifier', MultinomialNB())
])

# Обучаем классификатор
pipeline.fit(phrases, labels)

return pipeline

# Классификация намерения пользователя
def classify_intent(classifier, text):
# Предсказываем класс
intent = classifier.predict([text])[0]

# Получаем вероятности для всех классов
probas = classifier.predict_proba([text])[0]
classes = classifier.classes_

# Находим индекс максимальной вероятности
max_proba_index = np.argmax(probas)

# Проверяем, достаточно ли высока вероятность для определенного класса
if probas[max_proba_index] > 0.5: # Порог уверенности
return intent
else:
return "unknown"

import spacy

# Загрузка модели spaCy для русского языка
nlp = spacy.load("ru_core_news_sm")

# Функция для извлечения сущностей
def extract_entities(text):
doc = nlp(text)

entities = {
"LOC": [], # Локации
"PERSON": [], # Имена людей
"DATE": [], # Даты
"TIME": [], # Время
"ORG": [] # Организации
}

# Извлекаем именованные сущности
for ent in doc.ents:
if ent.label_ in entities:
entities[ent.label_].append(ent.text)

# Для запросов о погоде часто нужно извлечь название города
if "погода" in text.lower() and not entities["LOC"]:
for token in doc:
if token.text.lower() == "погода" and token.i + 2 < len(doc):
next_token = doc[token.i + 2]
if next_token.pos_ == "NOUN" or next_token.pos_ == "PROPN":
entities["LOC"].append(next_token.text)

return entities

class DialogContext:
def __init__(self):
self.current_topic = None
self.entities = {}
self.history = []
self.last_intent = None

def update(self, user_input, intent, new_entities):
self.history.append({"user": user_input})
self.last_intent = intent

for entity_type, values in new_entities.items():
if values:
self.entities[entity_type] = values

if intent != "unknown":
self.current_topic = intent

def get_context_info(self):
return {
"current_topic": self.current_topic,
"entities": self.entities,
"last_intent": self.last_intent
}

def add_bot_response(self, response):
self.history.append({"bot": response})

# Генерация ответа с учетом контекста
def generate_response(intent, entities, context):
if intent == "greeting":
return "Здравствуйте! Чем могу помочь?"

elif intent == "weather":
location = None
if "LOC" in entities and entities["LOC"]:
location = entities["LOC"][0]
elif "LOC" in context.entities:
location = context.entities["LOC"][0]

if location:
return f"Погода в {location} сейчас: +20°C, солнечно"
else:
return "В каком городе вас интересует погода?"

elif intent == "time":
import datetime
current_time = datetime.datetime.now().strftime("%H:%M")
return f"Текущее время: {current_time}"

elif intent == "goodbye":
return "До свидания! Обращайтесь, если понадобится помощь."

else:
return "Извините, я не совсем понимаю, что вы имеете в виду."

# Инициализация компонентов
intent_classifier = train_intent_classifier()
dialog_context = DialogContext()

# Функция обработки запроса пользователя
def process_user_request(user_input):
intent = classify_intent(intent_classifier, user_input)

entities = extract_entities(user_input)

dialog_context.update(user_input, intent, entities)

response = generate_response(intent, entities, dialog_context)

dialog_context.add_bot_response(response)

return response

# Цикл обработки диалога
def chat_loop():
print("Бот: Привет! Я умный бот с NLP. Напишите 'выход' для завершения.")

while True:
user_input = input("Вы: ")

if user_input.lower() == 'выход':
print("Бот: До свидания!")
break

response = process_user_request(user_input)
print(f"Бот: {response}")

if __name__ == "__main__":
chat_loop()

import unittest
from bot_functions import classify_intent, extract_entities, generate_response
from dialog_context import DialogContext

class BotTestCase(unittest.TestCase):

def setUp(self):
self.intent_classifier = train_intent_classifier()
self.context = DialogContext()

def test_intent_classification(self):
self.assertEqual(classify_intent(self.intent_classifier, "привет как дела"), "greeting")
self.assertEqual(classify_intent(self.intent_classifier, "какая погода в москве"), "weather")
self.assertEqual(classify_intent(self.intent_classifier, "сколько времени"), "time")

def test_entity_extraction(self):
entities = extract_entities("какая погода в Москве")
self.assertIn("Москве", entities["LOC"])

entities = extract_entities("погода в Санкт-Петербурге завтра")
self.assertIn("Санкт-Петербурге", entities["LOC"])
self.assertIn("завтра", entities["DATE"])

def test_context_management(self):
self.context.update("погода в Москве", "weather", {"LOC": ["Москве"]})
context_info = self.context.get_context_info()

self.assertEqual(context_info["current_topic"], "weather")
self.assertIn("LOC", context_info["entities"])
self.assertIn("Москве", context_info["entities"]["LOC"])

self.context.update("а завтра?", "weather", {"DATE": ["завтра"]})
context_info = self.context.get_context_info()

self.assertEqual(context_info["current_topic"], "weather")
self.assertIn("Москве", context_info["entities"]["LOC"])
self.assertIn("завтра", context_info["entities"]["DATE"])

def test_response_generation(self):
response = generate_response("greeting", {}, self.context)
self.assertIn("Здравствуйте", response)

response = generate_response("weather", {"LOC": ["Москва"]}, self.context)
self.assertIn("Москва", response)

if __name__ == '__main__':
unittest.main()

from sklearn.model_selection import cross_val_score, train_test_split
from sklearn.metrics import classification_report, confusion_matrix
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns

# Подготовка данных для оценки
phrases, labels = zip(*training_data)
X_train, X_test, y_train, y_test = train_test_split(phrases, labels, test_size=0.3, random_state=42)

# Создание и обучение классификатора
pipeline = Pipeline([
('tfidf', TfidfVectorizer()),
('classifier', MultinomialNB())
])

# Кросс-валидация
cv_scores = cross_val_score(pipeline, phrases, labels, cv=5)
print(f"Средняя точность по кросс-валидации: {np.mean(cv_scores):.3f}")

# Обучение на тренировочной выборке
pipeline.fit(X_train, y_train)

# Оценка на тестовой выборке
y_pred = pipeline.predict(X_test)
print(classification_report(y_test, y_pred))

# Визуализация матрицы ошибок
cm = confusion_matrix(y_test, y_pred)
plt.figure(figsize=(10, 8))
sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', 
xticklabels=pipeline.classes_, 
yticklabels=pipeline.classes_)
plt.xlabel('Предсказание')
plt.ylabel('Истинное значение')
plt.title('Матрица ошибок')
plt.show()

import cProfile
import functools
from time import time

# Профилирование функций
def profile_function(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
profile = cProfile.Profile()
try:
profile.enable()
result = func(*args, **kwargs)
profile.disable()
return result
finally:
profile.print_stats(sort='cumulative')
return wrapper

# Измерение времени выполнения
def timing_decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
start_time = time()
result = func(*args, **kwargs)
end_time = time()
print(f"Функция {func.__name__} выполнялась {end_time – start_time:.4f} секунд")
return result
return wrapper

# Простой механизм кэширования для функций NLP
class SimpleCache:
def __init__(self, max_size=100):
self.cache = {}
self.max_size = max_size

def get(self, key):
return self.cache.get(key)

def set(self, key, value):
if len(self.cache) >= self.max_size:
self.cache.pop(next(iter(self.cache)))
self.cache[key] = value

# Создаем кэш
intent_cache = SimpleCache()
entity_cache = SimpleCache()

# Оптимизированная функция классификации намерений
def optimized_classify_intent(classifier, text):
cached_result = intent_cache.get(text)
if cached_result:
return cached_result

intent = classifier.predict([text])[0]

intent_cache.set(text, intent)

return intent

# Оптимизированная функция извлечения сущностей
def optimized_extract_entities(text):
cached_result = entity_cache.get(text)
if cached_result:
return cached_result

entities = extract_entities(text)

entity_cache.set(text, entities)

return entities

import logging
import json
from datetime import datetime

# Настройка логирования
logging.basicConfig(
filename='bot_logs.log',
level=logging.INFO,
format='%(asctime)s – %(name)s – %(levelname)s – %(message)s'
)

# Класс для сбора обратной связи и метрик
class BotAnalytics:
def __init__(self, log_file='bot_analytics.json'):
self.log_file = log_file
self.interactions = []
self.load_data()

def load_data(self):
try:
with open(self.log_file, 'r') as f:
self.interactions = json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
self.interactions = []

def save_data(self):
with open(self.log_file, 'w') as f:
json.dump(self.interactions, f, indent=2)

def log_interaction(self, user_input, bot_response, intent, confidence, entities, feedback=None):
interaction = {
'timestamp': datetime.now().isoformat(),
'user_input': user_input,
'bot_response': bot_response,
'intent': intent,
'confidence': confidence,
'entities': entities,
'feedback': feedback
}
self.interactions.append(interaction)
self.save_data()
logging.info(f"Interaction logged: {user_input} -> {intent} ({confidence:.2f})")

def collect_feedback(self, interaction_id, feedback):
if 0 <= interaction_id < len(self.interactions):
self.interactions[interaction_id]['feedback'] = feedback
self.save_data()
logging.info(f"Feedback collected for interaction {interaction_id}: {feedback}")

def get_stats(self):
if not self.interactions:
return {"message": "No data available"}

total_interactions = len(self.interactions)
intents = {}
feedback_counts = {"positive": 0, "negative": 0, "neutral": 0, "none": 0}

for interaction in self.interactions:
intent = interaction['intent']
intents[intent] = intents.get(intent, 0) + 1

feedback = interaction.get('feedback')
if feedback == "positive":
feedback_counts["positive"] += 1
elif feedback == "negative":
feedback_counts["negative"] += 1
elif feedback == "neutral":
feedback_counts["neutral"] += 1
else:
feedback_counts["none"] += 1

return {
"total_interactions": total_interactions,
"intent_stats": {intent: count / total_interactions for intent, count in intents.items()},
"feedback_stats": {k: v / total_interactions for k, v in feedback_counts.items()}
}

# Пример использования системы аналитики
analytics = BotAnalytics()

def process_user_request_with_analytics(user_input):
intent, confidence = classify_intent_with_confidence(intent_classifier, user_input)

entities = extract_entities(user_input)

dialog_context.update(user_input, intent, entities)

response = generate_response(intent, entities, dialog_context)

analytics.log_interaction(user_input, response, intent, confidence, entities)

return response, len(analytics.interactions) – 1 # Возвращаем ID взаимодействия для обратной связи

# Модифицированная версия классификатора, возвращающая уверенность
def classify_intent_with_confidence(classifier, text):
intent = classifier.predict([text])[0]

probas = classifier.predict_proba([text])[0]
classes = classifier.classes_

max_proba_index = np.argmax(probas)
confidence = probas[max_proba_index]

return intent, confidence