cloud/cloud_llm_engine.py

import asyncio
from typing import AsyncIterator, Optional

import spaces
import torch
from transformers import AutoTokenizer, AutoModelForCausalLM


MODEL_ID = "Qwen/Qwen2.5-1.5B-Instruct"


@spaces.GPU(duration=120)
def _generate_sync(
    messages: list,
    max_tokens: int,
) -> str:

    tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)

    model = AutoModelForCausalLM.from_pretrained(
        MODEL_ID,
        torch_dtype=torch.float16,
        device_map="cuda",
    )
    print("\n========== CLOUD LLM PROMPT DEBUG ==========")
    print(f"Message count: {len(messages)}")

    for i, message in enumerate(messages):
        role = message.get("role", "UNKNOWN")
        content = message.get("content", "")

        print(
            f"[MESSAGE {i}] "
        f"role={role} "
        f"chars={len(content)}"
    )

    print("============================================\n")
    prompt = tokenizer.apply_chat_template(
        messages,
        tokenize=False,
        add_generation_prompt=True,
    )

    inputs = tokenizer(
        prompt,
        return_tensors="pt",
    ).to("cuda")
    print(
    f"[CLOUD LLM] Actual input tokens: "
    f"{inputs.input_ids.shape[1]}"
    )
    with torch.inference_mode():
        outputs = model.generate(
            **inputs,
            max_new_tokens=max_tokens,
            temperature=0.6,
            top_p=0.9,
            do_sample=True,
        )

    generated = outputs[0][inputs.input_ids.shape[1]:]

    response = tokenizer.decode(
        generated,
        skip_special_tokens=True,
    )

    del outputs
    del inputs
    del model
    del tokenizer

    torch.cuda.empty_cache()

    return response


class CloudLLMEngine:

    def __init__(self):
        self.last_usage = None

    async def stream_tokens(
        self,
        user_text: str,
    ) -> AsyncIterator[str]:

        messages = [
            {
                "role": "system",
                "content": (
                    "You are Edi, an AI engineering mentor."
                ),
            },
            {
                "role": "user",
                "content": user_text,
            },
        ]

        async for token in self.stream_tokens_from_messages(messages):
            yield token

    async def stream_tokens_from_messages(
        self,
        messages: list,
        session_id: str = "",
        max_tokens: Optional[int] = None,
    ) -> AsyncIterator[str]:

        response = await asyncio.to_thread(
            _generate_sync,
            messages,
            max_tokens or 250,
        )

        # Compatibility streaming.
        #
        # ZeroGPU generation completes before this adapter receives
        # the result. We emit small text chunks afterward so the
        # existing AgentController can consume an AsyncIterator.

        chunk_size = 12

        for index in range(0, len(response), chunk_size):
            yield response[index:index + chunk_size]
            await asyncio.sleep(0)

    async def aclose(self):
        return