「LangChainとLangGraphによるRAG・AIエージェント［実践］入門」でつまずいたことメモ:5章

Last updated at 2024-12-07Posted at 2024-11-24

はじめに

「LangChainとLangGraphによるRAG・AIエージェント［実践］入門」の第5章で私がつまずいたことのメモです。

（このメモのほかの章へ：1章 / 2章 / 3章 / 4章 / 5章 / 6章）

この記事は個人で作成したものであり、内容や意見は所属企業・部門見解を代表するものではありません。

第5章 LangChain Expression Language(LCEL)徹底解説

前章に続き大嶋さんが担当された章です。

5.1 RunnableとRunnableSequenceーLCELの最も基本的な構成要素

コラムの「LCELはどのように実現されているのか」が興味深かったです。演算子|で簡単に連鎖させられるのは便利ですが、その代わりに内部の実装はかなり泥臭くなっているんだろうなぁと思ったら、内部で使われているcoerce_to_runnable()のコードも意外にシンプルでした。いいですね。

いつものように脱線してしまいますが、Pythonでは__ror__という右オペランドをオーバーロードする仕組みがあるんですね。C++には私がコード書いていた頃にはなかったので（かなり前なので今の仕様はわかりませんが）非メンバー関数でオーバーロードしないといけなかったりして、なんかきれいに書けずモヤっていたのを思い出しました。

あと、本ではLangSmithでChainの内部動作が確認できるとサラッと書いてありますが、これがすごいです。最初は「なんで今動かしているコードのことをLangSmithが知ってるんだ！？」とびっくりしましたが、4.1節をおさらいして腹落ちしました。あの時に書いていた以下のおまじないによって、LangSmithにトレース情報が転送されているんですね。

os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com"
os.environ["LANGCHAIN_API_KEY"] = userdata.get("LANGCHAIN_API_KEY")
os.environ["LANGCHAIN_PROJECT"] = "agent-book"

LangSmithの画面を見ると、内部動作の確認どころかパフォーマンス関連の収集分析もやってくれていて、これはすごく便利そうです。ちなみに環境変数名がLANGCHAIN_ENDPOINTだったので、もしかしてプロキシみたいな仕組みでLangChainからの全処理がLangSmithを経由しちゃうの？とびっくりしましたが、あくまでもLangSmithに送られるのはトレース情報のみで、今やっているようなOpenAIのAPIコールはローカル（Google Colabの実行環境）から直接行われる模様です。そりゃそうですよね、全部経由してたらパフォーマンス面やセキュリティ面が面倒なことになります。おそらく、APM（アプリケーションパフォーマンスモニタリング）のサービスにおける情報収集Agentみたいな機能がLangChain自身に備わっているということですね。

5.2 RunnableLambdaー任意の関数をRunnableにする

関数をそのまま渡せるのも便利すぎます。前述のcoerce_to_runnable()がCallableをRunnableLambdaに変換してくれています。

あと、コラムの「独自の関数をstreamに対応させたい場合」でちょっと脱線です。
私も自社でLLMを使ったサービスを展開していたりするのですが、LLMのストリーミングの処理はちょっとやっかいなんですよね。本の例のように小文字を大文字に変換する程度なら問題ないのですが、処理の対象が文字列（文字ではなく）になると面倒になります。

たとえば、LLMを使った開発を効率化するためのライブラリについて質問し、回答文の中でメジャーなライブラリ名が出てきたら公式サイトのURLを付け足してあげることを考えます。

動かない例

from typing import Iterator

def enrich_url(input_stream: Iterator[str]) -> Iterator[str]:
    # 有名なライブラリは公式ページのURLを補足
    replacements = {
        "LangChain": "[LangChain](https://www.langchain.com/)",
        "Hugging Face Transformers": "[Hugging Face Transformers](https://huggingface.co/docs/transformers/ja/index)",
        "LlamaIndex": "[LlamaIndex](https://www.llamaindex.ai/)"
    }

    for text in input_stream:
        # 置換対象のライブラリ名があったらURL付きに置換
        for target, replacement in replacements.items():
            text = text.replace(target, replacement)
        yield text

chain = prompt | model | StrOutputParser() | enrich_url

for chunk in chain.stream({"input": "LLMを使った開発を効率化するためのライブラリで代表的なものの名前を5個ほど列挙して、それぞれ50文字程度で解説してください。"}):
    print(chunk, end="", flush=True)

実行してみましょう。

以下は、LLMを使った開発を効率化するための代表的なライブラリです。

1. **Hugging Face Transformers**  
   様々な事前学習済みモデルを提供し、自然言語処理タスクを簡単に実装可能。

2. **LangChain**  
   LLMを活用したアプリケーションの構築を支援するフレームワークで、チェーン構造を利用。

3. **Haystack**  
   検索エンジンやQAシステムの構築をサポートするライブラリで、LLMとの統合が容易。

4. **OpenAI API**  
   OpenAIのLLMを利用するためのAPIで、テキスト生成や対話システムの構築が可能。

5. **LlamaIndex**  
   LLMとデータソースを結びつけるためのインデックス作成ライブラリで、情報検索を効率化。

本当はLangChainやHugging Face TransformersやLlamaIndexに対してURLを補足してくれるはずだったのですが、なにもしてくれません。そうです、ストリーミングで渡されてくる文字が細切れになっていることが原因です。enrich_url関数の最終行のyieldをちょっと細工して、細切れ度合いを見てみましょう。

enrich_urlの最終行を変更

        yield text + "/"

/以下/は/、/LL/M/を/使/った/開/発/を/効/率/化/する/ため/の/代表/的/な/ライ/ブラ/リ/です/。

/1/./ **/H/ug/ging/ Face/ Transformers/**/  
/  / 様/々/な/事/前/学/習/済/み/モデル/を/提供/し/、/自然/言/語/処/理/タ/スク/を/簡/単/に/実/装/可能/。

/2/./ **/Lang/Chain/**/  
/  / L/LM/を/活/用/した/ア/プリ/ケ/ーション/の/構/築/を/支/援/する/フ/レ/ーム/ワ/ーク/で/、/チェ/ーン/構/造/を/利用/。

/3/./ **/Hay/stack/**/  
/  / 検/索/エ/ン/ジ/ン/や/QA/シ/ステ/ム/の/構/築/を/サ/ポ/ート/する/ライ/ブラ/リ/で/、/LL/M/との/統/合/が/容易/。

/4/./ **/Open/AI/ API/**/  
/  / Open/AI/の/LL/M/を/利用/する/ため/の/API/で/、/テ/キ/スト/生成/や/対/話/シ/ステ/ム/の/構/築/が/可能/。

/5/./ **/L/lama/Index/**/  
/  / L/LM/と/デ/ータ/ソ/ース/を/結/び/つ/け/る/ため/の/イン/デ/ックス/作/成/ライ/ブラ/リ/で/、/情報/検索/を/効/率/化/。//

かなりズタズタにされて渡ってくることがわかります。見た感じだと、OpenAIの場合はほぼトークン単位みたいです。ちなみにAzure OpenAIの場合はコンテンツフィルタリングの機能が働く関係で、ある程度の長さの文字列で出力されます。使うLLMやそこに備わるフィルタリング機能などによって変わりますので、ストリーミング出力が「必ず単語単位でやってくる」といった前提を勝手に期待することはできません¹。

というわけで、ズタズタにされて渡ってくることを前提にコードを直してみます。

これなら動く

from typing import Iterator

def enrich_url(input_stream: Iterator[str]) -> Iterator[str]:
    # 有名なライブラリは公式ページのURLを補足
    replacements = {
        "LangChain": "[LangChain](https://www.langchain.com/)",
        "Hugging Face Transformers": "[Hugging Face Transformers](https://huggingface.co/docs/transformers/ja/index)",
        "LlamaIndex": "[LlamaInde](https://www.llamaindex.ai/)"
    }
    buffer = "" # 文字がバラバラに取得されるのでバッファに溜めて処理
    max_len = max(len(key) for key in replacements)  # 一番長いライブラリ名の文字数

    for text in input_stream:
        buffer += text  # バッファに溜める
        # バッファ内に置換対象のライブラリ名があったらURL付きに置換
        for target, replacement in replacements.items():
            buffer = buffer.replace(target, replacement)
        # バッファには次回の置換処理に備えて最長ライブラリ名-1文字分だけ残しておけばいいので、そこまでを出力
        if len(buffer) > (max_len - 1):
            yield buffer[:-(max_len - 1)]
            buffer = buffer[-(max_len - 1):]
        else:
            pass    # 溜まっていない場合は続きの文字を待つ

    # 最後にバッファに残っている分を出力
    if buffer:
        yield buffer

chain = prompt | model | StrOutputParser() | enrich_url

for chunk in chain.stream({"input": "LLMを使った開発を効率化するためのライブラリで代表的なものの名前を5個ほど列挙して、それぞれ50文字程度で解説してください。"}):
    print(chunk, end="", flush=True)

バッファリングする処理を入れました。置換元の最長文字数分を常に確保できるようにしておくというやっつけ実装ですが²。

以下は、LLMを使った開発を効率化するための代表的なライブラリです。

1. **[Hugging Face Transformers](https://huggingface.co/docs/transformers/ja/index)**  
   様々な事前学習済みモデルを提供し、自然言語処理タスクを簡単に実装可能。

2. **[LangChain](https://www.langchain.com/)**  
   LLMを活用したアプリケーションの構築を支援するフレームワークで、チェーン構造を利用。

3. **Haystack**  
   検索エンジンやQAシステムの構築をサポートするライブラリで、LLMとの統合が容易。

4. **OpenAI API**  
   OpenAIのLLMを利用するためのAPIで、テキスト生成や対話システムの構築が可能。

5. **[LlamaInde](https://www.llamaindex.ai/)**  
   LLMとデータソースを結びつけるためのインデックス作成ライブラリで、情報検索を効率化。

いい感じ！

実際にいろいろ置換させるには、置換条件間の優先順位や、置換した結果をさらに置換しちゃっていいのか？などの考慮も必要です。ストリーミングの途中で何か処理を挟むのは大変ですね。もしかしたら何か便利なライブラリがあるかもしれません。情報がありましたらお待ちしております。

あ、１つ忘れていました。Google Colabのコード書いていると初期設定ではTabが2文字なっているのでイライラするかもしれません。4文字への変更方法は@ihafuさんの記事をどうぞ。

5.3 RunnableParallelー複数のRunnableを並列につなげる

dictが指定できるのも便利ですね。__or__や__ror__でMappingを受け付けるように実装されていて、いたせりつくせりです。

また、itemgetterを使うと2つの意見をまとめるところにtopicが伝えられるわけですね、なるほど！

5.4 RunnablePassthroughー入力をそのまま出力する

以前のLangChain本ではWebサイトの検索にDuckDuckGoを使っていましたが、今回はtavilyを使います。付録の「A.1 各種サービスのサインアップ」にtavilyのサインアップ手順がありませんが（許諾の関係？）、簡単なのでつまずくことはないでしょう。

さっそく最初のコードを実行してみたのですが、あれ？今日は2024年11月24日(日)なのに、古い情報が返ってきた。

東京の今日、2024年11月19日(火)の天気は「晴時々曇」で、最高気温は13℃、最低気温は8℃です。降水確率は0％となっています。

これは、続くassignを使ったコードを実行してみると原因がわかるのですが、検索対象になっているページが少し古い内容でした。実際にリアルタイムな天気予報の情報を使いたい場合は、専用のAPIと組み合わせる必要がありそうです。

あと、コラムのastream_eventsやastream_logによる解析は、現時点ではLangSmithで十分な感じがしているので、必要性を感じたところで深掘りしてみようと思います。

5.5 まとめ

LCELの仕組みを理解できてよかったです。次回は高度なRAGのお話です。

（このメモのほかの章へ：1章 / 2章 / 3章 / 4章 / 5章 / 6章）

実は日本語の場合、最低単位が「文字」かどうかすらちょっと怪しそうで、ジェネレータで文字列として処理するのは少し心配だったりします。ストリーミングにおける最低の出力単位が「文字」であることを明記したドキュメントなどがあれば安心なのですが。もしどなたかご存知の方がいましたらお知らせください。それにしても、マルチバイト文字の処理は大変ですね。次にエンジニアとして生まれ変わるなら、シングルバイト文字の圏内に生まれたいものです ↩
ボイヤー-ムーアなどの検索アルゴリズムをきちんと考えて取り込んだら、もっと最適化できそうな予感もします。でも、LLMの出力の方が処理時間のコストが高いでしょうから、パフォーマンスに与える効果はまったくないかもしれません。 ↩

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up