前回記事の続きになります。

前回、ChatGPTと対話を繰り返して、AIチャットアプリの要件整理を行いました。
- 使用するテックスタックの整理
- 簡易システム構成図の作成
- API設計
- テーブル設計
この記事では、実際にbackend側のコードをChatGPTに生成してもらいます。
6th prompt (Backend Models)
前回までのやり取りをChatGPTをくみ取ってくれるので、以下のプロンプトだけで実装するコードを生成してくれます。
はい、ではSQLAlchemyを用いて実装を進めてください。
以下、ChatGPTからの返答。
了解です!それでは、SQLAlchemy(+ FastAPI + PostgreSQL) を使って、先ほどの user
, conversation
, message
テーブルのモデルを実装していきます。
✅ 使用前提
- 使用するORM:
SQLAlchemy 2.0
スタイル(モダンな書き方) - DB接続:
SQLAlchemy + asyncpg
(Supabase互換) - マイグレーション管理:
Alembic
(後で導入可能)
📁 ディレクトリ構成(シンプルな例)
backend/
├── app/
│ ├── models/
│ │ ├── base.py
│ │ ├── user.py
│ │ ├── conversation.py
│ │ └── message.py
│ ├── db.py
│ ├── main.py
│ └── config.py
✅ config.py(環境設定)
import os
DATABASE_URL = os.getenv("DATABASE_URL", "postgresql+asyncpg://user:password@host:port/dbname")
✅ db.py(接続設定)
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker, declarative_base
from app.config import DATABASE_URL
engine = create_async_engine(DATABASE_URL, echo=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
Base = declarative_base()
✅ models/base.py
from app.db import Base
✅ models/user.py
from sqlalchemy import Column, String, DateTime
from sqlalchemy.dialects.postgresql import UUID
import uuid
from datetime import datetime
from app.db import Base
class User(Base):
__tablename__ = "user"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
email = Column(String, unique=True, index=True)
name = Column(String)
preferred_language = Column(String, default="ja")
created_at = Column(DateTime, default=datetime.utcnow)
✅ models/conversation.py
from sqlalchemy import Column, String, ForeignKey, DateTime
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
import uuid
from datetime import datetime
from app.db import Base
class Conversation(Base):
__tablename__ = "conversation"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
user_id = Column(UUID(as_uuid=True), ForeignKey("user.id", ondelete="CASCADE"))
title = Column(String, nullable=True)
created_at = Column(DateTime, default=datetime.utcnow)
messages = relationship("Message", back_populates="conversation")
✅ models/message.py
from sqlalchemy import Column, String, ForeignKey, DateTime, CheckConstraint
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
import uuid
from datetime import datetime
from app.db import Base
class Message(Base):
__tablename__ = "message"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
conversation_id = Column(UUID(as_uuid=True), ForeignKey("conversation.id", ondelete="CASCADE"))
role = Column(String, nullable=False)
content = Column(String, nullable=False)
translated_content = Column(String, nullable=True)
created_at = Column(DateTime, default=datetime.utcnow)
__table_args__ = (
CheckConstraint("role IN ('user', 'assistant')", name="valid_role"),
)
conversation = relationship("Conversation", back_populates="messages")
✅ main.py でテーブル作成確認
import asyncio
from app.db import engine, Base
from app.models import user, conversation, message
async def init_models():
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
if __name__ == "__main__":
asyncio.run(init_models())
このスクリプトを一度実行すれば、Supabase側にテーブルが作成されます!
次のステップ提案
- CRUDの作成(例:ユーザー登録、会話作成、履歴取得)
- APIエンドポイントと紐づけ
- Whisper・GPT・gTTSの処理の追加
- Alembicでマイグレーション対応(任意)
ご希望であれば、これらのAPI(例:会話開始 / メッセージ保存)も実装していきましょうか?
7th prompt(Backend APIs)
python 3.12 を利用するとして、requirements.txt を定義してください。利用するパッケージのバージョンも指定してください。
その上でAPIの実装を進めてください。
✅ requirements.txt
fastapi==0.110.0
uvicorn[standard]==0.29.0
# DB
SQLAlchemy==2.0.29
asyncpg==0.29.0
# gTTS(音声出力用)
gTTS==2.3.2
# Whisper用(OpenAI公式API利用)
openai==1.17.0
# 認証管理(任意)
python-dotenv==1.0.1
# その他(型チェックや便利系)
httpx==0.27.0
✅ 次に実装する内容
今回は以下のような流れでAPIを段階的に実装していきます。
📦 ステップ順
POST /transcribe
: 音声 → テキスト(Whisper API)POST /chat
: GPTで返答生成POST /tts
: gTTSで返答音声生成POST /translate
: テキスト翻訳POST /history
: 会話履歴保存GET /history
: 会話履歴取得
ステップ1:POST /transcribe(Whisper API)
📄 routers/transcribe.py
from fastapi import APIRouter, UploadFile, File, Form
from fastapi.responses import JSONResponse
import openai
import os
from dotenv import load_dotenv
load_dotenv()
openai.api_key = os.getenv("OPENAI_API_KEY")
router = APIRouter()
@router.post("/transcribe")
async def transcribe_audio(audio_file: UploadFile = File(...), language: str = Form(None)):
try:
contents = await audio_file.read()
response = openai.audio.transcriptions.create(
model="whisper-1",
file=(audio_file.filename, contents),
language=language
)
return {"transcription": response.text}
except Exception as e:
return JSONResponse(status_code=500, content={"error": str(e)})
ステップ2:POST /chat(GPTで返答)
📄 routers/chat.py
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
import openai
import os
router = APIRouter()
class ChatRequest(BaseModel):
message: str
history: list
language: str = "ja"
@router.post("/chat")
async def chat_with_gpt(data: ChatRequest):
try:
messages = [{"role": m["role"], "content": m["content"]} for m in data.history]
messages.append({"role": "user", "content": data.message})
response = openai.chat.completions.create(
model="gpt-3.5-turbo",
messages=messages
)
reply = response.choices[0].message.content
return {"reply": reply}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
ステップ3:POST /tts(gTTS)
📄 routers/tts.py
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from fastapi.responses import StreamingResponse
from gtts import gTTS
import io
router = APIRouter()
class TTSRequest(BaseModel):
text: str
language: str = "ja"
@router.post("/tts")
async def text_to_speech(data: TTSRequest):
try:
tts = gTTS(text=data.text, lang=data.language)
fp = io.BytesIO()
tts.write_to_fp(fp)
fp.seek(0)
return StreamingResponse(fp, media_type="audio/mpeg")
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
ステップ4:POST /translate(GPT翻訳)
📄 routers/translate.py
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
import openai
router = APIRouter()
class TranslateRequest(BaseModel):
text: str
target_language: str
@router.post("/translate")
async def translate_text(data: TranslateRequest):
try:
prompt = f"以下の文章を{data.target_language}に翻訳してください。\\n\\n{data.text}"
response = openai.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}]
)
translated = response.choices[0].message.content
return {"translated_text": translated}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
✅ ステップ5 & 6:履歴の保存と取得(POST/GET /history
)
こちらはすでに作ったSQLAlchemyモデルを用いて、次のターンでCRUDを組み立てます。
🔧 次ステップ
- 履歴(
/history
)の保存・取得をSQLAlchemyで実装 - ルーターの統合 &
main.py
でアプリ起動 .env.example
も作ってセキュアなAPIキー管理
続けて履歴系のAPIを作成してよろしいですか?また、ここまでのコードをファイルにまとめる形も可能です!
8th prompt (fix dependency error)
実際に生成してもらったコードが動作するか試して、パッケージの依存関係に関するエラーが出たので、それを解決するように依頼しています。
次に進む前に以下のエラーを解決したいです。
依存関係に競合があるようです。
ERROR: pip's dependency resolver does not currently take into account all the packages that are installed. This behaviour is the source of the following dependency conflicts.
aiohttp 3.8.4 requires async-timeout<5.0,>=4.0.0a3, but you have async-timeout 5.0.1 which is incompatible.
以下、ChatGPTからの返答。
❌ エラーの内容(要点)
このエラーは:
aiohttp==3.8.4 が async-timeout<5.0 を必要としているのに、現在 async-timeout==5.0.1 が入っている
という依存関係の競合によるものです。
✅ 解決方法(どれか一つ)
✅ 方法1:async-timeout
のバージョンを明示的に固定する(おすすめ)
requirements.txt
に以下を追加または修正します:
async-timeout==4.0.3
そして以下のようにインストール:
pip install -r requirements.txt --force-reinstall
✅ 方法2:aiohttp
を外す(もし未使用なら)
あなたのプロジェクトでは明示的に aiohttp
は使っていないため、aiohttp
をアンインストールするのもOKです:
pip uninstall aiohttp
(※ただし、依存パッケージのどれかが aiohttp を内部的に使っている場合はこの方法は不可)
✅ 方法3:pip
に依存関係を無視させる(緊急時のみ)
これはあまり推奨しませんが、下記のようにすれば強制インストールできます:
pip install -r requirements.txt --use-deprecated=legacy-resolver
🔁 あなたにおすすめの選択肢
✅ 方法1:「async-timeout==4.0.3」に下げる
この方法が最もクリーンで安全です。
aiohttp==3.8.x
は async-timeout==4.x
にしか対応していないため、5.x に上げてはいけません。
📌 修正後の requirements.txt(抜粋)
fastapi==0.110.0
uvicorn[standard]==0.29.0
SQLAlchemy==2.0.29
asyncpg==0.29.0
# async-timeout を aiohttp に合わせる
async-timeout==4.0.3
gTTS==2.3.2
openai==1.17.0
python-dotenv==1.0.1
httpx==0.27.0
修正後に再度インストールして、問題が解決するか確認してみてください!
もし pip freeze
の出力やエラー全文を貼っていただければ、さらに深堀りもできます。
9th prompt (step5,6: history)
async-timeout==4.0.3
に設定して進めました。
ステップ1~4までのメソッドの動作確認はできたので、ステップ5, 6: 履歴の保存と取得を実装してくれますか?
以下、ChatGPTからの返答。
✅ 前提
既に以下のSQLAlchemyモデルがあるものとします:
User
Conversation
Message
FastAPIにそれを組み込んで、以下のエンドポイントを作成します:
📘 API仕様
メソッド | エンドポイント | 説明 |
---|---|---|
POST |
/history |
履歴(会話・メッセージ)を保存 |
GET |
/history/{user_id} |
特定ユーザーの履歴を取得 |
📁 ディレクトリ例(前提)
app/
├── db.py
├── models/
│ ├── user.py
│ ├── conversation.py
│ └── message.py
├── routers/
│ └── history.py ← 今回ここに追加
└── main.py
ステップ5, 6: /history(履歴保存・取得)
✅ routers/history.py
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from typing import List
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from app.db import async_session
from app.models.user import User
from app.models.conversation import Conversation
from app.models.message import Message
import uuid
from datetime import datetime
router = APIRouter()
# Dependency
async def get_session() -> AsyncSession:
async with async_session() as session:
yield session
# --- Request Models ---
class MessageCreate(BaseModel):
role: str # 'user' or 'assistant'
content: str
translated_content: str | None = None
class HistoryCreate(BaseModel):
user_id: uuid.UUID
title: str
messages: List[MessageCreate]
# --- POST /history ---
@router.post("/history")
async def save_history(data: HistoryCreate, session: AsyncSession = Depends(get_session)):
try:
conversation = Conversation(
id=uuid.uuid4(),
user_id=data.user_id,
title=data.title,
created_at=datetime.utcnow()
)
session.add(conversation)
# reflect conversation.id in DB
await session.flush()
for m in data.messages:
message = Message(
id=uuid.uuid4(),
conversation_id=conversation.id,
role=m.role,
content=m.content,
translated_content=m.translated_content,
created_at=datetime.utcnow()
)
session.add(message)
await session.commit()
return {"conversation_id": str(conversation.id)}
except Exception as e:
await session.rollback()
raise HTTPException(status_code=500, detail=str(e))
# --- GET /history/{user_id} ---
@router.get("/history/{user_id}")
async def get_user_history(user_id: uuid.UUID, session: AsyncSession = Depends(get_session)):
try:
result = await session.execute(
select(Conversation).where(Conversation.user_id == user_id).order_by(Conversation.created_at.desc())
)
conversations = result.scalars().all()
history = []
for conv in conversations:
conv_result = await session.execute(
select(Message).where(Message.conversation_id == conv.id).order_by(Message.created_at)
)
messages = conv_result.scalars().all()
history.append({
"conversation_id": str(conv.id),
"title": conv.title,
"created_at": conv.created_at,
"messages": [
{
"role": msg.role,
"content": msg.content,
"translated_content": msg.translated_content,
"created_at": msg.created_at
} for msg in messages
]
})
return history
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
✅ main.py にルーターを登録
from fastapi import FastAPI
from app.routers import transcribe, chat, tts, translate, history
app = FastAPI()
app.include_router(transcribe.router)
app.include_router(chat.router)
app.include_router(tts.router)
app.include_router(translate.router)
app.include_router(history.router)
request body (POST /history)
{
"user_id": "f5b3d143-xxxx-yyyy-zzzz-abcdef123456",
"title": "初回会話",
"messages": [
{"role": "user", "content": "こんにちは", "translated_content": "Hello"},
{"role": "assistant", "content": "こんにちは、何かお手伝いしましょうか?", "translated_content": "Hello, how can I help?"}
]
}
✅ 次ステップ(希望があれば)
- Supabaseと接続して実データ保存確認
- 認証(JWT or Supabase Auth)
- 会話タイトルの自動生成(ChatGPT使う)
- 会話の削除 / 編集

Step1~4まではDBを必要しない処理で、Step5, 6では、ConversationとMessageモデルを使用するため、DB接続が必要になります。
10th prompt (fix error)
/history/{user_id}
というGETメソッドを定義してもらいましたが、これは最終的にはアプリで使わないようにしています。理由は、全てのconversationとmessageデータを取得する処理で大量のデータを返すことになってしまうためです。
1つconversationに対するmessageを取得処理する処理があれば十分なので、後で実装します。
次のステップに行く前に、post /history に問題がありますので、先に修正をしたいです。
messageには、conversation_idにforeign_keyがあります。
save_historyでは、最後にcommitしようとしますが、まだconversationテーブルにidが保存されていないせいか、メソッドを実行すると、AttributeError: NoForeignKeysErrorとなります。
まずエラーとならないように修正ください。
❗️原因の詳細
session.add(conversation)
は まだDBにINSERTしていない状態。conversation.id
を使ってメッセージにセットしているが、DB的にconversation.id
は未確定。- その状態で commit すると、SQLAlchemy が「外部キーに関連づけられる主キーが無い」と怒る →
NoForeignKeysError
✅ 解決方法:await session.flush()
を追加!
flush()
は 今のセッションの変更をDBに反映して、主キーなどを確定させる処理です(commit しない)。
✅ 修正済みの save_history
関数
@router.post("/history")
async def save_history(data: HistoryCreate, session: AsyncSession = Depends(get_session)):
try:
# 会話を作成
conversation = Conversation(
id=uuid.uuid4(),
user_id=data.user_id,
title=data.title,
created_at=datetime.utcnow()
)
session.add(conversation)
# 💡 FLUSHでconversation.idをDBに反映させる(ここが重要)
await session.flush()
# メッセージを追加
for m in data.messages:
message = Message(
id=uuid.uuid4(),
conversation_id=conversation.id, # ←ここで外部キー利用
role=m.role,
content=m.content,
translated_content=m.translated_content,
created_at=datetime.utcnow()
)
session.add(message)
# 最終的にコミット
await session.commit()
return {"conversation_id": str(conversation.id)}
except Exception as e:
await session.rollback()
raise HTTPException(status_code=500, detail=str(e))
🔄 補足
flush()
はコミットではないため、ロールバックも可能。flush()
でconversation.id
が確定され、以降で外部キーとして利用可能に。
これでエラーは解消されるはずです!
動作確認後、GET /history
と併せて会話履歴が正しく保存されているかも確認してみてください。
次に進む準備ができたら教えてください!
11th prompt (check flow)
次に進む前にフローを確認したいです。
今まで作成したメソッドがあれば、音声入力の受付からAIとの会話と音声データの返却までできているのでしょうか?
はい、フローの全体を振り返って確認するのはとても良い判断です!
ここまで実装してきた機能を整理しながら、音声入力 → 会話 → 音声返答までが成立しているかどうかを見ていきましょう。
✅ これまでに実装したステップの流れ
ステップ | 処理内容 | 実装状況 | エンドポイント |
---|---|---|---|
1 | 音声入力をテキストに変換(Whisper) | ✅ 完了 | POST /transcribe |
2 | ユーザーの入力に対してChatGPTで応答を生成 | ✅ 完了 | POST /chat |
3 | ChatGPTからの応答テキストを音声に変換(gTTS) | ✅ 完了 | POST /tts |
4 | 応答テキストの翻訳(任意) | ✅ 完了 | POST /translate |
5 | 会話内容(履歴)をDBに保存 | ✅ 完了(flush() 修正済) |
POST /history |
6 | 会話履歴をユーザー単位で取得 | ✅ 完了 | GET /history/{user_id} |
🔄 全体の処理フロー(API連携順)
以下のようなフローで 音声から音声の返答までが可能です:
- 🔴 音声入力(マイクで録音)
- 🎤
POST /transcribe
に音声ファイルを送信 → Whisperで文字に変換 - 💬
POST /chat
に文字列を送信 → OpenAI Chatで返答を生成 - 🌐 (任意)
POST /translate
で翻訳する(別言語UIなら) - 🔊
POST /tts
に返答テキストを送信 → gTTSで音声に変換 - 💾
POST /history
にユーザー・会話データを送信 → DBに保存
✅ 結論
はい、ここまでで:
- 音声入力
- AIとのやり取り
- 音声による返答
- 翻訳(オプション)
- 会話履歴の保存・取得
までの一連の音声AI会話体験が可能な状態になっています。
🧩 次に進めること(いくつかの候補)
- 🔐 認証機能(ユーザー登録、ログイン、トークン発行)
- 🌐 言語設定(ユーザー単位 or 会話単位の言語設定)
- 🧠 会話タイトル自動生成(ChatGPTを活用)
- 💻 フロントエンドの作成(Next.js + Tailwindで実装開始)
- 📦 Supabase 本番接続 or Railway デプロイ
- 🧪 テストコードの整備(pytestなど)
この後、どの部分を進めたいですか?フロントエンド?認証?もしくはデプロイ環境の設定?
12th prompt (ask about audio)
音声入力に関して、知識が乏しかったので、以下のような質問を投げています。
🔴 音声入力(マイクで録音
この部分はAPIは必要なく、フロントエンド側でできるということでしょうか?
またユーザーが入力した音声は、再生できる必要はないのですが、AIの返答と音声データはいつでも再生して聞けるようにしたいです。
今のmessageテーブルには、音声データとの関連性がないかと思います。
どのように実現できますか?
🎤 Q1. 音声入力(マイク録音)はAPIが必要か?
→ APIは不要、フロントエンドのみで完結します。
✅ フロントエンド(例:Next.js)で行うこと:
- ブラウザのマイクアクセス(
MediaRecorder
APIなど) - 録音された音声データを
.webm
か.wav
で取得 - 音声ファイルを
POST /transcribe
に送信してテキスト化
🔊 Q2. AIの返答音声を再生できるようにするには?
はい、返答(AIの返した音声データ)をいつでも再生できるようにするには:
- 音声ファイルをサーバー or ストレージに保存しておく
- DBにそのファイルのURLを保持する
🧩 解決案:Messageテーブルに audio_url
カラムを追加
✅ 修正後の Message
モデル(SQLAlchemy)
from sqlalchemy import Column, String, ForeignKey, DateTime, Enum
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
import uuid
from datetime import datetime
class Message(Base):
__tablename__ = "messages"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
conversation_id = Column(UUID(as_uuid=True), ForeignKey("conversations.id"), nullable=False)
role = Column(String, nullable=False) # 'user' or 'assistant'
content = Column(String, nullable=False)
translated_content = Column(String, nullable=True)
audio_url = Column(String, nullable=True) # 追加:返答音声のURL(gTTS後に保存)
created_at = Column(DateTime, default=datetime.utcnow)
conversation = relationship("Conversation", back_populates="messages")
🔁 返答音声の保存フロー(POST /tts
の改善)
📦 方法例1:サーバー内に保存
- 返答音声ファイル(mp3)を一時的なパスに保存
- ファイル名を UUID +
.mp3
にするなど Message.audio_url
にそのファイルのURL(/static/audio/...
)を保存
# 例: FastAPIでファイルを保存
from fastapi.staticfiles import StaticFiles
app.mount("/static", StaticFiles(directory="static"), name="static")
✅ 結論まとめ
課題 | 解決策 |
---|---|
音声録音のAPIは必要か? | ❌ 必要なし。ブラウザJS(MediaRecorder)で実装可能 |
AI返答の音声を再生したい | ✅ サーバーに音声を保存し、Messageテーブルに audio_url を追加 |
将来的に管理・再生するには? | ✅ /static/audio や S3/Supabase Storage に保存してURL管理 |
次ステップご提案
Message.audio_url
をDBに追加(マイグレーション)POST /tts
を修正:音声ファイル保存+URLを返すPOST /history
にそのURLを含めるGET /history
でそのURLを返し、フロントで再生できるように
進めたいステップを教えていただければ、具体的なコードを書いていきます!
たとえば「ローカルにmp3を保存してURLを返す方法を書いて」など、細かく指定いただいてもOKです。
ここまでのまとめ
ChatGPTと対話を繰り返して、概ね必要そうなbackendの処理を実装していきました。
Transcribe(Speech-to-Text) → LLM → TTS(Text-to-Speech) というステップでAIとの会話ができるようにメソッドを定義していきました。
また後から会話履歴を見れるようにmessageを保存する処理とaudio_urlを用意して、音声ファイルをいつでも再生できるようにカラムを追加しました。
次にaudio_urlの実ファイルをSupabase Storageに保存して、保存されたファイルのURLをaudio_urlに設定してきます。
Supabase StorageにFastAPIからファイルアップロードする備忘録

SQLAlchemy Table Migration with Alembic
