Новогодний подарок: Как я прикрутил LLM к scratch и порадовал ребёнка

После апгрейда компьютера Scratch не был установлен в системе. Поэтому первым делом я скачал его и... почти сразу же удалил. Дело в том что он не умеет в HTTP-запросы. А без них общение с внешними API невозможны. Казалось бы можно сворачивать проект и не страдать больше фигнёй, но тут на сцену вышел Turbowarp. По сути, это тот же самый Scratch, только на максималках. Его ключевая фишка - поддержка пользовательских расширений (extension). То есть мы можем написать собственный extensions и добавить недостающую функциональность.

В качестве мозга был выбран Gigachat. Во-первых он бесплатный. Во-вторых не надо заморачиваться с VPN, а в третьих он работает без vpn и бесплатен.

1. Зарегистрироваться в Studio по ссылке https://developers.sber.ru/studio/workspaces/

2. Создать проект, в инструментах выбрать GigaChat API и заполнить все необходимые поля.

3. Зайти в созданный проект и выбрать: "Настроить API"

4. Получить и сохранить ключ.

import time import uuid from fastapi import FastAPI, BackgroundTasks from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel import httpx import os # ===================== # APP # ===================== app = FastAPI() app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # ===================== # STATE # ===================== chat_history = [] last_answer = "" state = "idle" # idle | processing | ready # ===================== # GIGACHAT CONFIG # ===================== GIGACHAT_AUTH_URL = "https://ngw.devices.sberbank.ru:9443/api/v2/oauth" GIGACHAT_CHAT_URL = "https://gigachat.devices.sberbank.ru/api/v1/chat/completions" AUTH_KEY = os.getenv("GIGACHAT_AUTH_KEY") SCOPE = "GIGACHAT_API_PERS" MODEL_NAME = "GigaChat" access_token = None token_expires_at = 0 # ===================== # MODELS # ===================== class SendRequest(BaseModel): message: str # ===================== # TOKEN # ===================== async def ensure_token(): global access_token, token_expires_at now = time.time() if access_token and now < token_expires_at - 60: return access_token headers = { "Authorization": f"Basic {AUTH_KEY}", "Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json", "RqUID": str(uuid.uuid4()), } data = f"scope={SCOPE}" async with httpx.AsyncClient(verify=False, timeout=30) as client: response = await client.post( GIGACHAT_AUTH_URL, headers=headers, content=data ) response.raise_for_status() token_data = response.json() access_token = token_data["access_token"] token_expires_at = now + token_data.get("expires_in", 1800) return access_token # ===================== # GIGACHAT REQUEST # ===================== async def request_gigachat(): global chat_history, last_answer, state try: token = await ensure_token() headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json" } payload = { "model": MODEL_NAME, "messages": chat_history } async with httpx.AsyncClient(verify=False, timeout=60) as client: response = await client.post( GIGACHAT_CHAT_URL, headers=headers, json=payload ) response.raise_for_status() data = response.json() answer = data["choices"][0]["message"]["content"] chat_history.append({ "role": "assistant", "content": answer }) last_answer = answer state = "ready" except Exception as e: print("error: ", repr(e)) last_answer = "" state = "idle" # ===================== # API # ===================== @app.post("/clear") async def clear_chat(): global chat_history, last_answer, state chat_history = [] last_answer = "" state = "idle" print("Chat cleared") return {"status": "cleared"} @app.post("/send") async def send_message(req: SendRequest, background_tasks: BackgroundTasks): global chat_history, last_answer, state if state == "processing": return {"status": "busy"} chat_history.append({ "role": "user", "content": req.message }) last_answer = "" state = "processing" print("Message received, querying GigaChat") background_tasks.add_task(request_gigachat) return {"status": "accepted"} @app.get("/get") async def get_answer(): return { "answer": last_answer if state == "ready" else "" }

uvicorn server:api --host 127.0.0.1 --port 8000

Или в режиме разработки, для автоматической перезагрузки при изменении кода:

uvicorn server:app --host 127.0.0.1 --port 8000 --reload

(function (Scratch) { "use strict"; let serverUrl = ""; let lastAnswer = ""; class GigaChatProxy { getInfo() { return { id: "gigachatproxy", name: "GigaChat Proxy", blocks: [ { opcode: "connect", blockType: Scratch.BlockType.COMMAND, text: "подключиться к серверу [URL]", arguments: { URL: { type: Scratch.ArgumentType.STRING, defaultValue: "http://127.0.0.1:8000" } } }, { opcode: "clearChat", blockType: Scratch.BlockType.COMMAND, text: "очистить чат" }, { opcode: "sendMessage", blockType: Scratch.BlockType.COMMAND, text: "отправить в чат [TEXT]", arguments: { TEXT: { type: Scratch.ArgumentType.STRING, defaultValue: "Привет" } } }, { opcode: "getAnswer", blockType: Scratch.BlockType.REPORTER, text: "получить ответ" } ] }; } connect({ URL }) { serverUrl = URL; lastAnswer = ""; } async clearChat() { if (!serverUrl) return; try { await fetch(serverUrl + "/clear", { method: "POST" }); } catch (e) { // ошибки игнорируем } lastAnswer = ""; } async sendMessage({ TEXT }) { if (!serverUrl) return; lastAnswer = ""; try { await fetch(serverUrl + "/send", { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ message: TEXT }) }); } catch (e) { // ошибки игнорируем } } async getAnswer() { if (!serverUrl) return ""; try { const response = await fetch(serverUrl + "/get"); const data = await response.json(); if (data.answer && data.answer !== "") { lastAnswer = data.answer; } } catch (e) { // игнор } return lastAnswer; } } Scratch.extensions.register(new GigaChatProxy()); })(Scratch);