המדריך המלא ל-ChatGPT API – איך לבנות בוט שיעבוד בשבילכם

class AIChatBot:
def __init__(self):
self.api_key = os.getenv("OPENAI_API_KEY")
openai.api_key = self.api_key
self.conversation_history = []
self.max_history = 10
self.last_cleared = datetime.now()

def clear_old_history(self):
"""מנקה היסטוריה ישנה אחת ליום"""
if datetime.now() - self.last_cleared > timedelta(days=1):
self.conversation_history = []
self.last_cleared = datetime.now()

def chat(self, message: str, system_prompt: str = "") -> str:
# ניקוי היסטוריה אם צריך
self.clear_old_history()

# הוספת ההודעה החדשה
self.conversation_history.append({
"role": "user",
"content": message
})

# שמירה על גודל מקסימלי
if len(self.conversation_history) > self.max_history * 2:
self.conversation_history = self.conversation_history[-self.max_history:]

# בניית השיחה המלאה
messages = []
if system_prompt:
messages.append({
"role": "system",
"content": system_prompt
})
messages.extend(self.conversation_history)

try:
# שליחת הבקשה ל-API
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=messages,
temperature=0.7,
max_tokens=1000,
presence_penalty=0.6,
frequency_penalty=0.0
)

# שמירת התשובה
assistant_message = response.choices[0].message['content']
self.conversation_history.append({
"role": "assistant",
"content": assistant_message
})

return assistant_message

def learn_from_interaction(
self,
question: str,
answer: str,
success: bool = True,
importance: float = 1.0
) -> None:
# נרמול השאלה
normalized_q = self._normalize_text(question)

if normalized_q in self.knowledge_base:
# עדכון ידע קיים
kb_entry = self.knowledge_base[normalized_q]
old_rate = kb_entry['success_rate']
new_rate = (
old_rate * (1 - self.learning_rate) +
float(success) * self.learning_rate
) * importance

kb_entry.update({
'success_rate': new_rate,
'uses': kb_entry['uses'] + 1,
'last_used': datetime.now(),
'answer': answer if success else kb_entry['answer']
})
else:
# הוספת ידע חדש
self.knowledge_base[normalized_q] = {
'answer': answer,
'success_rate': float(success),
'uses': 1,
'last_used': datetime.now(),
'importance': importance
}

def _normalize_text(self, text: str) -> str:
"""מנרמל טקסט להשוואה קלה יותר"""
return ' '.join(text.lower().split())

def get_response(self, question: str) -> str:
normalized_q = self._normalize_text(question)

# חיפוש במאגר הידע
if normalized_q in self.knowledge_base:
kb_entry = self.knowledge_base[normalized_q]
if kb_entry['success_rate'] > self.confidence_threshold:
return kb_entry['answer']

def analyze_campaign(self, campaign_data: Dict[str, Any]) -> str:
# מחלץ ומנתח נתונים
metrics = self.calculate_metrics(campaign_data)
trends = self.analyze_trends(metrics)
anomalies = self.detect_anomalies(metrics)

# בונה פרומפט מותאם אישית
prompt = f"""
As a senior marketing expert, analyze this campaign data:

Current Metrics:
{metrics}

Historical Trends:
{trends}

Detected Anomalies:
{anomalies}

Focus on:
1. Performance changes vs historical data
2. Unexpected patterns or anomalies
3. Specific optimization opportunities
4. ROI improvement suggestions

Previous successful insights:
{str(self.get_relevant_knowledge())}
"""

# מקבל תובנות מהמודל
insights = self.chat(prompt)

# לומד מהתוצאות
self.learn_from_interaction(
str(campaign_data),
insights,
self.validate_insights(insights),
importance=self.calculate_importance(metrics)
)

return insights

def calculate_metrics(self, data: Dict[str, Any]) -> Dict[str, float]:
"""מחשב מדדי ביצוע מרכזיים"""
return {
'ctr': data.get('clicks', 0) / max(data.get('impressions', 1), 1) * 100,
'conversion_rate': data.get('conversions', 0) / max(data.get('clicks', 1), 1) * 100,
'cost_per_conversion': data.get('cost', 0) / max(data.get('conversions', 1), 1),
'roi': (data.get('revenue', 0) - data.get('cost', 0)) / max(data.get('cost', 1), 1) * 100,
'average_position': data.get('avg_position', 0),
'quality_score': data.get('quality_score', 0)
}

def analyze_trends(self, current_metrics: Dict[str, float]) -> str:
"""מנתח מגמות לאורך זמן"""
self.performance_history.append({
'timestamp': datetime.now(),
'metrics': current_metrics
})

# משווה לנתונים היסטוריים
if len(self.performance_history) > 1:
prev_metrics = self.performance_history[-2]['metrics']
changes = {
metric: ((current - prev_metrics[metric]) / prev_metrics[metric] * 100)
for metric, current in current_metrics.items()
}
return f"Changes from last period: {changes}"

def optimize_prompt(self, prompt: str) -> str:
"""אופטימיזציה של הפרומפט לחיסכון בטוקנים"""
# הסרת מילות קישור מיותרות
filler_words = ['please', 'kindly', 'could you', 'maybe']
for word in filler_words:
prompt = prompt.replace(f' {word} ', ' ')

# קיצור משפטים ארוכים
sentences = prompt.split('.')
optimized_sentences = [
sent if len(sent.split()) < 20
else ' '.join(sent.split()[:20])
for sent in sentences
]

return '.'.join(optimized_sentences)

def track_costs(self, response: Any) -> None:
"""מעקב אחרי עלויות"""
tokens_used = response.usage.total_tokens
cost = tokens_used * self.cost_per_token

day = datetime.now().strftime('%Y-%m-%d')
if day not in self.token_usage:
self.token_usage[day] = {
'tokens': 0,
'cost': 0.0
}

self.token_usage[day]['tokens'] += tokens_used
self.token_usage[day]['cost'] += cost