RAG ワークフローの概要

本チュートリアルでは、LangChain を利用して検索拡張生成（RAG）システムを構築します。このプロセスは主に以下の 2 つのフェーズで構成されます。

• インデキシング: ウェブソースから情報を抽出し、ベクトル化处理后にデータベースへ格納。
• RAG 実行: ユーザーのクエリに対し、まずベクトルストアから関連情報を検索（Retrieve）、その結果を文脈として追加（Augmented）し、最終的に LLM が回答を生成（Generate）します。

1. ウェブコンテンツの取得

ネットワーク上の非構造化データを収集するため、Python の BeautifulSoup ライブラリを活用します。これは HTML や XML パースにおいて柔軟なデータ抽出を可能にします。

# 依存パッケージのインストール例
pip install beautifulsoup4

WebBaseLoader を使用してローディング処理をカプセル化します。

import bs4
from langchain_community.document_loaders import WebBaseLoader

def fetch_source_data(target_url: str):
    # 不要なタグを除去するためのフィルタ設定
    filter_strainer = bs4.SoupStrainer()
    
    loader = WebBaseLoader(
        web_paths=(target_url,),
        bs_kwargs={
            "parse_only": filter_strainer
        }
    )
    return loader.load()

# テスト実行
if __name__ == "__main__":
    sample_url = "https://news.cctv.com/2026/01/04/example_article.shtml"
    documents = fetch_source_data(sample_url)
    
    print(f"ロードされたドキュメント数: {len(documents)}")
    if documents:
        meta_info = documents[0].metadata.get('title', 'Untitled')
        preview_text = documents[0].page_content[:100]
        print(f"タイトル: {meta_info}")
        print(f"プレビューテキスト: {preview_text}...")

2. テキストのチャンキング（分割）

取得した全文をそのまま埋め込むのではなく、意味のまとまりに基づいて適切なサイズに分割する必要があります。RecursiveCharacterTextSplitter を使用することで、階層的に区切ることが可能です。

from langchain_text_splitters import RecursiveCharacterTextSplitter

def chunk_text_content(doc_list):
    """
    ドキュメントリストを再帰的に分割する関数
    """
    # セパレーター順で分割を試み、オーバーラップを設定
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=128,
        separators=["\n\n", "\n", ".", " ", ""]
    )
    
    segments = splitter.split_documents(doc_list)
    print(f"生成されたセグメント数: {len(segments)}")
    return segments

3. ベクトルストレージの設定

今回はローカル環境で動作する Ollama サーバー内の bge-m3 モデルを埋め込みエンジンとして採用します。これにより、チャットのプライバシーやコストを抑えつつベクトル処理を行います。

# Ollama 側でのモデル準備
# ollama pull bge-m3:567m

ChromaDB を用いた永続的インデックスの作成処理です。

from langchain_ollama import OllamaEmbeddings
from langchain_chroma import Chroma

def setup_vector_index():
    # ローカル Ollama インスタンスへの接続定義
    embedder = OllamaEmbeddings(
        model="bge-m3:567m",
        base_url="http://192.168.6.133:11434"
    )
    
    db_instance = Chroma(
        collection_name="news_data_index",
        embedding_function=embedder,
        persist_directory="./db_storage_chroma"
    )
    return db_instance

分割済みのデータを自動的に埋め込み処理して保存します。

if __name__ == "__main__":
    target_link = "https://news.cctv.com/2026/01/04/example_article.shtml"
    raw_docs = fetch_source_data(target_link)
    chunks = chunk_text_content(raw_docs)
    
    index_store = setup_vector_index()
    index_store.add_documents(chunks)
    print("インデックス登録が完了しました。")

4. 検索機能のツール化

エージェントが外部機能を呼び出すためには、検索ロジックを「ツール」として定義する必要があります。ここではメタデータを保持しつつ、アージェントに提示可能な形式で返却するラッパーを作成します。

from langchain.tools import tool

def define_search_function(store_ref):
    @tool(response_format="content_and_artifact")
    def search_context(user_query: str):
        """与えられた質問に関連する背景情報を検索・取得します。"""
        matches = store_ref.similarity_search(user_query, k=2)
        
        formatted_output = "\n\n".join([
            f"出典：{doc.metadata}\n内容：{doc.page_content}" 
            for doc in matches
        ])
        return formatted_output, matches
        
    return search_context

このように関数内で store_ref を閉じ込めることで、各検索時に行先の参照を共有できます。

5. RAG エージェントの編成

最後に、上述した検索ツールとチャットモデルを組み合わせてエージェントを初期化し、実際の問いかけを行いうテストを実行します。

if __name__ == "__main__":
    # コンポーネントの準備
    vector_db = setup_vector_index()
    search_tool = define_search_function(vector_db)
    
    # LLM 接続設定（Ollama 上での実行想定）
    chat_engine = init_chat_model(
        model="ollama:qwen3-next:80b-cloud",
        base_url="http://127.0.0.1:11434"
    )
    
    system_instruction = (
        "あなたはブログ記事の内容を照会できるツールを持っています。\n"
        "ユーザーの質問には、必ずツールを用いて確認された事実に基づいて答えてください。"
    )
    
    # エージェント定義
    agent_runner = create_agent(
        model=chat_engine,
        system_prompt=system_instruction,
        tools=[search_tool]
    )
    
    # クエリ実行
    query_topic = "『固体废物综合治理行动计划』の発布日と発行元について教えて。"
    
    response_stream = agent_runner.stream(
        {"messages": [{"role": "user", "content": query_topic}]},
        stream_mode="values",
    )
    
    # ストリーム出力の監視
    for iteration in response_stream:
        final_msg = iteration["messages"][-1]
        final_msg.pretty_print()

上記の実行結果では、エージェントが内部で retrieve_context ツールをトリガーし、ベクトルデータベースから該当情報を抽出した後、それらを統合して自然言語の回答を生成していることが確認できます。これにより、単なる予測ではなく、根拠に基づいた回答が可能となります。