こんにちは、かつコーチです。
前回はpgvectorでベクトル検索の環境を構築しました。
今回はいよいよ、RAG（AI自身の知識ではなく、検索した文書を渡して回答させる仕組み。カンニングペーパーを渡してから質問に答えてもらうイメージです）のパイプライン全体を実装します。
チャンク分割からembedding生成、pgvectorへの保存、検索、生成までを一気通貫でコードにしていきます。
RAGパイプラインの全体像
RAGの実装は、大きく分けて次の5ステップで構成されます。
- 文書をチャンク（意味の塊）に分割する
- 各チャンクをembeddingに変換する
- embeddingをpgvectorに保存する
- ユーザーの質問をembedding化し、コサイン類似度で近いチャンクを検索する
- 検索結果をプロンプトに注入し、LLMに回答を生成させる
順番に実装していきます。
実装手順
文書のチャンク分割
チャンクは見出し・段落単位で、目安500トークン程度に分割します。
文の途中で割ると意味が崩れるため、段落の区切り（空行）を基準に分割し、それでも長すぎる場合のみ文単位でさらに分割します。
<?php
namespace App\Services\Rag;
class TextChunker
{
public function __construct(
private int $maxTokens = 500,
) {}
/**
* @return array<int, string>
*/
public function chunk(string $text): array
{
// 見出し・段落単位（空行区切り）でまず分割
$paragraphs = preg_split('/\n{2,}/u', trim($text));
$chunks = [];
$buffer = '';
foreach ($paragraphs as $paragraph) {
$paragraph = trim($paragraph);
if ($paragraph === '') {
continue;
}
// バッファに足しても上限内なら結合、超えるなら確定して次へ
if ($this->estimateTokens($buffer . $paragraph) <= $this->maxTokens) {
$buffer = $buffer === '' ? $paragraph : $buffer . "\n\n" . $paragraph;
continue;
}
if ($buffer !== '') {
$chunks[] = $buffer;
}
// 1段落だけで上限を超える場合は文単位でさらに分割
if ($this->estimateTokens($paragraph) > $this->maxTokens) {
array_push($chunks, ...$this->splitBySentence($paragraph));
$buffer = '';
} else {
$buffer = $paragraph;
}
}
if ($buffer !== '') {
$chunks[] = $buffer;
}
return $chunks;
}
private function splitBySentence(string $paragraph): array
{
$sentences = preg_split('/(?<=[。！？])/u', $paragraph, -1, PREG_SPLIT_NO_EMPTY);
$chunks = [];
$buffer = '';
foreach ($sentences as $sentence) {
if ($this->estimateTokens($buffer . $sentence) <= $this->maxTokens) {
$buffer .= $sentence;
continue;
}
if ($buffer !== '') {
$chunks[] = $buffer;
}
$buffer = $sentence;
}
if ($buffer !== '') {
$chunks[] = $buffer;
}
return $chunks;
}
private function estimateTokens(string $text): int
{
// 日本語混じりの簡易概算。厳密なトークン数はtiktoken等での計測を推奨
return (int) ceil(mb_strlen($text) / 1.5);
}
}
トークン数の厳密な計測にはtiktoken互換のライブラリを使うのが理想ですが、まずは概算でパイプラインを組み、後から精度を上げる方針で問題ありません。
embedding生成とpgvectorへの保存
チャンクごとにembeddingを生成し、document_chunksテーブルに保存します。
<?php
namespace App\Services\Rag;
use App\Models\Document;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Http;
class DocumentIndexer
{
public function __construct(
private TextChunker $chunker,
) {}
public function index(Document $document): void
{
$chunks = $this->chunker->chunk($document->body);
foreach ($chunks as $chunk) {
$embedding = $this->embed($chunk);
DB::table('document_chunks')->insert([
'document_id' => $document->id,
'content' => $chunk,
'token_count' => (int) ceil(mb_strlen($chunk) / 1.5),
'embedding' => DB::raw("'[" . implode(',', $embedding) . "]'::vector"),
'created_at' => now(),
'updated_at' => now(),
]);
}
}
/**
* @return array<int, float>
*/
private function embed(string $text): array
{
$response = Http::withToken(config('services.openai.key'))
->post('https://api.openai.com/v1/embeddings', [
'model' => 'text-embedding-3-small',
'input' => $text,
])
->throw();
return $response->json('data.0.embedding');
}
}
コサイン類似度での検索
質問文をembedding化し、<=>演算子で近いチャンクを取得します。
<?php
namespace App\Services\Rag;
use Illuminate\Support\Facades\DB;
class ChunkSearcher
{
public function __construct(
private DocumentIndexer $indexer,
) {}
/**
* @return array<int, object>
*/
public function search(string $query, int $limit = 5): array
{
$queryEmbedding = (new \ReflectionClass($this->indexer))
->getMethod('embed');
// 実運用ではembedメソッドを共通サービスに切り出して直接呼び出す
$queryEmbedding->setAccessible(true);
$vector = $queryEmbedding->invoke($this->indexer, $query);
$vectorLiteral = '[' . implode(',', $vector) . ']';
return DB::select(
"SELECT id, document_id, content, embedding <=> ?::vector AS distance
FROM document_chunks
ORDER BY embedding <=> ?::vector
LIMIT ?",
[$vectorLiteral, $vectorLiteral, $limit]
);
}
}
補足として、embedメソッドをリフレクションで呼ぶのは可読性が悪いため、実際のプロジェクトではEmbeddingClientのような共通サービスに切り出し、DocumentIndexerとChunkSearcherの両方から注入する設計にしてください。
検索結果をプロンプトに注入して生成
最後に、検索したチャンクをシステムプロンプトに埋め込み、LLMに回答させます。
<?php
namespace App\Services\Rag;
use Illuminate\Support\Facades\Http;
class RagAnswerGenerator
{
public function __construct(
private ChunkSearcher $searcher,
) {}
public function answer(string $question): string
{
$chunks = $this->searcher->search($question);
$context = collect($chunks)
->map(fn ($chunk) => "- {$chunk->content}")
->implode("\n");
$response = Http::withToken(config('services.openai.key'))
->post('https://api.openai.com/v1/chat/completions', [
'model' => 'gpt-4o-mini',
'messages' => [
[
'role' => 'system',
'content' => "以下の参考情報のみを根拠に、日本語で質問に答えてください。\n\n参考情報:\n{$context}",
],
['role' => 'user', 'content' => $question],
],
])
->throw();
return $response->json('choices.0.message.content');
}
}
よくあるつまずきポイント・エラー対処
実際に私がこのパイプラインを組んだとき、検索結果がまったく関係ない内容ばかり返ってくる現象に悩まされました。
原因を追ったところ、チャンク分割時に段落の改行コードが\r\nと\n混在していて、正規表現の分割が意図通りに効いていませんでした。
結果として1つの巨大なチャンクとして保存され、意味的にぼやけたembeddingになっていたのです。
- ❌Before：
preg_split('/\n{2,}/u', $text)のみで、Windows由来の\r\n混在テキストが分割されず1チャンク化 - ✅After：チャンク分割前に
str_replace("\r\n", "\n", $text)で改行コードを正規化してから分割
さらに、embedding APIのレスポンスが遅延する場合に備えて、index()メソッドは必ずキュー経由（ShouldQueue実装のJob）で非同期実行するようにしました。
同期実行のままだと、文書登録数が増えるほどHTTPリクエストがタイムアウトしやすくなります。
応用・一歩先の使い方
検索精度をさらに上げたい場合、コサイン類似度だけでなく、キーワード全文検索（PostgreSQLのtsvector）と組み合わせるハイブリッド検索が有効です。
SELECT id, content,
embedding <=> ?::vector AS vector_distance,
ts_rank(to_tsvector('simple', content), plainto_tsquery('simple', ?)) AS keyword_score
FROM document_chunks
ORDER BY vector_distance
LIMIT 10;
ベクトル距離とキーワードスコアを重み付けして再ランキングすることで、専門用語の完全一致が重要なドメインでも検索漏れを減らせます。
まとめ
この記事のポイント
- チャンク分割は段落単位を基本に、改行コードの正規化を忘れずに行う
- embedding生成はキュー経由の非同期処理にしてタイムアウトを防ぐ
- 検索は
<=>演算子でコサイン距離を計算し、近い順にプロンプトへ注入する - 精度向上にはベクトル検索とキーワード検索のハイブリッド構成が有効
次に読むべき記事
- 次回：AI呼び出しをテストでモックする（Http::fakeパターン）
コメント