【LangChain】CustomRetriever実装：パッケージ製品が自社業務に合わない時の「魔改造」術
はじめに：「ウチの会社は特殊だから」への処方箋
業務システムの導入プロジェクトで、現場から必ず出る言葉。「ウチの業務フローは特殊なんで、パッケージそのままじゃ使えません」。
RAG開発でも同じです。標準のベクトル検索だけでは、「最新の日報だけ検索したい」「部長以上しか見られない極秘文書を除外したい」といったドロドロした要件に対応できません。
そんな時こそ、LangChainの CustomRetriever の出番です。既存の仕組みに満足せず、自社のルールに合わせて検索ロジックを魔改造する。これぞ社内SEの腕の見せ所です。
基礎知識：Retrieverはただの「検索関数」である
難しく考える必要はありません。LangChainにおけるRetrieverとは、「クエリ文字列(str)を受け取り、関連ドキュメントのリスト(List[Document])を返す」ただのクラスです。
つまり、この入力と出力の規格さえ守れば、中身で何をしようが自由なんです。SQLを叩こうが、社内APIを呼ぼうが、grepしようが自由自在です。
実装・設定：BaseRetrieverを継承して俺俺ロジックを書く
実装は非常にシンプルです。_get_relevant_documents メソッドをオーバーライドするだけ。
from langchain_core.retrievers import BaseRetriever
class MyCompanyRetriever(BaseRetriever):
def _get_relevant_documents(self, query: str, *, run_manager, **kwargs):
# ここに独自の検索ロジックを書く
# 例：社内APIを叩く
results = internal_api.search(query, user_dept="sales")
# Document型に変換して返す
return [Document(page_content=r["text"]) for r in results]たったこれだけで、既存のLangChainエコシステム（RAGチェーンなど）にそのまま組み込めます。標準インターフェース準拠の強みですね。
応用テクニック：日付フィルターや権限チェックを挟む
実務でよくあるのが、「古い情報は出すな」という要件です。
CustomRetrieverなら、クエリ実行前に本日日付を取得し、「直近1年以内」というフィルタリング条件を動的に付与することが可能です。検索結果を取得した後で、Python側で並び替えたり間引いたりするのも容易です。
トラブルシューティング：非同期メソッド(ainvoke)の実装忘れ
Webアプリなどで使う場合、_aget_relevant_documents（非同期版）も実装しておかないと、メインの処理をブロックしてしまうことがあります。
同期処理しか書かないのは、片手落ち。「動くけど遅い」システムを作らないよう注意しましょう。
まとめ：痒い所に手が届くシステムこそが定着する
標準機能だけで作れるシステムは楽ですが、ユーザーの満足度は「あと一歩」になりがちです。
CustomRetrieverでラストワンマイルの要件（社内ルール、特殊なフィルター）を埋めてあげることで、初めて「これは使える！」と現場に受け入れられるツールになります。
コメント
コメントを投稿