はじめに
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
実装のポイント
- 内部チェーンと外部チェーンの組み合わせ
- 中間結果の活用
- 複雑な処理フローの制御
まとめと次のステップ
LangChainのRunnableは、AI処理パイプラインを構築する強力なツールです。この記事で学んだ内容を基に、以下のような発展的な実装に挑戦してみましょう：
- 並列処理の実装
- エラーリカバリーの強化
- カスタムチェーンの作成
- パフォーマンスの最適化
リポジトリ
参考リンク
実践的なコード例と詳細な説明を通じて、LangChainのRunnableの基本を理解できたことと思います。これらの基礎を応用することで、より複雑で実用的なAIアプリケーションの開発が可能になります。