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

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);