AI: RAG（検索拡張生成）

from openai import OpenAI

client = OpenAI()

def get_embedding(text: str, model: str = "text-embedding-3-small") -> list[float]:
    response = client.embeddings.create(input=text, model=model)
    return response.data[0].embedding

text = "国内出張の日当は 1 日 $45 です。"
vector = get_embedding(text)

print(f"Dimension: {len(vector)}")
print(f"First 5 values: {vector[:5]}")

Dimension: 1536
First 5 values: [0.0123, -0.0045, 0.0378, -0.0211, 0.0056]

import numpy as np

def cosine_similarity(a: list[float], b: list[float]) -> float:
    a_arr = np.array(a)
    b_arr = np.array(b)
    dot = np.dot(a_arr, b_arr)
    norm_a = np.linalg.norm(a_arr)
    norm_b = np.linalg.norm(b_arr)
    return float(dot / (norm_a * norm_b))

sentences = {
    "travel_allowance": get_embedding("国内出張の日当は 1 日 $45 です。"),
    "reimbursement": get_embedding("出張費の精算はどう申請しますか？"),
    "unrelated": get_embedding("今日の天気は晴れです。"),
}

sim1 = cosine_similarity(sentences["travel_allowance"], sentences["reimbursement"])
sim2 = cosine_similarity(sentences["travel_allowance"], sentences["unrelated"])

print(f"Related sentences:   {sim1:.4f}")
print(f"Unrelated sentences: {sim2:.4f}")

Related sentences:   0.8234
Unrelated sentences: 0.3012

def top_k_search(
    query_embedding: list[float],
    doc_embeddings: list[list[float]],
    doc_texts: list[str],
    k: int = 3,
) -> list[tuple[str, float]]:
    similarities = [
        (doc_texts[i], cosine_similarity(query_embedding, doc_embeddings[i]))
        for i in range(len(doc_texts))
    ]
    similarities.sort(key=lambda x: x[1], reverse=True)
    return similarities[:k]

documents = [
    "国内出張の日当は 1 日 $45 です。",
    "海外出張には事前に副社長の承認が必要です。",
    "従業員は 30 日以内に領収書を提出しなければなりません。",
    "在宅勤務ポリシーでは週最大 2 日の在宅を認めています。",
    "オフィスの給湯室には無料のコーヒーと軽食が備えられています。",
]

doc_embeddings = [get_embedding(doc) for doc in documents]
query = "出張で 1 日いくら使えますか？"
query_embedding = get_embedding(query)

results = top_k_search(query_embedding, doc_embeddings, documents, k=3)

for text, score in results:
    print(f"[{score:.4f}] {text}")

[0.8912] 国内出張の日当は 1 日 $45 です。
[0.7634] 従業員は 30 日以内に領収書を提出しなければなりません。
[0.7102] 海外出張には事前に副社長の承認が必要です。

from openai import OpenAI

client = OpenAI()

def rag_query(question: str, documents: list[str], k: int = 3) -> str:
    query_emb = get_embedding(question)
    doc_embs = [get_embedding(doc) for doc in documents]
    top_docs = top_k_search(query_emb, doc_embs, documents, k=k)

    context = "\n\n".join([f"[Doc {i+1}] {text}" for i, (text, _) in enumerate(top_docs)])

    prompt = f"""以下の文脈のみに基づいて質問に答えてください。
文脈に答えが含まれていない場合は「分かりません」と言ってください。

文脈:
{context}

質問: {question}
答え:"""

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
    )
    return response.choices[0].message.content

docs = [
    "国内出張の日当は 1 日 $45 です。",
    "海外出張には少なくとも 2 週間前に副社長の承認が必要です。",
    "従業員は出張終了後 30 日以内に経費報告書を提出しなければなりません。",
    "ホテルの精算上限は 1 泊 $200 です。",
    "航空券の予約は会社の出張ポータルを利用しなければなりません。",
]

answer = rag_query("出張の日当はいくらですか？", docs)
print(answer)

国内出張の日当は 1 日 $45 です。

def ask_without_rag(question: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": question}],
        temperature=0,
    )
    return response.choices[0].message.content

question = "Acme 社の出張日当はいくらですか？"

print("=== RAG なし ===")
print(ask_without_rag(question))

print("\n=== RAG あり ===")
print(rag_query(question, docs))

=== RAG なし ===
Acme 社の出張日当についての具体的な情報は持ち合わせていません。
企業の出張手当はさまざまですが、一般的な範囲は 1 日 $50〜$100 です。
正確な金額は自社の出張規定をご確認ください。

=== RAG あり ===
国内出張の日当は 1 日 $45 です。

from langchain.text_splitter import RecursiveCharacterTextSplitter

text = """Acme 社 出張規定

1. 国内出張
出張の日当は 1 日 $45 です。食事および雑費が含まれます。
ホテルの精算上限は 1 泊 $200 です。

2. 海外出張
すべての海外出張には少なくとも 2 週間前に副社長の承認が必要です。
海外出張の日当は 1 日 $75 です。
ホテルの精算上限は 1 泊 $300 です。

3. 経費報告
すべての経費報告書は出張完了後 30 日以内に提出しなければなりません。
$25 を超えるすべての経費には領収書が必要です。"""

splitter = RecursiveCharacterTextSplitter(
    chunk_size=200,
    chunk_overlap=50,
    separators=["\n\n", "\n", ". ", " "],
)

chunks = splitter.split_text(text)
for i, chunk in enumerate(chunks):
    print(f"--- Chunk {i+1} ---")
    print(chunk)
    print()

--- Chunk 1 ---
Acme 社 出張規定

1. 国内出張
出張の日当は 1 日 $45 です。食事および雑費が含まれます。
ホテルの精算上限は 1 泊 $200 です。

--- Chunk 2 ---
2. 海外出張
すべての海外出張には少なくとも 2 週間前に副社長の承認が必要です。
海外出張の日当は 1 日 $75 です。

--- Chunk 3 ---
海外出張の日当は 1 日 $75 です。
ホテルの精算上限は 1 泊 $300 です。

3. 経費報告
すべての経費報告書は出張完了後 30 日以内に提出しなければなりません。

import numpy as np
from openai import OpenAI

client = OpenAI()

# --- ステップ 1: 文書を準備 ---
documents = [
    {
        "title": "出張規定",
        "content": (
            "Acme 社 出張規定\n\n"
            "1. 国内出張\n"
            "国内出張の日当は 1 日 $45 です。"
            "食事および雑費が含まれます。"
            "ホテルの精算上限は 1 泊 $200 です。\n\n"
            "2. 海外出張\n"
            "すべての海外出張には少なくとも 2 週間前に副社長の承認が必要です。"
            "海外出張の日当は 1 日 $75 です。"
            "ホテルの精算上限は 1 泊 $300 です。"
        ),
    },
    {
        "title": "休暇規定",
        "content": (
            "Acme 社 休暇規定\n\n"
            "1. 年次有給休暇\n"
            "正社員は年間 15 日の有給休暇を付与されます。"
            "未使用の休暇は最大 5 日まで翌年に繰り越せます。\n\n"
            "2. 病気休暇\n"
            "従業員は年間 10 日の有給病気休暇を付与されます。"
            "3 日を超える連続欠勤には医師の診断書が必要です。"
        ),
    },
    {
        "title": "IT セキュリティ規定",
        "content": (
            "Acme 社 IT セキュリティ規定\n\n"
            "1. パスワード要件\n"
            "すべてのパスワードは少なくとも 12 文字で、大文字・小文字・数字・"
            "特殊文字を含まなければなりません。"
            "パスワードは 90 日ごとに変更する必要があります。\n\n"
            "2. デバイスポリシー\n"
            "個人デバイスは社内ネットワークに接続してはなりません。"
            "すべての社用デバイスには承認済みのウイルス対策ソフトを導入しなければなりません。"
        ),
    },
]

# --- ステップ 2: 文書をチャンク化 ---
def chunk_text(text: str, chunk_size: int = 200, overlap: int = 50) -> list[str]:
    words = text.split()
    chunks = []
    start = 0
    while start < len(words):
        end = start + chunk_size
        chunks.append(" ".join(words[start:end]))
        start += chunk_size - overlap
    return chunks

all_chunks: list[str] = []
chunk_sources: list[str] = []

for doc in documents:
    chunks = chunk_text(doc["content"])
    for chunk in chunks:
        all_chunks.append(chunk)
        chunk_sources.append(doc["title"])

print(f"Total chunks: {len(all_chunks)}")

# --- ステップ 3: 埋め込みを生成 ---
def get_embedding(text: str, model: str = "text-embedding-3-small") -> list[float]:
    response = client.embeddings.create(input=text, model=model)
    return response.data[0].embedding

chunk_embeddings = [get_embedding(chunk) for chunk in all_chunks]
print(f"Embedding dimension: {len(chunk_embeddings[0])}")

# --- ステップ 4: ベクトル検索 ---
def cosine_similarity(a: list[float], b: list[float]) -> float:
    a_arr = np.array(a)
    b_arr = np.array(b)
    return float(np.dot(a_arr, b_arr) / (np.linalg.norm(a_arr) * np.linalg.norm(b_arr)))

def search(query: str, k: int = 3) -> list[tuple[str, str, float]]:
    query_emb = get_embedding(query)
    scores = [
        (all_chunks[i], chunk_sources[i], cosine_similarity(query_emb, chunk_embeddings[i]))
        for i in range(len(all_chunks))
    ]
    scores.sort(key=lambda x: x[2], reverse=True)
    return scores[:k]

# --- ステップ 5: RAG クエリ ---
def rag_answer(question: str, k: int = 3) -> str:
    results = search(question, k=k)

    context_parts = []
    for i, (chunk, source, score) in enumerate(results):
        context_parts.append(f"[ソース: {source}, 関連度: {score:.2f}]\n{chunk}")
    context = "\n\n---\n\n".join(context_parts)

    prompt = f"""以下の文脈のみに基づいて質問に答えてください。
文脈に十分な情報が含まれていない場合は「情報が不足しています」と言ってください。
回答の根拠となったソース文書を必ず引用してください。

文脈:
{context}

質問: {question}
答え:"""

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
    )
    return response.choices[0].message.content

# --- ステップ 6: テスト ---
questions = [
    "国内出張の日当はいくらですか？",
    "従業員の年次有給休暇は何日ですか？",
    "パスワードの要件は何ですか？",
    "会社の在宅勤務ポリシーは何ですか？",
]

for q in questions:
    print(f"Q: {q}")
    print(f"A: {rag_answer(q)}")
    print()

Total chunks: 9
Embedding dimension: 1536

Q: 国内出張の日当はいくらですか？
A: 出張規定によれば、国内出張の日当は 1 日 $45 です。

Q: 従業員の年次有給休暇は何日ですか？
A: 休暇規定によれば、正社員は年間 15 日の有給休暇を付与されます。

Q: パスワードの要件は何ですか？
A: パスワードは少なくとも 12 文字で、大文字・小文字・数字・特殊文字を含まなければなりません。IT セキュリティ規定によれば、90 日ごとに変更する必要があります。

Q: 会社の在宅勤務ポリシーは何ですか？
A: 情報が不足しています。提供された文脈には在宅勤務ポリシーが含まれていません。

texts = [
    "猫は独立心で知られる人気のペットです。",
    "犬は遊ぶのが好きで忠実な相棒です。",
    "好業績を受けて本日の株価は 3% 上昇しました。",
    "ネコ科の動物は日当たりの良い場所で登ったり眠ったりするのを好みます。",
    "中央銀行は金利を引き上げました。",
]

def vector_search(
    query: str,
    corpus: list[str],
    k: int = 5,
    embedding_model: str = "text-embedding-3-small",
) -> list[dict]:
    """
    スコア付きの上位 k 件の類似文書を返す。
    各結果は dict: {"text": ..., "score": ..., "rank": ...}
    """
    # ここに実装を書く
    pass