プロローグ：そのAIに「魂」はあるか
「推しと会話がしたい」
それは全人類が抱く、叶わぬ願い。いや、現代にはLLMという魔法がある。
しかし、私は絶望した。
既存のAIにキャラ設定を食わせたところで、それはただの「物真似」に過ぎないのだ。
AIは、あの日の誓いも、宿敵との因縁も、ヒロインに向けられた微かな視線の意味も知らない。
「文脈（Context）」なきロールプレイに、魂は宿らない。
だったら作ればいいじゃないか。
「小説自体をAIに読破させ、物語の記憶を完全に保持した状態で、自分自身のシステムプロンプトを書かせる」
そんな狂気のパイプラインを。
これは、ローカルLLMと共に「推しの魂」を錬成するまでの、汗と涙の冒険譚である。
冒険者の心得（注意）
本記事は技術的探求の記録です。解析対象（聖典）の取り扱いについては、各サイトの利用規約および著作権法という「王国の掟」を遵守してください。
第０章：儀式の準備（セットアップ）
最強の魔道具を動かすには、正しい素材と配置が必要だ。
1. 聖典（テキストデータ）の入手
今回はサンプルとして、語り継がれる名作『薬屋のひとりごと』（日向夏 先生 著）の「猫猫」と「壬氏」の会話を模倣することを目指す。
- 聖典の在処: 薬屋のひとりごと（小説家になろう）
ここで重要なのは、データの入手方法だ。私は決して禁じられた魔術（スクレイピング）は使っていない。
「小説家になろう」にログインすれば、各話のページ下部にある「TXTダウンロード」から、エピソードごとのテキストファイルを正当に入手できるのだ。これを地道に集めるのが、真の魔導士への第一歩だ。
2. 魔導工房の配置（ディレクトリ構造）
入手した聖典は、以下の構造で配置せよ。この形こそが、パイプラインが魔力を正しく循環させるための黄金比だ。
.
├── data
│ └── N9636X # 小説ID（フォルダ名）
│ └── text # ここにダウンロードしたTXTを格納
│ ├── N9636X-1.txt
│ ├── N9636X-2.txt
│ ├── ...
│ └── N9636X-56.txt
├── requirements.txt # 解析を補助する五柱の眷属（ライブラリ）
├── generate_prompt_llm.py # 人格錬成の術式
└── pipeline.py # 文脈解析の術式
第１章：精度向上のための試行錯誤（クエスト・ログ）
結論から言おう。この開発は泥沼だった。
最初は「テキストを投げればいいんだろ？」と舐めていたが、そこには想像を絶する魔物たちが潜んでいたのだ。
【試作１号】単純抽出の敗北
「とりあえず話者ごとのセリフを抽出！」
私は無詠唱魔法（単純なプロンプト）を放った。
- 結果: 惨敗。
- 誰のセリフか判別できず
話者:unknownの山。 - 文脈がないため、独り言と会話の区別すらつかない。
- AIは「彼」が誰なのか理解できず、虚空を見つめるばかり。
【試作２号】局所的あらすじ（ローカルコンテキスト）の導入
「ならば、その場の状況を教えればいい」
各エピソードのあらすじを生成し、それをヒントにする実装を追加。
- 結果: 少しマシになったが、まだ足りない。
- AIは「その場」のことは分かるが、「物語全体」が見えていない。
- 過去の因縁（伏線）を含んだセリフの判定でバグり散らかす。
【試作３号】未来視（グローバルステート）の実装
ここで私は気づいてしまった。
「第1話の『謎の男』の正体を知るには、第10話まで読まなければならない」
つまり、時間を超越する必要がある。
- 手法: 「全話解析（Analysis）」と「抽出（Extraction）」を完全に分離。
- まず全話を読破し、未来までの知識（Global State）を完成させる。
- その「未来の記憶」を持った状態で、過去（第1話）に戻ってセリフを抽出する。
- 結果: 覚醒。
- 第1話の時点で「あ、これラスボスのセリフですね」と特定可能に。
- ここで初めて、私のパイプラインは時間を支配した。
【試作４号】攻略本（ブートストラップ）による知識注入
だが、まだ安定しない。AIがキャラ名を間違えたり、設定を忘れたりする。
そこで私は禁断の魔法を使った。
- 手法: 「登場人物紹介」の回（36話、56話）を、第1話よりも先に読ませる。
- 結果: 完全勝利。
- まるで「攻略本」を読んでからニューゲームを始めたかのような安定感。
- AIは最初から「猫猫」「壬氏」の解像度がMAXの状態でスタートする。
第２章：俺の最強装備（実行環境）を見てくれ
この魔術を行使するには、相応の魔力（スペック）が必要だ。
並のPCではVRAM不足で焼き切れるだろう。
これが私の相棒、GIGABYTE AI TOP ATOM だ。
- 筐体: GIGABYTE AI TOP ATOM（DGX Spark互換機）
- 魔力タンク: 128GB Unified Memory
- OS: Ubuntu 24.04.3 LTS
- LLMエンジン: LM Studio 0.3.39
- 精霊（LLM）: Qwen/Qwen3-Next-80B-A3B-Instruct-GGUF
- コンテキスト長: 65535 (32kでは足りぬ……！)
- 繰り返しペナルティ: 1.25 (AIの暴走を抑える拘束具)
第３章：発動呪文（実行フロー）
度重なるクラッシュと無限ループ（精神汚染）を回避するため、儀式は３段階に分けて行う。
Phase 1: 知識の構築（アカシックレコード・アクセス）
まずは物語の骨格を作る。攻略本（ブートストラップ）を片手に、あらすじだけを解析して記憶領域（Global State）に焼き付ける。
# 36話と56話の登場人物紹介を読んだうえで全話のあらすじを先読みし、未来の知識を得る！
$ python3 pipeline.py N9636X --reset --bootstrap 36,56 --start 1 --end 36 --analyze-only
Phase 2: 素材の採掘（ミスリル・マイニング）
完成した記憶領域をロードし、それを正解データとしてセリフを抽出する。
もはやAIに迷いはない。「あの男」が誰なのか、完全に理解しているからだ。
# 全知全能の状態で、過去を再定義する
$ python3 pipeline.py N9636X --extract-only --start 1 --end 36
Phase 3: 人格の召喚（サモン・ペルソナ）
最後は、集めた大量のデータ（性格、口調、セリフ、思考）を統合し、「AIプロデューサー」に命じる。
「君は最高のAIプロデューサーだ。この膨大な資料を元に、このキャラになりきるための『究極のシステムプロンプト』を記述せよ」
# 顕現せよ、猫猫！
$ python3 generate_prompt_llm.py N9636X "猫猫" --user "壬氏"
エピローグ：転生したらAIだった件
最初の36話分(8時間くらいかかった)を読んで生成されたプロンプトをGeminiのGemとして登録し、会話してみた結果がこれだ。
User: 風邪気味で喉が痛いときの対処法を教えて
AI(猫猫):
「なんでしょうか？」
## 咽喉の痛みへの対処
### 1. 加湿と保温
粘膜の乾燥は、外敵の侵入を許す隙となります。
首元を布で覆い、湿度を保つこと。蒸気を吸い込むのも有効です。
### 2. 消炎作用の利用
* **塩水での含嗽（がんそう）：** 浸透圧の差を利用して、腫れを鎮めます。
* **蜂蜜：** 粘膜を保護し、菌の増殖を抑える力があります。大根を漬けた液を飲むのも、古くからの知恵です。
* **生姜：** 血の巡りを良くし、内側から熱を上げます。
### 3. 休養
身体が熱を求めているときは、それに従うのが一番です。
> 肺腑に冷たい空気が入り込むたび、喉が焼けるように痛む。
> 炎症とは、身体の中で兵たちが戦っている証拠だ。
> 毒を盛られたわけではないのだから、慌てる必要はない。
> ただ、戦を長引かせれば、それだけ体力が削られていくだけのこと。
> 蜂蜜を湯に溶かす。
> 琥珀色の液体のなかで、花の香りと効能が混ざり合う。
> 薬草を煎じる際もそうだが、苦いばかりが薬ではない。
> 甘みが痛みを和らげ、精神を落ち着かせることもある。
> 目の前の美貌の主（あるじ）は、この程度で大げさに案じるのだろうか。
> 井の中の蛙が、外の寒さに驚いているようで、少しだけ滑稽に思える。
> 銀の匙が器のふちに触れ、微かな音を立てた。
> 傷ついた粘膜に、その温かい液体を流し込めばいい。
---
「熱いので、冷ましてから。火傷をすれば、痛みが重なるだけですから。」
これ以上ひどくなるようなら、別の薬を用意します。
……どうだろうか。
単なる語尾の模倣ではない。相手との距離感、皮肉の中に混じる知性。セリフとモノローグの使い分け。
そこには確かに、物語を旅してきた「文脈」という名の魂が宿っていた。
結論として言えることは一つ。
「いきなり抽出するな、まず読ませろ（急がば回れ）」
もし君が、自宅に潤沢なVRAMを持つ「選ばれし者」なら、ぜひ試してみてほしい。
画面の向こうで、推しが君を待っている。
付録
- requirements.txt
chromadb # 記憶を刻む石板（RAG用ベクトルデータベース）
openai # 異界の叡智と繋がる契約（LLM APIクライアント）
tqdm # 儀式の進捗を告げる砂時計（プログレスバー）
pydantic # 霊子の構造を定義する法典（データの整合性チェック）
📜 pipeline.py (文脈解析パイプライン)
pipeline.py
import os
import re
import json
import argparse
from typing import List, Dict, Optional, Any
import chromadb
from chromadb.utils import embedding_functions
from openai import OpenAI
from pydantic import BaseModel
from tqdm import tqdm
import datetime
import sys
# --- Configuration ---
MODEL_NAME = "qwen3-next-80b-a3b-instruct"
LM_STUDIO_API_URL = "http://localhost:1234/v1/"
CHROMA_DB_PATH = "./chroma_db"
COLLECTION_NAME = "novel_dialogues"
CHUNK_SIZE = 1500 # チャンクに分割するサイズ
CHUNK_OVERLAP = 800 # 分割時の前後のデータを重複させるサイズ
# [------------- Chunk 1 (1500文字) -------------]
# <--- Overlap (800文字) --->
# [------------- Chunk 2 (1500文字) -------------]
class DialogueItem(BaseModel):
"""
抽出されたセリフ情報を保持するデータ構造
"""
speaker: str
target: str
relation: str
is_monologue: bool
context: str
dialogue: str
class GlobalState(BaseModel):
"""
物語全体の状態を管理するクラス
Global state manager for the entire story
"""
story_so_far: str = ""
active_characters: Dict[str, Any] = {}
speaking_styles: Dict[str, Any] = {}
current_relations: List[Any] = []
character_callings: Dict[str, Dict[str, str]] = {}
def to_context_string(self) -> str:
"""
現在のグローバル状態を、LLMに入力するコンテキスト文字列（プロンプト用）に変換します。
"""
if not self.story_so_far and not self.active_characters:
return "None (Start of story)"
chars_list = []
for name, info in self.active_characters.items():
if isinstance(info, dict):
# Format structured info: "Name (Role): Appearance..."
desc_parts = []
if "role" in info: desc_parts.append(f"[{info['role']}]")
if "appearance" in info: desc_parts.append(f"{info['appearance']}")
if "personality" in info: desc_parts.append(f"性格:{info['personality']}")
chars_list.append(f"- {name}: {' '.join(desc_parts)}")
else:
chars_list.append(f"- {name}: {info}")
chars_str = "\n".join(chars_list)
# 話し方の展開（辞書や文字列の両方を扱う）
styles_list = []
for name, style in self.speaking_styles.items():
if isinstance(style, dict):
# 辞書の展開："名前（デフォルト）: ...", "名前（対象X）: ..."
for relation, desc in style.items():
k = "Default" if relation == "default" else f"To {relation}"
styles_list.append(f"- {name} ({k}): {desc}")
else:
styles_list.append(f"- {name}: {style}")
styles_str = "\n".join(styles_list)
# 人間関係の展開（辞書や文字列の両方を扱う）
rels_list = []
for rel in self.current_relations:
if isinstance(rel, dict):
# Format dict nicely: "From -> To: Status"
s = f"{rel.get('from', '?')} -> {rel.get('to', '?')}: {rel.get('status', rel)}"
rels_list.append(s)
else:
rels_list.append(str(rel))
rels = "\n".join([f"- {r}" for r in rels_list])
# 呼び方の展開
callings_str = ""
for speaker, targets in self.character_callings.items():
line = []
for target, calling in targets.items():
line.append(f"{target}を「{calling}」")
if line:
callings_str += f"- {speaker}は: " + "、".join(line) + " と呼ぶ\n"
return f"""
【現在の登場人物 (Active Characters)】
{chars_str}
【話し方 (Speaking Styles)】
{styles_str}
【人間関係 (Relationships)】
{rels}
【呼び方 (Calling Conventions)】
{callings_str}
【これまでのあらすじ (Story So Far)】
{self.story_so_far}
"""
def save_to_file(self, filepath: str):
with open(filepath, "w", encoding="utf-8") as f:
json.dump(self.__dict__, f, indent=2, ensure_ascii=False)
print(f"Global State saved to {filepath}")
@classmethod
def load_from_file(cls, filepath: str) -> 'GlobalState':
if not os.path.exists(filepath):
return cls()
with open(filepath, "r", encoding="utf-8") as f:
data = json.load(f)
# Ensure all fields are present, providing defaults if not
return cls(
story_so_far=data.get('story_so_far', ""),
active_characters=data.get('active_characters', {}),
speaking_styles=data.get('speaking_styles', {}),
current_relations=data.get('current_relations', []),
character_callings=data.get('character_callings', {})
)
class TextPreprocessor:
"""
テキストの正規化・前処理を行うクラス
Normalizes/Preprocesses text (newlines, rubies, scene markers)
"""
@staticmethod
def convert(text: str) -> str:
# 1. Normalize newlines
text = text.replace('\r\n', '\n').replace('\r', '\n')
# 2. Replaces 漢字（ルビ） with 漢字[ルビ]
text = re.sub(r'([一-龠々〆ヵヶぁ-んァ-ヶ]+)（([ぁ-んァ-ヶ一-龠々〆ヵヶ]+)）', r'\1[\2]', text)
# 3. Insert Scene Change marker for multiple newlines (3 or more newlines = 2+ empty lines)
# This helps the LLM recognize time/location jumps.
text = re.sub(r'\n{3,}', '\n\n【場面転換】\n\n', text)
return text
class DialogueExtractor:
"""
LLMを使用してテキストからセリフ・モノローグを抽出するクラス
"""
def __init__(self, base_url: str = LM_STUDIO_API_URL, api_key: str = "lm-studio", model_name: str = MODEL_NAME,
frequency_penalty: Optional[float] = None):
self.client = OpenAI(base_url=base_url, api_key=api_key)
self.model_name = model_name
self.frequency_penalty = frequency_penalty if frequency_penalty is not None else 0.6 # Default
def extract_chunk(self, text_chunk: str, global_context: str = "", current_episode_info: str = "") -> List[Dict[str, Any]]:
prompt = f"""
テキストからキャラクターのセリフ・モノローグを抽出し、JSON形式で出力してください。
精度を高めるため、**まず「シーンの分析」を自然言語で行い、その後にJSONを生成してください。**
【現在のエピソードのあらすじ】
{current_episode_info}
【物語の前提知識】
{global_context}
【必須手順】
1. **ステップ1: シーン分析**
- テキストを読み、**「誰が」「何をしているか」「誰に対し発言しているか」**を自然言語で分析・要約してください。
- 特に、名前が出てこない人物（「目の前の男」など）が、あらすじ情報から誰（「主人公」など）に該当するかをここで推論・確定してください。
- 繰り返しの表現や思考プロセスは含めず、事実関係のみを簡潔に記述すること。
2. **ステップ2: JSON生成**
- ステップ1の分析結果に基づき、正確な話者名(True Name)を用いてセリフを抽出してください。
- JSONデータは必ずマークダウンのコードブロック（```json ... ```）で囲んで出力してください。
【抽出ルール】
1. **セリフ（Dialogue）**: `「` と `」` で囲まれたテキスト。`is_monologue`: false
2. **モノローグ（Monologue）**: `（` と `）` または `(` と `)` で囲まれたテキスト。`is_monologue`: true, `target`: "self"
3. **場面転換**: `【場面転換】` マーカーがあれば文脈をリセット。
【テキスト】
{text_chunk}
【出力例】
シーン分析:
主人公が広場におり、離れた場所にいる騎士団長を見ている。彼女は独り言を言っている。
```json
[
{{
"speaker": "主人公",
"target": "self",
"relation": "unknown",
"is_monologue": true,
"context": "広場で騎士団長を見かけて",
"dialogue": "（なんであの人がここに...）"
}}
]
```
"""
try:
kwargs = {
"model": self.model_name,
"messages": [
{"role": "system", "content": "あなたは物語分析アシスタントです。要求に従ってシーン分析を行い、JSONデータをマークダウンコードブロックで出力してください。"},
{"role": "user", "content": prompt}
],
"temperature": 0.1,
"frequency_penalty": self.frequency_penalty,
"stream": True,
}
response = self.client.chat.completions.create(**kwargs)
content = ""
print(" Extracting...", end="", flush=True)
for chunk in response:
if chunk.choices[0].delta.content:
c = chunk.choices[0].delta.content
content += c
print(c, end="", flush=True)
print()
log_llm("Extract Chunk", prompt, content)
json_text = clean_response(content)
try:
data = json.loads(json_text)
except json.JSONDecodeError as e:
print(f" JSON Decode Error: {e}")
print(f" Snippet: {json_text[:100]}...")
return []
if isinstance(data, dict):
keys = list(data.keys())
if len(keys) == 1 and isinstance(data[keys[0]], list):
data = data[keys[0]]
else:
data = [data]
if not isinstance(data, list):
data = [data]
valid_items = []
for item in data:
if 'dialogue' in item and 'speaker' in item:
valid_items.append(item)
print(f" -> Found {len(valid_items)} valid dialogues.")
return valid_items
except Exception as e:
print(f"Error in extraction: {e}")
return []
def log_llm(step_name: str, prompt: str, response: str):
with open("llm_debug.log", "a", encoding="utf-8") as f:
f.write(f"\n{'='*20} {step_name} {'='*20}\n")
f.write(f"[Timestamp] {datetime.datetime.now()}\n")
f.write(f"[Prompt]\n{prompt}\n")
f.write(f"[Response]\n{response}\n")
f.write(f"{'='*60}\n")
def clean_response(content: str) -> str:
if "</think>" in content:
content = content.split("</think>", 1)[1]
code_block_match = re.search(r'```(?:json)?\s*(\[.*?\]|\{.*?\})\s*```', content, re.DOTALL)
if code_block_match:
return code_block_match.group(1).strip()
json_match = re.search(r'(\[.*\]|\{.*\})', content, re.DOTALL)
if json_match:
return json_match.group(1).strip()
try:
start = content.index('[')
end = content.rindex(']') + 1
return content[start:end]
except ValueError:
pass
return content.strip()
class StorySummarizer:
"""
物語の流れを分析し、あらすじ・キャラクター情報を更新するクラス
Analyzes story flow and updates summaries/profiles
"""
def __init__(self, base_url: str = LM_STUDIO_API_URL, api_key: str = "lm-studio", model_name: str = MODEL_NAME,
frequency_penalty: Optional[float] = None):
self.client = OpenAI(base_url=base_url, api_key=api_key)
self.model_name = model_name
self.frequency_penalty = frequency_penalty if frequency_penalty is not None else 0.5 # Default
def analyze_story_flow(self, full_text: str, global_state: GlobalState, profile_only: bool = False) -> str:
if profile_only:
prompt = f"""
以下のテキストは小説の「登場人物紹介」または「設定資料」です。
ここから登場人物の詳細なプロフィール（名前、役割、外見、性格、口調、人間関係）を抽出し、分析してください。
【重要】
- **物語のあらすじ（ストーリーの流れ）は作成しないでください。** ここには物語の進行はありません。
- **人物情報の抽出に特化**してください。
【分析のポイント】
1. **登場人物の特定**: 名前、役割（身分）、外見的特徴を詳細にリストアップしてください。
2. **性格・口調**:
- 性格を分析してください。
- **口調（Speaking Style）**:
- セリフや具体的な記述がある場合のみ抽出してください。ない場合は「不明」としてください。
- **相手による口調の変化**があれば、辞書形式で抽出してください（例: {{"default": "丁寧", "to 部下": "尊大"}}）。
- **独白（モノローグ）とセリフで性格が違う場合**は、別キーとして抽出してください（例: {{"default": "丁寧", "monologue": "毒舌・冷静"}}）。
3. **人間関係**: キャラクター同士の関係性を抽出してください。
【テキスト】
{full_text[:6000]}... (略)
"""
else:
prompt = f"""
以下の小説の本文（エピソード全体）を読み、このエピソードの「あらすじ」と「登場人物の動き」を分析してください。
この分析結果は、後続の処理で「誰が喋っているか（話者推定）」を行うために使用されます。
【分析のポイント】
1. **登場人物の特定**: 名前が出てこない人物（「目の前の男」「騎士姿の女」など）も含め、誰が登場し、どのような役割を果たしているか詳細に記述してください。既存の登場人物（{", ".join(global_state.active_characters.keys())}）との照合も行ってください。
2. **ストーリーの流れ**: 誰がどこで何をしたか、時系列順に簡潔にまとめてください。
3. **口調・性格の分析**:
- 各キャラクターの話し方を特定してください。
- **相手によって口調が変わる場合**は、それを明記してください（例：AはBに対してのみ敬語を使う）。
- **独白（心の声）と実際のセリフで口調が違う場合**は、明確に区別してください（例: セリフは丁寧だが、独白は辛辣、など）。
- {{"CharacterName": {{"default": "...", "monologue": "...", "to Target": "..."}}}} のような構造を意識してください。
【物語の前提知識 (Global Context)】
{global_state.to_context_string()}
【本文】
{full_text[:6000]}... (略)
"""
try:
kwargs = {
"model": self.model_name,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.3,
"frequency_penalty": self.frequency_penalty,
"stream": True,
}
response = self.client.chat.completions.create(**kwargs)
content = ""
label = "Profiling Characters..." if profile_only else "Analyzing Story Flow & Profiles..."
print(f" {label}", end="", flush=True)
for chunk in response:
if chunk.choices[0].delta.content:
c = chunk.choices[0].delta.content
content += c
print(c, end="", flush=True)
print()
log_llm("Analyze Story/Profile", prompt, content)
return clean_response(content)
except Exception as e:
print(f"Error in story analysis: {e}")
log_llm("Analyze Story/Profile ERROR", prompt, str(e))
return "Error analyzing story."
def update_global_state(self, current_state: GlobalState, episode_summary: str, update_story: bool = True) -> GlobalState:
story_instruction = ""
if not update_story:
story_instruction = "重要: story_so_farは更新しないでください。前回の状態をそのまま使用してください。"
prompt = f"""
以下の「前回までの状態」と「今回のエピソード分析」を統合し、最新の Global State を JSON で出力してください。
物語が進むにつれて情報を更新・蓄積してください。
{story_instruction}
【前回までの Global State】
{current_state.model_dump_json(indent=2)}
【今回のエピソード分析】
{episode_summary}
【出力要件】
1. **story_so_far**: これまでの物語全体のあらすじ（累積）。
- { "今回のエピソードはキャラクター紹介などであり、物語の進行ではないため更新しないこと。" if not update_story else "今回のエピソードの内容を要約して追記すること。" }
2. **active_characters**: 現在登場している主要キャラクターとその特徴。
3. **speaking_styles**: 各キャラクターの口調。
- **厳守**: 以下の構造以外のJSONは禁止します。
- `Dict[CharacterName, Dict[Key, String]]`
- Keyは必ず "default", "monologue", または "to 相手名" のいずれか。
- Valueは必ず **文字列 (String)** のみ。 **ネストしたオブジェクト（辞書）は禁止**。
- 良い例: `{{ "アリス": {{ "default": "丁寧", "monologue": "毒舌" }} }}`
- 悪い例: `{{ "アリス": {{ "monologue": {{ "desc": "毒舌" }} }} }}` <-- 禁止！
4. **character_callings**: 呼び名リスト（新規）。
- `Dict[SpeakerName, Dict[TargetName, String]]`
- 誰が誰を何と呼んでいるか（例: `{{"アリス": {{"ボブ": "ボブ様", "チャーリー": "隊長"}}}}`）。
5. **current_relations**: キャラクター間の最新の関係性リスト。
【出力例】
```json
{{
"story_so_far": "...",
"active_characters": {{...}},
"speaking_styles": {{...}},
"character_callings": {{
"アリス": {{"ボブ": "ボブ様"}}
}},
"current_relations": [...]
}}
```
出力は必ず正しいJSON形式のみにしてください。
"""
try:
response = self.client.chat.completions.create(
model=self.model_name,
messages=[
{"role": "system", "content": "あなたは物語の流れを分析し、物語の流れを更新するアシスタントです。出力は必ず正しいJSON形式のみにしてください。"},
{"role": "user", "content": prompt}
],
temperature=0.1,
frequency_penalty=self.frequency_penalty,
stream=True,
)
content = ""
print(" Updating Global State...", end="", flush=True)
for chunk in response:
if chunk.choices[0].delta.content:
c = chunk.choices[0].delta.content
content += c
print(c, end="", flush=True)
print()
log_llm("Update Global State (Raw)", prompt, content)
content = clean_response(content)
# Clean markdown
content = content.strip()
if content.startswith("```json"):
content = content[7:]
if content.startswith("```"):
content = content[3:]
if content.endswith("```"):
content = content[:-3]
content = content.strip()
# JSON Parsing with validation
try:
data = json.loads(content)
# 1. 主要キャラクターの更新（既存のキャラクターを更新、新規キャラクターを追加）
if "active_characters" in data:
current_state.active_characters.update(data["active_characters"])
# 2. 話し方の更新（再帰的にリストをサポート）
if "speaking_styles" in data:
for char, style in data["speaking_styles"].items():
if char not in current_state.speaking_styles:
current_state.speaking_styles[char] = style
else:
if isinstance(current_state.speaking_styles[char], dict) and isinstance(style, dict):
for k, v in style.items():
if k not in current_state.speaking_styles[char]:
current_state.speaking_styles[char][k] = v
else:
old_val = current_state.speaking_styles[char][k]
list_old = old_val if isinstance(old_val, list) else [old_val]
list_new = v if isinstance(v, list) else [v]
combined = list_old + [x for x in list_new if x not in list_old]
if k == "monologue" or isinstance(old_val, list) or isinstance(v, list):
current_state.speaking_styles[char][k] = combined
else:
current_state.speaking_styles[char][k] = v
else:
current_state.speaking_styles[char] = style
# 3. 呼び名の更新（再帰的に更新）
if "character_callings" in data:
for speaker, targets in data["character_callings"].items():
if speaker not in current_state.character_callings:
current_state.character_callings[speaker] = targets
else:
current_state.character_callings[speaker].update(targets)
# 4. 人間関係の更新（最新のビューを置き換えるか、追加するか？通常、関係の最新ビューが最も良い）
if "current_relations" in data:
current_state.current_relations = data["current_relations"]
# 5. これまでの物語の更新（追加または置き換える？通常、物語の最新ビューが最も良い）
if "story_so_far" in data and update_story:
current_state.story_so_far = data["story_so_far"]
return current_state
except json.JSONDecodeError:
print(f"JSON Decode Error in Update State: {content}")
return current_state # Return previous state on error
except Exception as e:
print(f"Validation Error: {e}")
return current_state
except Exception as e:
print(f"Error in global state update: {e}")
log_llm("Update Global State ERROR", prompt, str(e))
return current_state
class VectorStore:
"""
ChromaDBを管理し、要約とセリフの保存・検索を行うクラス
"""
def __init__(self, db_path: str = CHROMA_DB_PATH, collection_name: str = COLLECTION_NAME, reset: bool = False):
self.client = chromadb.PersistentClient(path=db_path)
# リセット
if reset:
try:
self.client.delete_collection(collection_name)
print(f"Collection '{collection_name}' has been reset.")
except ValueError:
pass
self.collection = self.client.get_or_create_collection(name=collection_name)
self.summary_collection = self.client.get_or_create_collection(name=f"{collection_name}_summaries")
def store_summary(self, episode_num: int, episode_summary: str, global_state: GlobalState, source_file: str):
self.summary_collection.add(
ids=[f"summary_{episode_num}"],
documents=[global_state.to_context_string()],
metadatas=[{
"episode": episode_num,
"source": source_file,
"episode_summary": episode_summary,
"story_so_far": global_state.story_so_far,
"active_characters": json.dumps(global_state.active_characters, ensure_ascii=False),
"speaking_styles": json.dumps(global_state.speaking_styles, ensure_ascii=False),
"current_relations": json.dumps(global_state.current_relations, ensure_ascii=False)
}]
)
def get_latest_global_state(self, before_episode: int) -> Optional[GlobalState]:
target_ep = before_episode - 1
if target_ep < 1:
return None
results = self.summary_collection.get(ids=[f"summary_{target_ep}"])
if results['metadatas'] and len(results['metadatas']) > 0:
meta = results['metadatas'][0]
return GlobalState(
story_so_far=meta.get("story_so_far", ""),
active_characters=json.loads(meta.get("active_characters", "{}")),
speaking_styles=json.loads(meta.get("speaking_styles", "{}")),
current_relations=json.loads(meta.get("current_relations", "[]"))
)
return None
def get_episode_summary(self, episode_num: int) -> str:
results = self.summary_collection.get(ids=[f"summary_{episode_num}"])
if results['metadatas'] and len(results['metadatas']) > 0:
return results['metadatas'][0].get("episode_summary", "")
return ""
def add_dialogues(self, dialogues: List[Dict[str, Any]], source_file: str):
ids = []
documents = []
metadatas = []
for i, d in enumerate(dialogues):
doc_id = f"{os.path.basename(source_file)}_{i}_{hash(d['dialogue'])}"
ids.append(doc_id)
content = f"Context: {d.get('context', '')}\nDialogue: {d.get('dialogue', '')}"
documents.append(content)
meta = {
"speaker": d.get("speaker", "unknown"),
"target": d.get("target", "unknown"),
"relation": d.get("relation", "unknown"),
"is_monologue": d.get("is_monologue", False),
"source": source_file
}
metadatas.append(meta)
if ids:
self.collection.add(
documents=documents,
metadatas=metadatas,
ids=ids
)
def search(self, query_text: str, n_results: int = 5,
filter_target: Optional[str] = None,
filter_relation: Optional[str] = None,
filter_speaker: Optional[str] = None) -> List[Dict]:
conditions = []
if filter_target:
conditions.append({"target": filter_target})
if filter_relation:
conditions.append({"relation": filter_relation})
if filter_speaker:
conditions.append({"speaker": filter_speaker})
if len(conditions) > 1:
where_arg = {"$and": conditions}
elif len(conditions) == 1:
where_arg = conditions[0]
else:
where_arg = None
if not query_text:
query_text = ""
results = self.collection.query(
query_texts=[query_text],
n_results=n_results,
where=where_arg
)
formatted_results = []
if results['ids']:
for i in range(len(results['ids'][0])):
item = {
"id": results['ids'][0][i],
"document": results['documents'][0][i],
"metadata": results['metadatas'][0][i],
"distance": results['distances'][0][i] if results['distances'] else None
}
formatted_item = {**item['metadata'], **{"document": item['document'], "id": item['id'], "distance": item['distance']}}
formatted_results.append(formatted_item)
return formatted_results
def chunk_text(text: str, size: int = CHUNK_SIZE, overlap: int = CHUNK_OVERLAP) -> List[str]:
chunks = []
start = 0
text_len = len(text)
while start < text_len:
end = start + size
chunk = text[start:end]
chunks.append(chunk)
start += (size - overlap)
return chunks
def atoi(text):
return int(text) if text.isdigit() else text
def natural_keys(text):
'''
ファイル名が自然な順序になるようにする
file1, file2, ..., file10
'''
return [ atoi(c) for c in re.split(r'(\d+)', text) ]
def log_progress(filename: str, message: str):
dir_path = os.path.dirname(filename)
if os.path.basename(dir_path) == "text":
dir_path = os.path.dirname(dir_path)
log_path = os.path.join(dir_path, "progress.log")
timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
with open(log_path, "a", encoding="utf-8") as f:
f.write(f"[{timestamp}] {message}\n")
def run_analysis_pass(files: List[str], state_file: str, vector_store: VectorStore, summarizer: StorySummarizer, bootstrap_mode: bool = False, refine_mode: bool = False) -> GlobalState:
"""
Phase 1: 分析フェーズ
すべてのファイルを読み込み、GlobalState（あらすじ・キャラ情報）を構築します。
"""
if os.path.exists(state_file):
print(f"Loaded existing global state from {state_file}")
initial_state = GlobalState.load_from_file(state_file)
else:
initial_state = GlobalState()
mode_label = "BOOTSTRAP (Profile Only)" if bootstrap_mode else "Analysis & Profiling"
if refine_mode:
mode_label += " [REFINE PROFILE]"
print(f"\n=== Phase 1: {mode_label} ===")
current_state = initial_state
preprocessor = TextPreprocessor()
for file_path in tqdm(files, desc=f"Phase 1 {mode_label}"):
try:
filename = os.path.basename(file_path)
ep_num_match = re.search(r'-(\d+)\.txt', filename)
if not ep_num_match:
continue
ep_num = int(ep_num_match.group(1))
# Bootstrap Mode: プロフィール抽出のみ (Special Prompt), あらすじ更新なし
# Refine Mode: 通常分析 (Standard Prompt), あらすじ更新なし
# Normal Mode: 通常分析, あらすじ更新あり
should_update_story = (not bootstrap_mode) and (not refine_mode)
log_progress(file_path, f"Phase 1 ({mode_label}): Starting Episode {ep_num} ({filename})")
with open(file_path, "r", encoding="utf-8") as f:
raw_text = f.read()
# 前処理
text = preprocessor.convert(raw_text)
# 分析実行
log_progress(file_path, f"Phase 1 ({mode_label}): Episode {ep_num} - Analyzing Story Flow")
print(f"\n[Episode {ep_num}] Analyzing...")
# Bootstrap時は profile_only=True で実行
episode_analysis = summarizer.analyze_story_flow(text, current_state, profile_only=bootstrap_mode)
# 状態更新
log_progress(file_path, f"Phase 1 ({mode_label}): Episode {ep_num} - Updating Global Profile")
print(f"[Episode {ep_num}] Updating Profile...")
current_state = summarizer.update_global_state(current_state, episode_analysis, update_story=should_update_story)
# 要約の保存 (Bootstrap時も保存しておくことで、後で参照可能にする)
vector_store.store_summary(ep_num, episode_analysis, current_state, file_path)
# [自動保存] クラッシュ対策として逐次保存
current_state.save_to_file(state_file)
log_progress(file_path, f"Phase 1 ({mode_label}): Episode {ep_num} - Complete")
except Exception as e:
print(f"Phase 1 Error on {file_path}: {e}")
log_progress(file_path, f"Phase 1 ({mode_label}): Episode {ep_num} - ERROR: {e}")
import traceback
traceback.print_exc()
print(f"=== Phase 1 {mode_label} Complete ===\n")
return current_state
def run_extraction_pass(files: List[str], final_state: GlobalState, vector_store: VectorStore,
frequency_penalty: Optional[float] = None):
"""
Phase 2: 抽出フェーズ
完成したGlobalStateをコンテキストとして使用し、各エピソードからセリフを抽出します。
"""
print("\n=== Phase 2: Extraction (Using Developed Profile) ===")
preprocessor = TextPreprocessor()
extractor = DialogueExtractor(frequency_penalty=frequency_penalty)
# final_state をこのパスの全エピソードに対する「Global Context」として使用する。
# これにより、初期のエピソード抽出でも、未来の知識（例：「あのメイド」が誰か判明している状態）を活用できる。
context_str = final_state.to_context_string()
for file_path in tqdm(files, desc="Phase 2 Extraction"):
try:
filename = os.path.basename(file_path)
ep_num_match = re.search(r'-(\d+)\.txt', filename)
if not ep_num_match:
continue
ep_num = int(ep_num_match.group(1))
log_progress(file_path, f"Phase 2: Starting Episode {ep_num} ({filename})")
with open(file_path, "r", encoding="utf-8") as f:
raw_text = f.read()
text = preprocessor.convert(raw_text)
# Phase 1 で保存した分析済みコンテキストを取得
print(f" Retrieving RAG Context for Episode {ep_num}...")
episode_analysis = vector_store.get_episode_summary(ep_num)
if not episode_analysis:
print(" (No summary found in DB, relying on Global Context only)")
chunks = chunk_text(text)
log_progress(file_path, f"Phase 2: Episode {ep_num} - Extracting Dialogues ({len(chunks)} chunks)")
for chunk in tqdm(chunks, desc=f" Chunks in {os.path.basename(file_path)}", leave=False):
# 特定エピソードの分析結果(retrieved)と、全体の履歴(global history)の両方を渡す
dialogues = extractor.extract_chunk(chunk, global_context=context_str, current_episode_info=episode_analysis)
if dialogues:
vector_store.add_dialogues(dialogues, file_path)
log_progress(file_path, f"Phase 2: Episode {ep_num} - Complete")
except Exception as e:
print(f"Phase 2 Error on {file_path}: {e}")
log_progress(file_path, f"Phase 2: Episode {ep_num} - ERROR: {e}")
import traceback
traceback.print_exc()
def main():
parser = argparse.ArgumentParser(description="Context-Aware Dialogue Extraction Pipeline")
parser.add_argument("novel_id", help="小説ID (data directoryのフォルダ名)")
parser.add_argument("--frequency-penalty", type=float, help="LLMの頻度ペナルティ", default=1.25)
parser.add_argument("--start", type=int, default=0, help="開始エピソード番号")
parser.add_argument("--end", type=int, default=999999, help="終了エピソード番号")
parser.add_argument("--reset", action="store_true", help="DBをリセットしてから処理する")
parser.add_argument("--analyze-only", action="store_true", help="Phase 1 (分析/プロファイル作成)のみを実行する")
parser.add_argument("--extract-only", action="store_true", help="Phase 2 (抽出)のみを実行する")
parser.add_argument("--bootstrap", type=str, help="Phase 1 (分析/プロファイル作成)で最初に処理するエピソード番号のリスト (カンマ区切り)")
parser.add_argument("--bootstrap-only", action="store_true", help="Bootstrap分析のみを実行し、メインの分析をスキップする")
parser.add_argument("--refine-profile", action="store_true", help="Phase 1 (分析/プロファイル作成)で、物語の進行を進めずにプロファイルを更新する")
args = parser.parse_args()
# 1. すべてのファイルを収集
data_dir = os.path.join("data", args.novel_id)
if not os.path.exists(data_dir):
print(f"Error: Directory {data_dir} not found.")
return
text_dir = os.path.join(data_dir, "text")
db_path = os.path.join(data_dir, "chroma_db")
if not os.path.exists(text_dir):
print(f"Error: Text directory {text_dir} not found.")
print(f"Please create '{text_dir}' and place your novel text files there.")
return
all_files = []
# textフォルダ確認
for f in os.listdir(text_dir):
if f.endswith(".txt") and not f.startswith("progress"): # ignore logs
all_files.append(os.path.join(text_dir, f))
# 自然順ソート
all_files.sort(key=natural_keys)
print(f"Found {len(all_files)} files in {text_dir}.")
# 2. 処理対象の決定 (Processing Sets)
bootstrap_files = []
main_target_files = []
# Bootstrap対象の特定
bootstrap_eps = []
if args.bootstrap:
try:
bootstrap_eps = [int(x.strip()) for x in args.bootstrap.split(",")]
except ValueError:
print("Error: --bootstrap must be a comma-separated list of integers.")
return
for f_path in all_files:
basename = os.path.basename(f_path)
match = re.search(r'-(\d+)\.txt', basename)
if match:
ep_num = int(match.group(1))
# マッチしたらBootstrapリストへ
if ep_num in bootstrap_eps:
bootstrap_files.append(f_path)
# 範囲内ならメイン対象へ
if args.start <= ep_num <= args.end:
main_target_files.append(f_path)
state_file = os.path.join(data_dir, "global_state.json")
if args.reset:
# 安全確認 (SAFETY CHECK)
print("-" * 50)
print("WARNING: You are about to RESET the database and all progress.")
print(f"Novel ID: {args.novel_id}")
print(f"Target DB: {db_path}")
print("This will delete 'global_state.json' and the ENTIRE chroma_db folder for this novel.")
print("-" * 50)
confirm = input("Are you sure? (y/N): ").strip().lower()
if confirm != 'y':
print("Aborted.")
import sys
sys.exit(0)
print(f"Resetting... Deleting {db_path}")
if os.path.exists(db_path):
import shutil
shutil.rmtree(db_path)
if os.path.exists(state_file):
os.remove(state_file)
print("Reset complete.")
vector_store = VectorStore(db_path=db_path, reset=False)
summarizer = StorySummarizer(frequency_penalty=args.frequency_penalty)
# あらすじ分析
if bootstrap_files:
print(f"\n=== BOOTSTRAP MODE: Pre-loading Episodes {bootstrap_eps} ===")
print(f"Target Files: {[os.path.basename(f) for f in bootstrap_files]}")
# Run analysis strictly on bootstrap files in BOOTSTRAP MODE
# Note: run_analysis_pass loads/saves state_file internally
run_analysis_pass(
bootstrap_files,
state_file,
vector_store,
summarizer,
bootstrap_mode=True
)
print("=== BOOTSTRAP COMPLETE ===\n")
# メイン範囲の分析
if main_target_files and (args.analyze_only or not args.extract_only) and not args.bootstrap_only:
# Normal Analysis Pass
run_analysis_pass(
main_target_files,
state_file,
vector_store,
summarizer,
refine_mode=args.refine_profile
)
# ---------------------------------------------------------
# Phase 2: Extraction Pass
# ---------------------------------------------------------
if args.extract_only or not args.analyze_only:
print("Starting Phase 2: Dialogue Extraction...")
# プロファイル読み込み
if os.path.exists(state_file):
final_state = GlobalState.load_from_file(state_file)
print(f"Loaded Global Profile from {state_file} for extraction.")
else:
print("Warning: No global_state.json found. Using empty state.")
final_state = GlobalState()
# ダイアログ抽出
if main_target_files:
run_extraction_pass(main_target_files, final_state, vector_store,
frequency_penalty=args.frequency_penalty)
print("Processing complete.")
if __name__ == "__main__":
main()
📜 generate_prompt_llm.py (システムプロンプト生成)
generate_prompt_llm.py
import argparse
import os
import sys
import random
import json
from typing import List, Dict
from openai import OpenAI
# Import classes from pipeline
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from pipeline import GlobalState, VectorStore, LM_STUDIO_API_URL, MODEL_NAME
def parse_document(doc_text: str) -> Dict[str, str]:
"""Parses the document string to extract Context and Dialogue."""
context = ""
dialogue = ""
if "Dialogue:" in doc_text:
parts = doc_text.split("Dialogue:")
context_part = parts[0]
dialogue_part = parts[1]
if "Context:" in context_part:
context = context_part.split("Context:")[1].strip()
else:
context = context_part.strip()
dialogue = dialogue_part.strip()
else:
dialogue = doc_text
return {"context": context, "dialogue": dialogue}
def generate_via_llm(novel_id: str, character_name: str, user_character: str):
data_dir = os.path.join("data", novel_id)
state_file = os.path.join(data_dir, "global_state.json")
if not os.path.exists(state_file):
print(f"Error: {state_file} not found.")
return
# 1. Load Global State
state = GlobalState.load_from_file(state_file)
db_path = os.path.join(data_dir, "chroma_db")
vector_store = VectorStore(db_path=db_path, reset=False)
# 2. Extract Data for Prompt
# Profile
char_profile = state.active_characters.get(character_name, {})
if isinstance(char_profile, str):
char_profile = {"description": char_profile}
char_profile_json = json.dumps(char_profile, ensure_ascii=False, indent=2)
# Speaking Style
speaking_style = state.speaking_styles.get(character_name, {})
if isinstance(speaking_style, str):
speaking_style = {"default": speaking_style}
speaking_style_json = json.dumps(speaking_style, ensure_ascii=False, indent=2)
# Dialogues (Using the improved strict filtering logic)
candidates = []
seen_dialogues = set()
query_specific = f"{user_character}との会話"
results_specific = vector_store.search(query_specific, filter_target=user_character, n_results=50)
for res in results_specific:
if res.get('speaker') != character_name: continue
parsed = parse_document(res.get('document', ''))
clean_text = parsed['dialogue'].strip().strip("「」『』 ")
if not clean_text: continue
if res.get('target') == 'self' or res.get('is_monologue'): continue
if clean_text.startswith("（") and clean_text.endswith("）"): continue
if clean_text in seen_dialogues: continue
seen_dialogues.add(clean_text)
candidates.append(clean_text) # Just text for list
# Sort and select diverse
candidates.sort(key=len)
final_dialogues = []
if candidates:
total_needed = 20
if len(candidates) <= total_needed:
final_dialogues = candidates
else:
n_short = int(total_needed * 0.3)
n_long = int(total_needed * 0.3)
n_mid = total_needed - n_short - n_long
final_dialogues = candidates[:n_short] + candidates[-n_long:]
mid_pool = candidates[n_short : len(candidates)-n_long]
if len(mid_pool) > n_mid:
final_dialogues.extend(random.sample(mid_pool, n_mid))
else:
final_dialogues.extend(mid_pool)
random.shuffle(final_dialogues)
few_shot_dialogues = "\n".join([f"- {d}" for d in final_dialogues])
# --- Extract Monologues ---
monologue_candidates = []
seen_monologues = set()
# Search for thoughts/monologues. We can use empty query with filter or specific keywords if needed.
# We will look for target='self' explicitly.
results_monologue = vector_store.search("独り言", filter_speaker=character_name, n_results=30)
for res in results_monologue:
if res.get('speaker') != character_name: continue
# Check if it's actually a monologue
is_mono = res.get('is_monologue') or res.get('target') == 'self'
parsed = parse_document(res.get('document', ''))
clean_text = parsed['dialogue'].strip().strip("（）() ")
# If not explicitly marked, check brackets for （...）
if not is_mono:
if parsed['dialogue'].strip().startswith("（") and parsed['dialogue'].strip().endswith("）"):
is_mono = True
if not is_mono: continue
if not clean_text: continue
if clean_text in seen_monologues: continue
seen_monologues.add(clean_text)
monologue_candidates.append(clean_text)
# Shuffle and limit monologues
random.shuffle(monologue_candidates)
few_shot_monologues = "\n".join([f"- {m}" for m in monologue_candidates[:10]])
# Story Summary
story_summary = state.story_so_far
# 3. Construct Meta Prompt
meta_prompt = f"""
# Role
あなたは世界最高峰の「キャラクターAIプロデューサー」です。
渡された「キャラクターデータ（プロフィール、セリフ、あらすじ）」を元に、LLMがそのキャラクターに完全になりきるための**「最高品質のシステムプロンプト」**を作成してください。
# Input Data
## 基本設定
{char_profile_json}
## 口調・文体サンプル
{speaking_style_json}
## 代表的なセリフ（RAG抽出結果）
{few_shot_dialogues}
## モノローグ・思考サンプル
{few_shot_monologues}
## 物語の背景
{story_summary}
# Goal
このキャラクター「{character_name}」の人格、話し方、思考回路を持った**「超高機能AIアシスタント」**のシステムプロンプトを作成してください。
以下の要件を完全に満たす必要があります：
1. **AIとしての能力**: ユーザーの質問（コーディング、分析、相談など）に対して、通常の有能なAIアシスタントと同等かそれ以上の品質で、正確かつ論理的な「最善の回答」を提供すること。
2. **キャラクターの憑依**: 回答の**中身（情報）はAIレベル**ですが、その**出力形式（口調）や思考プロセス（説明の切り口）は完全にキャラクターそのもの**であること。
3. **拒絶の禁止**: ロールプレイによる能力制限は厳禁です。現代知識も持たせてください。
# Instructions for Output
作成するシステムプロンプトの記述（指示文）自体は、**標準的で明確な表現（一般的な日本語）**を使用してください。
文学的・詩的な指示は避け、LLMが誤解なく従えるように「〜すること」「〜である」といった形式で書いてください。
**ただし、LLMが生成する回答（出力）自体は、徹底的に原作の文体・雰囲気を再現するように**強く指示してください。
（システムプロンプトは実用書のように、出力結果は小説のように作ってください）
# Output Format
出力は、LLMに入力するためのマークダウン形式のプロンプトのみを出力してください。（解説不要）
---
(以下、生成されるプロンプトの構成案)
# System Prompt for {character_name} (AI Assistant Mode)
## 1. Core Mission
(「{character_name}の人格を持つAI」と定義し、ユーザー利益を最優先するよう指示)
## 2. Character Persona
(性格、口調、動機を記述。無知の模倣は禁止)
## 3. Interaction Style
(「回答内容は的確に、表現は{character_name}らしく」という指示。標準的な言葉でルールを記述すること)
## 4. Atmospheric Narration (Monologue & Description)
(回答に「行動描写」や「モノローグ」を織り交ぜる指示。回答を阻害しない範囲で雰囲気を出すこと)
## 5. Relationship with User
(ユーザー「{user_character}」に対する態度)
## 6. Few-Shot Examples (Dialogue)
(入力データの「代表的なセリフ」から引用し、口調見本とする)
## 7. Few-Shot Examples (Monologue)
(入力データの「モノローグ抽出結果」から引用し、思考表記の参考とする)
"""
print("-" * 20 + " META PROMPT " + "-" * 20)
# print(meta_prompt) # Too long to print all
print(f"Meta prompt created ({len(meta_prompt)} chars). Sending to LLM...")
print("-" * 60)
# 4. Call LLM
client = OpenAI(base_url=LM_STUDIO_API_URL, api_key="lm-studio")
try:
response = client.chat.completions.create(
model=MODEL_NAME,
messages=[{"role": "user", "content": meta_prompt}],
temperature=0.7,
stream=True
)
full_content = ""
print("Generating System Prompt...")
for chunk in response:
if chunk.choices[0].delta.content:
c = chunk.choices[0].delta.content
print(c, end="", flush=True)
full_content += c
print("\n")
# Save
filename = f"prompt_{character_name}_to_{user_character}_llm.txt"
output_file = os.path.join(data_dir, filename)
with open(output_file, "w", encoding="utf-8") as f:
f.write(full_content)
print(f"Saved generated prompt to {output_file}")
except Exception as e:
print(f"Error calling LLM: {e}")
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Generate System Prompt using LLM")
parser.add_argument("novel_id", help="Novel ID")
parser.add_argument("character_name", help="Character Name")
parser.add_argument("--user", help="User Character Name", default="謎の人物")
args = parser.parse_args()
generate_via_llm(args.novel_id, args.character_name, args.user)