DEV Community: Yuyi Kimura (YK46)

AniGuess: Real-Time Multiplayer Anime Guessing Game

Yuyi Kimura (YK46) — Mon, 11 Aug 2025 02:51:03 +0000

This is a submission for the Redis AI Challenge: Beyond the Cache.

What I Built

I built AniGuess, a real-time multiplayer anime character guessing game that showcases Redis as a powerful multi-model platform beyond simple caching. The game supports up to 4 players competing simultaneously in rooms, where they guess anime characters based on attribute feedback similar to Wordle mechanics.

The application demonstrates Redis's versatility by using it as the primary database for game state management and session storage. Players join rooms with 6-character codes, compete across multiple rounds with configurable timers, and receive live updates as other players make guesses.

Demo

Play the game now: https://aniguess.onrender.com/

Repository: https://github.com/ypk46/aniguess

The game features:
• Real-time multiplayer gameplay with instant updates
• Smart hint system using Redis as storage
• Character images and detailed attribute comparisons
• Customizable game settings (1-10 rounds, 30-300 second timers)
• Complete scoring system with winner determination
• Responsive design for all devices

How I Used Redis 8

Rather than treating Redis as just a cache, I leveraged it as the primary database and real-time communication backbone for the entire multiplayer experience:

1. Primary Database for Game State

Redis serves as the main data store for all game sessions, eliminating the need for complex database queries during gameplay:

// Room state management in Redis
export class RoomService {
  async createRoom(settings: GameSettings): Promise<Room> {
    const room: Room = {
      code: this.generateRoomCode(),
      players: [],
      gameState: 'waiting',
      settings,
      currentRound: 0,
      // ... other properties
    };

    await this.redis.setex(
      `room:${room.code}`,
      3600, // 1 hour TTL
      JSON.stringify(room)
    );

    return room;
  }
}

2.Pub/Sub for Horizontal Scalability

The Redis adapter handles all the complexity of cross-instance
communication, making Socket.IO code work seamlessly across multiple servers.

// socket.service.ts
import { createAdapter } from '@socket.io/redis-adapter';

const pubClient = createClient({ url: redisConfig.url });
const subClient = pubClient.duplicate();
this.io = new SocketIOServer(server, {
  adapter: createAdapter(pubClient, subClient)
});

How It Works

Two Redis Connections: Publisher sends messages, Subscriber receives them
Cross-Instance Communication: When you emit to a room/broadcast, the adapter:
- Emits locally to connected clients
- Publishes to Redis channel
- Other server instances receive and emit to their clients

3. Session Management and Player State

Redis handles all player sessions and maintains game state with automatic expiration:

// Player session and game progress tracking
async submitGuess(roomCode: string, playerId: string, characterId: number) {
  const room = await this.getRoom(roomCode);

  // Store guess result in Redis with structured data
  const guessKey = `guess:${roomCode}:${playerId}:${room.currentRound}`;
  await this.redis.setex(guessKey, 3600, JSON.stringify({
    characterId,
    timestamp: Date.now(),
    attributes: comparedAttributes,
    isCorrect: guess.isCorrect
  }));

  // Update room state atomically
  await this.updateRoomState(roomCode, updatedRoom);
}

4. Multi-Model Data Structures

The application uses Redis's rich data types to optimize different aspects of the game:

• Hash structures for complex room configurations
• Sets for managing active players and room codes
• Sorted sets for leaderboards and scoring
• String operations with TTL for temporary game data

Key Benefits Achieved:

Zero Database Latency: All game operations happen in-memory with Redis, providing instant responses crucial for real-time gameplay
Horizontal Scalability: Pub/sub enables multiple server instances to share game state seamlessly
Automatic Cleanup: TTL on game rooms prevents memory bloat without manual intervention
Atomic Operations: Redis transactions ensure consistent game state even with concurrent player actions

This implementation showcases Redis 8 as a complete solution for real-time applications, proving it can serve as both the primary database and communication layer while maintaining the performance needed for engaging multiplayer experiences.

Clipper: Orchestrating Amazon Q with Algolia MCP for Read-Later Link Management

Yuyi Kimura (YK46) — Fri, 25 Jul 2025 22:44:37 +0000

This is a submission for the Algolia MCP Server Challenge

What I Built

I've created a CLI agent powered by Amazon Q that indexes your links (like articles or blog posts) and allows you to retrieve them using natural language, all powered by Algolia MCP.

This project offers a creative approach to building a useful CLI agent with minimal technical overhead. By using Amazon Q, we get a ready-made AI assistant with MCP support. When combined with Algolia MCP, it gains the superpower to save your data in a queryable format. This means you can retrieve your information using natural language, letting the LLM do the heavy lifting.

Algolia MCP provides LLMs with powerful indexing and search capabilities, while Amazon Q offers an interface to interact with an AI assistant directly within the terminal. Amazon Q's custom profiles and context allow you to guide your agent through specific workflows using natural language, further reducing technical overhead.

Additionally, I've developed a simple MCP server to open links and get their content. This server empowers the LLM to enrich URLs with extra content, such as summaries and keywords. This enriched data is then indexed in Algolia alongside the original URL.

Once indexed, you can now ask the agent to fetch links you've saved previously, simply by using natural language.

Using it is as easy as:

// Start your chat
q chat --profile clipper

// Index the link in Algolia
> Clip this link: <URL>

// Search the link in Algolia
> Fetch the link about deploying a React app

Demo

ypk46 / clipper-q-algolia-mcp

A creative orchestration of Amazon Q and Algolia MCP to save and search your read-later links.

Clipper: Orchestrating Amazon Q with Algolia MCP for Read-Later Link Management

Overview

Clipper is a creative CLI agent that leverages Amazon Q’s orchestration capabilities and Algolia’s Model Context Protocol (MCP) to help you save, enrich, and search your read-later links. Designed for the DEV code challenge, Clipper demonstrates how LLMs and MCP tools can be combined to build a personal knowledge manager for URLs.

How It Works

Custom Amazon Q Profile
- A custom profile is created using the Amazon Q CLI.
- The instructions.md file is added as context, guiding Q on how to process URLs, extract content, and interact with Algolia MCP.
Clipping a URL
- The agent receives a URL.
- It uses a custom MCP tool to open and extract the content of the link.
- Q generates a summary and relevant keywords from the article.
- The current date is recorded.
- An entry is added to the Algolia index clipper with
  - …

View on GitHub

How I Utilized the Algolia MCP Server

Set up Algolia MCP in the mcp.json for Amazon Q. You will need to follow the authentication process provided in the documentation at https://github.com/algolia/mcp-node.
Clipper will always fetch the list of applications using the MCP server. If there is only one, it will use it automatically; otherwise, it will ask which one to use.
Once the application is set, it will add entries to a "clipper" index. Each entry will have enriched data like a summary, keywords, and the date added for each URL.
When asked to fetch a link, Clipper will create queries based on your natural language request to search the Algolia index for the right link (or list of links).

Key Takeaways

🚀 The MCP server allows you to make integrations with Algolia blazingly fast. LLMs are powerful enough to understand how to use the tooling and perform actions on your behalf.
🔎 Algolia search capabilities are the real deal. This was my first time using Algolia, and I was honestly impressed with how easy and efficient it is.
🤖 Automating processes is now more accessible than ever. The whole goal of this project is to highlight the beauty in the simplicity of making useful tools without thousands of lines of code.

Sentinel AI: Chat Boundaries, Defined by Policy

Yuyi Kimura (YK46) — Sun, 04 May 2025 17:44:11 +0000

This is a submission for the Permit.io Authorization Challenge: AI Access Control

What I Built

I built Sentinel AI, an AI agent that leverages Permit.io for fine-grained access control within a RAG workflow. Users can interact with an AI assistant to ask questions about documents, but the AI's responses are constrained by the user's permissions, ensuring that sensitive information is only accessible to authorized individuals.

Demo

Here is a live demo: https://sentinel-ai-permit-challenge.onrender.com/

Project Repo

Project repository: https://github.com/ypk46/sentinel-ai-permit-challenge

My Journey

Building Sentinel AI was a real eye-opener. I learned so much about different authorization methods.

I really liked how Permit.io's docs are set up, they nail explaining the core ideas first, which is great before diving into the details. That's key!

Their SDK was easy to use, but the docs felt a little too basic – I wanted a full reference to see everything it can do and what's different from the REST API. It could use some improvement there, but overall, the docs and SDK are pretty good.

Authorization for AI Applications with Permit.io

Sentinel AI uses a combination of RBAC and ABAC models. These models enforce policies governing what data the RAG workflow can use as context.

Specifically, each user is assigned one or more roles. These roles determine the maximum sensitivity level of documents the user is permitted to access. Documents, the resources users wish to access, each have a sensitivity attribute defined within the Permit.io layer. ABAC rules are then applied to control document access based on the user's role and the document's sensitivity.

Permit.io simplifies the implementation of this access control. You can see how easily the entire policy structure for this project can be recreated by examining the scripts/setup_policy.py file in the project repository.

Once the policy structure is defined, we use the Permit.io REST API to evaluate user permissions based on ABAC. This evaluation identifies the specific documents the user is authorized to read, ensuring that the RAG workflow only utilizes context from these permitted documents.

Conclusion

Permit.io is designed to address the widespread need for access control across various applications. Its key advantage is the speed at which you can integrate it into your own application, allowing you to focus on your core value proposition instead of building access control from scratch.

Sentinel AI was developed specifically to demonstrate this rapid integration capability. We encourage you to visit the project repository. There, you can see how quickly the entire project, including the policy structure, document embeddings, and RAG access control, can be set up and run in just a few minutes.

Market Maestro: Your AI-Powered Stock Market Analyst

Yuyi Kimura (YK46) — Sun, 19 Jan 2025 17:40:03 +0000

This is a submission for the Agent.ai Challenge: Productivity-Pro Agent (See Details)

What I Built

I built an AI agent that can generate stock market reports for the requested company by symbol (like AAPL or TSLA). The report is sent by email and contains daily, weekly and monthly historical trends with charts along with trading suggestions for short, mid and long term investments.

I built this agent because I wanted to create a tool that would help people make informed decisions about their investments in the stock market. I believe this agent can be used by anyone who is interested in the stock market, regardless of their level of experience.

Demo

You can see a demo of my agent here: Market Maestro

Here are some screenshots:

Agent.ai Experience

I had a great experience using Agent.ai. The platform is very easy to understand and follow. I was definitely impressed with how fast I was able to go from idea to reality.

I had a lot of fun and learned a lot. This is definitely my first of many agents!

Ensuring Fair Processing with Celery - Part II

Yuyi Kimura (YK46) — Tue, 10 Dec 2024 12:30:00 +0000

This article explores task priorities in Celery, building upon the previous post about fair processing. Task priorities offer a way to enhance fairness and efficiency in background processing by assigning different priority levels to tasks based on custom criteria.

Why Task-Level Priority?

Task-level priority provides fine-grained control over task execution without complex implementation. By submitting all tasks to a single queue with assigned priority values, workers can process tasks based on their urgency. This ensures fair handling regardless of submission time.

For example, if one tenant submits 100 tasks and another submits 5 shortly after, task-level priority prevents the second tenant from waiting for all 100 tasks to complete.

This approach dynamically assigns priority based on a tenant's task count. Each tenant's first task starts with high priority, but with every 10 concurrent tasks, the priority decreases. This ensures that tenants with fewer tasks don't experience unnecessary delays.

Implementing Task Priority

First, install Celery and Redis:

pip install celery redis

Configure Celery to use Redis as the broker and enable priority-based task processing:

from celery import Celery

app = Celery(
    "tasks",
    broker="redis://localhost:6379/0",
    broker_connection_retry_on_startup=True,
)

app.conf.broker_transport_options = {
    "priority_steps": list(range(10)),
    "sep": ":",
    "queue_order_strategy": "priority",
}

Define a method to calculate dynamic priority using Redis to cache each tenant's task count:

import redis

redis_client = redis.StrictRedis(host="localhost", port=6379, db=1)

def calculate_priority(tenant_id):
    """
    Calculate task priority based on the number of tasks for the tenant.
    """
    key = f"tenant:{tenant_id}:task_count"
    task_count = int(redis_client.get(key) or 0)
    return min(10, task_count // 10)

Create a custom task class to decrement the task count upon successful completion:

from celery import Task

class TenantAwareTask(Task):
    def on_success(self, retval, task_id, args, kwargs):
        tenant_id = kwargs.get("tenant_id")

        if tenant_id:
            key = f"tenant:{tenant_id}:task_count"
            redis_client.decr(key, 1)

        return super().on_success(retval, task_id, args, kwargs)

@app.task(name="tasks.send_email", base=TenantAwareTask)
def send_email(tenant_id, task_data):
    """
    Simulate sending an email.
    """
    sleep(1)
    key = f"tenant:{tenant_id}:task_count"
    task_count = int(redis_client.get(key) or 0)
    logger.info("Tenant %s tasks: %s", tenant_id, task_count)

Trigger tasks for different tenants, ensuring the tenant_id is included in the task's keyword arguments:

if __name__ == "__main__":
    tenant_id = 1
    for _ in range(100):
        priority = calculate_priority(tenant_id)
        key = f"tenant:{tenant_id}:task_count"
        redis_client.incr(key, 1)
        send_email.apply_async(
            kwargs={"tenant_id": tenant_id, "task_data": {}}, priority=priority
        )


    tenant_id = 2
    for _ in range(10):
        priority = calculate_priority(tenant_id)
        key = f"tenant:{tenant_id}:task_count"
        redis_client.incr(key, 1)
        send_email.apply_async(
            kwargs={"tenant_id": tenant_id, "task_data": {}}, priority=priority
        )

You can see the full code here.

Start the Celery worker and trigger the tasks:

# Run the worker
celery -A tasks worker --loglevel=info

# Trigger the tasks
python tasks.py

This setup demonstrates how Celery's priority queue, combined with Redis, ensures fair task processing by dynamically adjusting priorities based on tenant activity. Let’s see a simplified output of the worker:

Conclusion

Task-level priority with Celery and Redis provides a robust solution for ensuring fair processing in multi-tenant systems. By dynamically assigning priorities and leveraging a single queue, you can maintain simplicity while meeting business requirements.

There are many ways to implement task-level priority, using RabbitMQ for example is more efficient since it support priority at its core but since we are also using Redis for task counting, it simplifies our overall architecture.

Hope you find this useful and see on the next one!

Ensuring Fair Processing with Celery — Part I

Yuyi Kimura (YK46) — Thu, 14 Nov 2024 21:25:09 +0000

If you’re familiar with Python, chances are you’ve heard of Celery. It’s often the go-to choice for handling tasks asynchronously, like image processing or sending emails.

Talking with some folks, I started noticing that many developers find Celery impressive at first, but as their projects scale and complexity increases, their excitement start to fade. While some move away from Celery for legitimate reasons, others may simply not explore its core deeply enough to tailor it to their needs.

In this blog, I want to discuss one of the reasons why some developers start looking for alternatives or even build custom background worker frameworks: fair processing. In environments where users/tenants submit tasks of varying sizes, the risk of one tenant’s heavy workload affecting others can create bottlenecks and lead to frustration.

I’ll walk you through strategies to implement fair processing in Celery, ensuring balanced task distribution so that no single tenant can dominate your resources.

The Problem

Let’s dive into a common challenge faced by multi-tenant applications, particularly those that handle batch processing. Imagine you have a system where users can queue their image processing tasks, allowing them to receive their processed images after a brief wait. This setup not only keeps your API responsive but also lets you scale your workers as needed to handle the load efficiently.

Everything runs smoothly—until one tenant decides to submit an enormous batch of images for processing. You’ve got multiple workers in place, and they can even auto-scale to accommodate increased demand, so you’re feeling confident about your infrastructure. However, the trouble begins when other tenants attempt to queue smaller batches—perhaps just a couple of images—and suddenly find themselves facing long wait times without any updates. Before you know it, support tickets start flooding in, with users complaining that your service is slow or even unresponsive.

This scenario is all too common because Celery, by default, processes tasks in the order they are received. When one tenant overwhelms your workers with a massive influx of tasks, even the best auto-scaling strategies may not be enough to prevent delays for other tenants. As a result, those users may experience service levels that fall short of what was promised or expected.

Rate Limiting with Celery

One effective strategy for ensuring fair processing is to implement rate limits. It allows you to control the number of tasks each tenant can submit within a specific time frame. This prevents any single tenant from monopolizing your workers and ensures that all tenants have a fair chance to process their tasks.

Celery has built-in functionality for rate limiting at the task level:

# app.py
from celery import Celery

app = Celery("app", broker="redis://localhost:6379/0")

@app.task(rate_limit="10/m") # Limit to 10 tasks per minute
def process_data(data):
    print(f"Processing data: {data}")

# Call the task
if __name__ == "__main__":
    for i in range(20):
        process_data.delay(f"data_{i}")

You can run the worker by executing:

celery -A app worker --loglevel=warning --concurrency 1 --prefetch-multiplier 1

Now, run the app.py script to trigger 20 tasks:

python app.py

If you manage to run it locally, you will notice that there is a delay between each task to ensure that the rate limit is enforced. Now you are probably thinking that this doesn't really help us with our problem, and you are totally right. This built-in rate limit by Celery is useful for scenarios in which our task may involve calls to external services that have strict rate limits.

This example highlights how the built-in feature may be too simple for complex scenarios. However, we can overcome this limitation by exploring Celery's framework in more depth. Let's see how we can set up a proper rate-limit with auto-retry per tenant.

We will be using Redis to track the rate-limit per tenant. Redis is a popular database and broker for Celery, so let's leverage this component that may probably be already in your stack.

Let's import a couple libraries:

import time
import redis
from celery import Celery, Task

Now we are going to implement a custom base task class for our rate limited task:

# Initialize a Redis client
redis_client = redis.StrictRedis(host="localhost", port=6379, db=0)

class RateLimitedTask(Task):
    def __init__(self, *args, **kwargs):
        # Set default rate limit
        if not hasattr(self, "custom_rate_limit"):
            self.custom_rate_limit = 10

        super().__init__(*args, **kwargs)

    def __call__(self, tenant_id, *args, **kwargs):
        # Rate limiting logic
        key = f"rate_limit:{tenant_id}:{self.name}"

        # Increment the count for this minute
        current_count = redis_client.incr(key)

        if current_count == 1:
            # Set expiration for the key if it's the first request
            redis_client.expire(key, 10)

        if current_count > self.custom_rate_limit:
            print(f"Rate limit exceeded for tenant {tenant_id}. Retrying...")
            raise self.retry(countdown=10)

        return super().__call__(tenant_id, *args, **kwargs)

This custom class will track the amount of tasks triggered by a specific tenant using Redis and set a TTL of 10 seconds. If the rate limit is exceeded, the task will be retried again in 10 seconds. So basically our default rate limit is 10 tasks within 10 seconds.

Let's define a sample task that emulate the processing:

@app.task(base=RateLimitedTask, custom_rate_limit=5)
def process(tenant_id: int, data):
    """
    Mock processing task that takes 0.3 seconds to complete.
    """
    print(f"Processing data: {data} for tenant: {tenant_id}")
    time.sleep(0.3)

Here we have defined a process task and you can see that I can change the custom_rate_limit at the task level. If we don't specify a custom_rate_limit, the default value of 10 would be assigned. Now our rate limit has changed to 5 tasks within 10 seconds.

Let's now trigger some tasks for different tenants:

if __name__ == "__main__":
    for i in range(20):
        process.apply_async(args=(1, f"data_{i}"))

    for i in range(10):
        process.apply_async(args=(2, f"data_{i}"))

We are defining 20 tasks for the tenant ID 1 and 10 tasks for the tenant ID 2.

So our complete code would look like this:

# app.py
import time
import redis
from celery import Celery, Task

app = Celery(
    "app",
    broker="redis://localhost:6379/0",
    broker_connection_retry_on_startup=False,
)

# Initialize a Redis client
redis_client = redis.StrictRedis(host="localhost", port=6379, db=0)


class RateLimitedTask(Task):
    def __init__(self, *args, **kwargs):
        if not hasattr(self, "custom_rate_limit"):
            self.custom_rate_limit = 10

        super().__init__(*args, **kwargs)

    def __call__(self, tenant_id, *args, **kwargs):
        # Rate limiting logic
        key = f"rate_limit:{tenant_id}:{self.name}"

        # Increment the count for this minute
        current_count = redis_client.incr(key)

        if current_count == 1:
            # Set expiration for the key if it's the first request
            redis_client.expire(key, 10)

        if current_count > self.custom_rate_limit:
            print(f"Rate limit exceeded for tenant {tenant_id}. Retrying...")
            raise self.retry(countdown=10)

        return super().__call__(tenant_id, *args, **kwargs)


@app.task(base=RateLimitedTask, custom_rate_limit=5)
def process(tenant_id: int, data):
    """
    Mock processing task that takes 0.3 seconds to complete.
    """
    print(f"Processing data: {data} for tenant: {tenant_id}")
    time.sleep(0.3)

if __name__ == "__main__":
    for i in range(20):
        process.apply_async(args=(1, f"data_{i}"))

    for i in range(10):
        process.apply_async(args=(2, f"data_{i}"))

Let's run our worker:

celery -A app worker --loglevel=warning --concurrency 1 --prefetch-multiplier 1

Now, run the app.py script to trigger the tasks:

python app.py

As you can see, the worker processes 5 tasks of the first tenant, and set up a retry for all the other tasks. It then take 5 tasks of the second tenant and set up a retry for the other tasks, and it keeps going.

This approach allows you to define a rate limit per tenant but as you can see in our example, for a task that runs very fast, being too strict with the rate limit ends up leaving the worker doing nothing for a while. Fine-tuning the rate limit parameters is crucial and depends on the specific task and volume. Don't hesitate to experiment until you find an optimal balance.

Conclusion

We’ve explored how Celery’s default task processing can lead to unfairness in multi-tenant environments and how rate limiting can help address this issue. By implementing tenant-specific rate limits, we can prevent any single tenant from monopolizing resources and ensure a more equitable distribution of processing power.

This approach provides a solid foundation for achieving fair processing in Celery. However, there are other techniques worth exploring to further optimize task handling in multi-tenant applications. While I’d initially planned to cover everything in one post, this topic is proving to be quite extensive! To ensure clarity and keep this article focused, I’ve decided to split it into two parts.

In the next part of this series, we’ll delve into task priorities as another mechanism to enhance fairness and efficiency. This approach allows you to assign different priority levels to tasks based on different criteria, ensuring that critical tasks are processed promptly even during high-demand periods.

Stay tuned for the next installment!

Chat with News: Interact with your articles

Yuyi Kimura (YK46) — Sun, 10 Nov 2024 23:26:07 +0000

This is a submission for the Open Source AI Challenge with pgai and Ollama

What I Built

This application enables you to fetch, embed, and interact with news articles from RSS sources. The CLI tool allows you to set up RSS links, fetch and embed available articles, and expose an API for seamless interaction.

A simple web application was also developed to provide a more user-friendly experience. You can ask questions about articles published on specific dates, and Ollama LLM will respond while retrieving the related article for fact-checking or obtaining more details from the original source.

Setting up RSS feeds, fetching articles, and creating embeddings takes only minutes. With just a few CLI commands, you can have your article database ready in no time.

Demo

Website

https://chat-with-news.onrender.com
The project is set up to answer questions based on the news published on the selected date. This help improves the RAG by limiting the numbers of articles while also giving context more relevant to the question.

Source code

https://github.com/ypk46/chat-with-news

Tools Used

pgvector: I'm using pgvector to store the article embeddings that are generated through llama3.2 model.
pgai: I use pgai to call llama3.2 model through SQL queries to get related articles and generate an answer.
pgai Vectorizer: I set up a CLI command to run track changes on the articles table and vectorize all pending articles using llama3.2 model. You can set up a cron job to run every X minutes, to have your articles embeddings always ready. See code here.
Ollama: I use Ollama to host the llama3.2 which serves a couple purposes:
- Generate an article's summary to display on the web.
- Generate the embeddings from the article full content.
- Generate answers to user's question related to any article on a given date.

Final Thoughts

This demo helped me establish some solid groundwork for a personal project involving news aggregation and AI. I believe the tools developed by the Timescale team truly empower developers, making them more efficient and simplifying tedious tasks in a consistent and user-friendly way.

The pgai Vectorizer has been a game-changer for me. Keeping an embeddings store up to date can be repetitive and time-consuming. However, building a vectorizer worker is straightforward and highly effective for ensuring that your real data and embeddings store remain synchronized.

Prize Categories

This submission qualifies for:

Open-source Models from Ollama
Vectorizer Vibe
All the Extensions