会社の業務をどう設計し、どう改善していくべきかという考え方・方法論がまとめられた一冊です。
読みながら、「AIによって属人化やブラックボックス化は確かに解消されていくのだろうけど、結局は組織の中で『理解』を共有していくことがとても重要になるのでは」ということを考えました。今回はその考えを整理してみたいと思います。

この本が扱っているのは「システム」ではなく「業務」

先に、この本のスタンスについて触れておきたいと思います。本書はシステム導入のノウハウ本ではなく、その手前にある「業務そのもの」をどう捉えるかを扱っています。

ここで言う業務とは、例えば見積書の作成のような、日々の仕事の流れそのものです。弊社のようにクライアント様のシステムを開発する会社であれば、まず営業担当がエンジニアと相談しながら、ある程度の見積金額を算出します。
次に、社内の決裁者からその金額で見積書を作ることの承認を得ます。
承認が下りたら実際の書類を作成します。金額の誤りやお客様の名前の間違いがあってはならないので、書類作成の担当者が内容の正しさを確認します。見積書を1枚作るだけでも、こうした業務の流れがあります。

では、見積書を作るという業務の目的は何でしょうか。見積書はクライアント様にとって、ビジネス上の意思決定の材料になります。どれくらいの金額でどれくらいのサービスを受けられるのかという情報をもとに、予算を確保したり、他のITベンダーの見積と比較したりする。そして実際にご発注いただけることになれば、その見積書の情報をもとに双方のお取引が成り立ち、結果として世の中に新しいサービスや価値を送り出すことになります。そう捉えると、見積書の作成は単に書類を1枚作る作業ではなく、大きなビジネス価値を生むことに繋がっていると言えます。

システム化は、業務の目的を達成するための「手段」

実際には、この業務をシステムなしで回している会社はほとんどないはずです。簡易的な決裁者確認はチャットツールで行うこともあれば、承認の専用システムで証跡を残すこともあり、多くの場合は両者を組み合わせて業務が動いています。

本書が繰り返し述べているのは「目的と手段の区別」です。目的にあたるのは、先ほど整理した業務の目的、つまりお客様の意思決定を支え、お取引につなげることです。そしてシステム化は、その目的を達成するための手段です。整理するとこうなります。

目的：業務の目的（お客様の意思決定を支え、お取引につなげる）
業務：その目的を果たすための仕事の流れ（金額の算出 → 社内承認 → 書類作成・確認）
手段：その業務を正確に・効率的に回すためのシステム化（チャットでの確認、専用システム）

システム化はあくまで手段であって、それ自体が目的になることはありません。だからこそ、まず業務とその目的があり、それを一旦システムから切り離して考えることが大事だと分かりました。

業務設計の2つの落とし穴 ― 属人化とブラックボックス化

業務設計における落とし穴として、本書では「属人化」と「ブラックボックス化」の2つが挙げられています。

業務が変化する中で、「属人化」と「ブラックボックス化」という2つの課題も生じます。担当者の裁量に任された業務は、その人しか把握していないルールが増え、異動や退職が発生した際に業務がストップするリスクを抱えます。また、業務の自動化は一見効率的に思えるものの、業務プロセスが見えにくくなり、結果として非効率な業務が固定化されることもあります。

属人化は、特定の誰かにしか分からない暗黙知が積み上がっていく状態です。マニュアルも引き継ぎもなく、「あの人に聞かないとこの情報をどう入力していたのか分からない」「その人がいないと業務が止まる」という状態ですね。

一方のブラックボックス化は、属人化と似ていますが別の概念です。Aさんがいなくても業務が回るように、自動化プログラムを組んだとします。確かに誰でも業務を行えるようになったけれど、そのプログラムの中身が何をしているのかは誰も分からない。ボタンを押せば結果は出てくるものの、解説ドキュメントもないため、それ以上業務を変えられない。こういう状態です。

Aさんがいないと回らないのが属人化、Aさんがいなくても回るけれど中身が誰にも分からないのがブラックボックス化。似ているようで、別の問題として整理されています。

AIはこの2つを解消してくれるのか

ここからは大部分が、書籍を読んで考えた私見になります。
AIは属人化・ブラックボックス化を解消してくれるのでしょうか。一定の水準までは、確かに解消しうると思っています。

属人化については、ある人が手作業でやっていた操作をAIの力で簡単にプログラム化できるようになりました。プロンプトや、一連の操作をひとまとまりにした「skills」を共有することで、仕事のやり方そのものを組織の資産にできます。

ブラックボックス化についても同様です。中身が何をしているか分からないソースコードをAIに読み込ませて解説してもらえば、どういう処理をしているのかを掴むことができます。そのプログラムの説明をドキュメント化する作業も、以前よりずっと簡単になりました。

AIが業務設計の落とし穴の解消に貢献してくれるのは間違いないと思います。

ただ、解消ではなく「移動」しているだけかもしれない

これで万事解決かというと、そうとも言い切れないと感じています。属人化・ブラックボックス化が消えたように見えて、実は「場所が移動しているだけ」ではないか、というのが自分なりの見立てです。

AIが解消してくれるのは、主に「具体的な手順」のレイヤーにある属人化・ブラックボックス化です。プロンプトの共有やコード化によって、操作や仕組みは誰でも扱えるようになります。しかしその一段上には、共有されにくいものが残ります。

ひとつは、そもそも「なぜその業務があるのか」という、業務そのものへの抽象度の高い理解です。
もうひとつは、ツールを生み出したときの考え方や経緯です。あるプログラムを作った、あるいは変更したとして、作った人がどのような意図でどう指示を出し、そのツールに至ったのか。この過程は、詳細にはなかなか共有されません。成果物だけが組織に残り、思考の過程は残らないことのほうが多いはずです。

しかもこの「抽象度の高い理解・考え方」は、引き継ぎやドキュメントで一度に継承が完了できる性質のものではありません。経営環境や外部要因の変化によって、業務の目的自体が変わりうるからです。常に変わっていくものであり、かつ抽象度も高い。「引き継ぎをしました」「ドキュメントを作りました」だけでは、組織に理解が伝播し定着している状態になることは難しいと思います。

だからこそ、目の前の属人化・ブラックボックス化がAIによって解消されるほど、より抽象度の高いところに移動していくと考えます。

※余談ですが、「AIに何をどう指示するかの『意図』が重要である」という論点については、「インテントエンジニア（Intent Engineer）」という新しい役割を提唱している記事がありました。

www.squer.io

要点は、AIには「これを作ってほしい」という解決策ではなく、何のためにそれが必要なのかという目的と文脈を伝えるべきだ、というものです。
AIに渡すべきものが手順ではなく目的であるならば、その目的の理解は、やはり人の側が持っておく必要があるということになります。

ひとつのアプローチとしての、非定型なコミュニケーション

では、業務への抽象度の高い理解を、どう組織に広げていくか。ここに銀の弾丸はないと思っています。アプローチのひとつとして有効だと感じているのが、非定型なコミュニケーションの場を持つことです。

自分の経験を振り返ると、新卒1年目の頃、先輩がエディタでプログラムを書いているところを横で見せてもらい、文法の理解など知識やテクニックに収まらない「仕事の考え方」を、見て、尋ねて学んだ覚えがあります。

また、先輩や上司がお客様と電話でやり取りしている様子をなんとなく聞いていて、いまその案件がどういう背景にあるのかをある程度推測する。会話からどのような意思決定のプロセスが取られているのかを感じ取り、自分がやっている業務のゴールや目的の理解に変えていく。こういう学び方もありました。人から発せられる情報を横から見たり聞いたりして、自分にインストールしていく感覚に近いと思います。

そう考えると、かしこまったミーティング以外で見聞きできる場には、固有の価値があるのだと思います。先輩や上司が考えていることを、とりとめもなく聞いて自分なりに落とし込む。AIがこれだけ定型業務を自動化してくれるからこそ、共有されにくいものは上のレイヤーに移っていき、非定型に情報が入る場から学ぶ意味は増しているのではないかと感じています。

「それでは暗黙知が広がるだけではないか」

ここまで書いておいて、自分でも思う反論があります。非定型なコミュニケーションを増やすというのは、結局のところ組織の中に暗黙知を広げることであり、属人化をむしろ助長するのではないか、というものです。

これはある程度その通りだと思います。ただ、ビジネスの外部環境の変化が激しく、業務の目的自体が変わりうることを前提にすると、抽象度の高い理解の共有は、常に一定量必要になり続けるものだと考えています。

だから目指したいのは、暗黙知を暗黙知のままにしておくことではありません。暗黙知が行き交う場をきちんと設けたうえで、そこで共有された理解を形式知に落とし込んでいく。この流れを繰り返し続けることだと思っています。非定型なコミュニケーションで言語化した理解を、「skills」やドキュメント、業務ルールとして形式知にしていく。形式知が外部環境の変化で合わなくなれば、また会話に戻ってアップデートする。この循環が止まっていないことこそが、組織が生きているということなのではないかと思います。

まとめ

『業務設計の教科書』が示しているのは、システム導入やデジタル化に飛びつく前に、業務の目的や価値に立ち返るべきだという視点でした。そして業務設計の落とし穴として、属人化とブラックボックス化の2つが挙げられていました。

AIは、この2つを一定水準まで解消してくれます。プロンプトや「skills」の共有、自動化、ドキュメント整備は、以前よりずっと手軽になりました。ただしこれは、属人化・ブラックボックス化が消えたというより、より抽象度の高い場所へ移動したと捉えるべきかもしれません。「なぜその業務があるのか」「どういう意図でそのツールを作ったのか」という「理解」は、ドキュメント化しにくく、しかも環境の変化とともに変わることがあります。

「理解」を組織に広げるアプローチのひとつが、非定型なコミュニケーションの場を持つことだと思っています。そこで生まれた理解を形式知に落とし込み、環境が変わればまた会話に戻って組み替える。この往復を続けられるかどうかが、AIを使ってもなお残る属人化・ブラックボックス化に向き合ううえで、組織としての差になっていくのではないかと考えています。

テコテックの採用活動について

テコテックでは新卒採用、中途採用共に積極的に募集をしています。採用サイトにて会社の雰囲気や福利厚生、募集内容をご確認いただけます。ご興味を持っていただけましたら是非ご覧ください。

www.tecotec.co.jp

2026-08-19

Liquid Glass Offという方針をとる場合の現実的な落とし所 — パーツ単位で妥協点を探った話

システム開発第二事業部の冨永です。

主にiOS/iPadアプリの開発を担当しております。

今回は、Liquid Glassをパーツ単位でOFFにする現実的な落とし所について書きたいと思います。

2026-08-13

AIで作業は「速く」なるのに、仕事は「早く」ならない理由

はじめに

AIで作業は「速く」なるのに、仕事はなぜか「早く」終わらない。そんな感覚を持ったことはないでしょうか。今回は、AIによる高速化で起きる「ボトルネックの移動」というテーマについて、前職で経験したリリーストラブルの話を交えながら、これからの時代に何を積み重ねておくべきかを考えてみます。

2026-08-07

大規模クローリングにおけるSQSキュー詰まりの解消法

はじめに
アーキテクチャ
キュー詰まり
まとめ
参考
テコテックの採用活動について

はじめに

システム開発第一事業部の大泉です。普段はエンジニアとして Python でバックエンドの開発を行っています。

AWS の SQS を使用して 1500 程度のサイトを対象に 1 日数百万ページを処理するクローラーを構築しました。開発初期、ページ数の多いサイトの URL でキューが埋め尽くされ、他のサイトのクローリングがほとんど行われない問題が発生しました。本記事ではこのようなキュー詰まりを解消し、すべてのサイトを平準化してクローリングする方法をご紹介します。

なお、サンプルコードは実装のイメージが伝わりやすいように簡略化したものであり、実際の実装とは異なります。

アーキテクチャ

SQS でクローリング対象の URL をキューイングするアーキテクチャです。1 つのメッセージに 1 つの URL を含めるようにします。

ECS: クローリング実行
SQS: URL をキューイング
Lambda: トリガーとなる URL をキューに送信
EventBridge Scheduler: Lambda を定期的に呼び出す
ElastiCache for Valkey: 単位時間あたりのクローリング回数を保持

ECS で行うクローリングの処理フローです。

クローリング処理フロー

キューにポーリングを行う
受信したメッセージから URL を取り出し、フェッチを行う
取得した HTML からリンクをすべて抽出する
抽出したリンクを 1 メッセージ 1 URL として SQS に送信する
1 に戻る

クローラーの立ち上げ時はキューにメッセージが何も入っていないため、トリガーとなるメッセージが必要です。それを投入するのが Lambda です。例えば以下のようなフローになります。

Lambda はサイト情報テーブルから https://example.com/ を取得
https://example.com/ をキューに送信
クローリング処理フロー の 1 に進む

キュー詰まり

問題になるのがリンクが多いサイトです。1 URL から大量の URL が抽出され、さらにそこから大量の URL が抽出され、というように一部のサイトの URL でキューが埋め尽くされます。その結果、他のサイトの URL を取り出すのに非常に時間がかかります。そこで以下を導入しました。

キューの分離
同時実行制限
可視性タイムアウト
遅延キュー

キューの分離

「何回目のフェッチで見つかった URL か」をクローリングの深さとします。例えばクローリングのトリガーとなる Lambda によってキューに送信された URL は深さ 0 です。深さ 0 の URL から取得した URL は深さ 1 です。

クローリングの深さによって優先度高キューと優先度低キューの 2 つに分離しました。優先度高キューには深さが小さい URL を入れます。優先度低キューには深さが大きい URL を入れます。

例えば、深さ 0 と 1 は優先度高キュー、深さ 2 以降は優先度低キューに入れるようにします。

優先度高キューのポーリングを先に行い、メッセージを受信しなければ優先度低キューのポーリングを行います。これにより深さが小さい URL のクローリングが優先して行われるようになります。

同時実行制限

同時実行制限は次の 2 種類を導入しました。

サーバ単位での同時実行制限: 1 サーバ内でのサイトごとのクローリング回数を平準化
時間単位での同時実行制限: 1 時間あたりのサイトごとのクローリング回数を平準化

2 についてはキャッシュサーバ（ElastiCache for Valkey）にメッセージ処理回数を保存します。キャッシュサーバを共有ストレージとして利用することで、各サーバ間でサイトごとのメッセージ処理回数を共有できるようになります。それぞれの同時実行制限に引っかかったメッセージは、次に述べる「メッセージ単位での可視性タイムアウト」を利用し、一定期間受信できないようにします。

サーバ単位での同時実行制限

サーバ内でサイトごとのメッセージ処理回数を管理するため、専用のクラスを実装しました。辞書型（マップ）で {"siteA": 1, "siteB": 2} のようにサイトごとのメッセージ処理回数を保持します。

以下はサンプルコードです。

class ConcurrencyManager:
    """サイトごとの同時実行数を管理"""

    def __init__(self):
        self.active_tasks = defaultdict(int)

    def can_acquire(self, site_id: int, limit: int) -> bool:
        """上限を超えていないかチェック"""
        if self.active_tasks[site_id] < limit:
            self.active_tasks[site_id] += 1
            return True
        return False

    def release(self, site_id: int):
        """クローリングが完了（メッセージ処理が完了）したときはデクリメント"""
        if self.active_tasks[site_id] > 0:
            self.active_tasks[site_id] -= 1

時間単位での同時実行制限

キャッシュサーバに処理回数を保存し、1 時間経過後にキャッシュが消えるように有効期限を設定しています（ttl_seconds = 3600）。

以下はサンプルコードです。制限を超えている場合は呼び出し元に False を返し、クローリングを行わないようにします。

def check_limit(site_id: int, limit: int) -> bool:
    """サイトごとのメッセージ処理回数をチェック"""
    key = f"count:{site_id}"
    ttl_seconds = 3600

    with client.pipeline(transaction=True) as pipe:
        pipe.incr(key)
        pipe.ttl(key)
        count, ttl = pipe.execute()

        # TTLが設定されていない場合、有効期限を設定
        if ttl == -1:
            client.expire(key, ttl_seconds)

        # 制限を超えているかチェック
        return count <= limit

メッセージ単位での可視性タイムアウト

可視性タイムアウトとは、あるコンシューマーが処理中のメッセージを他のコンシューマーが重複処理しないようにするため、一定期間キューから見えなくする機能です。SQS ではキュー自体の可視性タイムアウトとメッセージ単位での可視性タイムアウトを設定できます。

docs.aws.amazon.com

ここではメッセージ単位での可視性タイムアウトを利用します。同時実行制限に引っかかったサイトの URL に対してフェッチを行わず、可視性タイムアウトを延長してキューに戻します。これにより、制限に引っかかったメッセージが見えなくなっている時間が長くなり、他サイトのメッセージを先に取り出せるようになります。

docs.aws.amazon.com

可視性タイムアウトの設定時間は、あらかじめ決めた範囲（例: 5～10 秒）の中でランダムに決定します。ランダムに設定することで同じ条件のメッセージが同時に見えるようになることを防ぎます。

一斉にメッセージを処理するため、boto3 ライブラリの change_message_visibility_batch を使用して可視性タイムアウトを変更します。

boto3.amazonaws.com

以下はサンプルコードです。

entries = []
for i, receipt_handle in enumerate(receipt_handles):
    # 最小設定時間と最大設定時間を決定
    min_delay, max_delay = self.delay_calculator.determine_visibility_timeout()
    # ランダムに可視性タイムアウトを決定
    entries.append({
        "Id": str(i),
        "ReceiptHandle": receipt_handle,
        "VisibilityTimeout": random.randint(min_delay, max_delay)
    })
# バッチ(1度に10メッセージ)で可視性タイムアウト変更
for i in range(0, len(entries), 10):
    batch_entries = entries[i : i + 10]
    async with self._create_client() as client:
        await client.change_message_visibility_batch(
            QueueUrl=queue_url,
            Entries=batch_entries
        )

メッセージ単位での遅延キュー

遅延キューは、メッセージが初めてキューに入ってから一定期間キューから見えなくなる機能です。つまり、すぐにメッセージが取り出せないようにします。これを使うことで、クローリング回数の多いサイトの処理を抑制し、回数の少ないサイトのメッセージを優先的に取り出せるようにします。

docs.aws.amazon.com

なお、可視性タイムアウトはコンシューマーがメッセージを受信してから見えなくなるのに対し、遅延キューはメッセージがキューに入ってから見えなくなります。

フェッチレスポンスから抽出した URL （次のクローリング URL）をキューに送信するときに、以下の条件によって遅延キューの設定時間を決定します。

クローリングの深さ
サイトごとのメッセージ処理数によるパーセンタイル（メッセージ処理数が多いサイトを特定する）

これらを考慮し、最小設定時間～最大設定時間の間でランダムに設定します（ここでも揺らぎを持たせることで、同時にメッセージが取り出されるのを防ぎます）。これにより、リンクが多いサイトでは階層が深くなるほど新たに見つかる URL の遅延が長くなり、埋もれていた小規模サイトの URL を取り出しやすくなります。

例）

クローリングの深さ 1、サイトのメッセージ処理回数によるパーセンタイル 50% → 1 分～2 分
クローリングの深さ 2、サイトのメッセージ処理回数によるパーセンタイル 80% → 10 分～15 分

可視性タイムアウトと同様に、一斉にメッセージを処理するため、boto3 ライブラリの send_message_batch を使用して遅延キューを設定します。

boto3.amazonaws.com

以下はサンプルコードです。

for i in range(0, len(urls), 10):
    # クローリングで見つかったURLをバッチ送信のため分割
    batch_urls = urls[i : i + 10]

    # クローリングの深さとサイトごとのメッセージ処理数から最小設定時間と最大設定時間を決定
    min_delay_seconds, max_delay_seconds = self.delay_calculator.determine_delay_seconds()

    # キューに送信するためのエントリーを作成. 遅延キューの設定はDelaySeconds
    entries = []
    for idx, url in enumerate(batch_urls):
        base_site_info.url = url
        entries.append({
            "Id": str(idx),
            "MessageBody": json.dumps(base_site_info.to_dict(), ensure_ascii=False),
            "DelaySeconds": random.randint(min_delay_seconds, max_delay_seconds),
        })

    # バッチ(1度に10メッセージ)でキューに送信
    async with self._create_client() as client:
        await client.send_message_batch(
            QueueUrl=self.crawler_low_queue_url,
            Entries=entries,
        )

まとめ

SQS を利用した大規模クローラーにおけるキュー詰まりの問題とその解決策について紹介しました。クローリング平準化のため以下の 4 つのアプローチを組み合わせました。

キューの分離: 優先して処理したいメッセージとそうでないメッセージを分離する（本記事では、クローリングの深さによって分離した）
同時実行制限: 各サーバ内部で保持する同時実行数と、サーバ全体で共有する処理回数を管理することにより、クローリング回数の多いサイトの実行数を制限し、その分少ないサイトのクローリングを実行する
可視性タイムアウト: 同時実行制限に引っかかったメッセージの可視性タイムアウトを延長し、クローリング回数の少ないサイトのメッセージを取り出しやすくする
遅延キュー: クローリングの深さとサイトのメッセージ処理回数に応じて新規 URL の遅延を調整し、クローリング回数の少ないサイトのメッセージを取り出しやすくする

以上のように、SQS で大規模なメッセージを扱う際は単に流し込むだけでなく流量制御が重要になります。本記事が同様の課題を抱える方の参考になれば幸いです。