第03回学習メモ: LangChainによるRAG実装（LCEL版）¶

全体像¶

第02回の「ChromaDB を直叩きする RAG」を、LangChain の組み立て部品で書き直したもの。連載原典は LangChain 0.0.x の RetrievalQAWithSourcesChain（クラス引数の組み合わせで作るブラックボックス）を使っていたが、本フォルダでは現行 LangChain (0.3+) の LCEL（LangChain Expression Language）で書き直してある。| 演算子で部品をパイプ接続する Unix シェル風の書き方。

初期化フェーズ（setup_db.py、初回のみ）¶

flowchart LR
    Upstream["upstream GitHub<br/>events_2023.json"]
    Fetch["fetch_events<br/>HTTP GET"]
    Format["format_events<br/>### 2023年X月Y日 形式に整形<br/>metadatas に source 付与"]
    Embed["OpenAIEmbeddingFunction<br/>(text-embedding-3-small)"]
    Chroma["ChromaDB<br/>./data, collection=events_2023"]

    Upstream --> Fetch --> Format --> Embed --> Chroma

実行フェーズ（chatbot.py の LCEL チェイン）¶

dict が assigns を通るたびにキーが増えていく構造（LCEL の典型パターン）:

flowchart TD
    Input["入力 dict<br/>{question, chat_history}"]

    Input --> A1["RunnablePassthrough.assign<br/>(sources = itemgetter('question') | retriever)"]
    A1 --> D1["{question, chat_history, sources}"]

    D1 --> A2["RunnablePassthrough.assign(answer = answer_chain)"]

    subgraph answer_chain["answer_chain (LCEL)"]
        direction TB
        S1["RunnablePassthrough.assign<br/>(context = format_docs(sources))"]
        S2["ChatPromptTemplate<br/>system + {chat_history}<br/>+ user + {context}"]
        S3["ChatOpenAI(gpt-4o-mini)"]
        S4["StrOutputParser"]
        S1 --> S2 --> S3 --> S4
    end

    A2 --> Out["{question, chat_history, sources, answer}"]

    Out --> UI[Chainlit に表示]
    UI --> Step["cl.Step で sources を<br/>折りたたみ表示"]
    UI --> Msg["cl.Message で answer を<br/>通常メッセージ表示"]

要点: - assign(key=runnable) を重ねることで dict のキーが {question} → {…, sources} → {…, sources, answer} と段階的に成長する - answer_chain の中でも同じ手法を使い、{…, sources} から {…, sources, context} を作って prompt に渡す - MessagesPlaceholder("chat_history") がプロンプト内に「過去のメッセージリスト」の置き場所を作る（連載原典の ConversationBufferMemory の暗黙的な挿入と違って、置き場所がコードに見える）

第02回との対応関係¶

02 で自分で書いていた処理	03 で LangChain に任せた部分
`cl.user_session.get('history')` を手で append	`MessagesPlaceholder("chat_history")` でプロンプトに展開
`collection.query(query_texts=...)` を手で呼ぶ	`Chroma(...).as_retriever()`
distance しきい値で関連情報を選別	retriever 内部（デフォルト k=4 で上位採用）
関連情報を system message に手で挿入	`{context}` プレースホルダ + `format_docs` で連結
`response.choices[0].message.content` を手で取り出す	`StrOutputParser()`

「自前の組み立てを LangChain のチェイン構成に分解する」のが第03回の主題。LCEL ではその「分解」がコードに直接現れる。

使用ライブラリ・原理¶

LangChain の3つの抽象化¶

LangChain は LLM アプリを 3階層の部品に分けて抽象化する。03 のコードはこの3階層が全部登場する:

階層	このコードでの登場	役割
Model I/O	`ChatOpenAI`, `OpenAIEmbeddings`, `ChatPromptTemplate`, `StrOutputParser`	モデル呼び出しとプロンプト・出力の抽象化
Retrieval	`Chroma`, `as_retriever()`	ベクトルストア → リトリーバへの変換
Composition (LCEL)	`RunnablePassthrough.assign(...)`, `\\|` 演算子	上記を組み合わせて1本の `Runnable` に

PyTorch で言うと、Model I/O が nn.Linear、Retrieval が Dataset、Composition が nn.Sequential に近い役割分担。

LCEL（LangChain Expression Language）¶

「Runnable という共通プロトコルを実装したオブジェクトを | で繋ぐ」記法。Unix の cat file | grep | wc と同じ発想。すべての Runnable は以下のメソッドを持つ:

.invoke(input) / .ainvoke(input) — 1回実行（同期 / 非同期）
.stream(input) / .astream(input) — ストリーミング
.batch(inputs) / .abatch(inputs) — バッチ実行

このプロトコルが共通だから、prompt | llm | parser のように繋げられる。連載原典の RetrievalQAWithSourcesChain.from_chain_type(...) がブラックボックスだったのに対し、LCEL では各段がコードに直接現れるので、デバッグも差し替えも容易。

LCEL の主要部品¶

部品	役割
`RunnablePassthrough()`	入力をそのまま出力に流す（恒等関数）
`RunnablePassthrough.assign(key=runnable)`	入力（dict）にキーを追加して流す
`RunnableLambda(func)`	任意の Python 関数を Runnable 化
`itemgetter("key")`	dict から特定キーを取り出す（標準ライブラリ）
`StrOutputParser()`	LLM の応答 (`AIMessage`) から `.content` を取り出す
`MessagesPlaceholder("name")`	プロンプト内に「過去の会話メッセージリスト」を埋め込む特殊スロット

本コードのチェイン構成¶

retrieve_docs = itemgetter("question") | retriever

answer_chain = (
    RunnablePassthrough.assign(context=lambda x: format_docs(x["sources"]))
    | prompt
    | llm
    | StrOutputParser()
)

chain = (
    RunnablePassthrough.assign(sources=retrieve_docs)
    | RunnablePassthrough.assign(answer=answer_chain)
)

入力 {"question": str, "chat_history": list[BaseMessage]} が、各段でdict のキーが増えていく形で流れる:

1段目 assign(sources=retrieve_docs): question を retriever に通して sources キーを追加 → {question, chat_history, sources}
2段目 assign(answer=answer_chain): その dict を answer_chain に通して answer キーを追加 → {question, chat_history, sources, answer}

answer_chain の中では:

assign(context=...) で sources を format_docs で連結した文字列を context キーに追加
prompt が {context} {question} {chat_history} を埋め込んで ChatPromptValue を作る
llm がそれに応答
parser が応答から文字列を抜く

「dict が層を重ねて成長していく」のが LCEL の典型パターン。LangChain の create_retrieval_chain も内部はこの形をしている。

`MessagesPlaceholder("chat_history")`¶

プロンプト内に 「過去の会話メッセージリストをそのまま展開する」専用スロット。

prompt = ChatPromptTemplate.from_messages([
    ("system", SYSTEM_MESSAGE),
    MessagesPlaceholder("chat_history"),    # ← ここに過去メッセージが順に展開
    ("human", "{question}"),
])

連載原典の ConversationBufferMemory が「裏で chain にフックして履歴を埋める」抽象だったのに対し、LCEL ではプロンプトテンプレート上に履歴の置き場所が明示される。教育的にはこちらの方がはるかに分かりやすい。

履歴は HumanMessage / AIMessage のリストとして渡す（このコードでは cl.user_session に保持）。

`as_retriever()`¶

ベクトルストア（Chroma 等）は「中身を入れる箱」だが、検索API は Retriever という抽象に揃えてある。docsearch.as_retriever() でストア → リトリーバを変換することで、後段のチェインが「中身がChromaかFAISSかPineconeか」を気にしなくて済む。

デフォルトでは k=4（上位4件）のヒットを返す。第02回で手書きしていた「distance ≤ 0.4 だけ採用」のようなしきい値フィルタはデフォルトでは入っていない点に注意。as_retriever(search_type="similarity_score_threshold", search_kwargs={"score_threshold": 0.7}) のように指定する。

Chainlit との接続¶

@cl.on_chat_start でチェインを組んで cl.user_session.set("chain", chain) に保存、@cl.on_message で取り出して chain.ainvoke({"question": ..., "chat_history": ...}) を呼ぶ流れ。第02回の「履歴管理を Chainlit セッションに置く」発想と同じだが、置くものが list ではなく Runnable オブジェクトになっている。

履歴自体は cl.user_session の別キー (chat_history) に LangChain Message オブジェクトのリストとして保持し、毎ターン chain にハンドオフする。

ファイル別の役割¶

ファイル	役割
`setup_db.py`	upstream GitHub から `events_2023.json` をHTTP GET → `### 2023年X月Y日` 形式に整形 → ChromaDB に投入。第02回で欠けていた初期化スクリプトのLangChain版（埋め込みモデルは 3-small に更新済み）
`chatbot.py`	LangChain LCEL で構築する RAG チェイン本体。02 の手書きRAGを「`\|` でパイプ接続する Runnable」に置き換え
`data/`	`setup_db.py` の出力。ChromaDB の SQLite + parquet 永続化先（.gitignore 対象）

データソースは upstream（mahm/softwaredesign-llm-application の本家）の 02/events_2023.json を毎回ネットから取りに行く設計。これは「02のローカルファイルに依存させない」工夫だが、ネット接続が必要な点に注意。

行レベルの工夫¶

`setup_db.py:62` — 文書フォーマットに月日ヘッダを埋め込む¶

formatted_events.append(f"### 2023年{date}\n{event}")

第02回では生の記事文字列をそのまま投入していたが、03 では ### 2023年1月2日 の見出しを先頭に付けてから投入している。これは:

ベクトル検索のヒット精度が上がる（「2023年4月」みたいなクエリで日付が一致しやすい）
LLM への投入時に文書境界が明示される（複数文書を stuff 連結したときに見出しが区切り役になる）

地味だが RAG では効く工夫。

`setup_db.py:63` — メタデータ付与¶

metadatas.append({"source": f"2023年{date}"})

source というキーは LangChain の慣例的予約名で、連載原典の RetrievalQAWithSourcesChain が SOURCES: 行を自動生成するのに使っていた。本フォルダの LCEL 版ではそのチェインを使っていないので自動的な出典付与は無く、Document.metadata["source"] として参照できるだけになっている（cl.Step で表示するときの参考情報）。

`chatbot.py:50` — `{{context}}` のエスケープ¶

SYSTEM_MESSAGE = f"""...
----------------
関連情報:
{{context}}"""

f-string で {{ と書いているのは「f-string の波括弧」をエスケープして、リテラルの {context} を生成するため。これが LangChain の ChatPromptTemplate 側でプレースホルダとして解釈される。f-string と LangChain のプロンプト変数が2重に絡む箇所で、初見でハマりやすい（原典では {summaries} だったキーを、LCEL 版では {context} に変更している）。

`chatbot.py:81` — `itemgetter` でキーを取り出す¶

retrieve_docs = itemgetter("question") | retriever

itemgetter("question") は標準ライブラリ operator モジュールの関数で、「dict から "question" キーの値を取り出す Callable」を返す。LCEL は Callable も自動で Runnable に昇格するので、これをそのまま | で繋げる。lambda x: x["question"] でも同じだが、itemgetter の方がLangChain 公式チュートリアルでも頻出。

`chatbot.py:91-96` — assign で dict を成長させる¶

chain = RunnablePassthrough.assign(sources=retrieve_docs) | RunnablePassthrough.assign(
    answer=answer_chain,
)

assign(key=runnable) は「入力 dict はそのまま流しつつ、key というキーに runnable の出力を追加する」操作。これを連結することで、チェインを通過するたびに dict のキーが増えていく。最終的に {question, chat_history, sources, answer} という形で出てくる。

LCEL の典型パターンで、sources（検索結果）と answer（回答）の両方を返したいときの定石。連載原典で return_source_documents=True フラグでやっていたことを、LCEL では明示的な assign で実現している。

学んだこと（要点）¶

第02回 → 第03回の本質的な差: コードのロジックは変わらず、実装を自前から LangChain の部品組み立てに置き換えただけ。RAG の概念モデル（履歴 + 検索 + 注入 + 生成）を、抽象化された部品で表現する練習
LCEL は Unix パイプの発想: prompt | llm | parser は cat | grep | wc と同じ。各部品が Runnable という共通プロトコル（.invoke / .ainvoke / .stream）を実装しているから繋げられる
dict を成長させていく構成法: RunnablePassthrough.assign(key=runnable) を重ねることで {question} → {question, sources} → {question, sources, answer} と成長していく。最終結果と中間結果の両方が取り出せる
MessagesPlaceholder("chat_history"): 連載原典の ConversationBufferMemory のように暗黙的にプロンプトに差し込むのではなく、プロンプト内に履歴の置き場所を明示できる。教育的に分かりやすい
{context} の二重エスケープ: f-string で {{context}} と書くのは、f-string の波括弧をエスケープして LangChain プロンプト用のプレースホルダ {context} を残すため。f-string と LangChain プロンプトの2層が絡む典型的ハマりどころ
itemgetter("question") は Callable → Runnable に自動昇格: LCEL は Python の Callable を自動で Runnable に変換するので、operator.itemgetter や lambda を | で繋げる
Retrieverのデフォルト: しきい値フィルタなし、k=4 が初期値。02 の distance ≤ 0.4 のような厳しさはないので、ノイズが入りやすい
チェインを cl.user_session.set("chain", chain) に置く: 02 では list を置いていたが、03 では Runnable オブジェクトを置く。セッションは状態ではなく動作を持つようになる
Embedding API は意味の地図への変換器: テキストを 1536 次元ベクトルにマッピングし、コサイン類似度で「意味の近さ」を計算できるようにする。RAG の検索が「キーワード一致」を超えて働く根拠

拡張アイデア¶

as_retriever(search_type="similarity_score_threshold", search_kwargs={"score_threshold": 0.7}) で 02 と同様のしきい値フィルタを入れる
.stream() / .astream() で逐次表示: LCEL は標準でストリーミング対応。Chainlit の cl.Message().stream_token(...) と組み合わせると、回答が文字単位で出てくる体験になる
RunnableWithMessageHistory: 現在は cl.user_session で履歴管理しているが、LangChain 純正の履歴ラッパーに変えると Chainlit に依存しない再利用可能なチェインになる
MultiQueryRetriever / EnsembleRetriever: 単純な類似度検索の精度を上げる発展型。クエリを複数生成して並列検索する手法と、複数 retriever を組み合わせる手法
map_reduce 相当の長文対応: 検索ヒットが多すぎてコンテキストに収まらないとき、load_summarize_chain や手書きの map-reduce LCEL チェインで段階的に要約しながら回答する
第27回との比較: DSPy + GEPA で同じ RAG を自動最適化するとどれだけ性能が上がるか

連載原典からの移植で行った変更¶

旧 LangChain 0.0.x の RetrievalQAWithSourcesChain ベースから、現行 LangChain (0.3+) の LCEL 構成 に書き換えた。

箇所	原典	本フォルダ
import パス	`langchain.embeddings.openai` / `langchain.chat_models` / `langchain.vectorstores`	`langchain_openai` / `langchain_chroma` / `langchain_core`
チェイン構築	`RetrievalQAWithSourcesChain.from_chain_type(...)`	`RunnablePassthrough.assign(...) \\| ... \\| prompt \\| llm \\| StrOutputParser()` の LCEL
会話履歴	`ConversationBufferMemory(memory_key="chat_history", ...)`	`cl.user_session` で管理 + `MessagesPlaceholder("chat_history")` でプロンプトに展開
プレースホルダ名	`{summaries}`（チェインが暗黙に埋める）	`{context}`（`format_docs` で自前整形して埋める）
呼び出し	`chain.acall(message)`	`chain.ainvoke({"question": ..., "chat_history": ...})`
Embedding モデル	`text-embedding-ada-002`	`text-embedding-3-small`（コスト1/5・性能同等以上）
LLM	`ChatOpenAI(temperature=0.0)`（デフォルト gpt-3.5）	`ChatOpenAI(model="gpt-4o-mini", temperature=0.0)`
関連情報の表示	`cl.Message(author="relevant", indent=1)`	`cl.Step(type="retrieval")`（`indent` 引数は現行 Chainlit で削除）
APIキー取得	環境変数 / 直書き想定	1Password CLI 経由（01・02 と統一）
初期化スクリプト	あり（`setup_db.py`、ただし旧 `OpenAIEmbeddingFunction` 形式）	更新（`api_key` 必須、3-small、絶対パス、進捗表示）
`@cl.on_message` の引数型	`message: str`	`message: cl.Message`（`message.content` で本文取得）

LCEL に移行して見えるようになったこと¶

連載原典の RetrievalQAWithSourcesChain.from_chain_type(...) では「chain_type="stuff" だと裏で何が起きているか」がブラックボックスだったが、LCEL では各段がコードに直接現れる:

chain = (
    RunnablePassthrough.assign(sources=retrieve_docs)       # ① 検索
    | RunnablePassthrough.assign(answer=(
        RunnablePassthrough.assign(context=format_docs)     # ② 結果を文字列化
        | prompt                                            # ③ プロンプトに埋め込み
        | llm                                               # ④ LLM 呼び出し
        | StrOutputParser()                                 # ⑤ 文字列抜き出し
    ))
)

「stuff chain_type」という名前で覆われていた手順 ②〜⑤ が、コードの行として並ぶ。差し替えやデバッグが容易になる。

既知の不具合・注意点¶

setup_db.py がネット接続必須: upstream GitHub から JSON を毎回 HTTP GET するので、オフラインでは動かない。ローカル 02/events_2023.json を使う形に書き換える手もある
チェインのエラーハンドリングが薄い: 検索 0 件のときも空文字を {context} に流すだけ。SYSTEM_MESSAGE 側で「関連情報が空ならその旨を伝える」指示はあるが、retriever 自体に空のときの分岐ロジックはない
chat_history のトークン量上限なし: cl.user_session に履歴を貯め続けるので、長時間会話で gpt-4o-mini のコンテキスト上限 (128k) に当たる可能性。実用化するなら ConversationBufferWindowMemory 相当のロジック（直近 N ターンのみ保持）を入れる
source メタデータの活用が薄い: setup_db.py で {"source": "2023年X月Y日"} を付けているが、本フォルダの LCEL 版では cl.Step に文書本文を表示するだけで、日付情報を回答に出典として埋め込むロジックは入れていない

起動方法¶

詳細は README.md を参照。サマリだけ:

cd 03

# Step 1: 初回のみ。upstream → ChromaDB を構築
uv run --no-project \
  --with "chromadb>=0.5" --with "openai>=1.0" --with "requests" \
  python setup_db.py

# Step 2: チャットボット起動
uv run --no-project \
  --with "langchain>=0.3" \
  --with "langchain-openai>=0.2" \
  --with "langchain-chroma>=0.1" \
  --with "chromadb>=0.5" \
  --with "chainlit" \
  chainlit run chatbot.py -w

記事参照¶

Software Design 2023年〜の連載第03回「LangChainによるRAG実装」
関連: 第02回（手書き版RAG）, 第10〜12回（Corrective/Adaptive RAG など発展形）

作成: 2026-05-18 ／最終更新: 2026-06-10

第03回 学習メモ: LangChainによるRAG実装（LCEL版）¶