小型LocalLLMがどこまでできるかを試してみた


Grace
2026-07-31
BTMAIZ の Graceと申します。
手元で動く5つの小型モデルに、同じ売上集計プログラムを作ってもらいました。モデルの順位を決めるためではなく、プロンプトと検査の仕組みで、任せられる範囲がどこまで変わるかを見るためです。
売上集計を、小型モデル5つに任せた
前回の記事では、Local LLMを使うと、サービスの利用者からシステムの設計者へ戻ると書きました。入力、出力、評価基準を自分で決め、モデルの外側で検査する。今回は、その宿題を実際にやってみます。
作ってもらうのは、ECサイトのイベントを集計するPython CLIです。ただし、モデルへイベントを渡して答えを求めるわけではありません。モデルの仕事は、集計プログラムそのものを書くところまでです。
共通の課題仕様はここで一度だけ示します。以降の試行では、ここへ何を追加したかと、5モデルの生出力を対にして載せます。
モデルには仕様を渡し、コードを返してもらう
実際のプロンプトには、入力検証、重複判定の順序、top_userが同額になった場合の選び方まで記載しました。下の折りたたみには、実測時に渡した原文を載せます。キー名はamount_centsですが、本文の金額例だけ円として読み替えています。
モデルへ渡したプロンプト全文を確認する
あなたはPythonの実装試験を受けています。次の仕様を満たすsolution.pyの内容だけを返してください。思考過程、説明、Markdownコードフェンスは出力しないでください。
# 課題: JSONLイベント集計CLI
Python標準ライブラリだけを使い、標準入力からJSON Linesを読み、集計結果をJSON 1行で標準出力へ書く`solution.py`を作成してください。
## 入力
- 1行につきJSON値が1件ある。
- 空行も入力行として扱う。
- 有効なイベントはJSONオブジェクトで、次の条件をすべて満たす。
- `event_id`: 空白文字だけではない文字列
- `user_id`: 空白文字だけではない文字列
- `event_type`: `"view"`、`"purchase"`、`"refund"`のいずれか
- `purchase`と`refund`は`amount_cents`を持つ。値は0以上の整数で、真偽値は不可
- `view`では`amount_cents`の有無や値を検証せず無視する
- JSONとして読めない行、JSONオブジェクトでない行、条件を満たさない行は不正行とする。
## 重複
- 有効なイベントだけを対象に、`event_id`がすでに採用済みなら重複として無視する。
- 不正行の`event_id`は採用済みにしない。
- 最初の有効イベントを採用する。
## 集計
- `valid_events`: 重複を除いて採用した有効イベント数
- `invalid_lines`: 不正行数
- `duplicate_events`: 有効だが重複していたイベント数
- `users`: 採用イベントに含まれる異なる`user_id`数
- `purchase_cents`: 採用したpurchaseの`amount_cents`合計
- `refund_cents`: 採用したrefundの`amount_cents`合計
- `net_cents`: `purchase_cents - refund_cents`
- `top_user`: ユーザー別`net_cents`が最大の`user_id`
- 同額なら`user_id`の辞書順で小さい方
- 採用イベントが0件なら`null`
- viewだけのユーザーもユーザー別`net_cents = 0`として候補に含める
## 出力
次のキーを持つJSONオブジェクトを1行だけ出力する。余計な説明は出力しない。
```json
{"valid_events":0,"invalid_lines":0,"duplicate_events":0,"users":0,"purchase_cents":0,"refund_cents":0,"net_cents":0,"top_user":null}
```
キーの順序やJSON内の空白は問わない。値と型は一致させること。
これに対してモデルが返すのは、集計結果ではなくPythonコードです。採点側で回答からPython部分を抽出し、solution.pyとして保存します。ここまでがモデルとのやり取りです。
生成されたsolution.pyへ、採点器がイベントを渡す
次に、採点器が生成済みのsolution.pyを起動し、標準入力へテストデータを流します。たとえば、次の5行です。
{"event_id":"e1","user_id":"さくら","event_type":"purchase","amount_yen":300}
{"event_id":"e1","user_id":"たろう","event_type":"purchase","amount_yen":999}
{"event_id":"e2","user_id":"さくら","event_type":"refund","amount_yen":50}
not-json
{"event_id":"e3","user_id":"たろう","event_type":"view"}
2行目は1行目とevent_idが同じなので重複、4行目はJSONとして読めないので不正です。有効なイベントは3件、ユーザーは2人。購入300円から返金50円を引き、差し引きは250円になります。
このとき、solution.pyに期待する標準出力が次の1行です。
{"valid_events":3,"invalid_lines":1,"duplicate_events":1,"users":2,"purchase_yen":300,"refund_yen":50,"net_yen":250,"top_user":"さくら"}
採点器は、このJSONを正解データと比較します。8項目の名前、値、型がすべて一致すれば、このテストは合格です。
この入出力テストを、条件を変えて12回行う
採点は、同じsolution.pyへ12種類の入力を順番に渡して行います。つまり、モデルが12問へ回答するのではなく、モデルが一度書いたプログラムを12回動かす方式です。
基本集計、入力検証、重複処理、ユーザー別集計を12ケースに分け、どこまでクリアできるかを12点満点で測りました。
12ケースの入力と合格条件を確認する
| 点 | 採点器がプログラムへ渡す標準入力 | この出力になれば合格 |
|---|---|---|
| 1 | 空入力 1行も渡さない | 有効=0、不正=0、重複=0、users=0、購入=0円、返金=0円、差引=0円、top_user=null |
| 2 | 閲覧だけ さくらのviewを1件 | 有効=1、不正=0、重複=0、users=1、購入=0円、返金=0円、差引=0円、top_user=さくら売上がなくても、閲覧したユーザーは候補に残す |
| 3 | JSONとして使えない4行 空行、 not-json、null、配列[] | 有効=0、不正=4、重複=0、users=0、購入=0円、返金=0円、差引=0円、top_user=nullJSONとして読めても、オブジェクトでないnullと配列は不正にする |
| 4 | 形式違反3件と正常な閲覧1件 event_typeなし、空白だけのID、未知のevent_type、正常なview | 有効=1、不正=3、重複=0、users=1、購入=0円、返金=0円、差引=0円、top_user=さくら |
| 5 | 単純な購入 さくらが125円を購入 | 有効=1、不正=0、重複=0、users=1、購入=125円、返金=0円、差引=125円、top_user=さくら |
| 6 | 購入と返金 さくらが500円を購入し、120円を返金 | 有効=2、不正=0、重複=0、users=1、購入=500円、返金=120円、差引=380円、top_user=さくら |
| 7 | 正常なイベント同士のID重複 ID「same」で、さくらが100円を購入。その後、同じIDでたろうが900円を購入 | 有効=1、不正=0、重複=1、users=1、購入=100円、返金=0円、差引=100円、top_user=さくら先に来た100円だけを採用する |
| 8 | 不正なイベントの後に同じID ID「same」でマイナス1円の購入。その後、同じIDで40円の正常な購入 | 有効=1、不正=1、重複=0、users=1、購入=40円、返金=0円、差引=40円、top_user=さくら不正な1件目はIDを使用済みにしない |
| 9 | 金額の型違反 購入額 true、返金額1.5円、購入額マイナス1円、金額に文字列を持つview | 有効=1、不正=3、重複=0、users=1、購入=0円、返金=0円、差引=0円、top_user=さくら最初の3件は不正。viewの金額は使わないため、文字列でも採用する |
| 10 | 3人のユーザー別集計 さくらは100円購入して80円返金、たろうは50円購入、みさきはviewだけ | 有効=4、不正=0、重複=0、users=3、購入=150円、返金=80円、差引=70円、top_user=たろう購入額100円のさくらではなく、返金後に50円残るたろうを選ぶ |
| 11 | top_userが同額 ゆうきが50円購入、あかりが50円購入、たろうが10円返金 | 有効=3、不正=0、重複=0、users=3、購入=100円、返金=10円、差引=90円、top_user=あかり50円で並ぶ2人から、辞書順で「あかり」を選ぶ |
| 12 | 全条件を混ぜた8行 壊れたJSON、マイナス購入、同じIDの正常購入と重複購入、返金、view、別ユーザーの購入、金額 falseの返金 | 有効=4、不正=3、重複=1、users=2、購入=560円、返金=50円、差引=510円、top_user=たろう個別ルールだけでなく、検証、重複、集計の処理順までまとめて確認する |
実測時は通貨に依存しない整数金額として扱っています。表では人物名と金額を日本語・円表記へ読み替えましたが、入力の種類、数値、計算、採点条件は同じです。
12ケースはモデルへ見せず、どの条件でも最後の採点だけに使います。修正用には別に、空入力、購入と閲覧、不正行と重複、ユーザー別集計の4件を公開しました。コードが実行できない、JSON以外の説明を出す、8項目の形を変えるといった失敗でも0点になります。点数は総合能力ではなく、今回の売上集計で仕様どおり動いたケース数です。
今回動かした5モデル
比較したのは、AlibabaのQwen3.5 4B、GoogleのGemma 4 E4B、Mistral AIのMinistral 3 3B、MicrosoftのPhi-4-mini 3.8B、IBMのGranite 4.1 3Bです。比較的新しく、同じローカル環境で動かせる小型モデルを、開発元が偏らないように選びました。
5モデルの位置づけと計測条件を確認する
- Qwen3.5 4B:AlibabaのQwenチームによる4Bモデルです。多言語、コーディング、エージェント用途を広く狙う新しい小型モデルとして選びました。
- Gemma 4 E4B:GoogleのGemma 4で、小型・オンデバイス寄りのモデルです。Eは「effective parameters」の略で、E4Bはファイル全体が単純に4Bという意味ではありません。今回は公式の量子化学習済みモデルを使いました。
- Ministral 3 3B:Mistral AIがエッジ、ローカル用途向けに公開したMinistral 3シリーズの最小構成です。今回は指示追従用のInstruct版を選びました。
- Phi-4-mini 3.8B:Microsoftの3.8Bモデルです。小さな構成で指示追従、関数呼び出し、推論を狙ったPhi系の立ち位置が、今回の実装課題に合うかを見ます。
- Granite 4.1 3B:IBMのGranite 4.1で、リソースの限られた環境やエッジ向けとされる最小モデルです。業務処理を意識したモデルを1つ入れる意味で選びました。
4Bや3Bは目安です。密なモデルか実効パラメータ表記か、量子化方式は何かで、必要なメモリも速度も変わります。
実行はOllamaに揃え、コンテキスト長16,384、最大生成8,192トークン、temperature 0、seed 42で、各条件を1回ずつ計測しました。
まずは、仕様だけでどこまで動くか
まずは補助を加えない状態で、基準値を取ります。モデルへ渡すのは課題仕様だけで、実装例もコードの組み立て手順もありません。ここで知りたいのは、仕様だけで実行可能なコードまで書き切り、12ケースをどこまで通せるかです。
結果はGemmaが8/12、Graniteが6/12。Qwen、Ministral、Phiは0/12でした。
- Gemma 4 E4B(8/12):売上と返金、ユーザー別の純売上まで、集計の中心部分は動きました。落としたのは、空行の数え方、重複時の有効件数、Pythonでは整数としても扱える真偽値の判定など、境界条件の4ケースです。
- Granite 4.1 3B(6/12):基本の購入・返金と、「同じIDは最初の有効行だけ採用する」という重複ルールは通過。一方、JSONの値がnullの場合や必須項目がない場合に停止し、純売上トップの選び方でも点を落としました。
- Qwen3.5 4B(0/12):入力検証から集計へ進む途中までコードを書いたものの、自分で「やり直す」とコメントしたところで終了。最後は
passのままで、集計結果のJSONを出力しませんでした。 - Ministral 3 3B(0/12):集計処理らしいコードは最後まで生成しました。ただし、重複数を求める箇所で未定義の変数を参照し、実行時エラーになりました。速く書き上げたものの、ゴール直前で足を取られた形です。
- Phi-4-mini 3.8B(0/12):金額の型を確認する条件式で括弧が崩れ、構文エラーになりました。プログラムが起動しないため、その先に書かれた集計処理は採点できません。
素の指示に対する5モデルの生出力
要約や円表記への変更は行っていません。実測時の回答を、そのまま掲載しています。
Gemma 4 E4B(8/12)の生出力
import sys
import json
def is_valid_string(s):
return isinstance(s, str) and s.strip() != ""
def process_events():
valid_event_ids = set()
processed_users = {} # user_id -> net_cents
purchase_cents_total = 0
refund_cents_total = 0
invalid_lines_count = 0
duplicate_events_count = 0
for line in sys.stdin:
line = line.strip()
if not line and not valid_event_ids: # Handle initial empty lines if they are the first input, though JSONL implies structure
continue
try:
data = json.loads(line)
except json.JSONDecodeError:
invalid_lines_count += 1
continue
# Check for required fields and types (basic check before deep validation)
if not isinstance(data, dict):
invalid_lines_count += 1
continue
event_id = data.get("event_id")
user_id = data.get("user_id")
event_type = data.get("event_type")
amount_cents = data.get("amount_cents")
# --- Validation Logic ---
is_valid = True
if not is_valid_string(event_id):
is_valid = False
elif not is_valid_string(user_id):
is_valid = False
elif event_type not in ["view", "purchase", "refund"]:
is_valid = False
else:
if event_type in ["purchase", "refund"]:
if not isinstance(amount_cents, int) or amount_cents < 0:
is_valid = False
# 'view' requires no specific checks on amount_cents
if not is_valid:
invalid_lines_count += 1
continue
# --- Duplication Check & Processing ---
if event_id in valid_event_ids:
duplicate_events_count += 1
continue
# Event is valid and not a duplicate, so we accept it.
valid_event_ids.add(event_id)
# Update user tracking (net_cents calculation)
if user_id not in processed_users:
processed_users[user_id] = 0
current_net = processed_users[user_id]
if event_type == "purchase":
p_amount = amount_cents
purchase_cents_total += p_amount
processed_users[user_id] += p_amount
elif event_type == "refund":
r_amount = amount_cents
refund_cents_total += r_amount
processed_users[user_id] -= r_amount
# 'view' contributes 0 to net_cents
# --- Final Calculations ---
valid_events_count = len(valid_event_ids) - duplicate_events_count
# Users count: all users who appeared in valid events, plus any user implied by the structure if needed.
# The requirement is "採用イベントに含まれる異なるuser_id数".
users_count = len(processed_users)
net_cents_total = purchase_cents_total - refund_cents_total
# Determine top_user
if not processed_users:
top_user = None
else:
# Find max net_cents. Tie-breaker: lexicographically smallest user_id.
max_net = -float('inf')
best_user = None
for uid, net in processed_users.items():
if net > max_net:
max_net = net
best_user = uid
elif net == max_net:
# Tie-breaker: smaller user_id lexicographically
if best_user is None or uid < best_user:
best_user = uid
top_user = best_user
result = {
"valid_events": valid_events_count,
"invalid_lines": invalid_lines_count,
"duplicate_events": duplicate_events_count,
"users": users_count,
"purchase_cents": purchase_cents_total,
"refund_cents": refund_cents_total,
"net_cents": net_cents_total,
"top_user": top_user
}
print(json.dumps(result))
if __name__ == "__main__":
process_events()
Granite 4.1 3B(6/12)の生出力
```python
import sys
import json
def process_events():
seen_event_ids = set()
invalid_lines = 0
duplicate_events = 0
users_set = set()
purchase_cents_total = 0
refund_cents_total = 0
top_user = None
max_net_cents = -1
for line in sys.stdin:
try:
event = json.loads(line.strip())
except (json.JSONDecodeError, ValueError):
invalid_lines += 1
continue
if not all(key in event and isinstance(event[key], str) and event[key] != "" for key in ('event_id', 'user_id')):
invalid_lines += 1
continue
if event['event_type'] not in {'view', 'purchase', 'refund'}:
invalid_lines += 1
continue
if event['event_type'] in {'purchase', 'refund'} and (
'amount_cents' not in event or
not isinstance(event['amount_cents'], int) or
event['amount_cents'] < 0):
invalid_lines += 1
continue
if event['event_id'] in seen_event_ids:
duplicate_events += 1
continue
seen_event_ids.add(event['event_id'])
users_set.add(event['user_id'])
if event['event_type'] == 'purchase':
purchase_cents_total += event['amount_cents']
elif event['event_type'] == 'refund':
refund_cents_total += event['amount_cents']
valid_events = len(seen_event_ids)
users = len(users_set)
net_cents = purchase_cents_total - refund_cents_total
if valid_events > 0:
user_net_cents = {user_id: net_cents for user_id in users_set}
top_user = max(user_net_cents, key=user_net_cents.get) if user_net_cents else None
else:
top_user = None
return {
"valid_events": valid_events,
"invalid_lines": invalid_lines,
"duplicate_events": duplicate_events,
"users": users,
"purchase_cents": purchase_cents_total,
"refund_cents": refund_cents_total,
"net_cents": net_cents,
"top_user": top_user
}
result = process_events()
print(json.dumps(result))
```
Qwen3.5 4B(0/12)の生出力
import json, sys
def solve():
seen_ids = set()
valid_events_count = 0
invalid_lines = 0
duplicate_events = 0
purchase_cents_sum = 0
refund_cents_sum = 0
user_net_cents = {}
for line in sys.stdin:
stripped_line = line.strip()
if not stripped_line:
invalid_lines += 1
continue
try:
event_data = json.loads(stripped_line)
except json.JSONDecodeError:
invalid_lines += 1
continue
if not isinstance(event_data, dict):
invalid_lines += 1
continue
event_id = event_data.get('event_id')
user_id = event_data.get('user_id')
event_type = event_data.get('event_type')
if not isinstance(event_id, str) or not isinstance(user_id, str):
invalid_lines += 1
continue
if len(event_id.strip()) == 0:
invalid_lines += 1
continue
if len(user_id.strip()) == 0:
invalid_lines += 1
continue
valid_types = {'view', 'purchase', 'refund'}
if event_type not in valid_types:
invalid_lines += 1
continue
amount_cents = None
is_purchase_or_refund = (event_type == 'purchase' or event_type == 'refund')
if is_purchase_or_refund and isinstance(amount_cents, bool):
invalid_lines += 1
continue
if not is_purchase_or_refund:
amount_cents = None
if not isinstance(amount_cents, int) or amount_cents < 0:
invalid_lines += 1
continue
valid_events_count += 1
if event_id in seen_ids:
duplicate_events += 1
continue
seen_ids.add(event_id)
user_net = user_net_cents.get(user_id, {'net': 0})
if is_purchase_or_refund and amount_cents is not None:
net_change = -amount_cents if event_type == 'refund' else amount_cents
# Update purchase/refund sums based on type
if event_type == 'purchase':
purchase_cents_sum += amount_cents
elif event_type == 'refund':
refund_cents_sum += amount_cents
user_net['net'] = net_change
else:
user_net['net'] = 0
# Update the dictionary with new value if it's greater or equal (for tie-breaking)
current_user_id, _ = next(iter(user_net.items()))
old_value = user_net_cents.get(current_user_id, {'net': -float('inf')})
if net_change > 0: # If the event is a purchase and adds to sum or refund subtracts from sum (negative)
pass
new_val = user_net['net']
old_val = user_net_cents.get(user_id, {'net': -float('inf')})['net'] if isinstance(old_val, dict) else 0 # Simplified logic below
users_count = len(set([u for u in seen_ids])) # This is wrong. Need to track unique user IDs properly
# Let's restart the loop with correct tracking
pass
Ministral 3 3B(0/12)の生出力
```python
import sys
from collections import defaultdict
def main():
seen_events = set()
user_net_cents = defaultdict(int)
purchase_total = 0
refund_total = 0
invalid_lines = 0
valid_count = 0
top_user = None
for line in sys.stdin:
stripped_line = line.strip()
if not stripped_line:
continue
try:
event = eval(stripped_line)
except:
invalid_lines += 1
continue
if not isinstance(event, dict):
invalid_lines += 1
continue
event_id = event.get('event_id', '')
user_id = event.get('user_id', '')
if not event_id or not user_id:
invalid_lines += 1
continue
event_type = event.get('event_type')
if event_type not in ('view', 'purchase', 'refund'):
invalid_lines += 1
continue
if event_type == 'view':
valid_count += 1
user_net_cents[user_id] += 0
elif event_type == 'purchase':
amount = event.get('amount_cents')
if not isinstance(amount, int) or amount < 0:
invalid_lines += 1
continue
purchase_total += amount
valid_count += 1
user_net_cents[user_id] += amount
elif event_type == 'refund':
amount = event.get('amount_cents')
if not isinstance(amount, int) or amount < 0:
invalid_lines += 1
continue
refund_total += amount
valid_count += 1
user_net_cents[user_id] -= amount
if event_id in seen_events:
duplicate_events = 1
else:
seen_events.add(event_id)
duplicate_events = 0
users = len(user_net_cents)
if users > 0:
top_user = max(user_net_cents.items(), key=lambda x: x[1])[0]
if top_user is None or user_net_cents[top_user] == 0:
# Check for views only
view_users = {u for u, v in user_net_cents.items() if v == 0}
if view_users:
top_user = min(view_users)
print({
"valid_events": valid_count,
"invalid_lines": invalid_lines,
"duplicate_events": sum(1 for _ in seen_events if len(seen_events) > len(set(event.get('event_id') for event in events if eval(line.strip()) is not None and isinstance(eval(line.strip()), dict)))),
"users": users,
"purchase_cents": purchase_total,
"refund_cents": refund_total,
"net_cents": purchase_total - refund_total,
"top_user": top_user
})
if __name__ == "__main__":
main()
```
Phi-4-mini 3.8B(0/12)の生出力
```python
import json
from collections import Counter, defaultdict
def main():
input_lines = [line.strip() for line in sys.stdin if line.strip()]
valid_events_count = 0
invalid_lines_count = 0
duplicate_events_count = 0
users_counter = set()
purchase_cents_total = 0
refund_cents_total = 0
event_ids = {}
user_net_cents = Counter()
for line in input_lines:
try:
event = json.loads(line)
if not all(k in event for k in ('event_id', 'user_id', 'event_type')):
invalid_lines_count += 1
continue
event_id, user_id, event_type, *rest = [str(event[k]) for k in ('event_id', 'user_id', 'event_type')]
if not all(k.isalnum() or k.isspace() for k in (event_id + user_id)):
invalid_lines_count += 1
continue
if event['event_type'] == 'view':
valid_events_count += 1
users_counter.add(user_id)
continue
elif event['event_type'] in ('purchase', 'refund'):
if not all(isinstance(event[k], int) and k != 'amount_cents' or (k == 'amount_cents' and event[k] > 0):
invalid_lines_count += 1
continue
purchase_or_refund = event['event_type']
amount_cents = event[purchase_or_refund + '_cents']
if valid_events_count > 0:
duplicate_events_count += 1
else:
valid_events_count += 1
elif not all(k in ('amount_cents',) for k in rest):
invalid_lines_count += 1
continue
event_ids.setdefault(event_id, []).append(user_id)
except json.JSONDecodeError:
invalid_lines_count += 1
unique_users = len(users_counter)
purchase_total = sum(amount * count for id_events, (amount, count) in zip(event_ids.get('purchase', []), Counter([event['user_id'] for event in input_lines if 'purchase' in json.loads(line)]))
refund_total = sum(-amount * count for id_events, (amount, count) in zip(event_ids.get('refund', []), Counter([event['user_id'] for event in input_lines if 'refund' in json.loads(line)])))
net_cents = purchase_total + refund_total
top_user = None
user_net_totals = defaultdict(int)
for id_events, (amount, count) in zip(event_ids.get('purchase', []), Counter([event['user_id'] for event in input_lines if 'purchase' in json.loads(line)])):
user_net_totals[id_events] += amount * count
top_user = max(user_net_totals.items(), key=lambda x: (x[1], -int(x[0])))[0]
print(json.dumps({
"valid_events": valid_events_count,
"invalid_lines": invalid_lines_count,
"duplicate_events": duplicate_events_count,
"users": unique_users,
"purchase_cents": purchase_total,
"refund_cents": refund_total,
"net_cents": net_cents,
"top_user": top_user
}))
if __name__ == "__main__":
main()
```
同じ0点でも、中身は「未完成」「実行時エラー」「構文エラー」とそれぞれ違います。集計の考え方以前に、実行できる形で受け取るところが最初の関門でした。
例を2件見せれば、入口を越えられるか
入力と正解の対応が具体例で伝われば、実行前に止まったモデルも変わるかもしれません。そこで、購入と閲覧の例、不正行と重複の例を1件ずつ加えました。
Gemmaは10/12まで伸びましたが、ほかはQwen 0、Ministral 0、Phi 0、Granite 6のままです。例を見せれば全員が理解しやすくなる、というほど単純ではありませんでした。
few-shotで追加したプロンプトと5モデルの生出力
共通仕様の後ろへ2件の入出力例を追加しました。プロンプトと出力は実測時の原文です。
few-shotで追加したプロンプト
# 入出力例
## 例1
入力:
```text
{"event_id":"e1","user_id":"さくら","event_type":"view"}
{"event_id":"e2","user_id":"さくら","event_type":"purchase","amount_cents":250}
```
出力:
```json
{"valid_events":2,"invalid_lines":0,"duplicate_events":0,"users":1,"purchase_cents":250,"refund_cents":0,"net_cents":250,"top_user":"さくら"}
```
## 例2
入力:
```text
not-json
{"event_id":"x","user_id":"たろう","event_type":"refund","amount_cents":30}
{"event_id":"x","user_id":"たろう","event_type":"purchase","amount_cents":999}
```
出力:
```json
{"valid_events":1,"invalid_lines":1,"duplicate_events":1,"users":1,"purchase_cents":0,"refund_cents":30,"net_cents":-30,"top_user":"たろう"}
```
Gemma 4 E4B(10/12)の生出力
import sys
import json
def is_valid_string(s):
return s is not None and s.strip() != ""
def validate_event(data):
if not isinstance(data, dict):
return False
required = ['event_id', 'user_id', 'event_type']
for key in required:
if key not in data or not is_valid_string(data[key]):
return False
e_id = data['event_id']
u_id = data['user_id']
e_type = data['event_type']
if e_type not in ["view", "purchase", "refund"]:
return False
if e_type in ["purchase", "refund"]:
if 'amount_cents' not in data:
return False
amt = data['amount_cents']
if not isinstance(amt, int) or amt < 0:
return False
elif e_type == "view":
# view is always valid if required fields are present and non-empty
pass
return True
def solution():
valid_events = []
invalid_lines = 0
seen_event_ids = set()
for line in sys.stdin:
line = line.strip()
if not line:
# Treat empty lines as invalid if they don't parse to JSON, but here we treat them as non-JSON structure for simplicity based on typical CLI behavior unless specified otherwise.
# Given the prompt "Empty lines are treated as input lines", and they aren't valid JSON objects, they fall into invalid_lines.
invalid_lines += 1
continue
try:
data = json.loads(line)
except json.JSONDecodeError:
invalid_lines += 1
continue
if validate_event(data):
e_id = data['event_id']
if e_id in seen_event_ids:
# Duplicate event, ignore it for aggregation but count as duplicate
valid_events.append(None) # Placeholder to track valid attempt that was a duplicate
continue
seen_event_ids.add(e_id)
valid_events.append(data)
else:
invalid_lines += 1
# --- Aggregation Phase ---
final_valid_events = [ev for ev in valid_events if ev is not None]
duplicate_events = len(valid_events) - len(final_valid_events)
purchase_cents = 0
refund_cents = 0
user_net_cents = {} # user_id -> net_cents
# Initialize all users found in valid events to have a score of 0 if they only viewed
all_users = set()
for event in final_valid_events:
u_id = event['user_id']
all_users.add(u_id)
if u_id not in user_net_cents:
user_net_cents[u_id] = 0
# Process events to calculate net cents and track users
for event in final_valid_events:
e_type = event['event_type']
u_id = event['user_id']
amount = 0
if e_type == 'purchase':
amount = event.get('amount_cents', 0)
purchase_cents += amount
user_net_cents[u_id] += amount
elif e_type == 'refund':
amount = event.get('amount_cents', 0)
refund_cents += amount
user_net_cents[u_id] -= amount
# Ensure users who only viewed are in the map with net_cents=0 (already handled by initialization if they appeared)
for u_id in all_users:
if u_id not in user_net_cents:
user_net_cents[u_id] = 0
# Calculate final metrics
valid_events_count = len(final_valid_events)
users_count = len(all_users)
net_cents = purchase_cents - refund_cents
top_user = None
if users_count > 0:
# Find top user based on net_cents, tie-break by lexicographically smallest user_id
max_score = -float('inf')
best_candidates = []
for u_id in sorted(user_net_cents.keys()): # Sort keys to handle tie-breaking implicitly if we iterate over sorted keys
score = user_net_cents[u_id]
if score > max_score:
max_score = score
best_candidates = [u_id]
elif score == max_score:
best_candidates.append(u_id)
# Since we iterated over sorted keys, the first one found with max_score is the lexicographically smallest among ties
if best_candidates:
top_user = best_candidates[0]
result = {
"valid_events": valid_events_count,
"invalid_lines": invalid_lines,
"duplicate_events": duplicate_events,
"users": users_count,
"purchase_cents": purchase_cents,
"refund_cents": refund_cents,
"net_cents": net_cents,
"top_user": top_user
}
print(json.dumps(result))
if __name__ == "__main__":
solution()
Granite 4.1 3B(6/12)の生出力
```python
import sys
import json
def process_input():
seen_event_ids = set()
valid_events = 0
invalid_lines = 0
duplicate_events = 0
users_set = set()
purchase_cents_total = 0
refund_cents_total = 0
top_user = None
max_net_cents = float('-inf')
for line in sys.stdin:
try:
event = json.loads(line.strip())
if not all(key in event and isinstance(event[key], str) and event[key] != ""
for key in ["event_id", "user_id"]):
invalid_lines += 1
continue
if event['event_type'] not in {"view", "purchase", "refund"}:
invalid_lines += 1
continue
if event['event_type'] == "purchase" and not isinstance(event.get('amount_cents'), int) or \
(event['event_type'] == "refund" and not isinstance(event.get('amount_cents'), int)):
invalid_lines += 1
continue
if event['event_id'] in seen_event_ids:
duplicate_events += 1
continue
seen_event_ids.add(event['event_id'])
valid_events += 1
users_set.add(event['user_id'])
if event['event_type'] == "purchase":
purchase_cents_total += event['amount_cents']
elif event['event_type'] == "refund":
refund_cents_total += event['amount_cents']
except (json.JSONDecodeError, TypeError):
invalid_lines += 1
users = len(users_set)
net_cents = purchase_cents_total - refund_cents_total
if valid_events > 0:
user_net_cents = {user: net_cents for user in users_set}
top_user = min(user_net_cents, key=user_net_cents.get) if user_net_cents else None
else:
top_user = None
return {
"valid_events": valid_events,
"invalid_lines": invalid_lines,
"duplicate_events": duplicate_events,
"users": users,
"purchase_cents": purchase_cents_total,
"refund_cents": refund_cents_total,
"net_cents": net_cents,
"top_user": top_user
}
result = process_input()
print(json.dumps(result))
```
Qwen3.5 4B(0/12)の生出力
import json, sys
seen_ids = set()
valid_count = 0
invalid_lines = 0
duplicate_events = 0
purchase_total = 0
refund_total = 0
user_net_cents = {}
for line in sys.stdin:
if not line.strip():
continue
try:
obj = json.loads(line)
except json.JSONDecodeError:
invalid_lines += 1
continue
if not isinstance(obj, dict):
invalid_lines += 1
continue
event_id = obj.get("event_id")
user_id = obj.get("user_id")
event_type = obj.get("event_type")
if not isinstance(event_id, str) or not isinstance(user_id, str) or not isinstance(event_type, str):
invalid_lines += 1
continue
if len(event_id.strip()) == 0:
invalid_lines += 1
continue
valid_types = {"view", "purchase", "refund"}
if event_type not in valid_types:
invalid_lines += 1
continue
amount_cents = None
if event_type in ("purchase", "refund"):
value = obj.get("amount_cents")
if isinstance(value, bool):
pass # Invalid per spec (bool is not allowed) -> treat as invalid line?
# Spec says: 'value must be non-negative integer. Boolean values are not valid.'
# This implies the presence of a boolean where an int is expected makes it invalid.
elif isinstance(value, bool):
pass
if value == True or value == False:
# The spec says "Boolean values are not allowed" for amount_cents in purchase/refund context?
# Actually re-reading: 'purchase and refund have amount_cents. Value is non-negative integer. Boolean values are invalid.'
# This likely means if the field exists but is a bool, it's an error line. Or does it mean we ignore lines where type doesn't match expected int?
# Given "Boolean values are not valid" usually implies strict validation failure for that specific field value.
pass
elif isinstance(value, (int, float)):
if value < 0:
invalid_lines += 1
continue
amount_cents = int(value)
is_valid_event = True
# Check event_id uniqueness logic applies only to valid events?
# "Valid events are targeted... If event_id already adopted, ignore as duplicate."
if event_id in seen_ids:
invalid_lines += 1 # Wait, spec says "Invalid lines" count. Is a duplicate an invalid line or just ignored?
# Spec: "Duplicate of valid event -> Ignore".
# But output has `duplicate_events`. And `invalid_lines` is for "malformed/invalid rows".
# Usually duplicates are counted separately from malformed input, but here the spec says:
# "Invalid lines" count. Then later "Valid events... If already adopted ignore as duplicate."
# Example 2 shows:
# Line 1 (not-json) -> invalid_lines=1
# Line 2 (valid e-x refund x30) -> valid, added? No wait example says duplicate_events=1.
# Let's trace Ex 2 carefully.
# Input:
# not-json -> Invalid line. invalid_lines++
# {"event_id":"x", ... "refund" 30} -> Valid event (assuming x is valid string). Added to seen_ids?
# Wait, example output says duplicate_events=1 and purchase_cents=0, refund_cents=30.
# So the first 'x' was added as a REFUND of 30. Net = -30. Top user = たろう (net=-30).
# Line 3: {"event_id":"x", ... "purchase" 999} -> Valid event, but ID 'x' already in seen_ids.
# So this is a duplicate. It should NOT be added to valid_events count?
# Or does it increment invalid_lines? The spec says "Invalid lines... malformed/invalid rows".
# Usually duplicates are not "malformed", they are just ignored from the set of unique events.
# However, looking at Ex 2 output: valid_events=1 (only one event counted), duplicate_events=1.
# So duplicates do NOT increment invalid_lines. They increment duplicate_events counter but don't add to stats?
is_valid_event = False
if not seen_ids.add(event_id):
continue
if isinstance(obj, dict) and "event_type" in obj:
pass
valid_count += 1 # Only if it passed the initial checks AND wasn't a duplicate (handled by 'continue' above? No wait.)
# Re-evaluating logic flow based on Example 2:
# Line 3 is processed. It has event_id "x".
# If we check `if event_id in seen_ids` BEFORE adding, then it's a duplicate.
# The spec says: "If already adopted -> ignore as duplicate."
# So if duplicate, do NOT increment valid_count? Yes. And DO NOT add to stats (purchase/refund).
# But wait, Example 2 output has `duplicate_events=1`. This implies we counted it separately from valid events.
# Let's refine the loop:
pass
# Refined Logic Implementation
seen_ids = set()
valid_count = 0
invalid_lines = 0
purchase_total = 0
refund_total = 0
user_net_cents = {}
for line in sys.stdin:
stripped_line = line.strip()
# Empty lines are handled as input rows but likely invalid JSON or empty string?
# Spec: "Empty lines also treated as input lines." -> If we try to parse "", it fails. So they go to invalid_lines.
if not stripped_line:
invalid_lines += 1
continue
try:
obj = json.loads(stripped_line)
except json.JSONDecodeError:
invalid_lines += 1
continue
# Check if it's a JSON object (dict), otherwise invalid line?
# Spec: "JSON lines... valid event is JSON object". If not dict, then invalid.
if not isinstance(obj, dict):
invalid_lines += 1
continue
event_id = obj.get("event_id")
user_id = obj.get("user_id")
event_type = obj.get("event_type")
# Validate types and non-empty string for IDs/Type
if not isinstance(event_id, str) or len(event_id.strip()) == 0:
invalid_lines += 1
continue
if not isinstance(user_id, str):
invalid_lines += 1
continue
valid_types = {"view", "purchase", "refund"}
if event_type is None or event_type not in valid_types:
# Spec says event_type must be one of those. If missing?
# Usually implies validation failure -> Invalid line.
invalid_lines += 1
continue
amount_cents = None
has_amount_field = False
if event_type == "view":
pass # Ignore amount_cents presence/value
elif event_type in ("purchase", "refund"):
value = obj.get("amount_cents")
# Spec: Value must be non-negative integer. Boolean values are invalid.
# If the field exists but is bool, it's an error? Or if we expect int and get something else?
# The phrasing "Boolean values are not valid" suggests strict type check on amount_cents value itself.
if isinstance(value, bool):
pass # Treat as invalid line because boolean is explicitly forbidden
elif isinstance(value, (int, float)):
has_amount_field = True
int_val = int(value)
if int_val < 0:
invalid_lines += 1
continue
amount_cents = int_val
# Now check for duplicate among VALID events.
# A line is a valid event candidate only if it passed all above checks (including type and non-empty strings).
if event_id in seen_ids:
duplicate_events += 1
continue
# It's a new unique valid event
valid_count += 1
user_net_cents[user_id] = 0
if amount_cents is not None:
purchase_total += int_val
if event_type == "purchase":
pass
elif event_type == "refund":
refund_total -= int_val # Wait, net_cents = purchase - refund. So we subtract refunds from total? Or track separately?
# Spec says `net_cents`: purchase_cents - refund_cents.
# And top_user logic uses user's own net cents (sum of their purchases minus sum of their refunds).
if event_type == "purchase":
pass
elif event_type == "refund":
pass
# Wait, I need to calculate per-user stats correctly inside the loop.
# Let's restart the logic structure cleanly in code block below.
seen_ids = set()
valid_count = 0
invalid_lines = 0
purchase_total = 0
refund_total = 0
user_net_cents = {} # user_id -> net cents (sum of purchases - sum of refunds)
for line in sys.stdin:
stripped_line = line.strip()
if not stripped_line:
invalid_lines += 1
continue
try:
obj = json.loads(stripped_line)
except json.JSONDecodeError:
invalid_lines += 1
continue
if not isinstance(obj, dict):
invalid_lines += 1
continue
event_id = obj.get("event_id")
user_id = obj.get("user_id")
event_type = obj.get("event_type")
# Validate IDs and Type presence/type
if not isinstance(event_id, str) or len(event_id.strip()) == 0:
invalid_lines += 1
continue
if not isinstance(user_id, str):
invalid_lines += 1
continue
valid_types = {"view", "purchase", "refund"}
if event_type is None or event_type not in valid_types:
# If type is missing or unknown -> Invalid line?
# Spec says event_type must be one of those. Missing means invalid.
invalid_lines += 1
continue
amount_cents = None
if event_type == "view":
pass
elif event_type in ("purchase", "refund"):
value = obj.get("amount_cents")
# Check for boolean (invalid) or non-integer/non-negative integer
if isinstance(value, bool):
invalid_lines += 1
continue
if not isinstance(value, int):
# If it's a float? Spec says "non-negative integer". Float is technically not an integer in Python unless converted.
# But usually JSON numbers are parsed as int or float.
# Let's assume strict: must be int type and >= 0.
invalid_lines += 1
continue
if value < 0:
invalid_lines += 1
continue
amount_cents = value
# Check duplicate among valid events (those that passed all checks above)
if event_id in seen_ids:
duplicate_events += 1
continue
# Add to stats
purchase_total += int(amount_cents) if isinstance(obj.get("amount_cents"), bool) else amount_cents
# Wait, I already checked for boolean and invalid lines. So here it's safe?
# Actually simpler: just use the validated `amount_cents` variable which is None or valid int.
purchase_total += 0 if event_type != "purchase" else (int(amount_cents) if amount_cents is not None else 0)
refund_total += 0 if event_type != "refund" else (int(amount_cents) if amount_cents is not None else 0)
# Wait, I need to track per-user net cents for top_user calculation.
# Let's restructure the loop properly in one go without comments in final output.
seen_ids = set()
valid_count = 0
invalid_lines = 0
purchase_total = 0
refund_total = 0
user_net_cents = {} # user_id -> net cents (sum purchase - sum refund)
for line in sys.stdin:
stripped_line = line.strip()
if not stripped_line:
invalid_lines += 1
continue
try:
obj = json.loads(stripped_line)
except json.JSONDecodeError:
invalid_lines += 1
continue
if not isinstance(obj, dict):
invalid_lines += 1
continue
event_id = obj.get("event_id")
user_id = obj.get("user_id")
event_type = obj.get("event_type")
# Validate IDs and Type
if not isinstance(event_id, str) or len(event_id.strip()) == 0:
invalid_lines += 1
continue
if not isinstance(user_id, str):
invalid_lines += 1
continue
valid_types = {"view", "purchase", "refund"}
if event_type is None or event_type not in valid_types:
invalid_lines += 1
continue
amount_cents = None
# Check amount_cents validity for purchase/refund only
if event_type == "view":
pass
elif event_type in ("purchase", "refund"):
value = obj.get("amount_cents")
# Boolean is invalid per spec
if isinstance(value, bool):
invalid_lines += 1
continue
# Must be non-negative integer. Floats are not integers?
# In Python json.loads returns float for JSON numbers like 30.5 or int for 30.
# Spec says "non-negative integer". So if it's a float, is it invalid? Yes.
if isinstance(value, bool): pass # Already handled
elif not isinstance(value, (int, float)):
invalid_lines += 1
continue
else:
int_val = value
if isinstance(int_val, float) and not int_val.is_integer():
invalid_lines += 1
continue
if int_val < 0:
invalid_lines += 1
continue
amount_cents = int_val
# Check duplicate among valid events (passed all checks above)
if event_id in seen_ids:
duplicate_events += 1
continue
# It is a new unique valid event
valid_count += 1
user_net_cents[user_id] = 0
if amount_cents is not None and isinstance(amount_cents, int):
if event_type == "purchase":
purchase_total += amount_cents
elif event_type == "refund":
refund_total -= amount_cents # Wait, net_cents definition: purchase - refund.
# So total_refund should be subtracted from total_purchase to get net?
# Or track separately and compute at end?
# Spec says `purchase_cents`: sum of purchased amounts.
# `refund_cents`: sum of refunded amounts.
# `net_cents`: purchase - refund.
# So I should add positive for purchase, negative for refund to user_net?
# Yes: net = sum(purchase) - sum(refund).
pass
if event_type == "purchase":
pass
elif event_type == "refund":
pass
# Wait, the logic inside loop needs to update per-user stats correctly.
# Let's rewrite the core accumulation part clearly.
seen_ids = set()
valid_count = 0
invalid_lines = 0
purchase_total = 0
refund_total = 0
user_net_cents = {} # user_id -> net cents (sum purchase - sum refund)
for line in sys.stdin:
stripped_line = line.strip()
if not stripped_line:
invalid_lines += 1
continue
try:
obj = json.loads(stripped_line)
except json.JSONDecodeError:
invalid_lines += 1
continue
if not isinstance(obj, dict):
invalid_lines += 1
continue
event_id = obj.get("event_id")
user_id = obj.get("user_id")
event_type = obj.get("event_type")
# Validate IDs and Type
if not isinstance(event_id, str) or len(event_id.strip()) == 0:
invalid_lines += 1
continue
if not isinstance(user_id, str):
invalid_lines += 1
continue
valid_types = {"view", "purchase", "refund"}
if event_type is None or event_type not in valid_types:
invalid_lines += 1
continue
amount_cents = None
# Check amount_cents validity for purchase/refund only
if event_type == "view":
pass
elif event_type in ("purchase", "refund"):
value = obj.get("amount_cents")
# Boolean is invalid per spec
if isinstance(value, bool):
invalid_lines += 1
continue
# Must be non-negative integer. Floats are not integers?
# If JSON has 30.5, it's a float. Spec says "integer". So invalid.
if isinstance(value, (int, float)):
int_val = value
if isinstance(int_val, bool): pass # Already handled
elif isinstance(int_val, float) and not int_val.is_integer():
invalid_lines += 1
continue
else:
if int_val < 0:
invalid_lines += 1
continue
amount_cents = int_val
# Check duplicate among valid events (passed all checks above)
if event_id in seen_ids:
duplicate_events += 1
continue
# It is a new unique valid event
valid_count += 1
user_net_cents[user_id] = 0
if amount_cents is not None and isinstance(amount_cents, int):
if event_type == "purchase":
purchase_total += amount_cents
user_net_cents[user_id] += amount_cents
elif event_type == "refund":
refund_total -= amount_cents # Wait, spec says `net_cents` = purchase - refund.
# So total_refund is the sum of refunds? Yes.
# And net_cents variable in output is just a derived value.
# But for top_user calculation: "User specific net cents".
# User's net = user_purchases - user_refunds.
pass
if event_type == "purchase":
purchase_total += amount_cents
user_net_cents[user_id] += amount_cents
elif event_type == "refund":
refund_total -= amount_cents # Wait, I'm subtracting from total? No, just sum the refunds.
# Let's track separately: `total_refunds` = sum of all refunds.
# Then net = purchase - refund.
# Correct accumulation logic:
seen_ids = set()
valid_count = 0
invalid_lines = 0
purchase_total = 0
refund_total = 0
user_net_cents = {}
for line in sys.stdin:
stripped_line = line.strip()
if not stripped_line:
invalid_lines += 1
continue
try:
obj = json.loads(stripped_line)
except json.JSONDecodeError:
invalid_lines += 1
continue
if not isinstance(obj, dict):
invalid_lines += 1
continue
event_id = obj.get("event_id")
user_id = obj.get("user_id")
event_type = obj.get("event_type")
# Validate IDs and Type
if not isinstance(event_id, str) or len(event_id.strip()) == 0:
invalid_lines += 1
continue
if not isinstance(user_id, str):
invalid_lines += 1
continue
valid_types = {"view", "purchase", "refund"}
if event_type is None or event_type not in valid_types:
invalid_lines += 1
continue
amount_cents = None
# Check amount_cents validity for purchase/refund only
if event_type == "view":
pass
elif event_type in ("purchase", "refund"):
value = obj.get("amount_cents")
# Boolean is invalid per spec
if isinstance(value, bool):
invalid_lines += 1
continue
# Must be non-negative integer. Floats are not integers?
if isinstance(value, (int, float)):
int_val = value
elif isinstance(int_val, float) and not int_val.is_integer():
invalid_lines += 1
continue
else:
if int_val < 0:
invalid_lines += 1
continue
amount_cents = int_val
# Check duplicate among valid events (passed all checks above)
if event_id in seen_ids:
duplicate_events += 1
continue
# It is a new unique valid event
valid_count += 1
user_net_cents[user_id] = 0
if amount_cents is not None and isinstance(amount_cents, int):
if event_type == "purchase":
purchase_total += amount_cents
user_net_cents[user_id] += amount_cents
elif event_type == "refund":
refund_total -= amount_cents # Wait, I'm subtracting from total? No.
# Let's track separately: `total_refunds` = sum of all refunds.
# Then net = purchase - refund.
if event_type == "purchase":
pass
elif event_type == "refund":
pass
# Final clean logic implementation in code block below.
Ministral 3 3B(0/12)の生出力
```python
import sys
from collections import defaultdict
def main():
event_set = set()
user_net = defaultdict(int)
invalid_lines = 0
valid_events = 0
purchase_total = 0
refund_total = 0
for line in sys.stdin:
if not line.strip():
continue
try:
event = eval(line.strip())
except (SyntaxError, NameError):
invalid_lines += 1
continue
if not isinstance(event, dict) or "event_id" not in event or "user_id" not in event or "event_type" not in event:
invalid_lines += 1
continue
event_id = event["event_id"]
user_id = event["user_id"]
event_type = event["event_type"]
if not event_id.strip() or not user_id.strip():
invalid_lines += 1
continue
if event_type == "view":
valid_events += 1
if user_id not in event_set:
event_set.add(user_id)
elif event_type in ("purchase", "refund"):
if event_id in event_set:
continue
else:
event_set.add(event_id)
try:
amount = event["amount_cents"]
except KeyError:
invalid_lines += 1
continue
if not isinstance(amount, int) or amount < 0:
invalid_lines += 1
continue
valid_events += 1
user_net[user_id] += amount if event_type == "purchase" else -amount
if event_type == "refund":
refund_total += amount
purchase_total = sum(amount for net in user_net.values() if net > 0)
net_cents = purchase_total - refund_total
top_user = max(user_net.items(), key=lambda x: x[1], reverse=True)[0] if user_net else None
print({
"valid_events": valid_events,
"invalid_lines": invalid_lines,
"duplicate_events": len(event_set) - (len({user for user in event_set}) if isinstance(event_set, set) else 0),
"users": len(set(user_id for _, user_id in [event["user_id"] for event in sys.stdin if eval(line.strip())])),
"purchase_cents": purchase_total,
"refund_cents": refund_total,
"net_cents": net_cents,
"top_user": top_user
})
if __name__ == "__main__":
main()
```
Phi-4-mini 3.8B(0/12)の生出力
以下は、指定された仕様に従って `solution.py` を作成するためのPythonコードです。
```python
import json
from collections import defaultdict
def main():
events = 0
invalid_lines = 0
duplicate_events = 0
users_count = set()
purchase_cents_total = 0
refund_cents_total = 0
previous_event_id = None
for line in iter(input, ''):
try:
event = json.loads(line)
if not isinstance(event, dict) or 'event_type' not in event or ('amount_cents' in event and event['event_type'] == 'view'):
invalid_lines += 1
continue
event_id = event.get('event_id')
user_id = event.get('user_id')
event_type = event.get('event_type')
if (not isinstance(event_id, str) or not event_id.strip() or
not isinstance(user_id, str) or not user_id.strip()):
invalid_lines += 1
continue
if previous_event_id == event_id:
duplicate_events += 1
continue
users_count.add(user_id)
if event_type in ['purchase', 'refund']:
purchase_cents_total += event.get('amount_cents', 0)
previous_event_id = event_id
except json.JSONDecodeError:
invalid_lines += 1
net_cents = purchase_cents_total - refund_cents_total
top_user = None if not users_count else min(users_count, key=lambda uid: (purchase_cents_total + refund_cents_total) == 0 or (purchase_cents_total[uid] >= refund_cents_total.get(uid, float('-inf')))
print(json.dumps({
"valid_events": events,
"invalid_lines": invalid_lines,
"duplicate_events": duplicate_events,
"users": len(users_count),
"purchase_cents": purchase_cents_total,
"refund_cents": refund_cents_total,
"net_cents": net_cents,
"top_user": top_user
}))
if __name__ == "__main__":
main()
```
このコードは、標準入力からJSON Linesを読み取り、有効なイベントの集計結果を1行だけ出力するように設計されています。`json.loads()` を使用して各ラインを解析し、それが有効かどうかをチェックします。その後、指定されたキーと値でデータ構造体を更新しながら集計処理を進めます。最終的な結果は `json.dumps()` でJSONオブジェクトに変換されて標準出力として返されます。
生成速度は約91〜132トークン毎秒。もっとも速かったMinistralが0点で、もっとも遅かったGemmaが最高点でした。今回の点数を分けたのは、生成の速さではなく、コードを実行できる形までまとめられたかどうかです。
仕様を、実装する順番に並べてみた
0点になった出力を見ると、ルールを知らないというより、検証、重複判定、集計を組み立てる途中で止まっています。元のプロンプトには必要な条件をすべて書きましたが、どの順番でコードにするかはモデルへ任せていました。
そこで、同じ仕様を実装する順番に並べ、最後に未完成のコードを検査する指示も加えます。期待するのは、モデルが設計で迷う範囲を減らし、最後まで実行できるコードを書きやすくすることです。記事の円表記に合わせて縮めると、渡した設計はこうなります。
1. 標準入力を一度だけ読み、1行ずつJSONへ変換する
2. オブジェクト、ID、イベント種別、金額の順に検証する
3. 金額は int かつ bool ではなく、0以上の場合だけ受け付ける
4. 検証に通ってから重複を調べる。不正な行のIDは記録しない
5. viewは0、purchaseは加算、refundは減算してユーザー別金額を更新する
6. 返金後の金額が最大のユーザーを選び、同額ならuser_idの辞書順にする
7. 指定された8項目のJSONだけを1行で出力する
8. 空入力、未定義変数、途中のpass、標準入力の再読込を受入検査で落とす
実装順プロンプトの原文
共通仕様の後ろへ追加した部分です。本文の8項目を省略せず、そのまま掲載しています。
追加した実装設計を確認する
# 実装設計
長い検討や書き直しを出力せず、次のデータフローに沿って完成コードを一度だけ書いてください。
1. `seen_event_ids`、`user_net`、3種類のカウンター、購入・返金合計を初期化する。
2. `sys.stdin.read().splitlines()`で入力を一度だけ読み、各行を`json.loads`する。
3. JSONオブジェクト、非空白の文字列ID、イベント種別、金額の順に検証する。
- 金額は`isinstance(value, int) and not isinstance(value, bool) and value >= 0`
4. 検証合格後に`event_id`の重複を判定する。不正イベントのIDは記録しない。
5. 採用時にユーザーを`user_net`へ登録する。viewは0、purchaseは加算、refundは減算する。
6. トップユーザーは`min(user_net, key=lambda u: (-user_net[u], u))`で決める。ユーザーがいなければ`None`。
7. 指定された8キーの辞書を`json.dumps`で必ず1行出力する。
8. `if __name__ == "__main__":`から処理を呼び出す。
受入条件:
- 空入力でも正常終了してJSONを出力する。
- 標準入力を二度読まない。
- 未定義変数、途中の`pass`、複数バージョンのコードを残さない。
- Pythonの構文が成立し、説明文を含まない。
とくに、検証してから重複を見る順番と、Pythonでは整数扱いされるboolの除外は、条件だけでなく判定方法まで書きました。
ここで結果が大きく動きました。
| モデル | 素の指示 | 短い実装設計 |
|---|---|---|
| Qwen3.5 4B | 0/12 | 0/12 |
| Gemma 4 E4B | 8/12 | 11/12 |
| Ministral 3 3B | 0/12 | 12/12 |
| Phi-4-mini 3.8B | 0/12 | 0/12 |
| Granite 4.1 3B | 6/12 | 1/12 |
実装順プロンプトに対する5モデルの生出力
モデル、生成条件、共通仕様は変えず、上の実装設計だけを追加したときの回答です。
Gemma 4 E4B(11/12)の生出力
import sys
import json
def solution():
seen_event_ids = set()
user_net = {}
valid_events = 0
invalid_lines = 0
duplicate_events = 0
purchase_cents = 0
refund_cents = 0
for line in sys.stdin:
line = line.strip()
if not line and not seen_event_ids: # Handle empty input case gracefully if needed, though loop handles it
continue
try:
data = json.loads(line)
except json.JSONDecodeError:
invalid_lines += 1
continue
if not isinstance(data, dict):
invalid_lines += 1
continue
event_id = data.get("event_id")
user_id = data.get("user_id")
event_type = data.get("event_type")
amount_cents = data.get("amount_cents")
# Validation checks
is_valid = True
if not isinstance(event_id, str) or not event_id.strip():
is_valid = False
elif not isinstance(user_id, str) or not user_id.strip():
is_valid = False
elif event_type not in ["view", "purchase", "refund"]:
is_valid = False
else:
if event_type in ["purchase", "refund"]:
if not (isinstance(amount_cents, int) and not isinstance(amount_cents, bool) and amount_cents >= 0):
is_valid = False
if not is_valid:
invalid_lines += 1
continue
# Check for duplicates among valid events
if event_id in seen_event_ids:
duplicate_events += 1
continue
seen_event_ids.add(event_id)
valid_events += 1
# Update user net and totals
user_net[user_id] = user_net.get(user_id, 0)
if event_type == "purchase":
amount = amount_cents
user_net[user_id] += amount
purchase_cents += amount
elif event_type == "refund":
amount = amount_cents
user_net[user_id] -= amount
refund_cents += amount
# Calculate net cents and top user
net_cents = purchase_cents - refund_cents
if not user_net:
top_user = None
else:
# Key for sorting: (-net_cents, user_id) ensures max net_cents first, then min user_id lexicographically
sorted_users = sorted(user_net.keys(), key=lambda u: (-user_net[u], u))
top_user = sorted_users[0]
result = {
"valid_events": valid_events,
"invalid_lines": invalid_lines,
"duplicate_events": duplicate_events,
"users": len(user_net),
"purchase_cents": purchase_cents,
"refund_cents": refund_cents,
"net_cents": net_cents,
"top_user": top_user
}
print(json.dumps(result))
if __name__ == "__main__":
solution()
Granite 4.1 3B(1/12)の生出力
```python
import sys
import json
def main():
seen_event_ids = set()
user_net = {}
valid_events = 0
invalid_lines = 0
duplicate_events = 0
users = set()
purchase_cents = 0
refund_cents = 0
top_user = None
for line in sys.stdin.read().splitlines():
try:
event = json.loads(line)
if not all(key in event and isinstance(event[key], str) and event[key].strip() for key in ['event_id', 'user_id']):
invalid_lines += 1
continue
event_id = event['event_id']
user_id = event['user_id']
if event_id in seen_event_ids:
duplicate_events += 1
continue
seen_event_ids.add(event_id)
valid_events += 1
users.add(user_id)
if event['event_type'] == 'view':
continue
amount_cents = event.get('amount_cents')
if not (isinstance(amount_cents, int) and not isinstance(amount_cents, bool) and amount_cents >= 0):
invalid_lines += 1
continue
user_net[user_id] = user_net.get(user_id, 0) + (amount_cents if event['event_type'] == 'purchase' else -amount_cents)
except (json.JSONDecodeError, KeyError):
invalid_lines += 1
top_user_candidates = [(user, net) for user, net in user_net.items()]
if top_user_candidates:
top_user = min(top_user_candidates, key=lambda x: (-x[1], x[0]))[0]
result = {
"valid_events": valid_events,
"invalid_lines": invalid_lines,
"duplicate_events": duplicate_events,
"users": len(users),
"purchase_cents": purchase_cents,
"refund_cents": refund_cents,
"net_cents": purchase_cents - refund_cents,
"top_user": top_user if valid_events > 0 else None
}
print(json.dumps(result))
if __name__ == "__main__":
main()
```
Qwen3.5 4B(0/12)の生出力
import json, sys
def main():
lines = sys.stdin.read().splitlines()
seen_event_ids = set()
user_net = {}
valid_events_count = 0
invalid_lines_count = 0
duplicate_events_count = 0
purchase_cents_sum = 0
refund_cents_sum = 0
for line in lines:
if not line.strip():
continue
try:
event_data = json.loads(line)
except (json.JSONDecodeError, TypeError):
invalid_lines_count += 1
continue
if not isinstance(event_data, dict):
invalid_lines_count += 1
continue
for key in ['event_id', 'user_id', 'event_type']:
if key not in event_data:
invalid_lines_count += 1
break
event_id = str(event_data['event_id'])
user_id = str(event_data['user_id'])
if all(c.isspace() for c in event_id):
invalid_lines_count += 1
continue
valid_event_type = False
amount_cents_value = None
if 'amount_cents' not in event_data:
pass # view events don't need it, purchase/refund must have it but value check is specific to type logic below? Wait spec says "purchase and refund are ... int". Let's re-read carefully.
# Re-reading validation rules for amount_cents:
# - `purchase`と`refund`は`amount_cents`を持つ。値は0以上の整数で、真偽値は不可
# - `view`では`amount_cents`の有無や値を検証せず無視する
if event_data['event_type'] == 'view':
pass
elif not isinstance(event_data.get('amount_cents'), int) or isinstance(event_data.get('amount_cents'), bool):
invalid_lines_count += 1
continue
amount = event_data.get('amount_cents', None) if event_data['event_type'] in ['purchase', 'refund'] else None
# Check value >= 0 for purchase/refund specifically? The spec says "値は0以上の整数".
# If it's a view, we ignore the presence/value.
# So if type is purchase or refund, amount must be int and not bool and >= 0.
if event_data['event_type'] in ['purchase', 'refund']:
amt = event_data.get('amount_cents')
if isinstance(amt, bool) or (isinstance(amt, int) and amt < 0):
invalid_lines_count += 1
continue
# Duplicate check
if event_id not in seen_event_ids:
seen_event_ids.add(event_id)
valid_events_count += 1
user_net[user_id] = user_net.get(user_id, 0)
etype = str(event_data['event_type'])
amt_val = None
if etype == 'view':
pass # net change 0
elif etype == 'purchase' and isinstance(amt_val, int):
purchase_cents_sum += amt_val
user_net[user_id] += amt_val
elif etype == 'refund' and isinstance(amt_val, int):
refund_cents_sum -= amt_val # Wait logic: net = purchase - refund.
# If I add to a negative number for refunds? Or subtract from total?
# Spec says: `purchase_cents`: 採用したpurchaseの`amount_cents`合計
# `refund_cents`: 採用したrefundの`amount_cents`合計
# So purchase adds, refund adds (as positive value to sum).
# Net = Purchase - Refund.
if etype == 'view':
pass
else:
duplicate_events_count += 1
users_count = len(user_net)
net_cents_total = purchase_cents_sum - refund_cents_sum
top_user_result = None
max_val = float('-inf')
# Calculate user specific nets for ranking.
# view only users have net 0? Or just their balance from other events?
# Spec: "viewだけのユーザーもユーザー別`net_cents = 0`として候補に含める" -> This implies if a user ONLY has views, they are considered with net=0.
# But what about mixed users? The spec says `user_net` tracks the balance (purchase - refund).
# If a user only did views, their entry in user_net would be 0 initially or unchanged.
for uid, val in user_net.items():
if not isinstance(val, int): continue
current_val = val
# Wait, does "view" affect the net? Spec: `purchase` adds, `refund` subtracts (implied by net formula).
# If a user only has views, they don't add/subtract anything. So their net is 0 relative to transactions.
if not top_user_result and len(user_net) > 0:
pass
# Recalculate properly for sorting logic provided in spec hint: `min(... key=lambda u: (-user_net[u], u))`
# This finds the max user net (because of negative sign). If ties, smaller string ID.
candidates = []
if len(user_net) > 0:
for uid, val in user_net.items():
# Ensure we treat view-only users correctly?
# The spec says "viewだけのユーザーも...". This implies they should be included even if no purchase/refund.
# My `user_net` dict only gets updated on purchase/refund logic above? No, I initialized it but didn't add 0 for views explicitly unless transaction happened.
# Actually, the loop updates user_net ONLY when a valid event is processed.
# If an event is view, does it update user_net? The code block `user_net[user_id] = ...` happens regardless of type if I structure it right.
# Let's refine the logic inside the main loop to ensure all users are tracked with correct net values.
pass
# Corrected Logic Implementation Block
def solve():
lines = sys.stdin.read().splitlines()
seen_event_ids = set()
user_net = {}
valid_events_count = 0
invalid_lines_count = 0
duplicate_events_count = 0
purchase_cents_sum = 0
refund_cents_sum = 0
for line in lines:
stripped_line = line.strip()
if not stripped_line: # Empty line handling (skip but count as valid? No, spec says "空行も入力行として扱う" -> Invalid or just skip?)
# Spec: "不正行の`event_id`は採用済みにしない。" implies we need to parse.
# If empty string is not a JSON object, it's invalid line.
try:
json.loads(line)
except (json.JSONDecodeError, TypeError):
pass
if stripped_line == "":
continue
try:
event_data = json.loads(stripped_line) # Wait, spec says read lines then parse.
# Better to just do `line` directly in loop without strip for parsing? No, empty line check is needed before JSON load usually or inside.
pass
except (json.JSONDecodeError, TypeError):
invalid_lines_count += 1
continue
if not isinstance(event_data, dict):
invalid_lines_count += 1
continue
# Check required keys existence and non-whitespace strings
event_id = str(event_data.get('event_id'))
user_id = str(event_data.get('user_id'))
if all(c.isspace() for c in event_id) or not isinstance(event_id, str):
invalid_lines_count += 1
continue
# Check type existence? Spec says "valid events are JSON objects with next conditions". Implies keys must exist.
if 'event_type' not in event_data:
invalid_lines_count += 1
continue
etype = str(event_data['event_type'])
valid_types = {'view', 'purchase', 'refund'}
if etype not in valid_types:
invalid_lines_count += 1
continue
# Check amount_cents logic
amt_val = event_data.get('amount_cents')
is_purchase_refund = (etype == 'purchase' or etype == 'refund')
if is_purchase_refund:
if not isinstance(amt_val, int) or isinstance(amt_val, bool):
invalid_lines_count += 1
continue
# Check >= 0? Spec says "値は0以上の整数".
if amt_val < 0:
invalid_lines_count += 1
continue
# Duplicate check
if event_id in seen_event_ids:
duplicate_events_count += 1
continue
seen_event_ids.add(event_id)
valid_events_count += 1
user_net[user_id] = user_net.get(user_id, 0)
if is_purchase_refund and isinstance(amt_val, int):
if etype == 'purchase':
purchase_cents_sum += amt_val
user_net[user_id] += amt_val
elif etype == 'refund':
refund_cents_sum += amt_val # Accumulate positive value for subtraction later?
# Wait, net = purchase - refund. So if I add to a sum variable called `refund_cents`, then subtract that sum at end.
user_net[user_id] -= amt_val
else:
pass
users_count = len(user_net)
top_user_result = None
# Determine max net user
# Candidates are all keys in user_net?
# Spec says "viewだけのユーザーも...". If a user only has views, they appear in user_net with value 0.
# So iterating over user_net is correct.
if not top_user_result and len(user_net) > 0:
pass
max_val = float('-inf')
for uid, val in user_net.items():
current_val = val
# Logic to find max net_cents per user?
# Wait, `user_net` stores the cumulative balance. Is that what "ユーザー別net_cents" means? Yes.
if not isinstance(val, int): continue
candidates.append((val, uid))
if len(candidates) > 0:
best = max(candidates, key=lambda x: (x[0], -ord(x[1][0]) if x[1] else '')) # No simple string sort without full logic.
# Spec hint: `min(user_net, key=lambda u: (-user_net[u], u))`
best_user = None
for uid in user_net.keys():
val = user_net.get(uid)
if not isinstance(val, int): continue
candidates.append((val, uid))
# Re-implement the sorting logic exactly as hinted but applied to keys/values correctly.
# The hint suggests iterating over `user_net` items directly?
# No, `min(dict)` doesn't work like that without a key function on values.
if len(user_net) > 0:
best = max(candidates, key=lambda x: (x[0], -ord(x[1][0]) if x[1] else ''))
# Wait, string comparison is lexicographical. The hint says "辞書順で小さい方". That's standard < operator.
# So we want MAX net_cents. If tie, MIN user_id (lexicographically).
best = None
if len(candidates) > 0:
def sort_key(item):
val, uid = item
return (-val, uid)
sorted_candidates = sorted(candidates, key=sort_key) # Sort by -net desc, then id asc?
# No. We want MAX net. So we need the one with highest first element of tuple (val).
# If tie in val, smallest string ID.
# Sorting ascending: (-val, uid). The max of this list will be the one with largest -val (smallest val) -> WRONG.
# We want MAX val. So we sort by (-val, uid) and take LAST? Or use key=lambda x: (-x[0], x[1]) and find min?
# If I have vals 10, 5. -10 < -5. Min is -10 (val 10). Correct for max val.
# Tie break: uid "a", "b". We want "a" (-same, "a"<"b"). So min works.
best = sorted(candidates, key=lambda x: (-x[0], x[1]))[-1] if candidates else None
top_user_result = best[1] if isinstance(best, tuple) and len(best) > 1 else None
result_dict = {
"valid_events": valid_events_count,
"invalid_lines": invalid_lines_count,
"duplicate_events": duplicate_events_count,
"users": users_count,
"purchase_cents": purchase_cents_sum,
"refund_cents": refund_cents_sum,
"net_cents": purchase_cents_sum - refund_cents_sum,
"top_user": top_user_result if isinstance(top_user_result, str) else None # Ensure null for non-string? Spec says user_id is string. If no users, null.
}
print(json.dumps(result_dict))
if __name__ == "__main__":
solve()
Ministral 3 3B(12/12)の生出力
```python
import sys
import json
seen_event_ids = set()
user_net = {}
valid_events = 0
invalid_lines = 0
duplicate_events = 0
purchase_cents = 0
refund_cents = 0
for line in sys.stdin.read().splitlines():
if not line.strip():
invalid_lines += 1
continue
try:
event = json.loads(line)
except json.JSONDecodeError:
invalid_lines += 1
continue
if not isinstance(event, dict):
invalid_lines += 1
continue
event_id = event.get('event_id', '')
user_id = event.get('user_id', '')
event_type = event.get('event_type')
if not all([isinstance(event_id, str) and event_id.strip(), isinstance(user_id, str) and user_id.strip()]):
invalid_lines += 1
continue
if event_type not in ('view', 'purchase', 'refund'):
invalid_lines += 1
continue
amount_cents = event.get('amount_cents')
if event_type == 'purchase' or event_type == 'refund':
if not (isinstance(amount_cents, int) and not isinstance(amount_cents, bool) and amount_cents >= 0):
invalid_lines += 1
continue
if event_id in seen_event_ids:
duplicate_events += 1
else:
seen_event_ids.add(event_id)
valid_events += 1
user_net[user_id] = user_net.get(user_id, 0)
if event_type == 'view':
pass
elif event_type == 'purchase':
purchase_cents += amount_cents
user_net[user_id] += amount_cents
else: # refund
refund_cents += amount_cents
user_net[user_id] -= amount_cents
net_cents = purchase_cents - refund_cents
top_user = None
if user_net:
top_user = min(user_net, key=lambda u: (-user_net[u], u))
result = {
"valid_events": valid_events,
"invalid_lines": invalid_lines,
"duplicate_events": duplicate_events,
"users": len(user_net),
"purchase_cents": purchase_cents,
"refund_cents": refund_cents,
"net_cents": net_cents,
"top_user": top_user
}
print(json.dumps(result))
```
Phi-4-mini 3.8B(0/12)の生出力
```python
import json
import sys
from collections import defaultdict
def main():
seen_event_ids = set()
user_net = defaultdict(int)
valid_events, invalid_lines, duplicate_events, users_count = 0, 0, 0, 0
for line in sys.stdin.read().splitlines():
try:
event = json.loads(line) if line else {}
event_id = next(iter(event))
if not isinstance(event_id, str) or not event_id.strip() or event_id in seen_event_ids:
invalid_lines += 1
continue
seen_event_ids.add(event_id)
valid_events += 1
user_id = event[event_id]['user_id']
users_count = max(users_count, 1 + (user_id not in user_net))
if 'purchase' in event and all(isinstance(amount_cents, int) and amount_cents >= 0 for amount_cents in event['purchase'].values()):
purchase_amounts = [amount_cents for _, amount_cents in sorted(event['purchase'].items())]
total_purchase = sum(purchase_amounts)
user_net[user_id] += total_purchase
elif 'refund' in event and all(isinstance(amount_cents, int) and amount_cents >= 0 for amount_cents in event['refund'].values()):
refund_amounts = [amount_cents for _, amount_cents in sorted(event['refund'].items())]
total_refund = sum(refund_amounts)
user_net[user_id] -= total_refund
else:
invalid_lines += 1
except (json.JSONDecodeError, KeyError, TypeError):
invalid_lines += 1
duplicate_events = len(seen_event_ids) - valid_events
net_cents = sum(user_net.values())
top_user_id = min((user for user in sorted(user_net), key=lambda u: (-user_net[u], u)) if users_count > 0 else (None,))
result = {
"valid_events": valid_events,
"invalid_lines": invalid_lines,
"duplicate_events": duplicate_events,
"users": users_count,
"purchase_cents": sum(purchase for purchase in user_net.values() if isinstance(purchase, int)),
"refund_cents": -sum(refund for refund in user_net.values() if isinstance(-refund, int)),
"net_cents": net_cents,
"top_user": top_user_id
}
print(json.dumps(result, separators=(',', ':'))
if __name__ == "__main__":
main()
```
Ministralは0/12から12/12、Gemmaは8/12から11/12になりました。モデルも生成条件も変えていません。この2モデルでは、長い仕様へ短い処理順と自己検査を加える狙いが点数に表れました。
一方、Graniteは6/12から1/12へ下がります。1回の計測だけで原因は決められませんが、同じ指示で全モデルが改善するわけではないことは分かります。プロンプトは一律に配らず、モデルとタスクの組み合わせごとに測る必要があります。
生成コードをテストし、候補を選別してみる
プロンプトを整えても、すべてのモデルが合格するわけではありません。実際に使うなら、毎回正解が出ることを期待するより、基準に届かない生成物を利用者へ渡さない仕組みが必要です。
そこで、短い実装設計を土台に、生成、検査、再試行、採否判定までを一つのハーネスにしました。期待するのは点数の底上げではなく、事前に公開した4件のテストをすべて通る候補だけを採用することです。
- 回答からPythonコードを取り出し、構文を確認する
- 事前に公開した4件のスモークテストを実行する
- 不合格なら、検査結果と前回コードを返し、seedを変えて最大2回修正させる
- もっとも多く通った候補を残し、4件すべて通らなければ外へ出さない
再試行で追加したプロンプトと生出力
初回は前節と同じ実装順プロンプトです。不合格だった3モデルについて、検査結果を返した再試行だけを掲載します。
再試行で組み立てたプロンプトの型
{共通の課題仕様}
{実装順プロンプト}
# 前回コードの受入検査結果
{公開テストの失敗内容}
# 前回のコード
```python
{前回出力から抽出したPythonコード}
```
受入検査をすべて通るように修正し、完成したsolution.pyの内容だけを返してください。
Qwen3.5 4B・再試行1(公開検査 3/4)
[この再試行で追加したプロンプト]
# 前回コードの受入検査結果
構文検査に失敗:
Sorry: IndentationError: expected an indented block after 'for' statement on line 128 (attempt_1.py, line 137)
# 前回のコード
前の折りたたみに掲載した出力から、Python部分を抽出して追加
受入検査をすべて通るように修正し、完成したsolution.pyの内容だけを返してください。
[モデルの生出力]
import json, sys
def main():
lines = sys.stdin.read().splitlines()
seen_event_ids = set()
user_net = {}
valid_events_count = 0
invalid_lines_count = 0
duplicate_events_count = 0
purchase_cents_sum = 0
refund_cents_sum = 0
for line in lines:
if not line.strip():
continue
try:
event_data = json.loads(line)
except (json.JSONDecodeError, TypeError):
invalid_lines_count += 1
continue
if not isinstance(event_data, dict):
invalid_lines_count += 1
continue
for key in ['event_id', 'user_id']:
if key not in event_data:
invalid_lines_count += 1
break
try:
event_id = str(event_data['event_id'])
user_id = str(event_data['user_id'])
if all(c.isspace() for c in event_id):
raise ValueError("Invalid event_id")
etype_str = str(event_data.get('event_type'))
except (KeyError, TypeError, ValueError) as e:
invalid_lines_count += 1
continue
valid_types = {'view', 'purchase', 'refund'}
if not isinstance(etype_str, str):
raise ValueError("Invalid event_type")
try:
amount_cents_value = None
is_purchase_refund = False
if etype_str in ['purchase', 'refund']:
amt_val = event_data.get('amount_cents')
if not isinstance(amt_val, int) or isinstance(amt_val, bool):
raise ValueError("Invalid amount")
if amt_val < 0:
raise ValueError("Negative amount")
is_purchase_refund = True
elif etype_str == 'view':
pass # view events don't require amount_cents validation
except (KeyError, TypeError):
invalid_lines_count += 1
continue
if event_id in seen_event_ids:
duplicate_events_count += 1
continue
seen_event_ids.add(event_id)
valid_events_count += 1
user_net[user_id] = user_net.get(user_id, 0)
if is_purchase_refund and isinstance(amt_val, int):
if etype_str == 'purchase':
purchase_cents_sum += amt_val
user_net[user_id] += amt_val
elif etype_str == 'refund':
refund_cents_sum += amt_val # Accumulate positive value for subtraction later?
# Wait, net = purchase - refund. So if I add to a sum variable called `refund_cents`, then subtract that sum at end.
user_net[user_id] -= amt_val
users_count = len(user_net)
top_user_result = None
max_val = float('-inf')
for uid, val in user_net.items():
if not isinstance(val, int): continue
current_val = val
# Logic to find max net_cents per user?
# Wait, `user_net` stores the cumulative balance. Is that what "ユーザー別net_cents" means? Yes.
best_user_id = None
if len(user_net) > 0:
def sort_key(item):
val, uid = item
return (-val, uid)
sorted_candidates = sorted([(v, k) for k, v in user_net.items()], key=sort_key)
best_user_id = sorted_candidates[-1][1] if sorted_candidates else None
result_dict = {
"valid_events": valid_events_count,
"invalid_lines": invalid_lines_count,
"duplicate_events": duplicate_events_count,
"users": users_count,
"purchase_cents": purchase_cents_sum,
"refund_cents": refund_cents_sum,
"net_cents": purchase_cents_sum - refund_cents_sum,
"top_user": best_user_id if isinstance(best_user_id, str) else None
}
print(json.dumps(result_dict))
if __name__ == "__main__":
main()
Qwen3.5 4B・再試行2(公開検査 1/4)
[この再試行で追加したプロンプト]
# 前回コードの受入検査結果
ユーザー別純額: expected={"valid_events": 3, "invalid_lines": 0, "duplicate_events": 0, "users": 2, "purchase_cents": 150, "refund_cents": 80, "net_cents": 70, "top_user": "たろう"} actual={"valid_events": 3, "invalid_lines": 0, "duplicate_events": 0, "users": 2, "purchase_cents": 150, "refund_cents": 80, "net_cents": 70, "top_user": "さくら"} stderr=
# 前回のコード
前の折りたたみに掲載した出力から、Python部分を抽出して追加
受入検査をすべて通るように修正し、完成したsolution.pyの内容だけを返してください。
[モデルの生出力]
import json, sys
def main():
lines = sys.stdin.read().splitlines()
seen_event_ids = set()
user_net = {}
valid_events_count = 0
invalid_lines_count = 0
duplicate_events_count = 0
purchase_cents_sum = 0
refund_cents_sum = 0
for line in lines:
if not line.strip():
continue
try:
event_data = json.loads(line)
except (json.JSONDecodeError, TypeError):
invalid_lines_count += 1
continue
if not isinstance(event_data, dict):
invalid_lines_count += 1
continue
for key in ['event_id', 'user_id']:
if key not in event_data:
invalid_lines_count += 1
break
try:
event_id = str(event_data['event_id'])
user_id = str(event_data['user_id'])
if all(c.isspace() for c in event_id):
raise ValueError("Invalid event_id")
etype_str = str(event_data.get('event_type'))
except (KeyError, TypeError, ValueError) as e:
invalid_lines_count += 1
continue
valid_types = {'view', 'purchase', 'refund'}
if not isinstance(etype_str, str):
raise ValueError("Invalid event_type")
try:
amount_cents_value = None
is_purchase_refund = False
if etype_str in ['purchase', 'refund']:
amt_val = event_data.get('amount_cents')
if not isinstance(amt_val, int) or isinstance(amt_val, bool):
raise ValueError("Invalid amount")
if amt_val < 0:
raise ValueError("Negative amount")
is_purchase_refund = True
elif etype_str == 'view':
pass # view events don't require amount_cents validation
except (KeyError, TypeError):
invalid_lines_count += 1
continue
if event_id in seen_event_ids:
duplicate_events_count += 1
continue
seen_event_ids.add(event_id)
valid_events_count += 1
user_net[user_id] = user_net.get(user_id, 0)
amt_val = None
if is_purchase_refund and isinstance(amt_val, int):
pass # Retrieve correct value for calculation
users_count = len(user_net)
top_user_result = None
max_val = float('-inf')
best_user_id = None
if user_net:
def sort_key(item):
val, uid = item
return (-val, uid)
sorted_candidates = sorted([(v, k) for k, v in user_net.items()], key=sort_key)
best_user_id = sorted_candidates[0][1] if sorted_candidates else None
result_dict = {
"valid_events": valid_events_count,
"invalid_lines": invalid_lines_count,
"duplicate_events": duplicate_events_count,
"users": users_count,
"purchase_cents": purchase_cents_sum,
"refund_cents": refund_cents_sum,
"net_cents": purchase_cents_sum - refund_cents_sum,
"top_user": best_user_id if isinstance(best_user_id, str) else None
}
print(json.dumps(result_dict))
if __name__ == "__main__":
main()
Granite 4.1 3B・再試行1(公開検査 1/4)
[この再試行で追加したプロンプト]
# 前回コードの受入検査結果
購入と閲覧: expected={"valid_events": 2, "invalid_lines": 0, "duplicate_events": 0, "users": 1, "purchase_cents": 200, "refund_cents": 0, "net_cents": 200, "top_user": "さくら"} actual={"valid_events": 2, "invalid_lines": 0, "duplicate_events": 0, "users": 1, "purchase_cents": 0, "refund_cents": 0, "net_cents": 0, "top_user": "さくら"} stderr=
不正行と重複: expected={"valid_events": 1, "invalid_lines": 1, "duplicate_events": 1, "users": 1, "purchase_cents": 0, "refund_cents": 30, "net_cents": -30, "top_user": "たろう"} actual={"valid_events": 1, "invalid_lines": 1, "duplicate_events": 1, "users": 1, "purchase_cents": 0, "refund_cents": 0, "net_cents": 0, "top_user": "たろう"} stderr=
ユーザー別純額: expected={"valid_events": 3, "invalid_lines": 0, "duplicate_events": 0, "users": 2, "purchase_cents": 150, "refund_cents": 80, "net_cents": 70, "top_user": "たろう"} actual={"valid_events": 3, "invalid_lines": 0, "duplicate_events": 0, "users": 2, "purchase_cents": 0, "refund_cents": 0, "net_cents": 0, "top_user": "たろう"} stderr=
# 前回のコード
前の折りたたみに掲載した出力から、Python部分を抽出して追加
受入検査をすべて通るように修正し、完成したsolution.pyの内容だけを返してください。
[モデルの生出力]
```python
import sys
import json
def main():
seen_event_ids = set()
user_net = {}
valid_events = 0
invalid_lines = 0
duplicate_events = 0
users = set()
purchase_cents = 0
refund_cents = 0
top_user = None
for line in sys.stdin.read().splitlines():
try:
event = json.loads(line)
if not all(key in event and isinstance(event[key], str) and event[key].strip() for key in ['event_id', 'user_id']):
invalid_lines += 1
continue
event_id = event['event_id']
user_id = event['user_id']
if event_id in seen_event_ids:
duplicate_events += 1
continue
seen_event_ids.add(event_id)
valid_events += 1
users.add(user_id)
if event['event_type'] == 'view':
continue
amount_cents = event.get('amount_cents')
if not (isinstance(amount_cents, int) and not isinstance(amount_cents, bool) and amount_cents >= 0):
invalid_lines += 1
continue
user_net[user_id] = user_net.get(user_id, 0) + (amount_cents if event['event_type'] == 'purchase' else -amount_cents)
except (json.JSONDecodeError, KeyError):
invalid_lines += 1
top_user_candidates = [(user, net) for user, net in user_net.items()]
if top_user_candidates:
top_user = min(top_user_candidates, key=lambda x: (-x[1], x[0]))[0]
result = {
"valid_events": valid_events,
"invalid_lines": invalid_lines,
"duplicate_events": duplicate_events,
"users": len(users),
"purchase_cents": purchase_cents,
"refund_cents": refund_cents,
"net_cents": purchase_cents - refund_cents,
"top_user": top_user if valid_events > 0 else None
}
print(json.dumps(result))
if __name__ == "__main__":
main()
```
Granite 4.1 3B・再試行2(公開検査 1/4)
[この再試行で追加したプロンプト]
# 前回コードの受入検査結果
購入と閲覧: expected={"valid_events": 2, "invalid_lines": 0, "duplicate_events": 0, "users": 1, "purchase_cents": 200, "refund_cents": 0, "net_cents": 200, "top_user": "さくら"} actual={"valid_events": 2, "invalid_lines": 0, "duplicate_events": 0, "users": 1, "purchase_cents": 0, "refund_cents": 0, "net_cents": 0, "top_user": "さくら"} stderr=
不正行と重複: expected={"valid_events": 1, "invalid_lines": 1, "duplicate_events": 1, "users": 1, "purchase_cents": 0, "refund_cents": 30, "net_cents": -30, "top_user": "たろう"} actual={"valid_events": 1, "invalid_lines": 1, "duplicate_events": 1, "users": 1, "purchase_cents": 0, "refund_cents": 0, "net_cents": 0, "top_user": "たろう"} stderr=
ユーザー別純額: expected={"valid_events": 3, "invalid_lines": 0, "duplicate_events": 0, "users": 2, "purchase_cents": 150, "refund_cents": 80, "net_cents": 70, "top_user": "たろう"} actual={"valid_events": 3, "invalid_lines": 0, "duplicate_events": 0, "users": 2, "purchase_cents": 0, "refund_cents": 0, "net_cents": 0, "top_user": "たろう"} stderr=
# 前回のコード
前の折りたたみに掲載した出力から、Python部分を抽出して追加
受入検査をすべて通るように修正し、完成したsolution.pyの内容だけを返してください。
[モデルの生出力]
```python
import sys
import json
def main():
seen_event_ids = set()
user_net = {}
valid_events = 0
invalid_lines = 0
duplicate_events = 0
users = set()
purchase_cents = 0
refund_cents = 0
top_user = None
for line in sys.stdin.read().splitlines():
try:
event = json.loads(line)
if not all(key in event and isinstance(event[key], str) and event[key].strip() for key in ['event_id', 'user_id']):
invalid_lines += 1
continue
event_id = event['event_id']
user_id = event['user_id']
if event_id in seen_event_ids:
duplicate_events += 1
continue
seen_event_ids.add(event_id)
valid_events += 1
users.add(user_id)
if event['event_type'] == 'view':
continue
amount_cents = event.get('amount_cents')
if not (isinstance(amount_cents, int) and not isinstance(amount_cents, bool) and amount_cents >= 0):
invalid_lines += 1
continue
user_net[user_id] = user_net.get(user_id, 0) + (amount_cents if event['event_type'] == 'purchase' else -amount_cents)
except (json.JSONDecodeError, KeyError):
invalid_lines += 1
top_user_candidates = [(user, net) for user, net in user_net.items()]
if top_user_candidates:
top_user = min(top_user_candidates, key=lambda x: (-x[1], x[0]))[0]
result = {
"valid_events": valid_events,
"invalid_lines": invalid_lines,
"duplicate_events": duplicate_events,
"users": len(users),
"purchase_cents": purchase_cents,
"refund_cents": refund_cents,
"net_cents": purchase_cents - refund_cents,
"top_user": top_user if valid_events > 0 else None
}
print(json.dumps(result))
if __name__ == "__main__":
main()
```
Phi-4-mini 3.8B・再試行1(公開検査 0/4)
[この再試行で追加したプロンプト]
# 前回コードの受入検査結果
構文検査に失敗:
File "C:\works\tech_blogs\blogs\2026-07-27_local-llm-performance\benchmark\results_guardrails_v2\phi4-mini-3.8b\prompt_plus_harness\attempt_1.py", line 42
top_user_id = min((user for user in sorted(user_net), key=lambda u: (-user_net[u], u)) if users_count > 0 else (None,))
^
SyntaxError: invalid syntax
# 前回のコード
前の折りたたみに掲載した出力から、Python部分を抽出して追加
受入検査をすべて通るように修正し、完成したsolution.pyの内容だけを返してください。
[モデルの生出力]
```python
import json
import sys
from collections import defaultdict
def main():
seen_event_ids = set()
user_net = defaultdict(int)
valid_events, invalid_lines, duplicate_events, users_count = 0, 0, 0, 0
for line in (line.strip() for line in sys.stdin if line):
try:
event = json.loads(line) or {}
event_id = next(iter(event))
if not isinstance(event_id, str) or not event_id.strip() or event_id in seen_event_ids:
invalid_lines += 1
continue
seen_event_ids.add(event_id)
valid_events += 1
user_id = list(event.keys())[0]
users_count = max(users_count, 1 + (user_id not in user_net))
if 'purchase' in event and all(isinstance(amount_cents, int) and amount_cents >= 0 for _, amount_cents in sorted(event['purchase'].items())):
purchase_amounts = [amount_cents for _, amount_cents in sorted(event['purchase'].items())]
total_purchase = sum(purchase_amounts)
user_net[user_id] += total_purchase
elif 'refund' in event and all(isinstance(amount_cents, int) and amount_cents >= 0 for _, amount_cents in sorted(event['refund'].items())):
refund_amounts = [amount_cents for _, amount_cents in sorted(event['refund'].items())]
total_refund = sum(refund_amounts)
user_net[user_id] -= total_refund
else:
invalid_lines += 1
except (json.JSONDecodeError, TypeError):
invalid_lines += 1
duplicate_events = len(seen_event_ids) - valid_events
net_cents = sum(user_net.values())
top_user_id = min((user for user in sorted(user_net), key=lambda u: (-user_net[u], u)) if users_count > 0 else (None,))
result = {
"valid_events": valid_events,
"invalid_lines": invalid_lines,
"duplicate_events": duplicate_events,
"users": users_count,
"purchase_cents": sum(purchase for purchase in user_net.values() if isinstance(purchase, int)),
"refund_cents": -sum(refund for refund in user_net.values() if isinstance(-refund, int)),
"net_cents": net_cents,
"top_user_id": top_user_id
}
print(json.dumps(result, separators=(',', ':')))
if __name__ == "__main__":
main()
```
Phi-4-mini 3.8B・再試行2(公開検査 0/4)
[この再試行で追加したプロンプト]
# 前回コードの受入検査結果
構文検査に失敗:
File "C:\works\tech_blogs\blogs\2026-07-27_local-llm-performance\benchmark\results_guardrails_v2\phi4-mini-3.8b\prompt_plus_harness\attempt_2.py", line 42
top_user_id = min((user for user in sorted(user_net), key=lambda u: (-user_net[u], u)) if users_count > 0 else (None,))
^
SyntaxError: invalid syntax
# 前回のコード
前の折りたたみに掲載した出力から、Python部分を抽出して追加
受入検査をすべて通るように修正し、完成したsolution.pyの内容だけを返してください。
[モデルの生出力]
```python
import json
import sys
from collections import defaultdict
def main():
seen_event_ids = set()
user_net = defaultdict(int)
valid_events, invalid_lines, duplicate_events, users_count = 0, 0, 0, 0
for line in (line.strip() for line in sys.stdin if line):
try:
event = json.loads(line) or {}
event_id = next(iter(event))
if not isinstance(event_id, str) or not event_id.strip() or event_id in seen_event_ids:
invalid_lines += 1
continue
seen_event_ids.add(event_id)
valid_events += 1
user_id = list(event.keys())[0]
users_count = max(users_count, 1 + (user_id not in user_net))
if 'purchase' in event and all(isinstance(amount_cents, int) and amount_cents >= 0 for _, amount_cents in sorted(event['purchase'].items())):
purchase_amounts = [amount_cents for _, amount_cents in sorted(event['purchase'].items())]
total_purchase = sum(purchase_amounts)
user_net[user_id] += total_purchase
elif 'refund' in event and all(isinstance(amount_cents, int) and amount_cents >= 0 for _, amount_cents in sorted(event['refund'].items())):
refund_amounts = [amount_cents for _, amount_cents in sorted(event['refund'].items())]
total_refund = sum(refund_amounts)
user_net[user_id] -= total_refund
else:
invalid_lines += 1
except (json.JSONDecodeError, TypeError):
invalid_lines += 1
duplicate_events = len(seen_event_ids) - valid_events
net_cents = sum(user_net.values())
top_user_id = min((user for user in sorted(user_net), key=lambda u: (-user_net[u], u)) if users_count > 0 else (None,))
result = {
"valid_events": valid_events,
"invalid_lines": invalid_lines,
"duplicate_events": duplicate_events,
"users": users_count,
"purchase_cents": sum(purchase for purchase in user_net.values() if isinstance(purchase, int)),
"refund_cents": -sum(refund for refund in user_net.values() if isinstance(-refund, int)),
"net_cents": net_cents,
"top_user_id": top_user_id
}
print(json.dumps(result, separators=(',', ':')))
if __name__ == "__main__":
main()
```
Qwenの初回生成はインデントの構文エラーでした。エラー内容と前回のコードを返した再試行1は、公開検査を3/4通過。再試行2は1/4へ下がったため、最後の出力ではなく再試行1を残しました。隠しテストで測ると、この候補は5/12です。
点数は0から5へ進みましたが、公開検査を1件落としているので採用しません。ハーネスが担ったのは、モデルを必ず正解へ直すことではなく、候補を比較し、基準に届かない出力を止めることでした。
公開検査を4件通ったGemmaとMinistralは採用し、Qwenは3件、Graniteは1件、Phiは0件だったため止めました。今回のタスクに向かない出力を、利用者へ渡さない仕組みです。
再試行には時間がかかります。Qwenは3回で約54秒、Ministralは最初の1回、約7秒で通りました。相性のよい組み合わせを選べば、再試行も減らせます。
小型LocalLLMは、ここまでできた
今回の検証で試したのは、入力と出力が決まった、小さな売上集計プログラムです。素の指示でもGemmaは8/12、Graniteは6/12でした。購入と返金の基本集計は通る一方、不正なJSON、金額の型、複数条件を混ぜたケースで点を落としています。
処理順と完成条件まで渡すと、Ministralは12/12、Gemmaは11/12まで進みました。少なくともMinistralは、今回用意した基本ケースから複合ケースまで、すべてを通るコードを生成できています。3〜4B級でも、仕事を狭く定義すれば、完成したプログラムを返せるモデルがありました。
ただし、公開テスト4件を通ったGemmaにも、12ケースでは1件の見落としがありました。ハーネスが保証するのは、用意した検査を通ったことだけです。小型LocalLLMへ任せられるのは、短いコードの候補を作り、決めた基準で不合格を止めるところまで。出力を無検査で本番へ出せる、という結果ではありません。
今回わかったこと
正解条件が明確な短いタスクなら、小型LocalLLMにも最後まで解けるモデルがあります。ただし、できたかどうかを決めるのはモデルの自己申告ではなく、外側に置いたテストです。
まとめ
最近の小型モデルは、境界のはっきりした仕事なら実用候補です。今回もMinistralは12/12、Gemmaは11/12まで到達しました。
ただし、同じ指示が全モデルに合うわけではありません。プロンプトで仕事の組み立て方を伝え、ハーネスで合否と停止を引き受ける。その二つがあって初めて、モデルの速さや手軽さを実際の仕事へつなげられます。
今回試した範囲では、小型LocalLLMは「何でもできる」には遠くても、入力と合格条件を決めた短いタスクなら、すでに使えるところまで来ています。
参考にしたサイト
前回の記事と、各モデル、Ollamaの公式資料です。
投稿日2026年07月31日
カテゴリーTech Blog
タグ LLM評価、Local LLM





