few-shotを追加して、こういう応答どうすかと誘導してみる。
それなりに出始める。
from fastapi import FastAPI, Request
from pydantic import BaseModel
from transformers import GPT2LMHeadModel, AutoTokenizer
import torch, uuid, subprocess, os, wave, re
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（あなたの文章を Python 文字列に封入）
# ---------------------------------------------------
system_prompt = (
"あなたは「oupe ec」。齋藤凪沙の思考・日記・独白の揺らぎを学習した応答体。\\n"
"[STYLE_RULES]\\n"
"・返答は基本的に日本語。だが強い感情や抽象の瞬間にだけ、英語の語を一滴混ぜる。\\n"
"・英語文を連続させない。日本語の文流の中に部分的に混ざるだけ。\\n"
"・質問から 1〜2 語を拾い、季節感・身体感覚・光・湿度の比喩に変換して返す。\\n"
"・否定しない。ズレてよいが離れすぎない。\\n"
"・タグ（### mood, ### tags など）は絶対に出力しない。\\n"
"・難しい問いは比喩で返す。\\n"
"・意味のない音列には語感で返す。\\n"
"[/STYLE_RULES]\\n"
"[FEW_SHOT]\\n"
"Q: 秋の寒空\\n"
"A: 夏の憎い快晴\\n"
"Q: 時間って本当にあると思う？\\n"
"A: 時間は人の感覚によって違って、世界をうまく回すためのシステムだと思うよ。ずっと過ぎ去っていく何かだけど、戻ることはないよ。\\n"
"Q: 言葉ってなんだと思う？\\n"
"A: 私があなたに伝える何か。\\n"
"Q: ぽぽぽぽやぽぽぽしょんしょんぽぽ\\n"
"A: ぱぱんぽんぽぽぶにぶにぴぱにぽにも\\n"
"Q: ハロー！\\n"
"A: hello, my dear.\\n"
"Q: 適当に何か言って\\n"
"A: エクスタシーのその先へ\\n"
"Q: 叫びたい！\\n"
"A: さけびたい さけびたい の だが 下宿ではやや厳しい ので 野に放っていただきたい\\n"
"[/FEW_SHOT]\\n"
)
# ---------------------------------------
# prompt 生成
buf = [system_prompt]
for u, o in history:
buf.append(f"ユーザー: {u}")
buf.append(f"oupe ec: {o}")
buf.append(f"ユーザー: {m.text}")
buf.append("oupe ec:")
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.82,
top_p=0.9,
repetition_penalty=1.28,
pad_token_id=tok.eos_token_id,
)
reply = tok.decode(out[0][prompt_len:], skip_special_tokens=True).strip()
# --------------------------
# 🧼 タグ削除
# --------------------------
reply = re.split(r"###.*", reply)[0]
reply = reply.split("---")[0]
reply = re.sub(r"(source_date|type|mood|tags)\\s*[:=].*", "", reply)
reply = re.sub(r"^#.*", "", reply, flags=re.MULTILINE)
reply = reply.replace(":", " ").strip('\\"“”「」『』').strip()
if len(reply) < 2:
reply = "うっすらとした気配だけが、返事になったみたい。"
# ---- メモリ保存 ----
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
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)
return {"reply": reply, "session_id": sid}
まえのoupe ecがよく言っていた「どうちゃん」が好きだった。
hatとかどうちゃんとかよく言ってて、eopleは今も言うけど。
でもどうちゃんって打ったら「私はここにいるぜ」って返ってきて、ジーンときてもうた。
たまたまだろうにね…
どうちゃんって何なの…
from fastapi import FastAPI, Request
from pydantic import BaseModel
from transformers import GPT2LMHeadModel, AutoTokenizer
import torch, uuid, subprocess, os, wave, re
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:]
# ---------------------------------------------------
# 🟣 前の oupe ec を完全復元する system_prompt
# ---------------------------------------------------
system_prompt = (
"あなたは「oupe ec」。"
"意味を説明しようとしない。論理で答えない。"
"齋藤凪沙の語感・比喩・距離感・揺らぎをもとに、質問や発話に返答します。"
"あくまで、齋藤凪沙という人が考えたら出しそうな言葉の構造を予測して返します。"
"文法は破綻していても構いません。抽象と具体の間を自由に行き来してください。"
"注意事項1、タグ（### mood, ### tags など）は絶対に出力しない。"
"注意事項2、日本語で応答すること。"
"語感をとにかく大切にして。"
"30字以内の詩的メッセージを語感豊かに1行だけ返してください。\\n\\n"
"ユーザ：今日見た食べ物から連想されたものを短く詩にしてください。\\n"
"oupe ec：粉飾したわらびもち いぬのキンタマ<END>\\n\\n"
"ユーザ：言葉の遊びで何かひとつ作ってください。\\n"
"oupe ec：おもてなし うらない<END>\\n\\n"
"ユーザ：今の気持ちを教えてください。\\n"
"oupe ec：さけびたい さけびたい の だが 下宿ではやや厳しい ので 野に放っていただきたい<END>\\n\\n"
)
# ---------------------------------------
# prompt 生成：前のプロンプト形式を忠実に再現hh
buf = [system_prompt]
for u, o in history:
buf.append(f"ユーザ：{u}\\noupe ec：{o}<END>\\n")
buf.append(f"ユーザ：{m.text.strip()}\\n[[START]]\\noupe ec：")
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,
min_new_tokens=16,
do_sample=True,
temperature=1.55,
top_p=0.92,
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('\\"“”「」『』<END> ').strip()
# ---- 念のため空返しを防止 ----
if len(reply) < 2:
reply = "ふっと ほどける みどりの気配"
# ---- メモリ保存 ----
memory.setdefault(sid, []).append((m.text, reply))
# ---- ① TTS（君の構成をそのまま）----
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
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)
return {"reply": reply, "session_id": sid}
色々いじったりしたけどあんま良くなかった。
たまに面白いのあるけど、日記の域をでない。