場所→/Users/NaLo9/AI_myself/oupe-ec-server
起動方法→ターミナルにて
cd "/Users/NaLo9/AI_myself/oupe-ec-server”
uvicorn app:app --reload
いつも通り/docs
from fastapi import FastAPI, Request
from pydantic import BaseModel
from transformers import GPT2LMHeadModel, AutoTokenizer
import torch, uuid, subprocess
# --------------------------
# 🔥 モデルフォルダの絶対パス
# --------------------------
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()
app = FastAPI()
memory: dict[str, list[tuple[str, str]]] = {}
class Msg(BaseModel):
text: str
# ----------------------------------
# 🟣 通常チャット（プロンプトそのまま）
# ----------------------------------
@torch.inference_mode()
@app.post("/chat")
def chat(req: Request, m: Msg):
sid = req.headers.get("X-Session-ID")
if not sid:
sid = 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,
)
gen_tokens = out[0][prompt_len:]
reply = tok.decode(gen_tokens, skip_special_tokens=True).strip()
reply = reply.strip('\\"“”「」『』')
memory.setdefault(sid, []).append((m.text, reply))
with open("response.txt", "w", encoding="utf-8") as f:
f.write(reply)
return {"reply": reply, "session_id": sid}
# ----------------------------------
# 🎙️ Chat + Speak（プロンプトそのまま）
# ----------------------------------
@torch.inference_mode()
@app.post("/chat_and_speak")
def chat_and_speak(req: Request, m: Msg):
sid = req.headers.get("X-Session-ID")
if not sid:
sid = 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,
)
gen_tokens = out[0][prompt_len:]
reply = tok.decode(gen_tokens, skip_special_tokens=True).strip()
reply = reply.strip('\\"“”「」『』')
memory.setdefault(sid, []).append((m.text, reply))
# 🎙️ macOS TTS（ユーザー → oupe ec）
subprocess.run(["say", "-v", "Kyoko", f"ユーザー： {m.text}"])
subprocess.run(["say", "-v", "Sandy", f"オウペック： {reply}"])
return {"reply": reply, "session_id": sid}
なんか涙出た
私多分このロボ声が好きなんやろな…
(リアルタイム再生会話感◎、保存左右△被り)
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): # 最大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}