Posted inAI

Haystack 2.0によるモジュール式AIパイプラインの構築：本番向けRAGにおけるLangChainの明快な代替案

AI tutorial - IT technology blog
AI tutorial - IT technology blog

アプローチの比較：明示的なDAG vs. 抽象化されたマジックフレームワーク

昨年の初め、私たちのチームは約250万件の社内ドキュメントをインデックス化するマルチテナント型のエンタープライズ検索サービスを引き継ぎました。検索処理が滞ったり、p99レイテンシが4,000msを超えて急上昇したりするたびに、原因の特定は困難を極めました。トレースを追うには、深くネストされたラッパークラスや不透明なコールバック、そして内部で暗黙的に受け渡されるコンテキスト辞書を掘り下げる必要がありました。プロンプトの調整や精度の向上に時間を割くべきところを、フレームワーク内部の仕様との格闘に何時間も浪費してしまっていたのです。

こうした摩擦をきっかけに、私たちはdeepset社がオーケストレーションエンジンをゼロから再設計したHaystack 2.0の検証を始めました。ステップ間でデータがどのように受け渡されるかというアーキテクチャが根本から異なっています。

  • モノリシックなチェーンアプローチ（例：初期のLangChain）： 深い継承階層と暗黙的な共有ステートに基づいて構築されています。各ステップは緩い辞書ペイロードを受け渡し、キーは明確な契約なしに出現したり消失したりします。素早いプロトタイピングには適していますが、本番環境での動的分岐のデバッグや型エラーの追跡は急速に泥沼化します。
  • 明示的なDAGアプローチ（Haystack 2.0）： Retriever、プロンプトテンプレート、LLMクライアントを、有向非巡回グラフ（DAG）内の独立したノードとしてモデル化します。すべてのノードは標準のPython型ヒントを使用して入力と出力の型を厳密に宣言します。ソケットの接続は、pipeline.connect("node_a.output_key", "node_b.input_key")で直接行います。型が一致しない場合はビルド時にグラフの構築が失敗するため、不正なペイロードが本番環境に混入するのを未然に防ぐことができます。

メリットとデメリット：実システムにおけるHaystack 2.0の評価

暗黙的なチェーンから明示的なグラフへ移行することは、エンジニアリング面で明確な利点をもたらす一方で、相応のトレードオフも伴います。

メリット

  • 予測可能なデータフロー： ソケットの接続が1行ずつ明示的に宣言されるため、各ノードが何を消費し、何を出力するかが常に一目で分かります。隠れた辞書キーによって下流のプロンプト入力が意図せず上書きされる心配はありません。
  • シンプルなカスタムコンポーネント作成： ドメイン固有のフィルターやトークン削減処理を記述する場合も、@componentデコレータと型ヒント付きのrun()メソッドを持つ標準クラスを作成するだけです。深くネストされた基底クラスを継承する必要は一切ありません。
  • ネイティブなシリアライズと可視化： パイプラインは標準機能でYAML形式へのエクスポートやMermaid図の生成が可能です。リクエストを処理する前に、CI/CDチェックでグラフトポロジーを検証できます。
  • 軽量なフットプリント： コアとなるhaystack-aiパッケージは非常にスリムです。QdrantやMilvusなどのベクトルデータベース用のサードパーティ連携は、独立したオプションパッケージとして提供されています。

デメリット

  • 初期セットアップコードの増加： 6ノードのパイプラインを配線するには、5つの明示的なpipeline.connect()文が必要です。ワンライナーで書ける手軽なプロトタイピングを犠牲にして、堅牢で検証可能なアーキテクチャを手に入れるトレードオフとなります。
  • エコシステムの適用範囲： 主要プロバイダー（OpenAI、HuggingFace、Cohere、Pineconeなど）は標準でスムーズに動作します。しかし、登場したばかりのコミュニティ製ツールの場合は既製のコネクタがないことも多く、15行程度のカスタムラッパーコードを自前で実装する必要があります。

推奨アーキテクチャ：インジェスチョンと検索の分離

インデックス作成とランタイムでのクエリ処理を1つのスクリプトにまとめたモノリシックなパイプラインは避けるべきです。これらを2つの専用グラフに分離することで、障害の影響範囲を限定し、リソース消費を抑えることができます。

  1. インジェスチョンパイプライン： 生ファイルを解析し、テキストを300語のチャンクに分割し、バッチ単位で埋め込みを生成して、ストレージにベクトルを書き込みます。バックグラウンドの非同期ワーカーとして実行します。
  2. クエリパイプライン： ランタイムでユーザーのクエリをベクトルに変換し、top-kの候補を取得して、プロンプトテンプレートを展開し、LLMを呼び出します。低レイテンシが求められるAPIパス内で実行します。

Document Storeを共有インフラストラクチャとして扱い、インジェスチョンとサービングのライフサイクルを分離することで、安定したレイテンシを維持できます。バックグラウンドでバッチ単位の再インデックス処理を実行しても、受信中のユーザー検索がブロックされることはありません。

Python 3.10以上の仮想環境をセットアップし、必要な依存関係をインストールします。

# 仮想環境を作成して有効化
python3 -m venv .venv
source .venv/bin/activate

# Haystack 2.x と軽量なローカル埋め込みプロバイダーをインストール
pip install --upgrade pip
pip install haystack-ai fastembed-haystack
export OPENAI_API_KEY="sk-your-openai-api-key"

実装ガイド：モジュール式RAGとカスタム処理

ローカル埋め込みと明示的なグラフ配線を用いた、エンドツーエンドのRAG構成例を紹介します。

ステップ1：ドキュメントインジェスチョンパイプラインの構築

InMemoryDocumentStoreを設定し、ドキュメントを100語のチャンクに分割して、FastEmbed（bge-small-en-v1.5）でローカルに埋め込みを計算し、メモリ上にベクトルを書き込みます。

from haystack import Pipeline, Document
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.components.preprocessors import DocumentSplitter
from haystack.components.writers import DocumentWriter
from haystack_integrations.components.embedders.fastembed import FastembedDocumentEmbedder

# 1. 共有Document Storeを初期化
document_store = InMemoryDocumentStore(embedding_similarity_function="cosine")

# 2. 各コンポーネントをインスタンス化
splitter = DocumentSplitter(split_by="word", split_length=100, split_overlap=20)
embedder = FastembedDocumentEmbedder(model="BAAI/bge-small-en-v1.5")
writer = DocumentWriter(document_store=document_store)

# 3. DAGを構築
ingestion_pipeline = Pipeline()
ingestion_pipeline.add_component("splitter", splitter)
ingestion_pipeline.add_component("embedder", embedder)
ingestion_pipeline.add_component("writer", writer)

# 4. 明示的にソケットを接続（node.output -> node.input）
ingestion_pipeline.connect("splitter.documents", "embedder.documents")
ingestion_pipeline.connect("embedder.documents", "writer.documents")

# サンプルデータでインデックス作成を実行
raw_docs = [
    Document(content="Haystack 2.0はコンポーネントの明示的な接続と決定論的パイプラインのために有向グラフを採用しています。"),
    Document(content="モジュール式RAGアーキテクチャは信頼性向上のため、インジェスチョンと実行時クエリパイプラインを分離します。"),
    Document(content="Haystackのカスタムコンポーネントに必要なのは、@componentデコレータと型付きのrunパラメータのみです。")
]

ingestion_pipeline.run({"splitter": {"documents": raw_docs}})
print(f"{document_store.count_documents()} 件のドキュメントチャンクのインデックス作成が正常に完了しました。")

ステップ2：クエリおよび生成パイプラインの設計

クエリパイプラインは、受信したユーザーリクエストに応じて動作します。クエリを埋め込みベクトルに変換し、関連する上位ドキュメントを検索し、プロンプトを構築してOpenAIに送信します。

from haystack.components.builders import PromptBuilder
from haystack.components.generators import OpenAIGenerator
from haystack_integrations.components.embedders.fastembed import FastembedTextEmbedder
from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever

# 明示的なフォールバックルールを備えた厳密なテンプレートを定義
prompt_template = """
以下のコンテキストドキュメントをもとに、質問に対して正確に回答してください。
もしコンテキスト内に回答が見つからない場合は、「コンテキスト内に該当する情報がありません。」と回答してください。

コンテキスト:
{% for doc in documents %}
- {{ doc.content }}
{% endfor %}

質問: {{ query }}
回答:
"""

# クエリコンポーネントのインスタンス化
text_embedder = FastembedTextEmbedder(model="BAAI/bge-small-en-v1.5")
retriever = InMemoryEmbeddingRetriever(document_store=document_store, top_k=2)
prompt_builder = PromptBuilder(template=prompt_template)
generator = OpenAIGenerator(model="gpt-4o-mini")

# クエリDAGの構築
query_pipeline = Pipeline()
query_pipeline.add_component("text_embedder", text_embedder)
query_pipeline.add_component("retriever", retriever)
query_pipeline.add_component("prompt_builder", prompt_builder)
query_pipeline.add_component("generator", generator)

# コンポーネントを明示的に接続
query_pipeline.connect("text_embedder.embedding", "retriever.query_embedding")
query_pipeline.connect("retriever.documents", "prompt_builder.documents")
query_pipeline.connect("prompt_builder.prompt", "generator.prompt")

# クエリを実行
user_query = "Haystack 2.0はどのようにコンポーネントを接続しますか？"
response = query_pipeline.run({
    "text_embedder": {"text": user_query},
    "prompt_builder": {"query": user_query}
})

print("LLMの出力:")
print(response["generator"]["replies"][0])

ステップ3：安全性とルーティングのためのカスタムコンポーネント作成

カスタムノードの柔軟性こそが、本番環境でHaystackが真価を発揮するポイントです。検索スコアが許容しきい値を下回った場合にフォールバック処理へと直接ルーティングすることで、不要なLLMトークン消費とハルシネーションを防ぐことができます。

from typing import List
from haystack import component
from haystack.dataclasses import Document

@component
class RelevanceFilter:
    def __init__(self, min_score: float = 0.65):
        self.min_score = min_score

    @component.output_types(passed_docs=List[Document], fallback=bool)
    def run(self, documents: List[Document]):
        # モデルにプロンプトを渡す前に、信頼度の低いチャンクを除外
        filtered = [doc for doc in documents if (doc.score or 0.0) >= self.min_score]
        if not filtered:
            return {"passed_docs": [], "fallback": True}
        return {"passed_docs": filtered, "fallback": False}

RetrieverとPromptBuilderの間にRelevanceFilterを挟むことで、無関係なコンテキストがプロンプトを汚染するのを防ぐことができます。私たちのワークロードでは、このシンプルなチェックによって無駄なトークン使用量を18%削減できました。

本番運用の実践Tips

  • CIでグラフ構造を検証： query_pipeline.dumps()でパイプラインのスキーマをエクスポートするか、query_pipeline.draw(path="graph.png")でアーキテクチャ図を出力し、プルリクエスト時に接続の変更内容をレビューできるようにします。
  • 共有入力の曖昧さを解消： 2つの下流コンポーネントが同じ上流データを必要とする場合は、明示的なソケットマッピングキーを指定してルーティングの混同を防ぎます。
  • ステートレス性の維持： パイプラインインスタンスはステートレスな実行機構として扱ってください。カスタムコンポーネントが実行中に内部インスタンス属性を変更しない限り、並行ワーカースレッド間でrun()を安全に呼び出すことができます。
Share:
Tags: