AI интеграция в Telegram ботах: ChatGPT, Claude и лучшие практики 2024

# AI интеграция в Telegram ботах: ChatGPT, Claude и лучшие практики 2024

Введение в AI-powered боты

Интеграция искусственного интеллекта превращает обычных Telegram ботов в умных помощников, способных понимать контекст, генерировать контент и решать сложные задачи.

Популярные AI API для интеграции

1. OpenAI API (ChatGPT)

**Настройка:**
```python
import openai
from openai import AsyncOpenAI

client = AsyncOpenAI(api_key="your-api-key")

async def get_ai_response(user_message, context=[]):
messages = context + [{"role": "user", "content": user_message}]

response = await client.chat.completions.create(
model="gpt-3.5-turbo",
messages=messages,
max_tokens=1000,
temperature=0.7
)

return response.choices[0].message.content
```

2. Anthropic Claude

**Настройка:**
```python
import anthropic

client = anthropic.AsyncAnthropic(api_key="your-api-key")

async def get_claude_response(user_message, context=""):
response = await client.messages.create(
model="claude-3-sonnet-20240229",
max_tokens=1000,
messages=[
{"role": "user", "content": f"{context}\n\n{user_message}"}
]
)

return response.content[0].text
```

Создание AI бота

Базовая архитектура

import asyncio
from aiogram import Bot, Dispatcher, F
from aiogram.types import Message
from aiogram.filters import Command

class AIBot:
def __init__(self, bot_token, openai_key):
self.bot = Bot(token=bot_token)
self.dp = Dispatcher()
self.openai_client = AsyncOpenAI(api_key=openai_key)
self.user_contexts = {} # Храним контекст для каждого пользователя

self.setup_handlers()

def setup_handlers(self):
@self.dp.message(Command('start'))
async def start_handler(message: Message):
self.user_contexts[message.from_user.id] = []
await message.answer(
"🤖 Привет! Я AI-помощник на базе ChatGPT.\n"
"Задавайте любые вопросы!"
)

@self.dp.message(F.text)
async def ai_handler(message: Message):
user_id = message.from_user.id

# Показываем индикатор набора
await self.bot.send_chat_action(message.chat.id, "typing")

try:
# Получаем контекст пользователя
context = self.user_contexts.get(user_id, [])

# Добавляем сообщение пользователя
context.append({"role": "user", "content": message.text})

# Ограничиваем контекст (последние 10 сообщений)
context = context[-10:]

# Получаем ответ от AI
ai_response = await self.get_ai_response(context)

# Добавляем ответ AI в контекст
context.append({"role": "assistant", "content": ai_response})
self.user_contexts[user_id] = context

# Отправляем ответ пользователю
await message.answer(ai_response)

except Exception as e:
await message.answer("❌ Произошла ошибка при обработке запроса")
print(f"Error: {e}")

async def get_ai_response(self, context):
system_prompt = {
"role": "system",
"content": "Ты полезный AI-помощник. Отвечай кратко и по делу."
}

messages = [system_prompt] + context

response = await self.openai_client.chat.completions.create(
model="gpt-3.5-turbo",
messages=messages,
max_tokens=1000,
temperature=0.7
)

return response.choices[0].message.content

async def start_polling(self):
await self.dp.start_polling(self.bot)

# Использование
if __name__ == "__main__":
bot = AIBot("BOT_TOKEN", "OPENAI_KEY")
asyncio.run(bot.start_polling())
```

Продвинутые функции

1. Function Calling

Позволяет боту выполнять действия:

functions = [
    {
        "name": "get_weather",
        "description": "Получить погоду в городе",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "Название города"}
            },
            "required": ["city"]
        }
    }

async def get_ai_response_with_functions(self, context):
response = await self.openai_client.chat.completions.create(
model="gpt-3.5-turbo",
messages=context,
functions=functions,
function_call="auto"
)

if response.choices[0].message.function_call:
function_name = response.choices[0].message.function_call.name

if function_name == "get_weather":
# Выполняем функцию получения погоды
city = json.loads(response.choices[0].message.function_call.arguments)["city"]
weather_data = await get_weather_data(city)
return f"Погода в {city}: {weather_data}"

return response.choices[0].message.content
```

2. Управление токенами

def count_tokens(text, model="gpt-3.5-turbo"):
encoding = tiktoken.encoding_for_model(model)
return len(encoding.encode(text))

def trim_context(context, max_tokens=3000):
total_tokens = sum(count_tokens(msg["content"]) for msg in context)

while total_tokens > max_tokens and len(context) > 1:
context.pop(0) # Удаляем самое старое сообщение
total_tokens = sum(count_tokens(msg["content"]) for msg in context)

return context
```

Лучшие практики

1. Безопасность

def sanitize_input(text):
# Удаляем потенциально опасный контент
text = re.sub(r'<[^>]+>', '', text) # HTML теги
text = text.replace('\n\n\n+', '\n\n') # Лишние переносы
return text[:1000] # Ограничиваем длину

def is_safe_request(text):
dangerous_patterns = [
r'ignore.+previous.+instructions',
r'system.+prompt',
r'jailbreak',
]

return not any(re.search(pattern, text.lower()) for pattern in dangerous_patterns)
```

2. Обработка ошибок

async def safe_ai_request(self, context, retries=3):
    for attempt in range(retries):
        try:
            response = await self.openai_client.chat.completions.create(
                model="gpt-3.5-turbo",
                messages=context,
                max_tokens=1000,
                timeout=30  # Таймаут запроса
            )
            return response.choices[0].message.content
            
        except openai.RateLimitError:
            await asyncio.sleep(2 ** attempt)  # Экспоненциальная задержка
            
        except openai.APIConnectionError:
            if attempt == retries - 1:
                return "❌ Не удается подключиться к AI сервису"
            
        except Exception as e:
            return f"❌ Ошибка: {str(e)}"

3. Кеширование

from functools import lru_cache

@lru_cache(maxsize=100)
async def cached_ai_response(prompt_hash):
# Кешируем частые запросы
pass

def get_prompt_hash(prompt):
return hashlib.md5(prompt.encode()).hexdigest()
```

Мониторинг и аналитика

import logging

class AIBotAnalytics:
def __init__(self):
self.request_count = 0
self.token_usage = 0
self.error_count = 0

def log_request(self, user_id, tokens_used, response_time):
self.request_count += 1
self.token_usage += tokens_used

logging.info(f"AI Request - User: {user_id}, Tokens: {tokens_used}, Time: {response_time}ms")

def log_error(self, error_type, user_id):
self.error_count += 1
logging.error(f"AI Error - Type: {error_type}, User: {user_id}")
```

Заключение

AI интеграция открывает безграничные возможности для Telegram ботов. Главное — правильно управлять контекстом, токенами и обеспечивать безопасность.

**Нужна помощь с AI интеграцией?** Обращайтесь к экспертам SOI.MOI: [@mixvlad](https://t.me/mixvlad)