質問を1日1通にまとめる方法――重複排除と持ち越しを実測した待合室

迷うたびに確認事項を送る運用は、判断を速くするより先に、受信箱を迷いで満たします。そこで、質問をいったん待合室へ積み、重複を除き、上限までを1日1通に束ねる仕組みにしました。

自宅ラボで、定期処理やAIに一部を任せつつ、最終判断は人が持ちたい運用者向けの話です。読み終える頃には、即時通知と日次通知の線引き、重複質問の扱い、溢れた質問の持ち越しを、外部サービスなしで再現できます。

2026-08-17 09:32 JSTの実走では、質問4行から同一質問を除いて3件になり、上限2件を1通へ整形し、残り1件を次回へ持ち越しました。同じ日付で再度処理しても、2通目は作りませんでした。

課題は、質問を送る前に判断の質をそろえることです

今回の本体は、質問を即座に人へ渡すための道具ではありません。人へ届く前に、質問として残す価値があるかを機械的に絞る待合室です。

列へ入る候補には、すべての関所を通す設計にしています。あらかじめ定めた判断を負う役割だけが出せること、予測回答が添えられていること、今後も再利用できる判断になりそうなこと、秘密情報や内部識別情報を含まないこと。このどれかが欠ければ送らない、fail-safe側の構成です。

さらに、同じ話題と同じ質問文は二重に送らないようにしています。質問する側にも、丸投げではなく「私はこう考える」を添えさせる。その小さな抵抗が、受け取る側の判断力を残します。

図1で重要なのは、最初の門を通らない質問は通知経路へ入らないことです。即時扱いへ強制する指定も、日次便へ強制する指定もありますが、即時経路には1日6件の上限があります。緊急用の抜け道を作っても、抜け道を常用させないためです。

即時通知・日次便・送らないを分けます

判断をAIへ渡す場所があっても、「聞くかどうか」の関所までAI任せにする必要はありません。以前の見張り係を固定プログラムからAIに置き換えた運用では判断をAIへ渡す範囲を扱いました。今回は反対に、質問を人へ届ける入口を固定ロジックとして残しました。

「すぐ届けない」設計は、忘れないための仕組みとも相性があります。日付でまとめる発想は、日付指定リマインダでも扱いました。ただし待合室は、予定を思い出すためではなく、判断の候補を雑音から守るためのものです。

最小版は、追記・重複排除・持ち越しだけを切り出しました

元の仕組みには四門、即時・日次の振り分け、止める1コマンドがあります。止めるマーカーを置くと、新しい候補の追加とまとめ配送を休眠させ、外すと再開できます。定期処理の設定自体に触れず、ループだけを止めるためのものです。

一方で、日次の外部配送を担う定期処理は、導入時点では発火行をコメントアウトしたままです。候補を積む側は別の定期処理から既に呼ばれていますが、12:30の配送そのものは有効化前です。外部へメッセージを送る効果を持つ部分は、段階的に切り替えます。

ここで公開する最小版は、待合室の芯だけを取り出しました。日中の追記、日次の重複排除、上限までの整形、余りの持ち越し、stateとreceiptの保存です。外部送信は持たないため、生成した1通をどの配送手段へ渡すかは明確に分離されています。

4行は3件になり、2件を1通へ束ねて1件を残しました

検証日は2026-08-17です。Python 3標準ライブラリだけを使い、同一質問を1組含む4行を追加してから、上限を2件にして日次処理を実行しました。

図2の変化しない側、すなわち同日2回目がmessage=noneで止まった点が重要です。「日次」を定期処理の時刻だけに任せず、stateにも日付を残して二重送信を避けています。生成メッセージのSHA-256はstateとreceiptの両方に記録され、どの1通を後段へ渡したか照合できます。

観測した事実

raw 4行はunique 3件となり、上限2件で2件が配送済み、1件がpendingとして残りました。receiptにはcarryover=1が記録され、同日の再実行は2通目を作りませんでした。

解釈

追記台帳を消さずに重複だけを配送時に除くことで、記録の事実性と受信者の読みやすさを両立できます。上限外を持ち越すことで、1通を読み切れる量に保てます。

未確定

実際の外部配送を有効化した後、初回のダイジェストがどの程度の量になり、どの速度で消し込めるかはまだ確認できていません。判断原則そのものを自動更新する機能も、この仕組みには含めていません。

ハマり所は、台帳の行数と配送件数を混同しないことです

• 同じ話題と同じ質問文は同じIDになります。追記台帳には2行残るため、raw=4でも配送候補はunique=3です。
• --cap 2で3件を処理すると、先頭2件だけを配送済みにし、残りは次回へ持ち越します。
• 同じ--dayで2回実行しても、2通目も追加receiptも作りません。
• この最小版は外部サービスへ送信しません。生成されたdaily-message.txtを既存の配送手段へ渡す役は、別に用意します。

チャット越しにすぐ依頼や確認をする仕組みが必要なら、『10分おきに見てて』とチャットで頼める監視の契約のような即時性を優先する運用が合います。迷いをすべて即時化するより、重さで経路を分けたほうが、判断を読む時間は残ります。

コピペで再現

前提条件 Python 3と作業ディレクトリへの書き込み権限が必要です。追加ライブラリ、データベース、外部サービスは不要です。

ハマり所 同じtopicと質問文は同じIDになります。日付を固定して同じ日に2回drainすると、2回目はメッセージを作りません。上限外の質問は配送済みにせず、次回へ残ります。

まず、次の内容をe2-artifact.pyとして保存します。

#!/usr/bin/env python3
"""質問を日中に溜め、1日1回だけ1通へ束ねる最小の待合室。"""
from __future__ import annotations
import argparse, hashlib, json, os, sys
from datetime import date, datetime
from pathlib import Path

def now_iso(): return datetime.now().astimezone().isoformat(timespec="seconds")
def norm(s): return " ".join(s.split()).casefold()
def qid(topic, question): return hashlib.sha256(f"{norm(topic)}|{norm(question)}".encode()).hexdigest()[:12]
def rows(path):
    if not path.exists(): return []
    return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]
def write(path, text):
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_name(f".{path.name}.tmp-{os.getpid()}")
    tmp.write_text(text, encoding="utf-8")
    os.replace(tmp, path)
def write_json(path, value): write(path, json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n")
def state(path):
    if not path.exists(): return {"delivered": {}, "last_drain_day": None}
    value = json.loads(path.read_text(encoding="utf-8"))
    value.setdefault("delivered", {})
    value.setdefault("last_drain_day", None)
    return value
def pending(queue, delivered):
    seen, result = set(delivered), []
    for row in queue:
        if row.get("id") and row["id"] not in seen:
            seen.add(row["id"]); result.append(row)
    return result
def add(args):
    if not args.question.strip() or not args.predicted.strip():
        print("error: question and predicted must not be empty", file=sys.stderr); return 2
    record = {"id": qid(args.topic, args.question), "topic": args.topic.strip() or "general", "question": args.question.strip(), "predicted": args.predicted.strip(), "queued_at": now_iso()}
    args.queue.parent.mkdir(parents=True, exist_ok=True)
    with args.queue.open("a", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False, sort_keys=True) + "\n")
    print(f"queued id={record['id']}")
def drain(args):
    day, st = args.day or date.today().isoformat(), state(args.state)
    todo = pending(rows(args.queue), st["delivered"])
    if st["last_drain_day"] == day:
        print(f"already-drained day={day} message=none pending={len(todo)}"); return
    chosen, carry = todo[:args.cap], len(todo) - len(todo[:args.cap])
    if not chosen:
        st.update({"last_drain_day": day, "last_drain_at": now_iso()}); write_json(args.state, st)
        print(f"drained day={day} message=none pending=0"); return
    lines = [f"判断の待合室 — {day} ({len(chosen)}件)", ""]
    for n, row in enumerate(chosen, 1): lines += [f"{n}. [{row['topic']}] {row['question']}", f"   予測: {row['predicted']}"]
    message = "\n".join(lines + ["", f"持ち越し: {carry}件（次回の日次便へ）"]) + "\n"
    digest, at = hashlib.sha256(message.encode()).hexdigest(), now_iso()
    write(args.message, message)
    for row in chosen: st["delivered"][row["id"]] = at
    st.update({"last_drain_day": day, "last_drain_at": at, "last_message_sha256": digest}); write_json(args.state, st)
    with args.receipts.open("a", encoding="utf-8") as f: f.write(json.dumps({"day": day, "delivered": len(chosen), "carryover": carry, "message_sha256": digest}, ensure_ascii=False) + "\n")
    print(message, end=""); print(f"receipt: delivered={len(chosen)} carryover={carry} sha256={digest[:16]}")
def status(args):
    st, queue = state(args.state), rows(args.queue)
    print(f"status: raw={len(queue)} unique={len(pending(queue, {}))} delivered={len(st['delivered'])} pending={len(pending(queue, st['delivered']))} last_day={st.get('last_drain_day')}")
p = argparse.ArgumentParser(); sub = p.add_subparsers(dest="cmd", required=True)
a = sub.add_parser("add"); a.add_argument("--queue", type=Path, default=Path("questions.jsonl")); a.add_argument("--topic", required=True); a.add_argument("--question", required=True); a.add_argument("--predicted", required=True); a.set_defaults(fn=add)
d = sub.add_parser("drain"); d.add_argument("--queue", type=Path, default=Path("questions.jsonl")); d.add_argument("--state", type=Path, default=Path("delivery-state.json")); d.add_argument("--receipts", type=Path, default=Path("receipts.jsonl")); d.add_argument("--message", type=Path, default=Path("daily-message.txt")); d.add_argument("--cap", type=int, default=8); d.add_argument("--day"); d.set_defaults(fn=drain)
s = sub.add_parser("status"); s.add_argument("--queue", type=Path, default=Path("questions.jsonl")); s.add_argument("--state", type=Path, default=Path("delivery-state.json")); s.set_defaults(fn=status)
args = p.parse_args()
if getattr(args, "cap", 1) < 1: p.error("--cap must be at least 1")
sys.exit(args.fn(args) or 0)

$ python3 e2-artifact.py add --topic 公開 --question '記事のコードは全文載せる?' --predicted '再現に必要な部分は省略せず全文載せる'
$ python3 e2-artifact.py add --topic 運用 --question '週次の確認は黙る週を許す?' --predicted '重要な変化がない週は通知しない'
$ python3 e2-artifact.py add --topic 公開 --question '記事のコードは全文載せる?' --predicted '再現に必要な部分は省略せず全文載せる'
$ python3 e2-artifact.py add --topic 通知 --question '質問は何件まで一通に入れる?' --predicted '読み切れる上限を決め、余りは翌日に持ち越す'
$ python3 e2-artifact.py status
status: raw=4 unique=3 delivered=0 pending=3 last_day=None
$ python3 e2-artifact.py drain --cap 2 --day 2026-08-17
判断の待合室 — 2026-08-17 (2件)

1. [公開] 記事のコードは全文載せる?
   予測: 再現に必要な部分は省略せず全文載せる
2. [運用] 週次の確認は黙る週を許す?
   予測: 重要な変化がない週は通知しない

持ち越し: 1件（次回の日次便へ）
receipt: delivered=2 carryover=1 sha256=0af0b98f0ac67a11
$ python3 e2-artifact.py status
status: raw=4 unique=3 delivered=2 pending=1 last_day=2026-08-17
$ python3 e2-artifact.py drain --cap 2 --day 2026-08-17
already-drained day=2026-08-17 message=none pending=1

この最小版で再現できるのは、「迷うたびに即送信しない」待合室です。四門や外部配送を足す前に、まず1通の量と持ち越しが自分の運用に合うかを、この形で確かめてください。