File size: 7,578 Bytes
58c4fec |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 |
"""
CRANE AI - Reasoning Modülü
"""
from typing import Dict, Any
from core.base_module import BaseMicroModule
import logging
logger = logging.getLogger(__name__)
class ReasonModule(BaseMicroModule):
"""Mantık yürütme için özelleşmiş modül"""
def __init__(self, config: Dict[str, Any]):
super().__init__(
model_id="microsoft/Phi-3-mini-4k-instruct",
config=config
)
# Mantık yürütme anahtar kelimeleri
self.reasoning_keywords = {
"çünkü", "neden", "sebep", "niye", "nasıl", "analiz",
"karşılaştır", "değerlendir", "açıkla", "kanıtla",
"sonuç", "çıkarım", "mantık", "düşünce", "fikir",
"strateji", "plan", "çözüm", "problem", "sorun",
"avantaj", "dezavantaj", "fark", "benzerlik"
}
# Analiz türleri
self.analysis_types = {
"problem_solving", "comparison", "evaluation",
"cause_effect", "strategy", "pros_cons"
}
def can_handle(self, query: str, context: Dict[str, Any]) -> float:
"""Mantık yürütme sorguları için uygunluk skoru"""
query_lower = query.lower()
# Mantık yürütme anahtar kelimelerini kontrol et
reasoning_score = 0
for keyword in self.reasoning_keywords:
if keyword in query_lower:
reasoning_score += 0.12
# Soru türlerini kontrol et
if "neden" in query_lower or "çünkü" in query_lower:
reasoning_score += 0.25
# Karşılaştırma ifadeleri
if any(word in query_lower for word in ["karşılaştır", "fark", "benzer", "hangisi"]):
reasoning_score += 0.2
# Analiz gerektiren ifadeler
if any(word in query_lower for word in ["analiz", "değerlendir", "incele", "araştır"]):
reasoning_score += 0.2
# Problem çözme ifadeleri
if any(word in query_lower for word in ["problem", "sorun", "çözüm", "nasıl"]):
reasoning_score += 0.15
# Uzun ve karmaşık sorular
if len(query.split()) > 15:
reasoning_score += 0.1
# Maksimum 1.0 skor
return min(reasoning_score, 1.0)
async def process(self, query: str, context: Dict[str, Any]) -> Dict[str, Any]:
"""Mantık yürütme işlemi"""
try:
# Analiz türünü belirle
analysis_type = self._detect_analysis_type(query)
# Mantık yürütme için özel prompt
prompt = self._build_reasoning_prompt(query, context, analysis_type)
# Yanıt üretimi
response = await self.generate_response(
prompt,
max_tokens=self.config.get("max_tokens", 1024),
temperature=0.3 # Mantık yürütme için düşük temperature
)
return {
"response": response,
"module": "reason_module",
"confidence": self.can_handle(query, context),
"analysis_type": analysis_type,
"reasoning_steps": self._extract_reasoning_steps(response)
}
except Exception as e:
logger.error(f"Reasoning processing error: {str(e)}")
return {
"error": str(e),
"module": "reason_module"
}
def _build_reasoning_prompt(self, query: str, context: Dict[str, Any], analysis_type: str) -> str:
"""Mantık yürütme için prompt hazırlar"""
# Analiz türüne göre özel prompt
if analysis_type == "problem_solving":
system_prompt = """Sen bir problem çözme uzmanısın. Sorunları adım adım analiz edip mantıklı çözümler üretiyorsun."""
elif analysis_type == "comparison":
system_prompt = """Sen bir karşılaştırma uzmanısın. Farklı seçenekleri objektif bir şekilde karşılaştırıyorsun."""
elif analysis_type == "evaluation":
system_prompt = """Sen bir değerlendirme uzmanısın. Durumları kapsamlı bir şekilde analiz ediyorsun."""
elif analysis_type == "cause_effect":
system_prompt = """Sen bir sebep-sonuç analizi uzmanısın. Olaylar arasındaki ilişkileri açıklıyorsun."""
else:
system_prompt = """Sen bir mantık yürütme uzmanısın. Karmaşık konuları açık ve anlaşılır şekilde analiz ediyorsun."""
# Temel prompt
prompt = f"""{system_prompt}
Kullanıcı sorusu: {query}
Lütfen:
1. Konuyu kapsamlı analiz et
2. Mantık zincirini açıkla
3. Somut örnekler ver
4. Sonuçları net bir şekilde özetle
Analiz türü: {analysis_type}
"""
# Bağlam varsa ekle
if context.get("history"):
prompt += f"\nÖnceki konuşma: {context['history'][-1]}\n"
return prompt
def _detect_analysis_type(self, query: str) -> str:
"""Sorgudan analiz türünü tespit eder"""
query_lower = query.lower()
# Problem çözme
if any(word in query_lower for word in ["problem", "sorun", "çöz", "nasıl"]):
return "problem_solving"
# Karşılaştırma
if any(word in query_lower for word in ["karşılaştır", "fark", "hangisi", "seç"]):
return "comparison"
# Değerlendirme
if any(word in query_lower for word in ["değerlendir", "incele", "analiz"]):
return "evaluation"
# Sebep-sonuç
if any(word in query_lower for word in ["neden", "çünkü", "sebep", "sonuç"]):
return "cause_effect"
# Strateji
if any(word in query_lower for word in ["strateji", "plan", "yaklaşım"]):
return "strategy"
# Avantaj-dezavantaj
if any(word in query_lower for word in ["avantaj", "dezavantaj", "artı", "eksi"]):
return "pros_cons"
# Varsayılan
return "general_reasoning"
def _extract_reasoning_steps(self, response: str) -> list:
"""Yanıttan mantık yürütme adımlarını çıkarır"""
steps = []
# Numaralı adımları bul
import re
numbered_steps = re.findall(r'(\d+[\.\)])\s*([^\n]+)', response)
for num, step in numbered_steps:
steps.append({
"step": num,
"description": step.strip()
})
# Anahtar kelimelerle adımları bul
key_phrases = [
"ilk olarak", "öncelikle", "birincisi",
"ikincisi", "üçüncüsü", "son olarak",
"sonuç olarak", "özetle", "bu nedenle"
]
sentences = response.split('.')
for sentence in sentences:
sentence = sentence.strip()
if any(phrase in sentence.lower() for phrase in key_phrases):
if len(sentence) > 10: # Çok kısa olanları atla
steps.append({
"step": "auto",
"description": sentence
})
return steps[:10] # Maksimum 10 adım |