g1.py

import time
import json
import logging
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:
            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:
            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))
        
        # Добавляем результат шага в историю сообщений
        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)