はじめに
本記事はNTTテクノクロス Advent Calendar 2025の21日目です。
みなさん、どうもこんにちは。NTTテクノクロスの下本です。
2025年も終わろうとしている現在AI技術は様々な分野、業界に貢献しています。何らかの形で生成AIと関わっている方がほとんどではないでしょうか。日々業務の中で自分も非常にお世話になっています。
何か自身でやってみれそうなコンテンツを探した結果、今回はチャットボットのAI活用に焦点を当てていきたいと思います。
技術紹介
チャットボットとはテキストや音声形式でユーザと対話を行うプログラムやシステムのことを指します。用途としてはカスタマーサポートやFAQの自動応答等が挙げられます。オンラインサイト上で見かけたことがあると思います。チャットボットは大きく以下の2つのタイプに分類することが出来ます。
-
ルールベース型
- あらかじめ決められたルールに沿って応答する
- 質問に対する回答はパターン化されている
-
AI型
- 機械学習や深層学習、さらには生成AIを用いて応答する
- ユーザの質問の意図をくみ取り、柔軟で高度な対話を実現する
生成AIとは入力に対して文書や画像を作成するシステムのことを指します。学習データの規則性を掴むことでデータにない新しいデータを作成することを可能にしています。指示を与えることでデータを生成したり、対話に応答できるのが特徴です。ChatGPTは最も有名な対話型の大規模言語モデルの1つです。詳細は論文 1 も公開されています。
本記事ではルールベースとAIをどちらも使用したハイブリッド型を作ってみようと思います。事前に参考になりそうなサイトを調べていると、その多くがモデル部分は大手が公開しているAPIを用いていました。参考までにですが前述したChatGPTを開発しているOpenAIのリファレンスはこちらになります。今回精度は諦めていますが既存のモデルを使用せずに自前で用意してみようと思います。
実装
以下実装パートです
ざっくり以下のような構成で試してみました。動作確認には最低限のコードで実施しました。動作確認用のコードはこだわってもいないので本記事での言及は省略します。決して強力ではないローカル環境で動かしてみます。
root/
├── model.py
├── train.py
├── agent.py
├── tools.py
├── main.py # ターミナル上での動作確認用(省略)
├── app.py # Webブラウザ上での動作確認用(省略)
├── templates/
│ └── index.html # (省略)
├── static/
│ ├── style.css # (省略)
│ └── script.js # (省略)
├── trainde_model/
│ ├── model.pt # 学習時に生成
│ └── tokenizer.json # 学習時に生成
└──training_data.json # (省略)
学習部分は以下です。Transformer2言語モデルをベースにしています。動かすことに重きをおいたためアルゴリズムについては拡張できませんでした。今後の課題です。
class PositionalEncoding(nn.Module):
"""位置エンコーディング"""
def __init__(self, d_model: int, max_len: int = 5000):
super().__init__()
position = torch.arange(max_len).unsqueeze(1)
div_term = torch.exp(torch.arange(0, d_model, 2) * (-math.log(10000.0) / d_model))
pe = torch.zeros(max_len, 1, d_model)
pe[:, 0, 0::2] = torch.sin(position * div_term)
pe[:, 0, 1::2] = torch.cos(position * div_term)
self.register_buffer('pe', pe)
def forward(self, x: torch.Tensor) -> torch.Tensor:
"""
Args:
x: Tensor of shape [seq_len, batch_size, embedding_dim]
"""
x = x + self.pe[:x.size(0)]
return x
class TransformerLanguageModel(nn.Module):
"""Transformerベースの言語モデル"""
def __init__(
self,
vocab_size: int,
d_model: int = 256,
nhead: int = 8,
num_layers: int = 6,
dim_feedforward: int = 1024,
dropout: float = 0.1,
max_len: int = 512
):
super().__init__()
self.d_model = d_model
self.vocab_size = vocab_size
# Embedding層
self.embedding = nn.Embedding(vocab_size, d_model)
self.pos_encoder = PositionalEncoding(d_model, max_len)
# Transformer Decoder層
decoder_layer = nn.TransformerDecoderLayer(
d_model=d_model,
nhead=nhead,
dim_feedforward=dim_feedforward,
dropout=dropout,
batch_first=False
)
self.transformer_decoder = nn.TransformerDecoder(
decoder_layer,
num_layers=num_layers
)
# 出力層
self.fc_out = nn.Linear(d_model, vocab_size)
self.dropout = nn.Dropout(dropout)
self._init_weights()
def _init_weights(self):
# 重みの初期化
initrange = 0.1
self.embedding.weight.data.uniform_(-initrange, initrange)
self.fc_out.bias.data.zero_()
self.fc_out.weight.data.uniform_(-initrange, initrange)
def _generate_square_subsequent_mask(self, sz: int) -> torch.Tensor:
# 自己回帰マスクの生成
mask = torch.triu(torch.ones(sz, sz), diagonal=1).bool()
return mask
def forward(
self,
src: torch.Tensor,
src_mask: Optional[torch.Tensor] = None
) -> torch.Tensor:
"""
Args:
src: [seq_len, batch_size]
src_mask: [seq_len, seq_len]
Returns:
output: [seq_len, batch_size, vocab_size]
"""
seq_len = src.size(0)
if src_mask is None:
src_mask = self._generate_square_subsequent_mask(seq_len).to(src.device)
# Embedding + Positional Encoding
src = self.embedding(src) * math.sqrt(self.d_model)
src = self.pos_encoder(src)
src = self.dropout(src)
# Transformer Decoder
output = self.transformer_decoder(src, src, tgt_mask=src_mask)
output = self.fc_out(output)
return output
def generate(
self,
prompt_ids: List[int],
max_length: int = 100,
temperature: float = 1.0,
top_k: int = 50,
device: str = 'cpu'
) -> List[int]:
self.eval()
generated = prompt_ids.copy()
with torch.no_grad():
for _ in range(max_length):
input_ids = torch.tensor([generated]).T.to(device) # [seq_len, 1]
output = self.forward(input_ids) # [seq_len, 1, vocab_size]
logits = output[-1, 0, :] / temperature
if top_k > 0:
top_k_logits, top_k_indices = torch.topk(logits, top_k)
probs = F.softmax(top_k_logits, dim=-1)
next_token_idx = torch.multinomial(probs, 1).item()
next_token = top_k_indices[next_token_idx].item()
else:
probs = F.softmax(logits, dim=-1)
next_token = torch.multinomial(probs, 1).item()
generated.append(next_token)
if next_token == 0:
break
return generated
class Tokenizer:
def __init__(self):
self.token_to_id = {}
self.id_to_token = {}
self.vocab_size = 0
# 特殊トークン
self.pad_token = '<PAD>'
self.unk_token = '<UNK>'
self.bos_token = '<BOS>'
self.eos_token = '<EOS>'
def _tokenize(self, text: str) -> List[str]:
"""テキストをトークンに分割(文字単位 + 特殊パターン)"""
tokens = []
i = 0
while i < len(text):
if text[i:i+4] == 'TOOL':
end = text.find(')', i)
if end != -1:
tokens.append(text[i:end+1])
i = end + 1
continue
# 通常の文字
tokens.append(text[i])
i += 1
return tokens
def build_vocab(self, texts: List[str]:
token_set = set()
for text in texts:
token_set.update(self._tokenize(text))
# 特殊トークンを追加
special_tokens = [self.pad_token, self.unk_token, self.bos_token, self.eos_token]
vocab = special_tokens + sorted(list(token_set))
self.token_to_id = {token: idx for idx, token in enumerate(vocab)}
self.id_to_token = {idx: token for idx, token in enumerate(vocab)}
self.vocab_size = len(vocab)
self.vocab = vocab
def encode(self, text: str, add_special_tokens: bool = True) -> List[int]:
ids = []
if add_special_tokens:
ids.append(self.token_to_id[self.bos_token])
for token in self._tokenize(text):
ids.append(self.token_to_id.get(token, self.token_to_id[self.unk_token]))
if add_special_tokens:
ids.append(self.token_to_id[self.eos_token])
return ids
def decode(self, ids: List[int], skip_special_tokens: bool = True) -> str:
tokens = []
special_token_ids = {
self.token_to_id[self.pad_token],
self.token_to_id[self.unk_token],
self.token_to_id[self.bos_token],
self.token_to_id[self.eos_token]
}
for id in ids:
if skip_special_tokens and id in special_token_ids:
continue
token = self.id_to_token.get(id, self.unk_token)
tokens.append(token)
return ''.join(tokens)
def save(self, path: str):
data = {
'token_to_id': self.token_to_id,
'id_to_token': {str(k): v for k, v in self.id_to_token.items()},
'vocab_size': self.vocab_size
}
with open(path, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
def load(self, path: str):
with open(path, 'r', encoding='utf-8') as f:
data = json.load(f)
self.token_to_id = data['token_to_id']
self.id_to_token = {int(k): v for k, v in data['id_to_token'].items()}
self.vocab_size = data['vocab_size']
def save_model(model: TransformerLanguageModel, tokenizer: Tokenizer, path: str):
os.makedirs(path, exist_ok=True)
torch.save({
'model_state_dict': model.state_dict(),
'model_config': {
'vocab_size': model.vocab_size,
'd_model': model.d_model,
}
}, os.path.join(path, 'model.pt'))
tokenizer.save(os.path.join(path, 'tokenizer.json'))
def load_model(path: str, device: str = 'cpu') -> Tuple[TransformerLanguageModel, Tokenizer]:
tokenizer = Tokenizer()
tokenizer.load(os.path.join(path, 'tokenizer.json'))
checkpoint = torch.load(os.path.join(path, 'model.pt'), map_location=device)
model = TransformerLanguageModel(
vocab_size=checkpoint['model_config']['vocab_size'],
d_model=checkpoint['model_config']['d_model']
)
model.load_state_dict(checkpoint['model_state_dict'])
model.to(device)
return model, tokenizer
model.pyの中で学習データからトークンも取得します。トークン数の多さがモデルの精度に大きく関わってきます。
class ConversationDataset(Dataset):
def __init__(self, data_path: str, tokenizer: Tokenizer, max_length: int = 256):
self.tokenizer = tokenizer
self.max_length = max_length
with open(data_path, 'r', encoding='utf-8') as f:
self.data = json.load(f)
self.texts = []
for conversation in self.data:
for turn in conversation['conversation']:
text = f"{turn['role']}: {turn['content']}"
self.texts.append(text)
def __len__(self):
return len(self.texts)
def __getitem__(self, idx):
text = self.texts[idx]
# トークン化
tokens = self.tokenizer.encode(text, add_special_tokens=True)
if len(tokens) > self.max_length:
tokens = tokens[:self.max_length]
return torch.tensor(tokens, dtype=torch.long)
def collate_fn(batch: List[torch.Tensor]) -> Tuple[torch.Tensor, torch.Tensor]:
max_len = max(len(tokens) for tokens in batch)
padded_batch = []
for tokens in batch:
padding = torch.zeros(max_len - len(tokens), dtype=torch.long)
padded_tokens = torch.cat([tokens, padding])
padded_batch.append(padded_tokens)
batch_tensor = torch.stack(padded_batch).T
src = batch_tensor[:-1, :]
tgt = batch_tensor[1:, :]
return src, tgt
def train_epoch(
model: TransformerLanguageModel,
dataloader: DataLoader,
optimizer: optim.Optimizer,
criterion: nn.Module,
device: str
) -> float:
model.train()
total_loss = 0
progress_bar = tqdm(dataloader, desc="Training")
for src, tgt in progress_bar:
src, tgt = src.to(device), tgt.to(device)
optimizer.zero_grad()
output = model(src)
# 損失計算
output_flat = output.reshape(-1, model.vocab_size)
tgt_flat = tgt.reshape(-1)
loss = criterion(output_flat, tgt_flat)
# 逆伝播
loss.backward()
torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
optimizer.step()
total_loss += loss.item()
progress_bar.set_postfix({'loss': f'{loss.item():.4f}'})
return total_loss / len(dataloader)
def evaluate(
model: TransformerLanguageModel,
dataloader: DataLoader,
criterion: nn.Module,
device: str
) -> float:
model.eval()
total_loss = 0
with torch.no_grad():
for src, tgt in dataloader:
src, tgt = src.to(device), tgt.to(device)
output = model(src)
output_flat = output.reshape(-1, model.vocab_size)
tgt_flat = tgt.reshape(-1)
loss = criterion(output_flat, tgt_flat)
total_loss += loss.item()
return total_loss / len(dataloader)
def train_model(
data_path: str,
save_path: str = 'trained_model',
epochs: int = 10,
batch_size: int = 8,
learning_rate: float = 0.0001,
device: str = None
):
if device is None:
device = 'cuda' if torch.cuda.is_available() else 'cpu'
with open(data_path, 'r', encoding='utf-8') as f:
data = json.load(f)
all_texts = []
for conversation in data:
for turn in conversation['conversation']:
all_texts.append(f"{turn['role']}: {turn['content']}")
tokenizer = Tokenizer()
tokenizer.build_vocab(all_texts)
dataset = ConversationDataset(data_path, tokenizer)
# 訓練/検証分割
train_size = int(0.9 * len(dataset))
val_size = len(dataset) - train_size
train_dataset, val_dataset = torch.utils.data.random_split(
dataset, [train_size, val_size]
)
train_loader = DataLoader(
train_dataset,
batch_size=batch_size,
shuffle=True,
collate_fn=collate_fn
)
val_loader = DataLoader(
val_dataset,
batch_size=batch_size,
shuffle=False,
collate_fn=collate_fn
)
model = TransformerLanguageModel(
vocab_size=tokenizer.vocab_size,
d_model=256,
nhead=8,
num_layers=6,
dim_feedforward=1024,
dropout=0.1
).to(device)
total_params = sum(p.numel() for p in model.parameters())
optimizer = optim.Adam(model.parameters(), lr=learning_rate, weight_decay=0.01)
scheduler = optim.lr_scheduler.ReduceLROnPlateau(optimizer, mode='min', factor=0.5, patience=3)
criterion = nn.CrossEntropyLoss(ignore_index=0)
# 学習
best_val_loss = float('inf')
for epoch in range(epochs):
print(f"\nEpoch {epoch + 1}/{epochs}")
train_loss = train_epoch(model, train_loader, optimizer, criterion, device)
val_loss = evaluate(model, val_loader, criterion, device)
scheduler.step(val_loss)
if val_loss < best_val_loss:
best_val_loss = val_loss
save_model(model, tokenizer, save_path)
ここまでで学習のプロセスは終了です。このモデルを使用したチャットUIを以下で定義します。
class AIAgent:
def __init__(self, model_path: str = "trained_model", device: str = None):
self.device = device or ('cuda' if torch.cuda.is_available() else 'cpu')
# モデルとトークナイザーの読み込み
if os.path.exists(model_path):
self.model, self.tokenizer = load_model(model_path, self.device)
else:
raise FileNotFoundError(f"モデルが見つかりません: {model_path}")
self.conversation_history: List[Dict[str, str]] = []
self.tools: Dict[str, Callable] = {}
self.system_prompt = ""
def register_tool(self, name: str, func: Callable, description: str, parameters: Dict[str, Any]):
self.tools[name] = {
"function": func,
"schema": {
"type": "function",
"function": {
"name": name,
"description": description,
"parameters": parameters
}
}
}
def set_system_prompt(self, prompt: str):
# システムプロンプトを設定
self.system_prompt = prompt
def _parse_tool_call(self, text: str) -> Optional[Dict[str, Any]]:
pattern = r'TOOL\[(\w+)\]\((.*?)\)'
match = re.search(pattern, text)
if match:
tool_name = match.group(1)
args_str = match.group(2)
args = {}
if args_str:
for arg_pair in args_str.split(','):
if '=' in arg_pair:
key, value = arg_pair.split('=', 1)
key = key.strip()
value = value.strip().strip('"\'')
args[key] = value
return {"tool_name": tool_name, "args": args}
return None
def _detect_tool_by_rules(self, message: str) -> Optional[Dict[str, Any]]:
# ルールベースでツールを検出
message_lower = message.lower()
# 例:計算関連
if any(op in message for op in ['+', '-', '*', '/', '×', '÷']):
# 数式を抽出
import re
# 数字と演算子のみを抽出
expr_match = re.search(r'[\d\s\+\-\*\/\(\)\.]+', message)
if expr_match:
expr = expr_match.group().strip()
# *を×に、/を÷に置き換えた場合の処理
expr = message.replace('×', '*').replace('÷', '/')
# 数式部分を抽出
expr_match = re.search(r'[\d\s\+\-\*\/\(\)\.]+', expr)
if expr_match:
return {"tool_name": "calculator", "args": {"expression": expr_match.group().strip()}}
"""~~省略~~"""
# 必要な分だけ追記が必要
return None
def chat(self, user_message: str, max_iterations: int = 3) -> str:
"""
ユーザーメッセージを処理し、応答を返す
Args:
user_message: ユーザーからのメッセージ
max_iterations: 最大実行回数(ツール呼び出しのループ制限)
Returns:
エージェントの応答
"""
# ユーザーメッセージを履歴に追加
self.conversation_history.append({
"role": "user",
"content": user_message
})
# ルールベースのツール判定
detected_tool = self._detect_tool_by_rules(user_message)
if detected_tool:
tool_name = detected_tool["tool_name"]
tool_args = detected_tool["args"]
try:
result = self.tools[tool_name]["function"](**tool_args)
result_str = str(result)
except Exception as e:
result_str = f"エラー: {str(e)}"
self.conversation_history.append({
"role": "assistant",
"content": result_str
})
return result_str
iteration = 0
final_response = ""
while iteration < max_iterations:
iteration += 1
# 会話履歴をプロンプトに変換
prompt = self._build_prompt()
response_text = self._generate_response(prompt)
tool_call = self._parse_tool_call(response_text)
if tool_call and tool_call["tool_name"] in self.tools:
# ツールを実行
tool_name = tool_call["tool_name"]
tool_args = tool_call["args"]
try:
result = self.tools[tool_name]["function"](**tool_args)
result_str = str(result)
except Exception as e:
result_str = f"エラー: {str(e)}"
# ツール実行結果を返す
final_response = result_str
# ツール実行結果を履歴に追加
self.conversation_history.append({
"role": "assistant",
"content": final_response
})
return final_response
else:
clean_response = re.sub(r'TOOL\[.*?\]\(.*?\)', '', response_text).strip()
if not clean_response:
clean_response = response_text.strip()
self.conversation_history.append({
"role": "assistant",
"content": clean_response
})
return clean_response
# 最大反復回数に達した場合
if final_response:
return final_response
return "申し訳ございません。応答を生成できませんでした。"
def _build_prompt(self) -> str:
prompt_parts = [self.system_prompt]
# 利用可能なツールの情報を追加
if self.tools:
prompt_parts.append("\n利用可能なツール:")
for tool_name, tool_info in self.tools.items():
desc = tool_info["schema"]["function"]["description"]
prompt_parts.append(f"- TOOL[{tool_name}]: {desc}")
prompt_parts.append("\nツールを使用する場合は TOOL[tool_name](arg1=value1) の形式で記述してください。\n")
# 会話履歴を追加
# 最新5ターンのみを保持
recent_history = self.conversation_history[-10:] if len(self.conversation_history) > 10 else self.conversation_history
for msg in recent_history:
role = msg["role"]
content = msg["content"]
if content:
prompt_parts.append(f"{role}: {content}")
prompt_parts.append("assistant:")
return "\n".join(prompt_parts)
def _generate_response(self, prompt: str, max_length: int = 100) -> str:
# プロンプトをトークン化
prompt_ids = self.tokenizer.encode(prompt, add_special_tokens=False)
max_prompt_length = 412
if len(prompt_ids) > max_prompt_length:
prompt_ids = prompt_ids[-max_prompt_length:]
original_prompt_length = len(prompt_ids)
generated_ids = self.model.generate(
prompt_ids,
max_length=max_length,
temperature=0.7,
top_k=50,
device=self.device
)
new_tokens = generated_ids[original_prompt_length:]
response = self.tokenizer.decode(new_tokens, skip_special_tokens=True)
# 応答が空の場合のフォールバック
if not response.strip():
response = "申し訳ありません。適切な応答を生成できませんでした。"
return response.strip()
def reset_conversation(self):
# 会話履歴をリセット
self.conversation_history = []
def get_conversation_history(self) -> List[Dict[str, str]]:
# 会話履歴を取得
return self.conversation_history
def calculator(expression: str) -> float:
"""
例:数式を計算するツール
"""
try:
allowed_chars = set("0123456789+-*/()., ")
if not all(c in allowed_chars for c in expression):
raise ValueError("使用できない文字が含まれています")
result = eval(expression, {"__builtins__": {}}, {})
return float(result)
except Exception as e:
raise ValueError(f"計算エラー: {str(e)}")
# 必要な分だけ追記が必要
# ツール定義のスキーマ
TOOL_SCHEMAS = {
"calculator": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "計算する数式"
}
},
"required": ["expression"]
}
}
ローカル環境で動作させてみます。アクセスすると以下のような画面が表示されます。
画面下部のメッセージを入力してください...の欄に質問したい内容を書き込みます。書き込んだら送信ボタンを押します。
計算結果を返してくれました。
ここまで読んでいただいた方ならお分かりだと思いますが、AI部分の応答がほとんど機能していません。
応答に関してはルールベースで処理しています。以下の関数で応答に用いるツールをキーワードマッピングしています。
def _detect_tool_by_rules(self, message: str) -> Optional[Dict[str, Any]]:
ここで呼出されるツール群に関しても定義する必要があります。今回は動作確認用に計算ツールを例としてコード部分に残していますが、もし追加機能が必要であればその分だけルールを追加しなくてはいけません。
言語モデルは補助的な役割に留まっているのが現状です。モデルの学習が不十分なことが最たる原因です。
今回学習データについては精度を度外視して作成したので当たり前の結果ではあります。
何の比較にもなりませんが技術紹介の項で記載したChatGPTで使われているモデルGPT-3と今回自作したモデルの比較はこちらになります。GPT-3の値は論文1内の最大モデルとなります。自作モデルの内容は伏せていますが、いかに異次元の数量であるかが伝わってもらえればと思います。さらにGPT-3の後継モデルについて詳細は非公開ですがGPT-3を上回るデータと言われているので途方もないです。
| モデル | パラメータ数 | トークン数 |
|---|---|---|
| GPT-3 | 1,750億 | 3,000億 |
| 自作モデル | 430万 | 5000 |
まとめ
ルールベースとAIを使用したハイブリッド型のチャットボットを作成してみました。
当然ですが自作モデルなので使い物には程遠かったです。今後アップデートするのであれば既存モデルを使用してみることにも挑戦してみようと思います。それ以外にも大規模データで学習する、ファインチューニングを実施する、学習アルゴリズムに手を入れるなどなど挙げればキリがありません。
自作モデルで動くようにできたのは苦労したので、個人レベルでも第一歩は踏み出せることが伝わってもらえれば嬉しいです。
明日は@ms-ichiさんの記事が公開されます。興味がある方はそちらの記事も是非ご覧ください。
では、またどこかで。
論文のリンク
-
Tom B. Brown , Benjamin Mann , Nick Ryder , Melanie Subbiah , et al."Language Models are Few-Shot Learners" arXiv 2020.
https://arxiv.org/abs/2005.14165 ↩ ↩2 -
Ashish Vaswani , Noam Shazeer , Niki Parmar , Jakob Uszkoreit , Llion Jones , Aidan N. Gomez , Lukasz Kaiser , Illia Polosukhin "Attention Is All You Need" arXiv 2023.
https://arxiv.org/pdf/1706.03762 ↩


