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

🎯 この記事の狙い

前回の基本的なデバッグ手法の解説に続き、今回は複雑なチェーン構造を持つ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
  }
}