ユーザーの質問 → 関連ドキュメントを検索 → [ドキュメント+質問]をLLMに提供 → 回答を生成

from sentence_transformers import SentenceTransformer
import numpy as np

model = SentenceTransformer('all-MiniLM-L6-v2')

documents = [
    "Pythonはデータサイエンスに広く使われているプログラミング言語です。",
    "機械学習はデータからパターンを学習するAIの分野です。",
    "パリはフランスの首都です。",
    "ディープラーニングはニューラルネットワークを使った機械学習手法です。",
]

doc_embeddings = model.encode(documents)
print(f"埋め込みの形状: {doc_embeddings.shape}")  # (4, 384)

query = "人工知能とデータ分析"
query_embedding = model.encode([query])[0]

def cosine_similarity(a, b):
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

similarities = [cosine_similarity(query_embedding, emb) for emb in doc_embeddings]
ranked = sorted(zip(similarities, documents), reverse=True)

print("\n類似度でランク付け:")
for score, doc in ranked:
    print(f"  {score:.4f}: {doc}")

from langchain.text_splitter import CharacterTextSplitter

text = """
機械学習は人工知能のサブフィールドであり、明示的にプログラムすることなく
コンピュータがデータから学習できるようにします。手法には教師あり学習、
教師なし学習、強化学習などがあり、それぞれ異なる種類の問題を解きます。
"""

splitter = CharacterTextSplitter(
    chunk_size=200,
    chunk_overlap=20,
    separator="\n"
)
chunks = splitter.split_text(text)
for i, chunk in enumerate(chunks):
    print(f"チャンク {i}: {chunk[:80]}...")

from langchain.text_splitter import RecursiveCharacterTextSplitter

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

with open("document.txt", "r", encoding="utf-8") as f:
    text = f.read()

chunks = splitter.split_text(text)
print(f"合計チャンク数: {len(chunks)}")
print(f"平均チャンク長: {sum(len(c) for c in chunks) / len(chunks):.0f} 文字")

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

semantic_splitter = SemanticChunker(
    embeddings=OpenAIEmbeddings(),
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=95    # 上位5%の類似度変化点で分割
)

chunks = semantic_splitter.create_documents([text])
print(f"セマンティックチャンキング結果: {len(chunks)} チャンク")

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings

child_splitter = RecursiveCharacterTextSplitter(chunk_size=200)
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=1000)

vectorstore = Chroma(
    collection_name="full_documents",
    embedding_function=OpenAIEmbeddings()
)
store = InMemoryStore()

retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=store,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter,
)

import numpy as np
import faiss
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain.schema import Document

# ===== FAISSの直接使用 =====
d = 384       # ベクトル次元
n = 10000     # ドキュメント数
vectors = np.random.randn(n, d).astype('float32')

# Flat L2インデックス（完全探索）
index_flat = faiss.IndexFlatL2(d)
index_flat.add(vectors)
print(f"FAISSインデックスサイズ: {index_flat.ntotal}")

# 検索
query = np.random.randn(1, d).astype('float32')
k = 5
distances, indices = index_flat.search(query, k)
print(f"上位 {k} 件: {indices[0]}")

# HNSWインデックス（近似、高速）
index_hnsw = faiss.IndexHNSWFlat(d, 32)
index_hnsw.add(vectors)
distances_hnsw, indices_hnsw = index_hnsw.search(query, k)
print(f"HNSW結果: {indices_hnsw[0]}")

# ===== LangChain + Chroma =====
documents = [
    Document(page_content="Pythonはデータサイエンスに使われています。", metadata={"source": "doc1"}),
    Document(page_content="機械学習はデータからパターンを学習します。", metadata={"source": "doc2"}),
    Document(page_content="ディープラーニングはニューラルネットワークベースの機械学習です。", metadata={"source": "doc3"}),
    Document(page_content="NLPはテキストデータを分析します。", metadata={"source": "doc4"}),
]

embeddings = OpenAIEmbeddings()
chroma_db = Chroma.from_documents(
    documents,
    embeddings,
    persist_directory="./chroma_db"
)

results = chroma_db.similarity_search("AIと機械学習", k=2)
for doc in results:
    print(f"ソース: {doc.metadata['source']}, 内容: {doc.page_content}")

from langchain_community.document_loaders import PyPDFLoader, DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate

# ===== 1. ドキュメントの読み込み =====
loader = PyPDFLoader("company_handbook.pdf")
pages = loader.load()
print(f"読み込んだページ数: {len(pages)}")

# ===== 2. テキストの分割 =====
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", ".", "!", "?", ",", " "]
)
chunks = text_splitter.split_documents(pages)
print(f"作成されたチャンク数: {len(chunks)}")

# ===== 3. 埋め込みとベクトルDB保存 =====
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(
    chunks,
    embeddings,
    persist_directory="./rag_db"
)
print("ベクトルDB保存完了")

# ===== 4. RAGチェーンの設定 =====
prompt_template = """あなたは役に立つAIアシスタントです。
提供されたコンテキストのみを使用して質問に回答してください。
コンテキストに回答がない場合は、「提供されたドキュメントには見つかりませんでした。」と言ってください。

コンテキスト:
{context}

質問: {question}

回答:"""

PROMPT = PromptTemplate(
    template=prompt_template,
    input_variables=["context", "question"]
)

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 4}
)

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    chain_type_kwargs={"prompt": PROMPT},
    return_source_documents=True
)

# ===== 5. 質問応答 =====
query = "有給休暇のポリシーは何ですか？"
result = qa_chain.invoke({"query": query})

print(f"\n質問: {query}")
print(f"回答: {result['result']}")
print(f"\nソースドキュメント:")
for doc in result['source_documents']:
    print(f"  - ページ {doc.metadata.get('page', '?')}: {doc.page_content[:100]}...")

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings
from llama_index.core.node_parser import SentenceSplitter
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding

Settings.llm = OpenAI(model="gpt-4o-mini", temperature=0)
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")
Settings.node_parser = SentenceSplitter(chunk_size=512, chunk_overlap=20)

documents = SimpleDirectoryReader("./docs/").load_data()
index = VectorStoreIndex.from_documents(documents)

query_engine = index.as_query_engine(
    similarity_top_k=4,
    response_mode="compact"
)

response = query_engine.query("これらのドキュメントで扱われている主なトピックは何ですか？")
print(f"回答: {response}")
print("\nソースノード:")
for node in response.source_nodes:
    print(f"  - スコア: {node.score:.4f}")
    print(f"    テキスト: {node.text[:100]}...")

from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever

# BM25リトリーバー（キーワードベース）
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 4

# ベクトルリトリーバー（意味的）
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

# アンサンブル（ハイブリッド）
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.5, 0.5]
)

results = ensemble_retriever.invoke("Pythonプログラミングチュートリアル")
print(f"ハイブリッド検索結果: {len(results)}")

from langchain.retrievers.multi_query import MultiQueryRetriever
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(temperature=0)
multi_query_retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(),
    llm=llm
)

# 内部でLLMが質問を複数のバージョンに書き直す
# 例：「RAGの利点は何ですか？」
# → 「検索拡張生成の利点は何ですか？」
# → 「RAGは標準的なLLMよりなぜ優れているのか？」
# → 「RAGが役立つのはなぜか？」

results = multi_query_retriever.invoke("RAGの利点は何ですか？")
print(f"マルチクエリ検索結果: {len(results)}")

mmr_retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 4,             # 最終的に返す件数
        "fetch_k": 20,      # 初期候補プール
        "lambda_mult": 0.5  # バランス：0=多様性、1=類似度
    }
)

results = mmr_retriever.invoke("機械学習アルゴリズム")
print(f"MMR結果: {len(results)}")

from langchain.schema import Document

docs_with_metadata = [
    Document(
        page_content="2024年Q1の収益は1000万ドルでした。",
        metadata={"year": 2024, "quarter": "Q1", "category": "financial"}
    ),
    Document(
        page_content="2024年Q2の収益は1200万ドルでした。",
        metadata={"year": 2024, "quarter": "Q2", "category": "financial"}
    ),
    Document(
        page_content="技術ロードマップ：AI機能の強化を計画中。",
        metadata={"year": 2024, "quarter": "Q1", "category": "strategy"}
    ),
]

# メタデータフィルターで検索を絞り込む
filtered_results = vectorstore.similarity_search(
    "収益パフォーマンス",
    k=2,
    filter={"category": "financial", "year": 2024}
)

from sentence_transformers import CrossEncoder
import numpy as np

cross_encoder = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')

query = "機械学習アルゴリズムの種類"
initial_results = vectorstore.similarity_search(query, k=20)  # 多く検索

# クロスエンコーダーで再スコアリング
pairs = [[query, doc.page_content] for doc in initial_results]
scores = cross_encoder.predict(pairs)

# スコアでソート
ranked = sorted(zip(scores, initial_results), reverse=True)
top_k = [doc for _, doc in ranked[:5]]

print("リランキング後の上位結果:")
for score, doc in ranked[:3]:
    print(f"  スコア {score:.4f}: {doc.page_content[:80]}...")

from langchain.retrievers.document_compressors import CohereRerank
from langchain.retrievers import ContextualCompressionRetriever

compressor = CohereRerank(
    cohere_api_key="your-api-key",
    top_n=3,
    model="rerank-multilingual-v3.0"
)

compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 20})
)

results = compression_retriever.invoke("会社のポリシーについて教えてください")
print(f"リランキング後のドキュメント: {len(results)}")

from FlagEmbedding import FlagReranker

reranker = FlagReranker('BAAI/bge-reranker-large', use_fp16=True)

query = "RAGとは何ですか？"
passages = [
    "RAGはRetrieval-Augmented Generationの略です。",
    "ragは掃除に使う布切れのことです。",
    "RAGシステムは検索と生成を組み合わせてLLMの回答を向上させます。",
]

scores = reranker.compute_score([[query, p] for p in passages])
ranked = sorted(zip(scores, passages), reverse=True)

for score, passage in ranked:
    print(f"  {score:.4f}: {passage}")

クエリ → LLMが仮説的回答を生成 → 仮説的回答を埋め込む → 実際のドキュメントを検索

from langchain.chains import HypotheticalDocumentEmbedder
from langchain_openai import OpenAI, OpenAIEmbeddings, ChatOpenAI

llm = OpenAI()
embeddings = OpenAIEmbeddings()

# LangChain組み込みのHyDE
hyde_embeddings = HypotheticalDocumentEmbedder.from_llm(
    llm=llm,
    embeddings=embeddings,
    prompt_key="web_search"
)

# 手動HyDE実装
def manual_hyde(query, llm, embeddings, vectorstore, k=4):
    # 1. 仮説的ドキュメントを生成
    hypothetical_doc = llm.invoke(
        f"次の質問に対する詳細な回答を書いてください: {query}"
    )

    # 2. 仮説的ドキュメントを埋め込む
    hyp_embedding = embeddings.embed_query(hypothetical_doc.content)

    # 3. 仮説的埋め込みで検索
    results = vectorstore.similarity_search_by_vector(hyp_embedding, k=k)

    return results, hypothetical_doc.content

chat_llm = ChatOpenAI(temperature=0.7)
results, hyp_doc = manual_hyde(
    "ディープラーニングの歴史", chat_llm, embeddings, vectorstore
)
print(f"仮説的ドキュメント: {hyp_doc[:200]}...")
print(f"検索された実際のドキュメント: {len(results)}")

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

class SelfRAGSimulator:
    """Self-RAGの動作をシミュレート（実際のSelf-RAGには特別に学習されたモデルが必要）"""

    def __init__(self, retriever, llm):
        self.retriever = retriever
        self.llm = llm

    def should_retrieve(self, query: str) -> bool:
        prompt = ChatPromptTemplate.from_template("""
次の質問に回答するために外部ドキュメントを検索する必要があるかどうか判断してください。
一般的な知識や推論だけで回答できる場合は「NO」と答えてください。
特定の事実や専門知識が必要な場合は「YES」と答えてください。
YES か NO のみで答えてください。

質問: {query}
判断:""")
        response = self.llm.invoke(prompt.format_messages(query=query))
        return "YES" in response.content.upper()

    def is_relevant(self, query: str, doc_content: str) -> bool:
        prompt = ChatPromptTemplate.from_template("""
次のドキュメントが質問に関連しているかどうか判断してください。
RELEVANT か IRRELEVANT のみで答えてください。

質問: {query}
ドキュメント: {doc}
判断:""")
        response = self.llm.invoke(
            prompt.format_messages(query=query, doc=doc_content[:500])
        )
        return "RELEVANT" in response.content.upper()

    def generate_with_reflection(self, query: str) -> str:
        # 1. 検索が必要かどうかを判断
        need_retrieve = self.should_retrieve(query)
        print(f"検索が必要: {need_retrieve}")

        if not need_retrieve:
            response = self.llm.invoke(query)
            return response.content

        # 2. ドキュメントを検索
        docs = self.retriever.invoke(query)

        # 3. 関連性でフィルタリング
        relevant_docs = [d for d in docs if self.is_relevant(query, d.page_content)]
        print(f"関連ドキュメント: {len(relevant_docs)}/{len(docs)}")

        if not relevant_docs:
            return "関連ドキュメントが見つかりませんでした。一般的な知識で回答します: " + \
                   self.llm.invoke(query).content

        # 4. コンテキストで回答を生成
        context = "\n\n".join([d.page_content for d in relevant_docs[:3]])
        prompt = f"""コンテキストを使って質問に答えてください。
コンテキスト: {context}
質問: {query}
回答:"""
        return self.llm.invoke(prompt).content

from langchain_community.tools.tavily_search import TavilySearchResults
from typing import List, Tuple

class CorrectiveRAG:
    def __init__(self, retriever, llm):
        self.retriever = retriever
        self.llm = llm
        self.web_search = TavilySearchResults(max_results=3)

    def evaluate_documents(self, query: str, docs: list) -> Tuple[str, List]:
        """
        ドキュメントの関連性を評価します。
        戻り値: ("CORRECT"|"INCORRECT"|"AMBIGUOUS", フィルタリングされたドキュメント)
        """
        evaluation_prompt = """検索されたドキュメントが質問にどれだけ関連しているか評価してください。
- CORRECT: ドキュメントが質問に直接答えられる
- INCORRECT: ドキュメントが質問と関係ない
- AMBIGUOUS: 部分的に関連しているが不完全

質問: {query}
ドキュメント:
{docs}

評価 (CORRECT/INCORRECT/AMBIGUOUS):"""

        docs_text = "\n---\n".join([d.page_content[:300] for d in docs[:4]])
        response = self.llm.invoke(
            evaluation_prompt.format(query=query, docs=docs_text)
        )

        evaluation = response.content.strip().upper()
        if "CORRECT" in evaluation:
            return "CORRECT", docs
        elif "INCORRECT" in evaluation:
            return "INCORRECT", []
        else:
            return "AMBIGUOUS", docs

    def run(self, query: str) -> str:
        # 1. 初期検索
        docs = self.retriever.invoke(query)

        # 2. ドキュメント品質を評価
        status, filtered_docs = self.evaluate_documents(query, docs)
        print(f"ドキュメント評価: {status}")

        # 3. 各ケースを処理
        if status == "INCORRECT":
            print("ウェブ検索で補完中...")
            web_results = self.web_search.invoke(query)
            context = "\n".join([r['content'] for r in web_results])

        elif status == "AMBIGUOUS":
            web_results = self.web_search.invoke(query)
            web_context = "\n".join([r['content'] for r in web_results])
            doc_context = "\n".join([d.page_content for d in filtered_docs[:2]])
            context = doc_context + "\n\n[ウェブ検索補完]\n" + web_context

        else:
            context = "\n\n".join([d.page_content for d in filtered_docs[:4]])

        # 4. 最終回答を生成
        response = self.llm.invoke(
            f"コンテキスト:\n{context}\n\n質問: {query}\n回答:"
        )
        return response.content

class AdaptiveRAG:
    def __init__(self, simple_retriever, advanced_retriever, llm):
        self.simple_retriever = simple_retriever       # 基本的なベクトル検索
        self.advanced_retriever = advanced_retriever   # ハイブリッド + リランキング
        self.llm = llm

    def classify_query(self, query: str) -> str:
        prompt = f"""次の質問の複雑さを分類してください:
- simple: 単純な事実確認または直接答えられる質問
- complex: 複数のソースを組み合わせるか、多段階の推論が必要な質問

質問: {query}
分類 (simple/complex):"""

        response = self.llm.invoke(prompt)
        return "complex" if "complex" in response.content.lower() else "simple"

    def run(self, query: str) -> str:
        query_type = self.classify_query(query)
        print(f"クエリタイプ: {query_type}")

        if query_type == "simple":
            docs = self.simple_retriever.invoke(query)
        else:
            docs = self.advanced_retriever.invoke(query)

        context = "\n\n".join([d.page_content for d in docs])
        return self.llm.invoke(
            f"コンテキスト:\n{context}\n\n質問: {query}\n回答:"
        ).content

# GraphRAGのインストールと初期化
pip install graphrag

# プロジェクトの初期化
python -m graphrag.index --init --root ./ragtest

# 設定を編集してインデックス作成
python -m graphrag.index --root ./ragtest

# グローバル検索（全ドキュメントセットの理解が必要）
python -m graphrag.query --root ./ragtest --method global "主なテーマは何ですか？"

# ローカル検索（特定のエンティティに焦点）
python -m graphrag.query --root ./ragtest --method local "会社Xについて教えてください"

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_recall,
    context_precision
)
from datasets import Dataset

evaluation_data = {
    "question": [
        "会社の年次有給休暇ポリシーは何ですか？",
        "リモートワークポリシーは何ですか？",
    ],
    "answer": [
        "従業員は1年後に15日の有給休暇を受け取り、毎年1日追加されます。",
        "従業員は週3日リモートワークできます。",
    ],
    "contexts": [
        ["従業員は1年勤務後に15日の有給休暇が付与されます。その後毎年1日追加されます。"],
        ["従業員は週最大2日リモートワークできます。追加日数は許可を得れば承認される場合があります。"],
    ],
    "ground_truth": [
        "1年後に15日、毎年1日追加",
        "デフォルトで2日リモートワーク、承認があれば追加可能",
    ]
}

dataset = Dataset.from_dict(evaluation_data)

result = evaluate(
    dataset,
    metrics=[faithfulness, answer_relevancy, context_recall, context_precision]
)

print(result)
# faithfulness: 0.75（回答とコンテキストの不一致を検出）
# answer_relevancy: 0.92
# context_recall: 0.85
# context_precision: 0.78

import hashlib
import json
import redis

class CachedRAGSystem:
    def __init__(self, rag_chain, redis_client=None, ttl=3600):
        self.rag_chain = rag_chain
        self.redis = redis_client
        self.ttl = ttl

    def _get_cache_key(self, query: str) -> str:
        return f"rag:{hashlib.md5(query.encode()).hexdigest()}"

    def query(self, query: str) -> dict:
        cache_key = self._get_cache_key(query)

        # キャッシュを確認
        if self.redis:
            cached = self.redis.get(cache_key)
            if cached:
                print("キャッシュヒット!")
                return json.loads(cached)

        # RAGを実行
        result = self.rag_chain.invoke({"query": query})
        response = {
            "answer": result["result"],
            "sources": [d.metadata for d in result["source_documents"]]
        }

        # キャッシュに保存
        if self.redis:
            self.redis.setex(cache_key, self.ttl, json.dumps(response))

        return response

from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
from langchain_openai import ChatOpenAI
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

streaming_llm = ChatOpenAI(
    model="gpt-4o-mini",
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()]
)

async def generate_rag_stream(query: str):
    docs = retriever.invoke(query)
    context = "\n\n".join([d.page_content for d in docs])

    async for chunk in streaming_llm.astream(
        f"コンテキスト:\n{context}\n\n質問: {query}\n回答:"
    ):
        if chunk.content:
            yield f"data: {chunk.content}\n\n"

@app.get("/rag/stream")
async def rag_stream_endpoint(query: str):
    return StreamingResponse(
        generate_rag_stream(query),
        media_type="text/event-stream"
    )