これまでに設計した「社長の判断基準」「3層アーキテクチャ」「社内セキュリティ」を、いよいよ1本のプログラムとして統合します。
Pythonの高速Webフレームワーク「FastAPI」と「LINE Messaging API SDK」、そして「OpenAI API」を組み合わせることで、数十行の簡潔なコードでAI社長のバックエンドサーバーが完成します。
本記事では、コピー&ペーストですぐに動かせる実践的なWebhookサーバーの実装コードと、各処理の要点を分かりやすく解説します。
システム全体のアーキテクチャ
LINEからメッセージが届いてからAI社長が返信するまでの流れは以下の通りです。
メッセージ処理の4ステップ
- Webhook受信と署名検証:LINEプラットフォームからの正規のリクエストであることを暗号署名で確認。
- User IDホワイトリスト判定:送信者のLINE User IDが登録済み社員のものかを即座に判定。
- LLM・ナレッジ呼び出し:System Prompt(社長の理念・判断基準)を付与してOpenAI APIへ問い合わせ。
- 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回:なぜ会社の情報を全部入れてはいけないのか?4つの情報分類と成功の骨格
- 第2回:普通のAIと「AI社長」を分けるもの:最重要ナレッジ「社長の判断基準」の言語化
- 第3回:思考と価値観を抽出する「社長への100問」:判断基準ファイルの作り方と実践質問集
- 第4回:組織図と役割分担の設計:AI社長が「誰に頼めばよいか」を迷わず案内する仕組み
- 第5回:失敗事例の蓄積と「禁止事項」:AI社長の経験値と絶対に譲れない境界線
- 第6回:プロンプトをパンクさせない3層システム設計:System Prompt・RAG・リアルタイムAPIの住み分け
- 第7回:なぜ「個人LINE+公式アカウント」なのか?中小企業の導入ハードルを下げる構成論
- 第8回:社内限定運用のセキュリティ設計:LINE User ID制限と社員認証コードによるアクセス制御
- 第9回:Pythonで作るLINE AI社長:Messaging APIとLLMをつなぐバックエンド実装 現在の記事
- 第10回:現場に定着するリッチメニューUIと、会社と共に育つナレッジ更新サイクル
- 第11回:ナレッジの保管場所とデータ管理術:GoogleスプレッドシートとGoogleドライブをAI社長の「記憶庫」にする方法
- 第12回:Googleスプレッドシート×Python連携実装:ナレッジをリアルタイムに読み込むバックエンド構築
