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)