🎯 この記事の狙い
前回の基本的なデバッグ手法の解説に続き、今回は複雑なチェーン構造を持つLangChainアプリケーションのデバッグ手法について、実践的な例を交えて解説します。
複雑なチェーン構造とは
LangChainの高度な使用例では、以下のような複雑な構造が出現します：
- 並列実行チェーン
- 選択的結果利用（pick機能）
- 連鎖的実行フロー
- 条件分岐を含む処理
チェーン構造の例
final_chain = RunnableParallel(
{
"description": base_chain.pick("description"),
"fun_fact": base_chain.pick("fun_fact"),
"habitat": base_chain.pick("habitat"),
"summary": summary_input_chain | summary_chain,
"animal": RunnablePassthrough()
}
)
高度なデバッグ手法
1. 構造化されたデバッグ出力
def format_dict(d: Dict[str, Any], indent: int = 2) -> str:
"""
デバッグ出力用の辞書整形関数
Args:
d: 整形する辞書オブジェクト
indent: インデントのスペース数
Returns:
str: 整形されたJSON文字列
"""
return json.dumps(d, ensure_ascii=False, indent=indent)
2. 実行時間の計測
@measure_execution_time
def execute_chain(chain, input_data):
logger.debug(f"[Debug] 入力データ:\n{format_dict(input_data)}")
result = chain.invoke(input_data)
logger.debug(f"[Debug] 実行結果:\n{format_dict(result)}")
return result
3. 高度なコールバックハンドラー
class DebugCallbackHandler(BaseCallbackHandler):
def on_llm_start(self, serialized: Dict[str, Any], prompts: List[str], **kwargs) -> None:
formatted_prompts = {
"prompts": prompts,
"serialized": {
"name": serialized.get("name", "unknown"),
"type": serialized.get("type", "unknown")
}
}
logger.debug(f"[Debug] LLM開始:\n{format_dict(formatted_prompts)}")
def on_llm_end(self, response, **kwargs) -> None:
formatted_response = {
"generations": [[{
"text": gen.text,
"message_type": gen.message.__class__.__name__
} for gen in gens] for gens in response.generations],
"token_usage": response.llm_output.get("token_usage", {})
}
logger.debug(f"[Debug] LLM終了:\n{format_dict(formatted_response)}")
パフォーマンス最適化
1. 並列実行の活用
base_chain = RunnableParallel(
description=description_prompt | model | parser,
fun_fact=fact_prompt | model | parser,
habitat=habitat_prompt | model | parser
)
2. 選択的なデータ利用
summary_input_chain = RunnableParallel(
{
"description": base_chain.pick("description"),
"fun_fact": base_chain.pick("fun_fact"),
"animal": RunnablePassthrough()
}
)
実装例：複雑なチェーンの構築と監視
1. チェーン構造の定義
def create_multi_chain():
"""
複数のチェーンを組み合わせた処理を作成
"""
# モデルとパーサーの設定
model = ChatOpenAI(temperature=0.7, callbacks=[DebugCallbackHandler()])
parser = StrOutputParser()
# プロンプトの作成
description_prompt = ChatPromptTemplate.from_messages([
("human", "{animal}について1文で説明してください。")
])
# ... その他のプロンプト定義 ...
# チェーンの構築
base_chain = RunnableParallel(...)
summary_chain = summary_prompt | model | parser
return final_chain
2. 実行フローの監視
def execute_chain(chain, input_data):
try:
logger.debug(f"[Debug] 入力データ:\n{format_dict(input_data)}")
result = chain.invoke(input_data)
logger.debug(f"[Debug] 実行結果:\n{format_dict(result)}")
return result
except Exception as e:
logger.error(f"チェーン実行エラー: {str(e)}")
raise
ベストプラクティス
-
デバッグ情報の階層化
-
重要度に応じたログレベルの使用
-
構造化されたJSON出力
-
コンテキスト情報の付加
-
パフォーマンス監視
-
実行時間の計測
-
トークン使用量の追跡
-
ボトルネックの特定
-
エラーハンドリング
-
詳細なエラー情報の記録
-
コンテキストの保持
-
リカバリー戦略の実装
-
チェーン構造の最適化
-
必要な処理の並列化
-
不要なデータの削除
-
メモリ使用量の最適化
デバッグ出力例
並列チェーン実行時の出力
[Debug] LLM開始:
{
"prompts": [
"象について1文で説明してください。",
"象についての面白い豆知識を1つ教えてください。",
"象の主な生息地について教えてください。"
],
"serialized": {
"name": "ChatOpenAI",
"type": "llm"
}
}
[Debug] LLM終了:
{
"generations": [...],
"token_usage": {
"completion_tokens": 156,
"prompt_tokens": 45,
"total_tokens": 201
}
}
まとめ
高度なLangChainアプリケーションのデバッグには以下が重要です：
-
構造化されたアプローチ
-
チェーンの構造を理解
-
適切なデバッグポイントの設定
-
系統的な情報収集
-
効率的な監視
-
パフォーマンスメトリクスの収集
-
ボトルネックの特定
-
リソース使用の最適化
-
実践的なツール活用
-
カスタムコールバックの実装
-
ログ出力の最適化
-
エラーハンドリングの強化
リポジトリ
GitHub - Sunwood-ai-labs/langchain-sandbox
Contribute to Sunwood-ai-labs/langchain-sandbox development by creating an account on GitHub.
github.com
コメント