Zenn
🎨

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

2024/11/14に公開
Python
パイプライン
LangChain
runnable
tech

はじめに

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は以下の特徴を持ちます：

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

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

最も基本的な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

実装のポイント

  • 内部チェーンと外部チェーンの組み合わせ
  • 中間結果の活用
  • 複雑な処理フローの制御

まとめと次のステップ

LangChainのRunnableは、AI処理パイプラインを構築する強力なツールです。この記事で学んだ内容を基に、以下のような発展的な実装に挑戦してみましょう：

  1. 並列処理の実装
  2. エラーリカバリーの強化
  3. カスタムチェーンの作成
  4. パフォーマンスの最適化

リポジトリ

https://github.com/Sunwood-ai-labs/langchain-sandbox

参考リンク

実践的なコード例と詳細な説明を通じて、LangChainのRunnableの基本を理解できたことと思います。これらの基礎を応用することで、より複雑で実用的なAIアプリケーションの開発が可能になります。

Discussion