Update g1.py

g1.py

@@ -1,61 +1,236 @@
-import groq
 import time
 import json
 import logging
-from typing import Generator
-from tqdm import tqdm
-
-logging.basicConfig(
-    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
-    level=logging.INFO
-)
-logger = logging.getLogger(__name__)
-
-class ResponseGenerator:
-    def __init__(self):
-        self.model = "llama3-70b-8192"
-        self.max_tokens = {
-            'intermediate': 2000,
-            'final': 1000
-        }
-
-    def _make_api_call(self, messages, max_tokens, is_final=False):
+from typing import List, Tuple, Generator, Optional, Any
+import groq  # Предполагается, что модуль groq доступен
+
+# Константы для параметризации
+DEFAULT_MAX_TOKENS = 8192
+FINAL_MAX_TOKENS = 4096
+RETRY_DELAY = 2  # базовая задержка в секундах
+
+# Выбор моделей на основе анализа:
+MODEL_ITERATIONS = "llama-3.3-70b-versatile"  # Для итеративных рассуждений
+MODEL_FINAL = "llama-3.3-70b-versatile"         # Для финального ответа
+MODEL_CODE = "llama-3.1-8b-instant"             # Для генерации кода
+
+logging.basicConfig(level=logging.DEBUG)
+
+# Замените 'your_api_key' на ваш реальный API ключ
+client = groq.Groq(api_key='your_api_key')
+
+def parse_json_response(raw_response: str) -> dict:
+    """
+    Разбирает строку JSON и проверяет наличие обязательных ключей: title, content, next_action.
+    
+    Аргументы:
+        raw_response (str): Строка, содержащая JSON.
+    
+    Возвращает:
+        dict: Распарсенный JSON-объект с ключами title, content и next_action.
+        
+    Выбрасывает:
+        ValueError: Если JSON некорректный или отсутствуют обязательные ключи.
+    """
+    logging.debug("RAW JSON RESPONSE: %s", raw_response)
+    try:
+        parsed_response = json.loads(raw_response)
+    except json.JSONDecodeError as e:
+        raise ValueError(f"Invalid JSON: {e}, content: {raw_response}")
+    
+    required_keys = {"title", "content", "next_action"}
+    if not required_keys.issubset(parsed_response):
+        raise ValueError("JSON missing required keys")
+    return parsed_response
+
+def make_api_call(
+    messages: List[dict], 
+    max_tokens: int, 
+    is_final_answer: bool = False, 
+    custom_client: Optional[Any] = None
+) -> Any:
+    """
+    Отправляет API-запрос с заданными сообщениями и возвращает результат.
+    
+    При is_final_answer=True возвращает строку финального ответа.
+    Иначе ожидается, что ответ будет в формате JSON с ключами "title", "content", "next_action".
+    
+    Аргументы:
+        messages (List[dict]): История сообщений для запроса.
+        max_tokens (int): Максимальное число токенов в ответе.
+        is_final_answer (bool): Флаг финального ответа.
+        custom_client (Any, optional): Альтернативный клиент для API-вызова.
+        
+    Возвращает:
+        Результат API-вызова (str для финального ответа или dict для промежуточного шага).
+    """
+    current_client = custom_client if custom_client is not None else client
+    # Выбираем модель в зависимости от типа ответа
+    model = MODEL_FINAL if is_final_answer else MODEL_ITERATIONS
+
+    for attempt in range(3):
         try:
-            # Исправлено: добавлена закрывающая скобка
-            response = groq.Groq().chat.completions.create(
-                model=self.model,
-                messages=messages,
-                max_tokens=max_tokens,
-                temperature=0.5,
-                **({"response_format": {"type": "json_object"}} if not is_final else {})
-            )
-            return response.choices[0].message.content
+            if is_final_answer:
+                response = current_client.chat.completions.create(
+                    model=model,
+                    messages=messages,
+                    max_tokens=max_tokens,
+                    temperature=0.7
+                )
+                logging.info("Успешно получен финальный ответ.")
+                return response.choices[0].message.content
+            else:
+                response = current_client.chat.completions.create(
+                    model=model,
+                    messages=messages,
+                    max_tokens=max_tokens,
+                    temperature=0.7,
+                    response_format={"type": "json_object"}
+                )
+                raw_response = response.choices[0].message.content
+                logging.info("Успешно получен промежуточный ответ.")
+                return parse_json_response(raw_response)
+        # TODO: При появлении информации о конкретных исключениях заменить Exception на более специфичные
         except Exception as e:
-            logger.error(f"API Error: {str(e)}")
-            return str(e)
-
-    def generate_response(self, prompt: str) -> Generator:
-        messages = [
-            {
-                "role": "system", 
-                "content": "You're an AI assistant. Explain your reasoning step-by-step in JSON format."
-            },
-            {"role": "user", "content": prompt}
-        ]
+            logging.exception("Попытка %d завершилась неудачно", attempt + 1)
+            if attempt == 2:
+                error_message = (
+                    f"Не удалось сгенерировать "
+                    f"{'финальный ответ' if is_final_answer else 'шаг'} за 3 попытки. Ошибка: {str(e)}"
+                )
+                if is_final_answer:
+                    return f"Error: {error_message}"
+                else:
+                    return {
+                        "title": "Error",
+                        "content": error_message,
+                        "next_action": "final_answer"
+                    }
+            time.sleep(RETRY_DELAY * (2 ** attempt))
+
+def generate_response(
+    prompt: str, 
+    context: Optional[str] = None, 
+    custom_client: Optional[Any] = None, 
+    max_steps: int = 25
+) -> Generator[Tuple[List[Tuple[str, str, float]], Optional[float]], None, None]:
+    """
+    Генерирует цепочку рассуждений и финальный ответ по заданному запросу.
+    
+    Новый подход:
+    1. **Анализ вопроса:** Определите ключевые компоненты вопроса и разбейте его на под-вопросы.
+    2. **Разбиение на под-вопросы:** Для каждого под-вопроса сформируйте 2-3 промежуточные мысли, такие как:
+       - Переосмысление вопроса для лучшего понимания
+       - Контекст, необходимый для ответа
+       - Предположения и связь концепций
+    3. **Оценка мыслей:** Оцените ясность, релевантность, логическую последовательность и охват концепций каждой мысли.
+    4. **Построение цепочки рассуждений:** Соедините самые сильные мысли в логичную цепочку.
+    5. **Обратный ход и альтернативные пути:** Если цепочка не отвечает полностью на вопрос, рассмотрите альтернативные варианты рассуждений.
+    
+    Аргументы:
+        prompt (str): Запрос для генерации ответа.
+        context (str, optional): Контекст для запроса.
+        custom_client (Any, optional): Альтернативный клиент для API-вызовов.
+        max_steps (int, optional): Максимальное количество шагов рассуждений.
+    
+    Возвращает:
+        Генератор, yield'ящий кортеж (steps, total_thinking_time), где:
+          - steps: список кортежей (название шага, содержание, время обработки)
+          - total_thinking_time: общее время обработки (либо None до финального шага)
+    """
+    system_message = (
+        "Вы – интеллектуальный помощник и эксперт по разработке на Python. "
+        "Ваша задача — ответить на вопрос, исследуя несколько путей рассуждений. "
+        "Сначала проанализируйте вопрос, выдельте ключевые информационные компоненты и разбейте его на логические под-вопросы. "
+        "Для каждого под-вопроса сформируйте 2-3 промежуточные мысли, включающие: переосмысление, контекст, предположения и связь концепций. "
+        "Оцените ясность, релевантность, логическую последовательность и охват концепций каждой мысли. "
+        "Постройте итоговую цепочку рассуждений, объединяя лучшие мысли, и при необходимости рассмотрите альтернативные пути. "
+        "Ответ должен быть в формате JSON с обязательными ключами: "
+        '"title" — краткое название шага, "content" — описание действий, "next_action" — "continue" или "final_answer". '
+        "Пример:\n"
+        '{"title": "Анализ задачи", "content": "Выделение ключевых элементов вопроса, разбивка на под-вопросы и формирование промежуточных мыслей...", "next_action": "continue"}\n'
+        "Дополнительные требования: используйте русский язык и избегайте Unicode-кодировок (например, пишите \"Привет\", а не \\\"\\u041f\\u0440\\u0438...\\\").'
+    )
+    
+    # Формирование начальных сообщений
+    user_content = f"Я хотел бы узнать про autogen. Контекст: {context}. {prompt}" if context else prompt
+    messages = [
+        {"role": "system", "content": system_message},
+        {"role": "user", "content": user_content},
+        {"role": "assistant", "content": "Спасибо! Начинаю анализ вопроса и построение цепочки рассуждений..."}
+    ]
+
+    steps: List[Tuple[str, str, float]] = []
+    total_thinking_time = 0.0
+    step_count = 1
+
+    # Итеративная генерация шагов цепочки рассуждений
+    while step_count <= max_steps:
+        start_time = time.time()
+        step_data = make_api_call(messages, max_tokens=DEFAULT_MAX_TOKENS, custom_client=custom_client)
+        elapsed = time.time() - start_time
+        total_thinking_time += elapsed
+
+        step_title = step_data.get('title', 'Без названия')
+        step_content = step_data.get('content', '')
+        steps.append((f"Step {step_count}: {step_title}", step_content, elapsed))
         
-        with tqdm(total=5, desc="Generating response") as pbar:
-            for _ in range(5):
-                result = self._make_api_call(messages, self.max_tokens['intermediate'])
-                messages.append({"role": "assistant", "content": result})
-                pbar.update(1)
-                yield result
-
-        final_answer = self._make_api_call(messages, self.max_tokens['final'], is_final=True)
-        yield final_answer
-
-def gradio_interface(prompt: str) -> str:
-    gen = ResponseGenerator()
-    response = []
-    for chunk in gen.generate_response(prompt):
-        response.append(chunk)
-    return "\n\n".join(response)
+        # Добавляем результат шага в историю сообщений
+        messages.append({"role": "assistant", "content": json.dumps(step_data, ensure_ascii=False)})
+
+        if step_data.get('next_action') == 'final_answer':
+            break
+
+        step_count += 1
+        yield steps, None  # Промежуточный вывод без общего времени
+
+    # Запрос на получение окончательного ответа
+    messages.append({
+        "role": "user",
+        "content": "Предоставьте окончательный ответ без формата JSON, сохранив исходное форматирование из подсказки."
+    })
+
+    start_time = time.time()
+    final_answer = make_api_call(messages, max_tokens=FINAL_MAX_TOKENS, is_final_answer=True, custom_client=custom_client)
+    final_elapsed = time.time() - start_time
+    total_thinking_time += final_elapsed
+
+    steps.append(("Final Answer", final_answer, final_elapsed))
+    yield steps, total_thinking_time
+
+def generate_code(prompt: str, custom_client: Optional[Any] = None) -> str:
+    """
+    Генерирует код на основе предоставленного запроса, используя модель, оптимизированную для генерации кода.
+    
+    Аргументы:
+        prompt (str): Запрос для генерации кода.
+        custom_client (Any, optional): Альтернативный клиент для API-вызова.
+    
+    Возвращает:
+        str: Сгенерированный код.
+    """
+    current_client = custom_client if custom_client is not None else client
+    response = current_client.chat.completions.create(
+        model=MODEL_CODE,
+        messages=[{"role": "user", "content": prompt}],
+        max_tokens=DEFAULT_MAX_TOKENS,
+        temperature=0.7
+    )
+    return response.choices[0].message.content
+
+if __name__ == "__main__":
+    # Пример использования: генерация цепочки рассуждений
+    context = "https://console.groq.com/docs/autogen"
+    prompt = "учитывай контекст autogen в ответ"
+    
+    for steps, total_time in generate_response(prompt, context):
+        if total_time is not None:
+            print(f"Общее время обработки: {total_time:.2f} секунд")
+        for title, content, t in steps:
+            print(f"{title}: {content} (Время: {t:.2f} секунд)")
+    
+    # Пример использования: генерация кода
+    code_prompt = "Напиши функцию на Python, которая сортирует список чисел."
+    generated_code = generate_code(code_prompt)
+    print("\nСгенерированный код:")
+    print(generated_code)