""" Консольное приложение с четырьмя функциями: 1) describe <путь_к_картинке> — описание изображения (vision-модель Qwen); 2) plot <описание> — matplotlib-график (сохраняется в plot.png); 3) arch <описание системы> — архитектурная схема через Graphviz (arch.png); 4) repro <путь_к_картинке> — воспроизводит график/схему с картинки (repro.png). Команды вводятся в интерактивном режиме. Для выхода: exit / quit / Ctrl+C. """ import os import io import re import json import base64 import logging import asyncio import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt import numpy as np import graphviz from dotenv import load_dotenv from openai import AsyncOpenAI # ---------- конфиг ---------- load_dotenv() LLM_API_KEY = os.getenv("LLM_API_KEY") LLM_BASE_URL = os.getenv("LLM_BASE_URL", "https://llm.lambda.coredump.ru/v1") LLM_MODEL = os.getenv("LLM_MODEL", "qwen3.5-122b") DISABLE_THINKING = os.getenv("LLM_DISABLE_THINKING", "false").lower() in ("1", "true", "yes") if not LLM_API_KEY: raise RuntimeError("В .env должна быть задана переменная LLM_API_KEY") logging.basicConfig( format="%(asctime)s | %(levelname)s | %(message)s", level=logging.WARNING, # в консоли не засоряем вывод INFO-логами ) logger = logging.getLogger("qwen-app") client = AsyncOpenAI(api_key=LLM_API_KEY, base_url=LLM_BASE_URL) def _chat_kwargs(max_tokens: int) -> dict: """Собирает общие kwargs для chat.completions.create, включая отключение reasoning если попросили. Параметры передаются через extra_body и игнорируются провайдерами, которые их не поддерживают.""" kwargs: dict = {"max_tokens": max_tokens} if DISABLE_THINKING: # OpenRouter: reasoning.exclude. DashScope/vLLM с Qwen3: enable_thinking=False kwargs["extra_body"] = { "reasoning": {"exclude": True}, "enable_thinking": False, "chat_template_kwargs": {"enable_thinking": False}, } return kwargs # ---------- 1. Описание изображения ---------- async def describe_image(image_bytes: bytes) -> str: """Отправляет картинку в vision-модель и возвращает текстовое описание.""" b64 = base64.b64encode(image_bytes).decode("utf-8") response = await client.chat.completions.create( model=LLM_MODEL, messages=[ { "role": "user", "content": [ { "type": "text", "text": ( "Опиши подробно, что изображено на картинке: " "объекты, их расположение, цвета, контекст и общее настроение." ), }, { "type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}, }, ], } ], **_chat_kwargs(max_tokens=8192), ) return _extract_content(response).strip() # ---------- 2. Генерация графиков ---------- PLOT_SYSTEM_PROMPT = """Ты генерируешь Python-код для построения графиков matplotlib. Строгие правила: - в окружении доступны только: matplotlib.pyplot как `plt` и numpy как `np`; - фигура уже создана заранее, просто рисуй через plt; - НЕ вызывай plt.show() и ничего не сохраняй в файлы; - НЕ используй import — модули уже импортированы; - возвращай ТОЛЬКО исполняемый python-код, без markdown-обрамления, без комментариев и пояснений.""" def _strip_code_fences(text: str) -> str: """Убирает ```python ... ``` если модель всё же обернула код.""" match = re.search(r"```(?:[a-zA-Z0-9_+-]*)?\s*(.*?)```", text, re.DOTALL) return (match.group(1) if match else text).strip() def _extract_content(response) -> str: """Достаёт текстовый ответ из choices[0], кидая понятную ошибку на пустой ответ.""" choice = response.choices[0] content = choice.message.content if not content: finish = getattr(choice, "finish_reason", "unknown") raise RuntimeError( f"Модель вернула пустой ответ (finish_reason={finish}). " "Возможные причины: thinking-модель упёрлась в max_tokens и не дошла до финального ответа; " "провайдер отфильтровал запрос; кончилась квота." ) return content async def generate_plot_code(request: str) -> str: response = await client.chat.completions.create( model=LLM_MODEL, messages=[ {"role": "system", "content": PLOT_SYSTEM_PROMPT}, {"role": "user", "content": request}, ], **_chat_kwargs(max_tokens=8192), ) return _strip_code_fences(_extract_content(response)) def render_plot(code: str) -> bytes: """ Выполняет сгенерированный код и возвращает PNG-байты. ВНИМАНИЕ: exec используется в ограниченном неймспейсе, но это не настоящая песочница. Для публичного приложения лучше изолировать исполнение (отдельный процесс с resource-лимитами, Docker, nsjail и т. п.). """ plt.close("all") plt.figure(figsize=(8, 5)) safe_builtins = { "range": range, "len": len, "abs": abs, "min": min, "max": max, "sum": sum, "enumerate": enumerate, "zip": zip, "map": map, "filter": filter, "sorted": sorted, "reversed": reversed, "any": any, "all": all, "list": list, "tuple": tuple, "dict": dict, "set": set, "float": float, "int": int, "str": str, "bool": bool, "round": round, "isinstance": isinstance, "type": type, "print": print, } safe_globals = {"plt": plt, "np": np, "__builtins__": safe_builtins} exec(code, safe_globals) buf = io.BytesIO() plt.savefig(buf, format="png", dpi=120, bbox_inches="tight") plt.close("all") buf.seek(0) return buf.read() # ---------- 3. Генерация архитектурных схем ---------- ARCH_SYSTEM_PROMPT = """Ты генерируешь архитектурные схемы на языке Graphviz DOT. Правила: - пиши корректный DOT-код для ориентированного графа (digraph); - давай узлам осмысленные метки; подписывай стрелки (протокол, формат данных, действие) где уместно; - группируй связанные компоненты в subgraph cluster_* с подписями; - используй разные shape и цвета заливки для разных типов компонентов: * клиент/пользователь — shape=actor или oval; * сервисы/бэкенды — shape=box, style=filled; * базы данных — shape=cylinder; * очереди/брокеры — shape=box3d; * внешние API и облачные сервисы — shape=cloud; * хранилища файлов — shape=folder; - задавай rankdir=LR для горизонтальной компоновки, если компонентов больше 4; - верни ТОЛЬКО валидный DOT-код без markdown-обрамления, без комментариев и пояснений.""" async def generate_arch_dot(description: str) -> str: response = await client.chat.completions.create( model=LLM_MODEL, messages=[ {"role": "system", "content": ARCH_SYSTEM_PROMPT}, {"role": "user", "content": description}, ], **_chat_kwargs(max_tokens=8192), ) return _strip_code_fences(_extract_content(response)) def render_architecture(dot_source: str) -> bytes: """Рендерит DOT-описание в PNG через локальный бинарь graphviz.""" src = graphviz.Source(dot_source, engine="dot", format="png") return src.pipe() # ---------- 4. Воспроизведение графика/схемы с фото ---------- REPRO_SYSTEM_PROMPT = """Ты анализируешь изображение и воспроизводишь его кодом. Определи тип: - "plot" — график любого вида (line, bar, scatter, histogram, pie, heatmap и т.п.); - "architecture" — архитектурная схема, блок-схема, граф компонентов, диаграмма потоков; - "other" — всё остальное. Если "plot" — сгенерируй Python-код на matplotlib, максимально близко воспроизводящий оригинал: - тот же тип графика, подписи осей, заголовок, легенда (если есть), примерная форма данных; - доступны только `plt` и `np`; БЕЗ import, БЕЗ plt.show() и savefig — фигура создана заранее. Если "architecture" — сгенерируй валидный DOT-код для Graphviz digraph: - те же узлы с теми же метками, те же связи и направления, та же группировка в subgraph cluster_*; - подходящие shape (cylinder для БД, cloud для внешних API, box3d для очередей и т.п.). Верни СТРОГО JSON-объект без markdown-обрамления, без пояснений до или после: {"type": "plot|architecture|other", "notes": "<краткое описание>", "code": "<исполняемый код>"} Поле `code` должно быть одной строкой JSON с экранированными переводами строк (\\n).""" def _extract_json(text: str) -> dict: """Парсит JSON из ответа модели, терпимо снимая markdown-обёртку и мусор по краям.""" cleaned = _strip_code_fences(text) try: return json.loads(cleaned) except json.JSONDecodeError: # резервный план: вытащить первый JSON-объект из текста match = re.search(r"\{.*\}", cleaned, re.DOTALL) if not match: raise RuntimeError(f"Модель вернула не-JSON:\n{text[:500]}") return json.loads(match.group(0)) async def analyze_and_reproduce(image_bytes: bytes) -> tuple[str, str, str]: """Возвращает (type, code, notes).""" b64 = base64.b64encode(image_bytes).decode("utf-8") response = await client.chat.completions.create( model=LLM_MODEL, messages=[ {"role": "system", "content": REPRO_SYSTEM_PROMPT}, { "role": "user", "content": [ {"type": "text", "text": "Проанализируй изображение и воспроизведи его кодом."}, { "type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}, }, ], }, ], **_chat_kwargs(max_tokens=8192), ) data = _extract_json(_extract_content(response)) return ( data.get("type", "other"), data.get("code", ""), data.get("notes", ""), ) # ---------- вспомогательные утилиты ---------- def read_image(path: str) -> bytes: """Читает файл изображения с диска.""" if not os.path.isfile(path): raise FileNotFoundError(f"Файл не найден: {path}") with open(path, "rb") as f: return f.read() def save_png(png: bytes, filename: str) -> str: """Сохраняет PNG-байты в файл рядом со скриптом и возвращает путь.""" out_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), filename) with open(out_path, "wb") as f: f.write(png) return out_path def print_help() -> None: print( "\nДоступные команды:\n" " describe <путь> — описать изображение\n" " plot <описание> — построить график (сохраняется в plot.png)\n" " arch <описание> — нарисовать архитектурную схему (arch.png)\n" " repro <путь> — воспроизвести график/схему с картинки (repro.png)\n" " help — эта справка\n" " exit / quit — выход\n" ) # ---------- обработчики команд ---------- async def handle_describe(arg: str) -> None: if not arg: print("Укажи путь к картинке. Пример: describe photo.jpg") return print("⏳ Анализирую изображение...") image_bytes = read_image(arg) text = await describe_image(image_bytes) print(f"\n{text}\n") async def handle_plot(arg: str) -> None: if not arg: print("Опиши, какой график построить. Пример: plot sin(x) на [0, 2π]") return print("⏳ Генерирую код и строю график...") code = await generate_plot_code(arg) logger.debug("Сгенерирован код:\n%s", code) png = await asyncio.to_thread(render_plot, code) path = save_png(png, "plot.png") print(f"✅ График сохранён: {path}\n") async def handle_arch(arg: str) -> None: if not arg: print( "Опиши архитектуру системы. Пример:\n" " arch React-фронт, FastAPI-бэкенд, Postgres и Redis за nginx" ) return print("⏳ Генерирую архитектурную схему...") dot = await generate_arch_dot(arg) logger.debug("Сгенерирован DOT:\n%s", dot) png = await asyncio.to_thread(render_architecture, dot) path = save_png(png, "arch.png") print(f"✅ Схема сохранена: {path}\n") async def handle_repro(arg: str) -> None: if not arg: print("Укажи путь к картинке. Пример: repro chart.png") return print("⏳ Анализирую и воспроизвожу...") image_bytes = read_image(arg) kind, code, notes = await analyze_and_reproduce(image_bytes) logger.debug("repro type=%s notes=%s\ncode:\n%s", kind, notes, code) if kind == "plot": png = await asyncio.to_thread(render_plot, code) path = save_png(png, "repro.png") label = f"📊 {notes}" if notes else "📊 Воспроизведённый график" print(f"✅ {label}\n Сохранён: {path}\n") elif kind == "architecture": png = await asyncio.to_thread(render_architecture, code) path = save_png(png, "repro.png") label = f"🏛 {notes}" if notes else "🏛 Воспроизведённая схема" print(f"✅ {label}\n Сохранена: {path}\n") else: msg = "Это не похоже ни на график, ни на архитектурную схему." if notes: msg += f"\n{notes}" print(f"ℹ️ {msg}\n") # ---------- главный цикл ---------- async def main() -> None: print("🤖 Qwen Console App") print_help() while True: try: line = input(">>> ").strip() except (EOFError, KeyboardInterrupt): print("\nВыход.") break if not line: continue parts = line.split(maxsplit=1) cmd = parts[0].lower() arg = parts[1].strip() if len(parts) > 1 else "" try: if cmd in ("exit", "quit"): print("Выход.") break elif cmd == "help": print_help() elif cmd == "describe": await handle_describe(arg) elif cmd == "plot": await handle_plot(arg) elif cmd == "arch": await handle_arch(arg) elif cmd == "repro": await handle_repro(arg) else: print(f"Неизвестная команда: {cmd!r}. Введи help для справки.") except FileNotFoundError as e: print(f"❌ {e}") except graphviz.ExecutableNotFound: print("❌ Graphviz не установлен. Установи: apt install graphviz / brew install graphviz / winget install graphviz") except Exception as e: print(f"❌ Ошибка: {e}") if __name__ == "__main__": asyncio.run(main())