Интеграция ChatGPT API в Python-приложения: пошаговое руководство

# Создаем виртуальное окружение
python -m venv openai-env

# Активируем виртуальное окружение
# Windows:
openai-env\Scripts\activate
# macOS/Linux:
source openai-env/bin/activate

# Устанавливаем библиотеку OpenAI
pip install openai

import openai

# Устанавливаем API ключ
openai.api_key = "ваш_api_ключ"

# Альтернативный способ — через переменную окружения
# import os
# openai.api_key = os.getenv("OPENAI_API_KEY")

from openai import OpenAI

# Инициализация клиента
client = OpenAI(api_key="ваш_api_ключ")

# Функция для генерации текста
def generate_text(prompt, model="gpt-3.5-turbo"):
response = client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": "Вы полезный ассистент."},
{"role": "user", "content": prompt}
],
temperature=0.7,
max_tokens=500
)

return response.choices[0].message.content

# Пример использования
prompt = "Объясни концепцию рекурсии в программировании простыми словами."
result = generate_text(prompt)
print(result)

def chat_with_gpt(conversation_history, user_input, model="gpt-3.5-turbo"):
# Добавляем новое сообщение пользователя
conversation_history.append({"role": "user", "content": user_input})

response = client.chat.completions.create(
model=model,
messages=conversation_history,
temperature=0.7
)

# Получаем ответ ассистента
assistant_response = response.choices[0].message.content

# Добавляем ответ ассистента в историю
conversation_history.append({"role": "assistant", "content": assistant_response})

return assistant_response, conversation_history

# Пример использования
conversation = [
{"role": "system", "content": "Вы дружелюбный ассистент, который помогает с программированием на Python."}
]

user_question = "Как использовать декораторы в Python?"
response, conversation = chat_with_gpt(conversation, user_question)
print("Ответ:", response)

# Продолжаем диалог
follow_up = "А как передать аргументы в декоратор?"
response, conversation = chat_with_gpt(conversation, follow_up)
print("Ответ на уточнение:", response)

response = client.chat.completions.create(
model="gpt-4",
messages=conversation_history,
temperature=0.2, # Низкая температура для более предсказуемых ответов
max_tokens=1000, # Увеличенный лимит токенов
top_p=0.9, # Nucleus sampling
frequency_penalty=0.5, # Снижение вероятности повторений
presence_penalty=0.5, # Поощрение разнообразия тем
stop=["\n\n"] # Остановка генерации при двойном переносе строки
)

import tiktoken

def num_tokens_from_messages(messages, model="gpt-3.5-turbo"):
"""Подсчет количества токенов в сообщениях"""
encoding = tiktoken.encoding_for_model(model)
num_tokens = 0
for message in messages:
# Каждое сообщение имеет базовую стоимость в токенах
num_tokens += 4 # базовая стоимость сообщения
for key, value in message.items():
if key == "role" or key == "content":
num_tokens += len(encoding.encode(str(value)))
num_tokens += 2 # финальная стоимость сообщения
return num_tokens

def manage_conversation_context(conversation_history, max_tokens=4000, model="gpt-3.5-turbo"):
"""Управление контекстом разговора, чтобы не превышать лимит токенов"""
# Сохраняем системное сообщение
system_message = next((msg for msg in conversation_history if msg["role"] == "system"), None)

# Убираем системное сообщение из подсчета и сохранения
working_messages = [msg for msg in conversation_history if msg["role"] != "system"]

# Сокращаем контекст, если необходимо
while num_tokens_from_messages([system_message] + working_messages, model) > max_tokens:
# Удаляем самые старые сообщения пользователя и ассистента
# (сохраняя целостность пар вопрос-ответ)
if len(working_messages) >= 2:
working_messages = working_messages[2:]
else:
break

# Возвращаем обновленный контекст с системным сообщением в начале
return [system_message] + working_messages

def extract_key_information(message, client):
"""Извлечение ключевой информации из сообщения"""
extraction_prompt = f"""
Извлеки и обобщи ключевую информацию из следующего сообщения.
Включи только самые важные факты, предпочтения и требования.
Сообщение: {message}
"""

response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": extraction_prompt}],
temperature=0.3,
max_tokens=200
)

return response.choices[0].message.content

def smart_context_management(conversation_history, user_input, client, model="gpt-3.5-turbo"):
"""Умное управление контекстом с сохранением важной информации"""
# Извлекаем ключевую информацию из нового сообщения пользователя
key_info = extract_key_information(user_input, client)

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

# Проверяем размер контекста
if num_tokens_from_messages(conversation_history, model) > 4000:
# Сохраняем системное сообщение и последние 4 сообщения
system_msg = next((msg for msg in conversation_history if msg["role"] == "system"), None)
recent_messages = conversation_history[-4:]

# Создаем новый контекст с обобщенной информацией
summary = "\n".join([extract_key_information(msg["content"], client) for msg in conversation_history 
if msg["role"] in ["user", "assistant"] and msg not in recent_messages])

# Формируем новый контекст
conversation_history = [
system_msg,
{"role": "system", "content": f"Краткое содержание предыдущей беседы: {summary}"},
*recent_messages
]

return conversation_history

from flask import Flask, request, jsonify, render_template
import os
from openai import OpenAI
import json

app = Flask(__name__)

# Инициализация клиента OpenAI
client = OpenAI(api_key=os.getenv('OPENAI_API_KEY'))

# Словарь для хранения диалогов пользователей
user_conversations = {}

@app.route('/')
def index():
return render_template('index.html')

@app.route('/api/chat', methods=['POST'])
def chat():
data = request.get_json()
user_id = data.get('user_id', 'default')
message = data.get('message', '')

# Инициализация диалога для нового пользователя
if user_id not in user_conversations:
user_conversations[user_id] = [
{"role": "system", "content": "Вы полезный ассистент по программированию на Python."}
]

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

# Управление контекстом диалога (ограничение количества сообщений)
if len(user_conversations[user_id]) > 10:
# Сохраняем системное сообщение и последние 5 обменов
system_message = user_conversations[user_id][0]
user_conversations[user_id] = [system_message] + user_conversations[user_id][-10:]

try:
# Отправка запроса к API
response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=user_conversations[user_id],
temperature=0.7,
max_tokens=500
)

# Получение ответа
assistant_response = response.choices[0].message.content

# Сохранение ответа в диалоге
user_conversations[user_id].append({"role": "assistant", "content": assistant_response})

return jsonify({
"response": assistant_response,
"conversation_length": len(user_conversations[user_id])
})

except Exception as e:
return jsonify({"error": str(e)}), 500

if __name__ == '__main__':
app.run(debug=True)

<!-- templates/index.html -->
<!DOCTYPE html>
<html>
<head>
<title>Python Chatbot</title>
<style>
/* Стили для чата */
#chat-container {
width: 600px;
margin: 0 auto;
border: 1px solid #ddd;
padding: 20px;
height: 500px;
display: flex;
flex-direction: column;
}
#chat-messages {
flex-grow: 1;
overflow-y: auto;
margin-bottom: 20px;
}
.message {
margin: 10px 0;
padding: 10px;
border-radius: 5px;
}
.user-message {
background: #e6f7ff;
margin-left: 50px;
}
.bot-message {
background: #f6f6f6;
margin-right: 50px;
}
#user-input {
width: 80%;
padding: 8px;
}
</style>
</head>
<body>
<h1>Python Programming Assistant</h1>
<div id="chat-container">
<div id="chat-messages"></div>
<div>
<input type="text" id="user-input" placeholder="Задайте вопрос..." />
<button onclick="sendMessage()">Отправить</button>
</div>
</div>

<script>
// Генерируем уникальный ID для пользователя
const userId = 'user_' + Math.random().toString(36).substr(2, 9);

function sendMessage() {
const userInput = document.getElementById('user-input');
const message = userInput.value.trim();

if (!message) return;

// Добавляем сообщение пользователя в чат
addMessage(message, 'user');
userInput.value = '';

// Отправляем запрос на сервер
fetch('/api/chat', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify({
user_id: userId,
message: message
}),
})
.then(response => response.json())
.then(data => {
// Добавляем ответ бота в чат
addMessage(data.response, 'bot');
})
.catch(error => {
console.error('Error:', error);
addMessage('Произошла ошибка при обработке запроса.', 'bot');
});
}

function addMessage(text, sender) {
const messagesDiv = document.getElementById('chat-messages');
const messageDiv = document.createElement('div');
messageDiv.className = `message ${sender}-message`;
messageDiv.textContent = text;
messagesDiv.appendChild(messageDiv);

// Прокрутка вниз
messagesDiv.scrollTop = messagesDiv.scrollHeight;
}

// Обработка нажатия Enter
document.getElementById('user-input').addEventListener('keypress', function(e) {
if (e.key === 'Enter') {
sendMessage();
}
});

// Приветственное сообщение
addMessage('Привет! Я твой ассистент по Python. Чем могу помочь?', 'bot');
</script>
</body>
</html>

# Пример асинхронного обработчика запросов для высокой нагрузки
import asyncio
import aiohttp
from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
from typing import List, Dict
import redis.asyncio as redis
import logging
import time
import uuid

app = FastAPI()

# Модель данных для запроса
class ChatRequest(BaseModel):
user_id: str
message: str
model: str = "gpt-3.5-turbo"

# Модель данных для ответа
class ChatResponse(BaseModel):
response: str
conversation_id: str
processing_time: float

# Соединение с Redis для кеширования и управления состояниями
redis_client = redis.Redis(host='localhost', port=6379, db=0)

# Функция для ограничения частоты запросов (rate limiting)
async def rate_limit(user_id: str, limit: int = 10, period: int = 60):
"""Ограничение до {limit} запросов за {period} секунд"""
key = f"rate_limit:{user_id}"
current = await redis_client.get(key)

if current and int(current) >= limit:
raise HTTPException(status_code=429, detail="Too many requests")

pipe = redis_client.pipeline()
await pipe.incr(key).expire(key, period).execute()

@app.post("/api/chat", response_model=ChatResponse)
async def chat_endpoint(request: ChatRequest):
start_time = time.time()

# Применяем ограничение частоты запросов
await rate_limit(request.user_id)

# Получаем историю диалога из Redis
conversation_key = f"conversation:{request.user_id}"
conversation_json = await redis_client.get(conversation_key)

if conversation_json:
conversation = json.loads(conversation_json)
else:
# Создаем новый диалог
conversation = [
{"role": "system", "content": "Вы полезный ассистент по программированию на Python."}
]

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

# Формируем запрос к OpenAI API
try:
async with aiohttp.ClientSession() as session:
async with session.post(
"https://api.openai.com/v1/chat/completions",
headers={
"Authorization": f"Bearer {os.getenv('OPENAI_API_KEY')}",
"Content-Type": "application/json"
},
json={
"model": request.model,
"messages": conversation,
"temperature": 0.7,
"max_tokens": 500
}
) as response:
if response.status != 200:
error_text = await response.text()
raise HTTPException(status_code=response.status, detail=f"API Error: {error_text}")

data = await response.json()
assistant_response = data["choices"][0]["message"]["content"]
except Exception as e:
logging.error(f"API request error: {str(e)}")
raise HTTPException(status_code=500, detail=f"Failed to process request: {str(e)}")

# Добавляем ответ ассистента в диалог
conversation.append({"role": "assistant", "content": assistant_response})

# Сохраняем обновленный диалог в Redis
await redis_client.set(conversation_key, json.dumps(conversation), ex=3600) # Истекает через 1 час

processing_time = time.time() – start_time

return ChatResponse(
response=assistant_response,
conversation_id=request.user_id,
processing_time=processing_time
)

import tiktoken
from openai import OpenAI

client = OpenAI()

def count_tokens(text, model="gpt-3.5-turbo"):
"""Подсчет количества токенов в тексте"""
try:
encoding = tiktoken.encoding_for_model(model)
return len(encoding.encode(text))
except Exception:
# Для моделей, не поддерживаемых напрямую, используем cl100k_base
encoding = tiktoken.get_encoding("cl100k_base")
return len(encoding.encode(text))

def optimize_prompt(prompt, max_tokens=500):
"""Оптимизация промпта для уменьшения количества токенов"""
tokens = count_tokens(prompt)

if tokens <= max_tokens:
return prompt

# Если превышен лимит, пробуем сжать текст
compression_prompt = f"""
Сократи следующий текст, сохранив основной смысл и ключевые детали.
Максимальный размер: {max_tokens} токенов.

Текст для сжатия: {prompt}
"""

response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": compression_prompt}],
temperature=0.3,
max_tokens=max_tokens
)

return response.choices[0].message.content

def chunk_large_text(text, chunk_size=1000):
"""Разбивает большой текст на части подходящего размера"""
words = text.split()
chunks = []
current_chunk = []
current_size = 0

for word in words:
word_size = count_tokens(word)

if current_size + word_size <= chunk_size:
current_chunk.append(word)
current_size += word_size
else:
chunks.append(' '.join(current_chunk))
current_chunk = [word]
current_size = word_size

if current_chunk:
chunks.append(' '.join(current_chunk))

return chunks

import time
import random

def call_api_with_retry(prompt, max_retries=5, base_delay=1, max_delay=60):
"""Вызов API с экспоненциальной задержкой при повторных попытках"""
retries = 0

while retries <= max_retries:
try:
response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}],
temperature=0.7
)
return response
except Exception as e:
retries += 1

if retries > max_retries:
raise Exception(f"Максимальное количество попыток исчерпано: {str(e)}")

# Экспоненциальная задержка с случайным отклонением (jitter)
delay = min(max_delay, base_delay * (2 ** (retries – 1)))
delay = delay * (1 + random.uniform(-0.1, 0.1)) # Добавляем 10% jitter

print(f"Ошибка API: {str(e)}. Повторная попытка через {delay:.2f} секунд...")
time.sleep(delay)

# Пример использования
try:
response = call_api_with_retry("Расскажи о методах оптимизации кода на Python")
print(response.choices[0].message.content)
except Exception as e:
print(f"Не удалось получить ответ: {str(e)}")

class TokenUsageTracker:
def __init__(self, redis_client, budget_limit=100000):
self.redis = redis_client
self.budget_limit = budget_limit # Лимит токенов на день

async def track_usage(self, user_id, input_tokens, output_tokens):
"""Отслеживание использования токенов"""
today = time.strftime("%Y-%m-%d")

# Ключи для хранения статистики
daily_key = f"usage:{today}"
user_key = f"usage:{today}:{user_id}"

# Обновляем статистику
async with self.redis.pipeline() as pipe:
# Обновляем общую статистику
await pipe.hincrby(daily_key, "input_tokens", input_tokens)
await pipe.hincrby(daily_key, "output_tokens", output_tokens)
await pipe.expire(daily_key, 86400 * 7) # Храним 7 дней

# Обновляем статистику пользователя
await pipe.hincrby(user_key, "input_tokens", input_tokens)
await pipe.hincrby(user_key, "output_tokens", output_tokens)
await pipe.expire(user_key, 86400 * 7) # Храним 7 дней

results = await pipe.execute()

# Проверяем лимиты
total_daily = int(results[0]) + int(results[1])
if total_daily > self.budget_limit:
# Отправляем оповещение о превышении бюджета
await self.send_budget_alert(today, total_daily)

async def send_budget_alert(self, date, total_usage):
"""Отправка оповещения о превышении бюджета"""
# Здесь логика оповещения (email, Slack и т.д.)
print(f"ALERT: Budget exceeded on {date}. Total usage: {total_usage} tokens")

# Можно также сохранить информацию о превышении
await self.redis.set(f"budget_alert:{date}", "1", ex=86400)

async def get_user_statistics(self, user_id, days=30):
"""Получение статистики использования по пользователю"""
stats = []
today = time.time()

for i in range(days):
date = time.strftime("%Y-%m-%d", time.localtime(today – 86400 * i))
user_key = f"usage:{date}:{user_id}"

data = await self.redis.hgetall(user_key)
if data:
stats.append({
"date": date,
"input_tokens": int(data.get(b"input_tokens", 0)),
"output_tokens": int(data.get(b"output_tokens", 0))
})

return stats