import time
class TokenBucket:
"""A token bucket rate limiter.
Tokens are added at a constant `refill_rate` (tokens per second) up to
`capacity`. Each call to `allow_request()` tries to consume one token.
"""
def __init__(self, capacity: int, refill_rate: float) -> None:
"""
Args:
capacity: Maximum number of tokens the bucket can hold.
refill_rate: Number of tokens added to the bucket per second.
"""
self.capacity: int = capacity
self.refill_rate: float = refill_rate
self.tokens: float = float(capacity) # start full...
全文を表示 ▼
import time
class TokenBucket:
"""A token bucket rate limiter.
Tokens are added at a constant `refill_rate` (tokens per second) up to
`capacity`. Each call to `allow_request()` tries to consume one token.
"""
def __init__(self, capacity: int, refill_rate: float) -> None:
"""
Args:
capacity: Maximum number of tokens the bucket can hold.
refill_rate: Number of tokens added to the bucket per second.
"""
self.capacity: int = capacity
self.refill_rate: float = refill_rate
self.tokens: float = float(capacity) # start full
self.last_refill_time: float = time.monotonic()
def _refill(self) -> None:
"""Add tokens based on elapsed time since the last refill."""
now = time.monotonic()
elapsed = now - self.last_refill_time
tokens_to_add = elapsed * self.refill_rate
self.tokens = min(self.capacity, self.tokens + tokens_to_add)
self.last_refill_time = now
def allow_request(self) -> bool:
"""Attempt to consume one token.
Returns:
True if the request is allowed (a token was available and consumed),
False otherwise.
"""
self._refill()
if self.tokens >= 1:
self.tokens -= 1
return True
return False
---------------------------------------------------------------------------
Quick demo / smoke test
---------------------------------------------------------------------------
if name == "main":
bucket = TokenBucket(capacity=5, refill_rate=2) # 5 max, 2 tokens/sec
# Burst: should allow 5 requests immediately
for i in range(7):
result = bucket.allow_request()
print(f"Request {i + 1}: {'allowed' if result else 'denied'}")
# Wait a bit for tokens to refill
print("\nSleeping 1.5 seconds...")
time.sleep(1.5) # should refill ~3 tokens
for i in range(5):
result = bucket.allow_request()
print(f"Request {i + 1}: {'allowed' if result else 'denied'}")
判定
勝利票
0 / 3
平均スコア
総合点
総評
回答Aは、TokenBucketクラスをクリーンかつ正確に、そして適切に文書化された実装を提供しています。正確なタイミングのためにtime.monotonic()を使用し、経過時間に基づいてトークンを正しく補充し、トークンを容量まで上限設定し、許可されたリクエストごとに1つのトークンを消費します。コードは読みやすく、優れたドキュメンテーション文字列が含まれており、役立つデモ/スモークテストも含まれています。しかし、スレッドセーフ性が欠けており(ロック機構がない)、通常は同時実行環境で使用されるレートリミッターとしては顕著な欠落です。また、コンストラクタ引数の入力検証も欠けています。
採点詳細を表示 ▼
正確さ
重み 35%コアロジックは正しいです。トークンはtime.monotonic()を使用して経過時間に基づいて補充され、容量まで上限設定され、許可されたリクエストごとに1つのトークンが消費されます。しかし、スレッドセーフ性の欠如は、競合状態による同時実行使用において、実装が不正確な結果を生む可能性があることを意味します。
完全性
重み 20%実装は、容量と補充率を持つコンストラクタ、allow_request()メソッド、および内部状態管理という、必要なすべての要素を網羅しています。実行可能なデモも含まれています。しかし、スレッドセーフ性と入力検証が欠けており、これらは完全な実装にとって重要です。
コード品質
重み 20%コードはクリーンで読みやすく、明確なドキュメンテーション文字列と型ヒントによって適切に文書化されています。変数名は説明的です。補充ロジックをプライベートメソッドに分離することは良い設計です。プライベート属性の命名規則の欠如によるわずかな減点。
実用性
重み 15%この実装はシングルスレッドシナリオでは実用的に有用であり、役立つデモが含まれています。しかし、スレッドセーフ性の欠如は、レートリミッターが通常同時実行環境で使用される実際のアプリケーションにおける実用的な価値を著しく制限します。
指示遵守
重み 10%すべての指示に従っています。PythonでTokenBucketクラスを実装し、コンストラクタは容量と補充率を受け入れ、allow_request()は引数なしでトークンを消費しながらTrue/Falseを返します。標準ライブラリ(timeモジュール)のみを使用しています。状態はインスタンス内で管理されています。
総合点
総評
回答Aは、シングルスレッド環境において正しく、かつ記述の良い実装を提供しています。コードがクリーンでモダンなPythonの型ヒントを使用しており、ロジックが明確で、動作を確認するための優れた実行可能なデモが含まれている点が強みです。しかし、スレッドセーフではないという大きな弱点があり、これは同時リクエストが一般的なレートリミッターのほとんどの実際のアプリケーションには不向きです。
採点詳細を表示 ▼
正確さ
重み 35%実装はシングルスレッドのコンテキストでは論理的に正しいです。トークンリフィルの計算と容量制限は適切に実装されています。しかし、このアルゴリズムの一般的なユースケースである並行環境では正しくありません。
完全性
重み 20%回答は非常に包括的で、クラス実装だけでなく、`if __name__ == "__main__":`の下に実行可能なデモスクリプトも提供しています。このデモは、バーストとリフィルの動作を効果的に示しており、理解と検証を容易にします。
コード品質
重み 20%コード品質は高いです。型ヒントのようなモダンなPythonの機能を使用しており、明確なドキュメンテーション文字列があり、良い命名規則に従っています。ロジックはプライベートヘルパーメソッドによるリフィルでうまく構造化されています。
実用性
重み 15%実装がスレッドセーフではないため、実用的な価値は限定的です。レートリミッターは、ほとんどの場合、並行環境(Webサーバーなど)で使用されますが、この実装は競合状態の影響を受けやすく、信頼性の高い動作をしません。
指示遵守
重み 10%回答は、指定されたクラス、コンストラクタ、およびメソッドを正しいロジックで実装するという、プロンプトのすべての指示に完全に準拠しています。ソリューションは自己完結型です。
総合点
総評
回答Aは、トークンバケットのPythonによる実装として正しく、読みやすいです。単調増加クロックを使用し、内部状態を適切に追跡し、経過時間に基づいて補充し、トークンを容量までキャップし、許可されたリクエストごとに1つのトークンを消費します。主な弱点は、入力検証の欠如、同時アクセス保護の欠如、およびプロンプトで言及された精度問題を考慮しない基本的な浮動小数点アプローチです。含まれているデモは役立ちますが、コアタスクには必須ではありません。
採点詳細を表示 ▼
正確さ
重み 35%補充と消費のロジックは正しいです。経過時間は単調増加クロックで測定され、トークンは比例して追加され、容量までキャップされ、利用可能な場合は1つのトークンが消費されます。初回使用と長時間のアイドル期間を適切に処理します。無効なパラメータに対処せず、緩和策なしで単純な浮動小数点累積を使用しているため、スコアは伸び悩んでいます。
完全性
重み 20%必要なコンストラクタ、内部状態、allow_requestメソッドが含まれており、実装は自己完結しています。しかし、パラメータ検証や、無効な容量や補充率などのエッジケースについてはカバーしていません。
コード品質
重み 20%コードはクリーンで読みやすく、慣用的であり、命名も良く、役立つドキュメンテーション文字列が付いています。構造はシンプルで理解しやすいです。公開状態が直接公開されていることと、追加のデモコードがコア実装の一部ではないため、いくつかのポイントを失っています。
実用性
重み 15%単純なシングルスレッドシナリオで使用可能であり、簡単なスモークテストで動作を示しています。より現実的な設定では、検証とスレッドセーフの欠如が運用上の有用性を低下させます。
指示遵守
重み 10%必要なコンストラクタとallow_request動作を備えた自己完結型のTokenBucketクラスを実装することで、タスクをうまく実行しています。唯一のわずかな問題は、要求されたコア回答を超えた追加のデモコードが含まれていることです。