LangGraphのループと条件分岐設計パターン完全ガイド

AIエージェントアプリケーションにおいて、複雑な会話フローをシンプルに管理することは永遠のテーマです。私は以前、ECサイトのAIカスタマーサービスでLangGraphを採用しましたが、特にループ構造と条件分岐の実装で 많은紆余曲折がありました。本記事では、LangGraphを使った実践的な設計パターンを、HolySheep AIのAPIと組み合わせて詳しく解説します。

なぜLangGraphなのか

LangGraphは、LangChainをベースとしたirected Cyclic Graph（DAG）ベースのフレームワークです。ノード間の依存関係を明確に定義でき、ループや条件分岐を宣言的に記述 가능합니다。特に以下のシナリオで真価を発揮します：

• 複雑な会話を要するカスタマーサポート
• 人間の介入了必要な承認ワークフロー
• 外部APIを繰り返し呼び出すデータ取得処理

私も実際に利用しているHolySheep AIは、レートが¥1=$1という破格のコスト効率を提供しており（月額公式比85%節約）、DeepSeek V3.2は$0.42/MTokという極限の低コストで運用できます。WeChat PayやAlipayにも対応しており、日本円の変動リスクなく安定利用が可能なのも大きなポイントです。

基本的なグラフ構造を理解する

LangGraphでは、StateGraphというクラスを使ってグラフを定義します。Stateはアプリケーション全体の状態を表し、各ノードは状態を受け取り、新しい状態を返す関数として実装されます。

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

状態の定義
class AgentState(TypedDict):
    messages: list
    intent: str
    retry_count: int
    max_retries: int

def initialize(state: AgentState) -> AgentState:
    """初期化ノード"""
    print(f"会話開始 - 現在のメッセージ数: {len(state['messages'])}")
    return state

def analyze_intent(state: AgentState) -> AgentState:
    """意図分析ノード"""
    last_message = state["messages"][-1]["content"]
    
    # HolySheep AIで意図分析
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "ユーザーの意図を categorize: 'order_inquiry', 'refund_request', 'product_question', 'human_escalation'"},
            {"role": "user", "content": last_message}
        ]
    )
    
    intent = response.choices[0].message.content.strip().lower()
    return {"intent": intent}

グラフの構築
workflow = StateGraph(AgentState)
workflow.add_node("initialize", initialize)
workflow.add_node("analyze_intent", analyze_intent)

workflow.set_entry_point("initialize")
workflow.add_edge("initialize", "analyze_intent")
workflow.add_edge("analyze_intent", END)

app = workflow.compile()

条件分岐による動的フロー制御

LangGraphの真価は、条件分岐にあります。router関数を使って、次哪个ノードに進むかを動的に決定できます。以下の例では、ECサイトのカスタマーサービスシナリオを想定しています：

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from typing import Literal

状態の定義（拡張版）
class ECServiceState(TypedDict):
    messages: list
    intent: str
    order_id: str | None
    requires_human: bool
    loop_count: int
    tools_used: list[str]

条件分岐路由器
def route_based_on_intent(state: ECServiceState) -> Literal[
    "handle_order_inquiry",
    "handle_refund", 
    "handle_product_question",
    "escalate_to_human",
    "check_resolution"
]:
    intent = state.get("intent", "")
    
    # 繰り返しチェック（無限ループ防止）
    if state.get("loop_count", 0) >= 5:
        return "escalate_to_human"
    
    if "order" in intent or "配送" in intent:
        return "handle_order_inquiry"
    elif "返金" in intent or "refund" in intent:
        return "handle_refund"
    elif "商品" in intent or "product" in intent:
        return "handle_product_question"
    elif "人" in intent or "human" in intent:
        return "escalate_to_human"
    else:
        return "check_resolution"

各処理ノードの実装
def handle_order_inquiry(state: ECServiceState) -> ECServiceState:
    """注文Inquiry処理"""
    order_id = extract_order_id(state["messages"][-1]["content"])
    
    # 外部API呼び出し（例：注文管理システム）
    order_data = get_order_from_api(order_id)
    
    response = client.chat.completions.create(
        model="deepseek-chat",
        messages=[
            {"role": "system", "content": "あなたはECサイトの注文サポートです。"},
            {"role": "user", "content": f"注文 {order_id} の状況を説明してください: {order_data}"}
        ]
    )
    
    return {
        "messages": state["messages"] + [{"role": "assistant", "content": response.choices[0].message.content}],
        "order_id": order_id,
        "loop_count": state.get("loop_count", 0) + 1,
        "tools_used": state.get("tools_used", []) + ["order_api"]
    }

def escalate_to_human(state: ECServiceState) -> ECServiceState:
    """人間へのエスカレーション"""
    return {
        "requires_human": True,
        "messages": state["messages"] + [
            {"role": "assistant", "content": "申し訳ありません。人間の担当者を呼び出します。少々お待ちください。"}
        ]
    }

グラフ構築
workflow = StateGraph(ECServiceState)
workflow.add_node("initialize", initialize)
workflow.add_node("analyze_intent", analyze_intent)
workflow.add_node("handle_order_inquiry", handle_order_inquiry)
workflow.add_node("handle_refund", handle_refund)
workflow.add_node("handle_product_question", handle_product_question)
workflow.add_node("escalate_to_human", escalate_to_human)
workflow.add_node("check_resolution", check_resolution)

workflow.add_edge("initialize", "analyze_intent")
workflow.add_conditional_edges(
    "analyze_intent",
    route_based_on_intent,
    {
        "handle_order_inquiry": "handle_order_inquiry",
        "handle_refund": "handle_refund",
        "handle_product_question": "handle_product_question",
        "escalate_to_human": "escalate_to_human",
        "check_resolution": "check_resolution"
    }
)

ループバック（解決しなかった場合）
workflow.add_edge("handle_order_inquiry", "analyze_intent")
workflow.add_edge("handle_refund", "analyze_intent")
workflow.add_edge("handle_product_question", "analyze_intent")

app = workflow.compile()

実際のAPI呼び出し例（HolySheep AI統合）

ここからは、HolySheep AIのAPIを使った実践的なコード例を示します。DeepSeek V3.2を使うことで、コストを大幅に削減できます。

from openai import OpenAI
import os

HolySheep AIクライアント初期化
client = OpenAI(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1"  # HolySheep APIエンドポイント
)

def call_with_retry(messages: list, max_retries: int = 3) -> str:
    """リトライ機構付きのAPI呼び出し"""
    for attempt in range(max_retries):
        try:
            # DeepSeek V3.2を使用（$0.42/MTok — 業界最安値）
            response = client.chat.completions.create(
                model="deepseek-chat",  # DeepSeek V3.2
                messages=messages,
                temperature=0.7,
                max_tokens=1000
            )
            return response.choices[0].message.content
            
        except Exception as e:
            if attempt == max_retries - 1:
                raise RuntimeError(f"API呼び出し失敗: {str(e)}")
            import time
            time.sleep(2 ** attempt)  # 指数バックオフ
    
    return ""

実際の使用例
def process_customer_message(user_input: str) -> dict:
    """顧客メッセージの処理"""
    messages = [
        {"role": "system", "content": "あなたは親身なカスタマーサポートです。"},
        {"role": "user", "content": user_input}
    ]
    
    result = call_with_retry(messages)
    
    # レイテンシ測定（HolySheepは<50msを保証）
    return {
        "response": result,
        "model": "deepseek-chat",
        "cost_estimate": len(result) / 1000 * 0.42  # 簡易コスト計算
    }

ループ構造の実装パターン

データ取得のように、条件が満たされるまで繰り返し処理を行うケースを考えます。以下は、複数ソースから情報を収集するループの例です：

def data_collection_loop(state: ECServiceState) -> ECServiceState:
    """データ収集ループノード"""
    collected = state.get("collected_data", {})
    sources = ["inventory", "pricing", "shipping", "reviews"]
    
    for source in sources:
        if source not in collected:
            try:
                # 各ソースからデータを取得
                data = fetch_from_source(source, state.get("product_id"))
                collected[source] = data
            except Exception as e:
                print(f"ソース {source} からの取得に失敗: {e}")
                collected[source] = None
    
    all_collected = all(v is not None for v in collected.values())
    
    return {
        "collected_data": collected,
        "all_sources_ready": all_collected
    }

def route_collection(state: ECServiceState) -> Literal["data_collection_loop", "synthesize_results"]:
    """コレクション状態に基づいてルート決定"""
    if state.get("all_sources_ready", False):
        return "synthesize_results"
    return "data_collection_loop"

ループグラフの構築
workflow = StateGraph(ECServiceState)
workflow.add_node("data_collection_loop", data_collection_loop)
workflow.add_node("synthesize_results", synthesize_results)

workflow.set_entry_point("data_collection_loop")
workflow.add_conditional_edges(
    "data_collection_loop",
    route_collection,
    {
        "data_collection_loop": "data_collection_loop",
        "synthesize_results": "synthesize_results"
    }
)
workflow.add_edge("synthesize_results", END)

よくあるエラーと対処法

エラー1: 無限ループによるスタックオーバーフロー

❌ 誤った実装（loop_count的增加がない）
def bad_node(state):
    if not state.get("resolved"):
        return {"resolved": False}  # 無限に実行される

✅ 正しい実装（ループ回数制限付き）
def safe_node(state):
    loop_count = state.get("loop_count", 0)
    if loop_count >= 5 or state.get("resolved"):
        return {"resolved": True, "loop_count": loop_count}
    return {"resolved": False, "loop_count": loop_count + 1}

エラー2: 条件分岐での状態不整合

❌ 誤った実装（intentが未定義の場合）
def bad_route(state):
    if "order" in state["intent"]:  # KeyError発生の可能性
        return "handle_order"

✅ 正しい実装（デフォルト値とNoneチェック）
def safe_route(state):
    intent = state.get("intent") or ""
    if "order" in intent:
        return "handle_order"
    elif intent == "":
        return "initialize"  # フォールバック
    return "general_response"

エラー3: API呼び出しのレート制限

❌ 誤った実装（リトライ機構なし）
def fast_api_call(message):
    return client.chat.completions.create(
        model="deepseek-chat",
        messages=[{"role": "user", "content": message}]
    )

✅ 正しい実装（指数バックオフ付きリトライ）
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def safe_api_call(message):
    try:
        return client.chat.completions.create(
            model="deepseek-chat",
            messages=[{"role": "user", "content": message}]
        )
    except RateLimitError:
        raise  # tenacityがリトライ

エラー4: ベースURL設定の誤り

❌ 誤った設定（openai.comへの接続を試みる）
client = OpenAI(api_key="xxx")  # デフォルトでapi.openai.comを使用

✅ 正しい設定（HolySheepのエンドポイントを明示）
client = OpenAI(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1"  # 必ず指定
)

エラー5: 状態の型の不整合

❌ 誤った実装（型の混在）
class BadState(TypedDict):
    count: int

def bad_increment(state):
    state["count"] = "5"  # 文字列代入 → 型エラー

✅ 正しい実装（型の明示）
class GoodState(TypedDict):
    count: int
    items: list[str]

def good_increment(state):
    state["count"] = state.get("count", 0) + 1
    state["items"] = state.get("items", []) + ["new_item"]

パフォーマンス最適化のポイント

LangGraphアプリケーションのパフォーマンスを最大化するには、以下の点を意識しています：

• モデル選択：HolySheep AIではDeepSeek V3.2（$0.42/MTok）が最もコスト効率が高く、通常処理に十分使えます。複雑な推論が必要な場合のみGPT-4.1（$8/MTok）に切り替えるのが賢明です
• 状態管理の最適化：불필요な状態保持を避け、必要なデータのみを保持することで、グラフの実行効率が向上します
• 並列処理：独立したノードはParallelNodeを使って同時に実行可能です
• チェックポインティング：checkpoint引数を使って、中間状態を保存できます

まとめ

LangGraphのループと条件分岐は、複雑なAIエージェントアプリケーションを構築する上で不可欠な機能です。状態管理を慎重に行い、ループ回数に上限を設けることで 안정的なシステムを構築できます。

特にHolySheep AIを活用すれば、DeepSeek V3.2の$0.42/MTokという破格のコストで大規模運用が可能になります。Gemini 2.5 Flashの$2.50/MTokやClaude Sonnet 4.5の$15/MTokと比較しても、85%以上のコスト削減が実現できます。

私も実際にECサイトのカスタマーサービス基盤を構築していますが、HolySheep AIの安定したレイテンシ（<50ms）と柔軟な料金体系に大変満足しています。

👉 HolySheep AI に登録して無料クレジットを獲得