Zenn
🐱

【実践編】LangChainの高度なデバッグテクニック：複雑なチェーンの可視化と最適化

2024/11/14に公開
並列処理
LangChain
デバッグ手法
複雑なアプリケーション
tech

🎯 この記事の狙い

前回の基本的なデバッグ手法の解説に続き、今回は複雑なチェーン構造を持つLangChainアプリケーションのデバッグ手法について、実践的な例を交えて解説します。

複雑なチェーン構造とは

LangChainの高度な使用例では、以下のような複雑な構造が出現します：

  1. 並列実行チェーン
  2. 選択的結果利用（pick機能）
  3. 連鎖的実行フロー
  4. 条件分岐を含む処理

チェーン構造の例

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

ベストプラクティス

  1. デバッグ情報の階層化

    • 重要度に応じたログレベルの使用
    • 構造化されたJSON出力
    • コンテキスト情報の付加

  2. パフォーマンス監視

    • 実行時間の計測
    • トークン使用量の追跡
    • ボトルネックの特定

  3. エラーハンドリング

    • 詳細なエラー情報の記録
    • コンテキストの保持
    • リカバリー戦略の実装

  4. チェーン構造の最適化

    • 必要な処理の並列化
    • 不要なデータの削除
    • メモリ使用量の最適化

デバッグ出力例

並列チェーン実行時の出力

[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アプリケーションのデバッグには以下が重要です：

  1. 構造化されたアプローチ

    • チェーンの構造を理解
    • 適切なデバッグポイントの設定
    • 系統的な情報収集

  2. 効率的な監視

    • パフォーマンスメトリクスの収集
    • ボトルネックの特定
    • リソース使用の最適化

  3. 実践的なツール活用

    • カスタムコールバックの実装
    • ログ出力の最適化
    • エラーハンドリングの強化

リポジトリ

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

参考リンク

Discussion