ClaudeでRAGを実装する｜検索拡張生成の設計パターンと実践例

defchunk_text(text, chunk_size=1000, overlap=100):
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunks.append(text[start:end])
        start = end - overlap
    return chunks

system_prompt = """あなたは社内文書に基づいて質問に回答するアシスタントです。

ルール:
- 提供されたコンテキストに基づいてのみ回答する
- コンテキストに情報がない場合は「この情報は見つかりませんでした」と答える
- 回答の根拠となるドキュメント名を明記する"""

user_prompt = f"""以下のコンテキストを参照して、質問に答えてください。

<context>
{retrieved_documents}
</context>

質問:{user_question}"""

context_parts = []
for i, doc in enumerate(retrieved_docs):
    context_parts.append(f"""<document index="{i+1}" source="{doc['source']}">
{doc['content']}
</document>""")

context = "nn".join(context_parts)

importanthropic

client = anthropic.Anthropic()

response = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    messages=[
        {
            "role": "user",
            "content": [
                {
                    "type": "document",
                    "source": {
                        "type": "text",
                        "media_type": "text/plain",
                        "data": "社内規定: 有給休暇は年間20日付与される。..."
                    },
                    "title": "就業規則.txt",
                    "citations": {"enabled": True}
                },
                {
                    "type": "text",
                    "text": "有給休暇は何日もらえますか？"
                }
            ]
        }
    ]
)

for block in response.content:
    if block.type == "text":
        print(block.text)
        if hasattr(block, "citations") and block.citations:
            for cite in block.citations:
                print(f"  出典:{cite.document_title} -{cite.cited_text}")

importanthropic
importnumpyasnp
frompathlibimport Path

# --- 1. ドキュメントの読み込みとチャンク化 ---
defload_and_chunk(directory: str, chunk_size: int = 800) -> list[dict]:
    chunks = []
    for path in Path(directory).glob("*.md"):
        text = path.read_text(encoding="utf-8")
        # 見出しで分割
        sections = text.split("n## ")
        for section in sections:
            if len(section.strip()) > 50:
                chunks.append({
                    "source": path.name,
                    "content": section.strip()
                })
    return chunks

# --- 2. embeddingの生成（OpenAI） ---
fromopenaiimport OpenAI

openai_client = OpenAI()

defget_embeddings(texts: list[str]) -> list[list[float]]:
    response = openai_client.embeddings.create(
        model="text-embedding-3-small",
        input=texts
    )
    return [item.embedding for item in response.data]

# --- 3. 検索 ---
defsearch(query: str, chunks: list[dict], embeddings: np.ndarray, top_k: int = 10) -> list[dict]:
    query_embedding = get_embeddings([query])[0]
    query_vec = np.array(query_embedding)

    # コサイン類似度
    similarities = np.dot(embeddings, query_vec) / (
        np.linalg.norm(embeddings, axis=1) * np.linalg.norm(query_vec)
    )

    top_indices = np.argsort(similarities)[-top_k:][::-1]
    return [chunks[i] for i in top_indices]

# --- 4. 回答生成 ---
claude_client = anthropic.Anthropic()

defanswer(query: str, retrieved: list[dict]) -> str:
    context = "nn".join([
        f"[{doc['source']}]n{doc['content']}" for doc in retrieved
    ])

    response = claude_client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=1024,
        system="提供されたコンテキストに基づいて回答してください。情報がない場合はその旨伝えてください。",
        messages=[
            {
                "role": "user",
                "content": f"<context>n{context}n</context>nn質問:{query}"
            }
        ]
    )
    return response.content[0].text