NEKO.Q — Async AI Cat Avatar Pipeline

A message-queue-driven AI persona engine

What Is This?

NEKO.Q is a containerized async pipeline that turns a language model into a persistent AI persona — in this case, a cat influencer that can reply to fans and write social posts.

The interesting part isn't the LLM call itself (that's a thin wrapper around gpt-4o-mini with structured outputs). The interesting part is everything around it: a production-style task queue architecture where every AI generation request is brokered through RabbitMQ, processed by Celery workers, cached in Redis, and exposed through a non-blocking REST API. The frontend is a lightweight Tkinter GUI forwarded over X11 — functional, retro, and deliberately minimal.

Why Build It This Way?

Most "chat with AI" demos call the API synchronously and block until it returns. That works for a toy — but not for anything real. NEKO.Q demonstrates the pattern you'd actually use in production:

• Decouple request from execution — the API returns immediately with a task ID
• Scale workers independently — spin up more Celery workers without touching the API
• Survive failures gracefully — RabbitMQ persists unprocessed tasks; Redis caches results
• Poll or webhook for results — clients stay responsive, no long-hanging connections

Architecture

Sequence Diagram

sequenceDiagram
    participant UI as Frontend (Tkinter)
    participant API as FastAPI
    participant MQ as RabbitMQ
    participant W as Celery Worker
    participant LLM as OpenAI API
    participant R as Redis

    UI->>API: POST /avatar/reply {prompt}
    API->>MQ: Publish task to queue
    API-->>UI: 202 {task_id, status: "queued"}

    MQ->>W: Deliver task message
    W->>LLM: Structured output request (gpt-4o-mini)
    LLM-->>W: Parsed response (Pydantic schema)
    W->>R: Store result (task_id → JSON)

    loop Poll every 1s
        UI->>API: GET /tasks/{task_id}
        API->>R: Lookup result
        R-->>API: Result or PENDING
        API-->>UI: {status, data?}
    end

    Note over UI: Display response when status = SUCCESS

Loading

Services

Service	Role	Port
web	FastAPI REST API server	8000
celery_worker	Async task executor (OpenAI calls)	—
rabbitmq	AMQP message broker	5672 / 15672 (management UI)
redis	Task result backend + caching	6379
chat-mac	Tkinter GUI client (X11 forwarded)	—

Quick Start

1. Clone & configure

git clone https://github.com/youruser/neko-q.git
cd neko-q/backend
cp .env.example .env

Edit .env with your credentials:

OPENAI_API_KEY=sk-...
RABBITMQ_USER=user
RABBITMQ_PASS=password
CELERY_BROKER_URL=amqp://user:password@rabbitmq:5672//
CELERY_BACKEND_URL=redis://redis:6379/0
TIMEZONE=Asia/Taipei

2. Launch (macOS)

./run.sh

The startup script handles Docker Desktop, XQuartz, socat bridging (TCP 6000 → UNIX socket), and docker compose --profile mac up --build. A window titled NEKO.Q will appear.

3. Explore the API

Open http://localhost:8000/docs for the interactive Swagger UI. Key endpoints:

Method	Endpoint	Description
POST	/api/avatar/reply	Queue a reply-style generation
POST	/api/avatar/post	Queue a social-post generation
GET	/api/tasks/{task_id}	Poll task status and retrieve result

Project Structure

neko-q/
├── backend/
│   ├── app.py                 # FastAPI entrypoint
│   ├── api/
│   │   ├── controller.py      # Route definitions
│   │   └── schemas.py         # Request/response models
│   ├── services/
│   │   └── worker.py          # Celery tasks + OpenAI integration
│   ├── prompts/
│   │   └── framework_prompts.py
│   ├── docker-compose.yml
│   ├── Dockerfile
│   ├── .env.example
│   └── run.sh
├── frontend/
│   ├── chat_gui.py            # Tkinter X11 client
│   └── Dockerfile
└── README.md

How the Task Pipeline Works

1. Client sends a prompt via POST /avatar/reply or /avatar/post
2. FastAPI calls task.delay(prompt) — this publishes a message to RabbitMQ and returns a task ID instantly
3. RabbitMQ holds the message in a durable queue until a worker picks it up
4. Celery Worker consumes the message, calls OpenAI with structured output parsing (Pydantic), and writes the result to Redis
5. Client polls GET /tasks/{task_id} until it sees status: "success", then renders the response

The worker uses OpenAI's beta.chat.completions.parse() with Pydantic models as response_format, guaranteeing type-safe, schema-validated outputs — no regex parsing, no JSON-mode guessing.

Requirements

• macOS 13+ (for X11 GUI forwarding)
• Docker Desktop (latest)
• Homebrew packages: socat, xquartz
• OpenAI API key (Pay-As-You-Go billing, not ChatGPT Plus)