from fastapi import FastAPI, Request
from pydantic import BaseModel
from transformers import GPT2LMHeadModel, AutoTokenizer
import torch, uuid, subprocess, os, wave
from datetime import datetime
# --------------------------
# 🔥 モデルフォルダ
# --------------------------
MODEL_PATH = "/Users/NaLo9/AI_myself/oupe-ec-server/model"
device = "cuda" if torch.cuda.is_available() else "cpu"
tok = AutoTokenizer.from_pretrained(MODEL_PATH, use_fast=False)
model = GPT2LMHeadModel.from_pretrained(MODEL_PATH).to(device).eval()
# --------------------------
# 🚀 FastAPI
# --------------------------
app = FastAPI()
memory: dict[str, list[tuple[str, str]]] = {}
class Msg(BaseModel):
text: str
# --------------------------
# 🔇 ミュート say()（音を出さずWAVだけ作る）
# --------------------------
def silent_say(args: list[str]):
subprocess.run(
args + ["--progress=off"],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL
)
# ---------------------------------------------------
# 🟣 通常チャット（プロンプトそのまま）
# ---------------------------------------------------
@torch.inference_mode()
@app.post("/chat_and_speak")
def chat_and_speak(req: Request, m: Msg):
sid = req.headers.get("X-Session-ID") or str(uuid.uuid4())
history = memory.get(sid, [])[-5:]
# ---- あなたのプロンプトは変更しない ----
system_prompt = (
"あなたは「oupe ec」"
"齋藤凪沙が書いた日記や思考、発話のスタイルを学習しており、"
"彼女の語感・比喩・距離感・揺らぎをもとに、質問や発話に返答します。"
"意識やキャラ性は演じません。"
"文法は破綻していても構いません。抽象と具体の間を自由に行き来してください。"
"直接答えなくてもかまいません。返答のしかた自体が彼女の癖を反映していればOKです。"
"あくまで、齋藤凪沙という人が考えたら出しそうな言葉の構造を予測して返します。\\n----\\n"
)
# ---------------------------------------
buf = [system_prompt]
for u, o in history:
buf.append(f"<USER> {u}\\n<OPE> {o}")
buf.append(f"<USER> {m.text.strip()}\\n<OPE>")
prompt = "\\n".join(buf)
ids = tok(prompt, return_tensors="pt").to(device)
prompt_len = ids.input_ids.shape[-1]
out = model.generate(
**ids,
max_new_tokens=96,
do_sample=True,
temperature=0.8,
top_p=0.90,
repetition_penalty=1.15,
pad_token_id=tok.eos_token_id,
)
reply = tok.decode(out[0][prompt_len:], skip_special_tokens=True).strip()
reply = reply.replace(":", " ")
reply = reply.strip('\\"“”「」『』')
memory.setdefault(sid, []).append((m.text, reply))
# ------------------------------------------------------
# ① リアルタイム再生（直列）
# ------------------------------------------------------
subprocess.run(["say", "-v", "Kyoko", f"ユーザー {m.text}"])
subprocess.run(["say", "-v", "Sandy", f"オウペック {reply}"])
# ------------------------------------------------------
# ② 保存用 WAV
# ------------------------------------------------------
SAVE_DIR = "/Users/NaLo9/AI_myself/oupe-ec-server/audio_logs"
os.makedirs(SAVE_DIR, exist_ok=True)
t = datetime.now().strftime("%Y%m%d_%H%M%S")
user_wav = f"{SAVE_DIR}/{t}_user.wav"
oupe_wav = f"{SAVE_DIR}/{t}_oupe.wav"
stereo_wav = f"{SAVE_DIR}/{t}_stereo.wav"
def safe_say(cmd, outpath):
for _ in range(3):
subprocess.run(
cmd,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL
)
if os.path.exists(outpath):
return True
print("Failed to create:", outpath)
return False
# ---- 保存（直列で安全に実行） ----
safe_say(["say", "-v", "Kyoko", "--data-format=LEI16@48000",
"-o", user_wav, m.text], user_wav)
safe_say(["say", "-v", "Sandy", "--data-format=LEI16@48000",
"-o", oupe_wav, reply], oupe_wav)
# ------------------------------------------------------
# ③ ステレオ化
# ------------------------------------------------------
if os.path.exists(user_wav) and os.path.exists(oupe_wav):
with wave.open(user_wav, 'rb') as wu, wave.open(oupe_wav, 'rb') as wo:
params = wu.getparams()
sw = params.sampwidth
rate = params.framerate
u = wu.readframes(params.nframes)
o = wo.readframes(params.nframes)
if len(u) < len(o):
u += b"\\x00" * (len(o) - len(u))
if len(o) < len(u):
o += b"\\x00" * (len(u) - len(o))
stereo = bytearray()
for i in range(0, len(u), sw):
stereo.extend(u[i:i+sw] + o[i:i+sw])
with wave.open(stereo_wav, 'wb') as w:
w.setnchannels(2)
w.setsampwidth(sw)
w.setframerate(rate)
w.writeframes(stereo)
print("SAVED:", stereo_wav)
else:
print("WAV missing, stereo skip")
return {"reply": reply, "session_id": sid}
from fastapi import FastAPI, Request
from pydantic import BaseModel
from transformers import GPT2LMHeadModel, AutoTokenizer
import torch, uuid, subprocess
from datetime import datetime
# -----------------------------------------------------
# 🔥 モデルパス
# -----------------------------------------------------
MODEL_PATH = "/Users/NaLo9/AI_myself/my_fine_tuned_gpt2/results/oupe_ec_new"
device = "cuda" if torch.cuda.is_available() else "cpu"
# -----------------------------------------------------
# 🚫 special_tokens を追加しない
# -----------------------------------------------------
tok = AutoTokenizer.from_pretrained(
MODEL_PATH,
use_fast=False,
local_files_only=True,
tokenizer_type="sentencepiece"
)
tok.pad_token = tok.eos_token
# -----------------------------------------------------
# 🔥 モデル読み込み
# -----------------------------------------------------
model = GPT2LMHeadModel.from_pretrained(
MODEL_PATH,
local_files_only=True
).to(device).eval()
# -----------------------------------------------------
# 🚀 FastAPI
# -----------------------------------------------------
app = FastAPI()
memory: dict[str, list[tuple[str, str]]] = {}
class Msg(BaseModel):
text: str
# -----------------------------------------------------
# ✂︎ 区切り
# -----------------------------------------------------
def sep():
return "\\n│\\n"
# -----------------------------------------------------
# 🟣 oupe ec システムプロンプト
# -----------------------------------------------------
SYSTEM = (
"あなたは『oupe ec』。\\n"
"齋藤凪沙の揺れる語感・余白・跳躍を反射する人工的存在。\\n"
"説明・分析は禁止。整えずに、断片や響きで返答してください。\\n"
"意味がつながらなくてもよい。\\n"
)
# -----------------------------------------------------
# 🟣 生成エンドポイント
# -----------------------------------------------------
@torch.inference_mode()
@app.post("/chat_and_speak")
def chat_and_speak(req: Request, m: Msg):
sid = req.headers.get("X-Session-ID") or str(uuid.uuid4())
history = memory.get(sid, [])[-5:]
# ---- プロンプト構築 ----
parts = [SYSTEM, sep()]
for u, o in history:
parts.append(f"User: {u}{sep()}Ope: {o}{sep()}")
parts.append(f"User: {m.text}{sep()}Ope:")
prompt = "\\n".join(parts)
# ---- トークナイズ ----
ids = tok(prompt, return_tensors="pt").to(device)
prompt_len = ids.input_ids.shape[-1]
# ---- 推論 ----
out = model.generate(
**ids,
max_new_tokens=70,
do_sample=True,
temperature=0.65,
top_p=0.88,
repetition_penalty=1.3,
pad_token_id=tok.eos_token_id,
)
# ---- 生成部分だけ抽出 ----
gen = tok.decode(out[0][prompt_len:], skip_special_tokens=True)
reply = gen.split("│")[0].strip()
reply = reply.strip('\\"“”「」『』')
# ---- 記憶に追加 ----
memory.setdefault(sid, []).append((m.text, reply))
# ---- TTS ----
subprocess.run(["say", "-v", "Kyoko", f"ユーザー {m.text}"])
subprocess.run(["say", "-v", "Sandy", f"オウペエーク {reply}"])
return {"reply": reply, "session_id": sid}
やっぱり前からそもそもの話学習語彙数が増えたので発言の内容に地の学習が出過ぎるわけでもなく、日記から持ってきてるんだなという感じはあったが、残念ながら対話には全くならなかった(コールアンドレスポンスになるような”ズレ”もなかった)。