Update g1.py

g1.py

@@ -4,58 +4,69 @@ import logging
 from typing import List, Tuple, Generator, Optional, Any
 import groq
 
-# Константы
-DEFAULT_MAX_TOKENS = 2000
-FINAL_MAX_TOKENS = 1000
-RETRY_DELAY = 1
+# Конфигурация
+DEFAULT_ITER_MODEL = "llama3-70b-8192"
+DEFAULT_FINAL_MODEL = "mixtral-8x7b-32768"
+MAX_TOKENS_LIMIT = 8192
+RETRY_DELAY = 1.5
 MAX_RETRIES = 3
-DEFAULT_MODEL = "llama3-8b-8192"
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
 class APIError(Exception):
-    """Кастомное исключение для ошибок API"""
+    """Базовое исключение для API ошибок"""
     pass
 
+class APIConnectionError(APIError):
+    """Ошибка подключения к API"""
+    pass
+
+def adjust_max_tokens(model: str, max_tokens: int) -> int:
+    """Автоматическая корректировка max_tokens в зависимости от модели"""
+    model_limits = {
+        "llama3-8b-8192": 8192,
+        "llama3-70b-8192": 8192,
+        "mixtral-8x7b-32768": 32768
+    }
+    return min(max_tokens, model_limits.get(model, 4096))
+
 def parse_json_response(raw_response: str) -> dict:
-    """Улучшенный парсер с валидацией структуры и обработкой Unicode"""
-    logger.debug("Parsing JSON response: %s", raw_response[:200])
-    
+    """Улучшенный парсер с полной валидацией"""
     try:
-        parsed = json.loads(raw_response)
-    except json.JSONDecodeError as e:
-        logger.error("JSON decode error: %s", e)
-        raise ValueError(f"Invalid JSON: {e}")
-
-    required = {"title", "content", "next_action"}
-    if missing := required - parsed.keys():
-        raise ValueError(f"Missing keys: {missing}")
-    
-    if parsed["next_action"] not in {"continue", "final_answer"}:
-        raise ValueError("Invalid next_action value")
-    
-    return parsed
+        data = json.loads(raw_response)
+        assert isinstance(data, dict), "Ответ должен быть объектом"
+        assert all(k in data for k in ["title", "content", "next_action"]), "Отсутствуют обязательные ключи"
+        assert data["next_action"] in {"continue", "final_answer"}, "Некорректное действие"
+        return data
+    except (json.JSONDecodeError, AssertionError) as e:
+        logger.error(f"Ошибка парсинга: {str(e)}")
+        return {
+            "title": "Ошибка парсинга",
+            "content": str(e),
+            "next_action": "final_answer"
+        }
 
 def make_api_call(
     messages: List[dict],
     max_tokens: int,
     is_final_answer: bool = False,
     custom_client: Optional[Any] = None,
-    model: str = DEFAULT_MODEL
+    iter_model: str = DEFAULT_ITER_MODEL,
+    final_model: str = DEFAULT_FINAL_MODEL
 ) -> Any:
-    """Унифицированный метод вызова API с улучшенной обработкой ошибок"""
+    """Улучшенный метод вызова API"""
     client = custom_client or groq.Client()
+    model = final_model if is_final_answer else iter_model
+    max_tokens = adjust_max_tokens(model, max_tokens)
     
     params = {
         "model": model,
         "messages": messages,
         "max_tokens": max_tokens,
         "temperature": 0.5,
+        "response_format": {"type": "json_object"} if not is_final_answer else None
     }
-    
-    if not is_final_answer:
-        params["response_format"] = {"type": "json_object"}
 
     for attempt in range(1, MAX_RETRIES + 1):
         try:
@@ -63,128 +74,110 @@ def make_api_call(
             content = response.choices[0].message.content
             
             if response.choices[0].finish_reason == "length":
-                logger.warning("Response truncated! Consider increasing max_tokens")
-            
+                logger.warning(f"Ответ обрезан! Рекомендуемый max_tokens: {max_tokens * 2}")
+                
             return content if is_final_answer else parse_json_response(content)
-
+            
         except groq.APIConnectionError as e:
-            logger.error("Connection error: %s. Attempt %d/%d", e, attempt, MAX_RETRIES)
+            logger.error(f"Сетевая ошибка (попытка {attempt}/{MAX_RETRIES}): {str(e)}")
             if attempt == MAX_RETRIES:
-                raise APIError("Failed to connect to API after multiple attempts")
-            time.sleep(RETRY_DELAY * attempt)
-        
+                raise APIConnectionError(str(e))
+            time.sleep(RETRY_DELAY * (2 ** attempt))
+            
         except groq.APIError as e:
-            logger.error("API error [%s]: %s", e.status_code, e.message)
+            logger.error(f"Ошибка API (код {e.status_code}): {e.message}")
             if e.status_code >= 500:
                 time.sleep(RETRY_DELAY * attempt)
             else:
-                raise APIError(f"API Error: {e.message}")
-        
-        except ValueError as e:
-            logger.error("Parsing error: %s. Attempt %d/%d", e, attempt, MAX_RETRIES)
-            if attempt == MAX_RETRIES:
-                raise APIError("Failed to parse response after multiple attempts")
-            time.sleep(RETRY_DELAY * attempt)
+                raise APIError(e.message)
 
 def generate_response(
     prompt: str,
     context: Optional[str] = None,
     custom_client: Optional[Any] = None,
-    max_steps: int = 25,
-    model: str = DEFAULT_MODEL
+    max_steps: int = 10,
+    iter_model: str = DEFAULT_ITER_MODEL,
+    final_model: str = DEFAULT_FINAL_MODEL
 ) -> Generator[Tuple[List[Tuple[str, str, float]], Optional[float]], None, None]:
-    """Улучшенная генерация ответа с валидацией ввода"""
-    # Валидация входных параметров
-    if not prompt.strip():
-        raise ValueError("Prompt cannot be empty")
-    if max_steps < 1:
-        raise ValueError("max_steps must be at least 1")
-
-    system_template = (
-        "Вы – эксперт по Python. Анализируйте вопрос по шагам:\n"
-        "1. Разбейте вопрос на под-вопросы\n"
-        "2. Для каждого сгенерируйте 2-3 варианта рассуждений\n"
-        "3. Оцените варианты по критериям ясности и релевантности\n"
-        "4. Выберите оптимальную цепочку\n"
-        "Формат ответа: JSON с ключами title, content, next_action"
+    """Улучшенная генерация ответа"""
+    system_prompt = (
+        "Вы — AI-ассистент для анализа технических вопросов. "
+        "Формат ответа: JSON с ключами title, content, next_action. "
+        "Используйте русский язык и технические термины."
     )
     
-    user_content = f"Context: {context}\n\nQuestion: {prompt}" if context else prompt
     messages = [
-        {"role": "system", "content": system_template},
-        {"role": "user", "content": user_content}
+        {"role": "system", "content": system_prompt},
+        {"role": "user", "content": f"Контекст: {context}\n\nВопрос: {prompt}" if context else prompt}
     ]
-
+    
     steps = []
     total_time = 0.0
-    step_count = 1
-
-    while step_count <= max_steps:
-        start = time.monotonic()
-        
-        try:
-            response = make_api_call(
-                messages=messages,
-                max_tokens=DEFAULT_MAX_TOKENS,
-                is_final_answer=False,
-                custom_client=custom_client,
-                model=model
-            )
-        except APIError as e:
-            logger.error("Critical error in step %d: %s", step_count, e)
-            steps.append(("Error", str(e), time.monotonic() - start))
-            break
-
-        elapsed = time.monotonic() - start
-        total_time += elapsed
-
-        step_entry = (
-            f"Step {step_count}: {response['title']}",
-            response["content"],
-            elapsed
-        )
-        steps.append(step_entry)
-        messages.append({"role": "assistant", "content": json.dumps(response, ensure_ascii=False)})
-
-        if response["next_action"] == "final_answer" or step_count == max_steps:
-            break
-
-        step_count += 1
-        yield steps, None
-
-    # Финализация ответа
-    messages.append({
-        "role": "user",
-        "content": "Сформируй окончательный ответ с детальным объяснением и примерами кода, если необходимо."
-    })
-
+    
     try:
+        for step in range(1, max_steps + 1):
+            start = time.monotonic()
+            
+            try:
+                response = make_api_call(
+                    messages=messages,
+                    max_tokens=MAX_TOKENS_LIMIT,
+                    is_final_answer=False,
+                    custom_client=custom_client,
+                    iter_model=iter_model
+                )
+            except APIError as e:
+                logger.error("Критическая ошибка: %s", str(e))
+                yield [("Ошибка", str(e), time.monotonic() - start)], None
+                return
+                
+            elapsed = time.monotonic() - start
+            total_time += elapsed
+            
+            steps.append((f"Шаг {step}: {response['title']}", response["content"], elapsed))
+            messages.append({"role": "assistant", "content": json.dumps(response, ensure_ascii=False)})
+            
+            if response["next_action"] == "final_answer":
+                break
+                
+            yield steps, None
+            
+        # Финализация
+        messages.append({"role": "user", "content": "Сформируйте финальный ответ с примерами кода и пояснениями."})
+        
+        final_start = time.monotonic()
         final_answer = make_api_call(
             messages=messages,
-            max_tokens=FINAL_MAX_TOKENS,
+            max_tokens=MAX_TOKENS_LIMIT,
             is_final_answer=True,
             custom_client=custom_client,
-            model=model
+            final_model=final_model
         )
-    except APIError as e:
-        final_answer = f"Error generating final answer: {str(e)}"
-    
-    final_time = time.monotonic() - start
-    total_time += final_time
-    steps.append(("Final Answer", final_answer, final_time))
-    
+        total_time += time.monotonic() - final_start
+        
+        steps.append(("Финальный ответ", final_answer, time.monotonic() - final_start))
+        
+    except Exception as e:
+        logger.exception("Непредвиденная ошибка:")
+        steps.append(("Критическая ошибка", str(e), 0.0))
+        
     yield steps, total_time
 
 if __name__ == "__main__":
-    context = "https://console.groq.com/docs/autogen"
-    prompt = "учитывай контекст autogen в ответ"
-    
     try:
-        for steps, total_time in generate_response(prompt, context):
+        for steps, total_time in generate_response(
+            prompt="Опишите процесс обработки запросов в Groq API",
+            context="Используйте официальную документацию Groq",
+            iter_model="llama3-70b-8192",
+            final_model="mixtral-8x7b-32768"
+        ):
             if total_time is not None:
-                print(f"\nОбщее время обработки: {total_time:.2f} секунд")
-                print("="*50)
+                print(f"\n✅ Готово за {total_time:.2f}s")
+                print("=" * 60)
             for title, content, t in steps:
-                print(f"\n{title} ({t:.2f}s)\n{content}")
-    except Exception as e:
-        print(f"Critical error: {str(e)}")
+                print(f"\n🔹 {title} [{t:.2f}s]\n{content}")
+                
+    except APIConnectionError as e:
+        print(f"🚨 Ошибка подключения: {str(e)}")
+    except APIError as e:
+        print(f"🚨 Ошибка API: {str(e)}")