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

## はじめに

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

### この記事で学べること

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

## 環境構築

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

```bash
pip install langchain langchain-openai python-dotenv loguru art
```

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

```plaintext
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の使用例として、テキスト分析を行う処理を実装します。

```python
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化
- 明確な入力と出力の型定義
- シンプルで再利用可能な処理単位

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

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

```python
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を組み合わせて、より複雑な処理を実現します。

```python
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
```

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

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

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

```python
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公式ドキュメント](https://python.langchain.com/docs/get_started/introduction)
- [GitHub Repository](https://github.com/langchain-ai/langchain)

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