LangChain Runnable入門：基本的な使い方と実装パターン

はじめに
1. この記事で学べること
環境構築
Runnableの基本概念
シンプルな変換処理の実装
1. 実装のポイント
パススルーチェーンの活用
1. 実装のポイント
結合チェーンの構築
1. 実装のポイント
ネストされたチェーンの高度な実装
1. 実装のポイント
まとめと次のステップ
リポジトリ
1. 参考リンク
2. 関連

はじめに

LangChainのRunnableは、AI処理パイプラインを構築する上で非常に重要なコンポーネントです。この記事では、LangChainのRunnableの基本的な使い方から、より複雑な実装パターンまでを、実践的なコード例を交えて解説します。

この記事で学べること

Runnableの基本的な概念と使い方
段階的に複雑さを増す実装パターン
実践的なエラーハンドリングとログ管理
チェーンの構築と最適化のベストプラクティス

環境構築

まず、必要なパッケージをインストールします：

pip install langchain langchain-openai python-dotenv loguru art

プロジェクトの基本構造：

sandbox/
└── runnable/
    └── basic/
        ├── 01_simple_transform.py
        ├── 02_passthrough_chain.py
        ├── 03_combined_chain.py
        ├── 04_nested_chain.py
        ├── logger_setup.py
        └── README.md

Runnableの基本概念

LangChainのRunnableは以下の特徴を持ちます：

チェーン化可能 : パイプライン演算子 | を使用して処理を連結
型安全 : 入力と出力の型を明確に定義可能
再利用性 : モジュール化された処理を複数の場所で再利用可能
エラーハンドリング : 統一された例外処理の仕組み

シンプルな変換処理の実装

最も基本的なRunnableの使用例として、テキスト分析を行う処理を実装します。

from langchain_core.runnables import RunnableLambda
from typing import Dict, Any

def text_analyzer(text: str) -> Dict[str, Any]:
    """テキストの基本的な分析を行う関数"""
    return {
        "original_text": text,          # 元のテキスト
        "character_count": len(text),   # 文字数
        "word_count": len(text.split()),# 単語数
        "is_question": "?" in text      # 疑問文かどうか
    }

# RunnableLambdaでラップ
transform = RunnableLambda(text_analyzer)

# 使用例
result = transform.invoke("LangChainは素晴らしいですか?")
print(result)

実装のポイント

RunnableLambdaで通常の関数をラップして、Runnable化
明確な入力と出力の型定義
シンプルで再利用可能な処理単位

パススルーチェーンの活用

データの流れを制御するためのパススルーチェーンを実装します。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

def create_chain_with_passthrough():
    def add_context(input_dict: Dict) -> Dict:
        """入力テキストに追加情報を組み合わせる"""
        combined_text = f"{input_dict['input_text']}（{input_dict['additional_info']}）"
        return {"text": combined_text}

    prompt = ChatPromptTemplate.from_template(
        "以下のトピックについて簡単に説明してください：{text}"
    )

    chain = (
        RunnableLambda(add_context)  # テキストの加工
        | prompt                     # プロンプトの生成
        | ChatOpenAI()              # ChatGPTでの処理
        | StrOutputParser()         # 文字列への変換
    )

    return chain

# 使用例
chain = create_chain_with_passthrough()
result = chain.invoke({
    "input_text": "Python",
    "additional_info": "プログラミング言語"
})

実装のポイント

パイプライン演算子 | による処理の連結
中間データの加工と受け渡し
ChatGPTとの統合

結合チェーンの構築

複数のRunnableを組み合わせて、より複雑な処理を実現します。

def create_combined_chain():
    def add_prefix(text: str) -> Dict[str, str]:
        """トピックにプレフィックスを追加"""
        return {"topic": f"最新の{text}について"}

    def add_requirements(data: Dict) -> Dict[str, str]:
        """要件を追加"""
        return {
            "topic": data["topic"],
            "requirements": "具体例を2つ含めてください。"
        }

    prompt = ChatPromptTemplate.from_template(
        "{topic}\n{requirements}"
    )

    chain = (
        RunnableLambda(add_prefix)     # プレフィックスの追加
        | RunnableLambda(add_requirements) # 要件の追加
        | prompt                       # プロンプトの作成
        | ChatOpenAI()                # ChatGPTでの生成
        | StrOutputParser()           # 文字列への変換
    )

    return chain

実装のポイント

複数の処理を段階的に適用
各処理の役割を明確に分離
エラーハンドリングの考慮

ネストされたチェーンの高度な実装

より複雑な処理フローを実現するために、チェーンをネストします。

def create_nested_chain():
    # 内部チェーン: トピックの説明を生成
    inner_prompt = ChatPromptTemplate.from_template(
        "{formatted_topic}の説明を{style}書いてください。"
    )
    inner_chain = RunnableLambda(format_input) | inner_prompt

    # 外部チェーン: 説明を箇条書きに変換
    outer_prompt = ChatPromptTemplate.from_template(
        "以下の説明文を箇条書きにしてください：\n{text}"
    )

    chain = (
        inner_chain                    # 内部チェーンで説明を生成
        | ChatOpenAI()                # ChatGPTで処理
        | StrOutputParser()           # 文字列に変換
        | {"text": RunnablePassthrough()} # 中間結果の保持
        | outer_prompt                # 外部チェーンで箇条書きに変換
        | ChatOpenAI()                # 再度ChatGPTで処理
        | StrOutputParser()           # 最終的な文字列に変換
    )

    return chain