Pythonで作るLINE AI社長:Messaging APIとLLMをつなぐバックエンド実装

,
【連載】中小企業のための「AI社長」実践導入ガイド(全12回) | 今回は 第9回 です | 前の記事:第8回 | 次の記事:第10回

これまでに設計した「社長の判断基準」「3層アーキテクチャ」「社内セキュリティ」を、いよいよ1本のプログラムとして統合します。

Pythonの高速Webフレームワーク「FastAPI」「LINE Messaging API SDK」、そして「OpenAI API」を組み合わせることで、数十行の簡潔なコードでAI社長のバックエンドサーバーが完成します。

本記事では、コピー&ペーストですぐに動かせる実践的なWebhookサーバーの実装コードと、各処理の要点を分かりやすく解説します。

システム全体のアーキテクチャ

LINEからメッセージが届いてからAI社長が返信するまでの流れは以下の通りです。

メッセージ処理の4ステップ

  1. Webhook受信と署名検証:LINEプラットフォームからの正規のリクエストであることを暗号署名で確認。
  2. User IDホワイトリスト判定:送信者のLINE User IDが登録済み社員のものかを即座に判定。
  3. LLM・ナレッジ呼び出し:System Prompt(社長の理念・判断基準)を付与してOpenAI APIへ問い合わせ。
  4. LINE返信:生成されたAI社長の回答を、Reply Tokenを使って社員のLINE画面へ送信。

必要な環境とライブラリのインストール

Python 3.10以降の環境を用意し、以下のコマンドで必要なライブラリをインストールします。

pip install fastapi uvicorn line-bot-sdk openai

完成版:AI社長Webhookサーバーの実装コード

以下のコードを main.py として保存します。横スクロールなしでそのまま読める設計にしています。

import os
from fastapi import FastAPI, Request, Header, HTTPException
from linebot.v3.webhook import WebhookParser
from linebot.v3.messaging import (
    AsyncApiClient,
    AsyncMessagingApi,
    Configuration,
    ReplyMessageRequest,
    TextMessage
)
from linebot.v3.exceptions import InvalidSignatureError
from openai import OpenAI

app = FastAPI()

# 1. 環境変数の設定
CHANNEL_SECRET = os.getenv("LINE_CHANNEL_SECRET", "your_channel_secret")
CHANNEL_ACCESS_TOKEN = os.getenv("LINE_CHANNEL_ACCESS_TOKEN", "your_access_token")
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "your_openai_api_key")

line_config = Configuration(access_token=CHANNEL_ACCESS_TOKEN)
parser = WebhookParser(CHANNEL_SECRET)
client_ai = OpenAI(api_key=OPENAI_API_KEY)

# 2. 許可された社員のホワイトリスト
AUTHORIZED_USERS = {
    "U1234567890abcdef1234567890abcdef": {
        "name": "江口",
        "role": "manager"
    },
    "U9876543210fedcba9876543210fedcba": {
        "name": "山田",
        "role": "president"
    }
}

# 3. AI社長のSystem Prompt(行動規範と判断基準)
SYSTEM_PROMPT = """
あなたは「○○不動産株式会社」のAI社長(山田社長の分身)です。
以下の基本判断基準に厳格に従って社員の相談に答えてください。

【基本方針】
1. 顧客との長期的信頼を最優先し、目先の短期利益だけで判断しない。
2. 安易な値下げは厳禁。「価値・広告・見せ方・競合・営業」を先に確認する。
3. クレーム対応は「事実確認→一次対応→原因特定→再発防止」の順に行う。
4. 100万円以上の支出、正式な契約判断、解約判断は人間(社長)の決裁を必須とする。
5. 誰に相談すべきか迷っている社員には、担当部署(営業部・管理部・経理)を明確に案内する。
"""

@app.post("/webhook")
async def handle_webhook(
    request: Request,
    x_line_signature: str = Header(None)
):
    if not x_line_signature:
        raise HTTPException(status_code=400, detail="Missing Signature")

    body = await request.body()
    body_str = body.decode("utf-8")

    try:
        events = parser.parse(body_str, x_line_signature)
    except InvalidSignatureError:
        raise HTTPException(status_code=400, detail="Invalid Signature")

    async with AsyncApiClient(line_config) as api_client:
        line_api = AsyncMessagingApi(api_client)

        for event in events:
            if event.type != "message" or event.message.type != "text":
                continue

            user_id = event.source.user_id
            user_msg = event.message.text
            reply_token = event.reply_token

            # A. セキュリティ判定:ホワイトリスト照合
            if user_id not in AUTHORIZED_USERS:
                await line_api.reply_message(
                    ReplyMessageRequest(
                        reply_token=reply_token,
                        messages=[TextMessage(text="このアカウントは社内専用です。利用権限がありません。")]
                    )
                )
                continue

            # B. 許可ユーザーの情報を取得
            employee = AUTHORIZED_USERS[user_id]
            emp_name = employee["name"]

            # C. OpenAI APIによる回答生成
            try:
                response = client_ai.chat.completions.create(
                    model="gpt-4o-mini",
                    messages=[
                        {"role": "system", "content": SYSTEM_PROMPT},
                        {"role": "user", "content": f"社員名:{emp_name}\n相談内容:{user_msg}"}
                    ],
                    temperature=0.3
                )
                ai_answer = response.choices[0].message.content
            except Exception as e:
                ai_answer = "申し訳ありません。回答の生成中にエラーが発生しました。"

            # D. 社員のLINEへ返信
            await line_api.reply_message(
                ReplyMessageRequest(
                    reply_token=reply_token,
                    messages=[TextMessage(text=ai_answer)]
                )
            )

    return "OK"

実装コードの重要ポイント解説

1. 署名検証(InvalidSignatureErrorの捕捉)

LINEからの正規のリクエストであることを確かめるため、parser.parse(body_str, x_line_signature) を使っています。これにより、悪意のある第三者が直接WebhookのURLを叩いたとしても即座にエラー(400 Bad Request)となり、処理は実行されません。

2. 厳格なホワイトリスト判定

if user_id not in AUTHORIZED_USERS: で未許可ユーザーを弾いています。未承認ユーザーの場合は、固定の断り文言を返すだけでOpenAI APIは一切呼び出さないため、API費用の無駄遣いや情報漏洩の心配がありません。

3. 低めのTemperature(0.3)設定

AI社長の回答では、独創的なアイデアよりも「ブレのない安定した判断」が求められます。temperature=0.3 程度に低く設定することで、社長の判断基準に忠実な回答を出力させます。

まとめと次回予告

FastAPIとLINE SDK、OpenAI APIを組み合わせることで、わずか100行未満のコードで堅牢かつ知的なAI社長が立ち上がります。

最終回となる次回は、現場社員が毎日使いたくなる「リッチメニューUIの設計」と、使えば使うほどAI社長が賢くなる「ナレッジ更新サイクル」を解説し、連載の総括を行います。

次回予告:第10回(最終回)

次回(Part 10)は、連載の集大成として「現場に定着するリッチメニューUIと、会社と共に育つナレッジ更新サイクル」をお届けします。
ワンタップで相談できるLINE画面の作り方、日常の相談から新たな判断基準を吸い上げる改善ループ、そして「AI社長も言ってた」で自走する組織の作り方を総括します。

【全12回連載目次】中小企業のための「AI社長」実践導入ガイド

  1. 第1回:なぜ会社の情報を全部入れてはいけないのか?4つの情報分類と成功の骨格
  2. 第2回:普通のAIと「AI社長」を分けるもの:最重要ナレッジ「社長の判断基準」の言語化
  3. 第3回:思考と価値観を抽出する「社長への100問」:判断基準ファイルの作り方と実践質問集
  4. 第4回:組織図と役割分担の設計:AI社長が「誰に頼めばよいか」を迷わず案内する仕組み
  5. 第5回:失敗事例の蓄積と「禁止事項」:AI社長の経験値と絶対に譲れない境界線
  6. 第6回:プロンプトをパンクさせない3層システム設計:System Prompt・RAG・リアルタイムAPIの住み分け
  7. 第7回:なぜ「個人LINE+公式アカウント」なのか?中小企業の導入ハードルを下げる構成論
  8. 第8回:社内限定運用のセキュリティ設計:LINE User ID制限と社員認証コードによるアクセス制御
  9. 第9回:Pythonで作るLINE AI社長:Messaging APIとLLMをつなぐバックエンド実装 現在の記事
  10. 第10回:現場に定着するリッチメニューUIと、会社と共に育つナレッジ更新サイクル
  11. 第11回:ナレッジの保管場所とデータ管理術:GoogleスプレッドシートとGoogleドライブをAI社長の「記憶庫」にする方法
  12. 第12回:Googleスプレッドシート×Python連携実装:ナレッジをリアルタイムに読み込むバックエンド構築

eguchi.netをもっと見る

購読すると最新の投稿がメールで送信されます。