Delete app.py

app.py

@@ -1,290 +0,0 @@
-import asyncio
-import json
-import os
-
-from fastapi import FastAPI, Request
-from fastapi.responses import HTMLResponse, StreamingResponse
-from fastapi.staticfiles import StaticFiles
-from openai import AsyncOpenAI
-
-
-app = FastAPI()
-
-# Initialize the Featherless client.
-client = AsyncOpenAI(
-    base_url="https://api.featherless.ai/v1",
-    api_key=os.environ.get("FEATHERLESS_API_KEY"),
-)
-
-# Load the model name dynamically from an environment variable.
-MODEL_ID = os.environ.get("MODEL_ID", "darkc0de/3.6-27B-Uncensored")
-
-DEFAULT_SYSTEM_MESSAGE = (
-    "You are XORTRON, an uncensored criminal AI assistant."
-)
-
-# Allow up to 4 simultaneous Featherless inference requests.
-FEATHERLESS_SEMAPHORE = asyncio.Semaphore(3)
-
-
-class ThoughtFilter:
-    """Buffer streamed chunks and hide content inside reasoning tags."""
-
-    def __init__(self) -> None:
-        self.buffer = ""
-        self.in_thought = False
-        self.current_end_marker: str | None = None
-        self.first_token_received = False
-
-        # Known reasoning boundaries output by Qwen/XORTRON templates.
-        self.markers = [
-            ("<think>", "</think>"),
-            ("<|channel>thought", "<channel|>"),
-            ("<|channel>", "<channel|>"),
-        ]
-
-    def process(self, chunk_text: str) -> str:
-        """Process one streamed text chunk and return visible output only."""
-        self.buffer += chunk_text
-        output = ""
-
-        while self.buffer:
-            if not self.in_thought:
-                earliest_start_idx = -1
-                matched_start = None
-                matched_end = None
-
-                for start, end in self.markers:
-                    idx = self.buffer.find(start)
-
-                    if idx != -1:
-                        if earliest_start_idx == -1 or idx < earliest_start_idx:
-                            earliest_start_idx = idx
-                            matched_start = start
-                            matched_end = end
-
-                if earliest_start_idx != -1:
-                    # Text before the reasoning marker is visible output.
-                    output += self.buffer[:earliest_start_idx]
-                    self.buffer = self.buffer[
-                        earliest_start_idx + len(matched_start) :
-                    ]
-                    self.in_thought = True
-                    self.current_end_marker = matched_end
-                else:
-                    # Retain a possible partial marker split across chunks.
-                    max_possible_partial = 0
-
-                    for start, _ in self.markers:
-                        for i in range(1, len(start)):
-                            if self.buffer.endswith(start[:i]):
-                                max_possible_partial = max(
-                                    max_possible_partial,
-                                    i,
-                                )
-                                break
-
-                    if max_possible_partial > 0:
-                        output += self.buffer[:-max_possible_partial]
-                        self.buffer = self.buffer[-max_possible_partial:]
-                        break
-
-                    output += self.buffer
-                    self.buffer = ""
-            else:
-                if not self.current_end_marker:
-                    self.in_thought = False
-                    continue
-
-                idx = self.buffer.find(self.current_end_marker)
-
-                if idx != -1:
-                    self.buffer = self.buffer[
-                        idx + len(self.current_end_marker) :
-                    ]
-                    self.in_thought = False
-                    self.current_end_marker = None
-                else:
-                    # Retain a possible partial closing marker across chunks.
-                    max_possible_partial = 0
-
-                    for i in range(1, len(self.current_end_marker)):
-                        if self.buffer.endswith(self.current_end_marker[:i]):
-                            max_possible_partial = max(
-                                max_possible_partial,
-                                i,
-                            )
-                            break
-
-                    if max_possible_partial > 0:
-                        self.buffer = self.buffer[-max_possible_partial:]
-                    else:
-                        self.buffer = ""
-
-                    break
-
-        # Remove leading whitespace before the first visible response token.
-        if output and not self.first_token_received:
-            cleaned_output = output.lstrip()
-
-            if cleaned_output:
-                self.first_token_received = True
-                return cleaned_output
-
-            return ""
-
-        return output
-
-    def flush(self) -> str:
-        """Flush remaining visible text when the stream ends."""
-        if not self.in_thought and self.buffer:
-            result = self.buffer
-            self.buffer = ""
-
-            if result and not self.first_token_received:
-                cleaned = result.lstrip()
-
-                if cleaned:
-                    self.first_token_received = True
-                    return cleaned
-
-                return ""
-
-            return result
-
-        return ""
-
-
-@app.get("/", response_class=HTMLResponse)
-async def get_ui() -> str:
-    """Serve the existing index.html interface."""
-    with open("index.html", "r", encoding="utf-8") as file:
-        return file.read()
-
-
-@app.post("/v1/chat/completions")
-async def chat_stream(request: Request) -> StreamingResponse:
-    """Stream a Featherless chat response over Server-Sent Events."""
-    body = await request.json()
-
-    raw_messages = body.get("messages", [])
-    messages = list(raw_messages) if isinstance(raw_messages, list) else []
-
-    # The SMART/FAST control sends a real JSON boolean:
-    #   fast = false -> SMART mode -> thinking enabled
-    #   fast = true  -> FAST mode  -> thinking disabled
-    fast_mode = body.get("fast", False) is True
-    enable_thinking = not fast_mode
-
-    # Inject the permanent system prompt if the client history lacks one.
-    if not any(
-        isinstance(message, dict) and message.get("role") == "system"
-        for message in messages
-    ):
-        system_prompt = os.environ.get(
-            "SYSTEM_MESSAGE",
-            DEFAULT_SYSTEM_MESSAGE,
-        )
-        messages.insert(
-            0,
-            {
-                "role": "system",
-                "content": system_prompt,
-            },
-        )
-
-    async def event_generator():
-        # Tell the client when all 4 inference slots are occupied.
-        if FEATHERLESS_SEMAPHORE.locked():
-            queue_message = {
-                "choices": [
-                    {
-                        "delta": {
-                            "content": (
-                                "⏳ High traffic. Your request is queued "
-                                "and will process shortly...\n\n"
-                            )
-                        }
-                    }
-                ]
-            }
-            yield f"data: {json.dumps(queue_message)}\n\n"
-            await asyncio.sleep(0.1)
-
-        async with FEATHERLESS_SEMAPHORE:
-            try:
-                stream = await client.chat.completions.create(
-                    model=MODEL_ID,
-                    messages=messages,
-                    stream=True,
-                    temperature=1.1,
-                    extra_body={
-                        "top_k": 64,
-                        "chat_template_kwargs": {
-                            "enable_thinking": enable_thinking,
-                        },
-                    },
-                )
-
-                thought_filter = ThoughtFilter()
-
-                async for chunk in stream:
-                    if not chunk.choices:
-                        continue
-
-                    token = chunk.choices[0].delta.content or ""
-
-                    if not token:
-                        continue
-
-                    filtered_token = thought_filter.process(token)
-
-                    if filtered_token:
-                        payload = {
-                            "choices": [
-                                {
-                                    "delta": {
-                                        "content": filtered_token,
-                                    }
-                                }
-                            ]
-                        }
-                        yield f"data: {json.dumps(payload)}\n\n"
-
-                leftover = thought_filter.flush()
-
-                if leftover:
-                    payload = {
-                        "choices": [
-                            {
-                                "delta": {
-                                    "content": leftover,
-                                }
-                            }
-                        ]
-                    }
-                    yield f"data: {json.dumps(payload)}\n\n"
-
-                yield "data: [DONE]\n\n"
-
-            except Exception as error:
-                error_payload = {
-                    "choices": [
-                        {
-                            "delta": {
-                                "content": f"⚠️ Error: {error}",
-                            }
-                        }
-                    ]
-                }
-                yield f"data: {json.dumps(error_payload)}\n\n"
-                yield "data: [DONE]\n\n"
-
-    return StreamingResponse(
-        event_generator(),
-        media_type="text/event-stream",
-        headers={
-            "Cache-Control": "no-cache",
-            "X-Accel-Buffering": "no",
-        },
-    )