LangGraphは、LLMアプリケーションに状態管理と制御フローを組み込むための強力なフレームワークです。特に複雑な会話型AIを構築する際、条件分支とループ構造は避けて通れないテーマです。私は以前、複数のECサイト向けAIカスタマーサービスを開発しましたが、その際にLangGraphの条件分支機構が大きく活躍しました。本稿では、ECのAIカスタマーサービスを具体例として、動的ルーティングと循環制御フローの設計手法を詳しく解説します。
なぜ条件分支が重要か
実際のAIカスタマーサービスでは、ユーザーの入力内容によって処理流程が全く異なります。例えば、「注文状況を確認したい」「商品をキャンセルしたい」「返金について知りたい」といった多様なインテントに対して、適切な分支処理が必要です。LangGraphでは、この条件分支をStateGraphとConditionalEdgeを用いて直感的に実装できます。
特に注目すべき点是、HolySheep AIのようなマルチモデル対応APIを活用することで、各分支先に最適なモデルを選択できることです。DeepSeek V3.2は$0.42/MTokという低コストながら高い推論能力を持ち、简易な分類タスクには最適です。一方、複雑な推論が必要な場合はClaude Sonnet 4.5（$15/MTok）を選択する柔軟な設計が可能です。
LangGraph 条件分支の実装
プロジェクト構成
まず、ECサイトのAIカスタマーサービスにおける条件分支の全体像を把握しましょう。以下の図のようなフローを実装します：
- ユーザー入力 → 自然言語理解（NLU）
- NLU結果に基づく条件分支（注文確認/キャンセル/返金/商品検索/その他）
- 各分支先で専門処理を実行
- 必要に応じてループ возвращение（追加質問）
- 最終応答生成
前提条件とインストール
# 必要なパッケージのインストール
pip install langgraph langchain-core langchain-holySheep python-dotenv
環境変数の設定
export HOLYSHEEP_API_KEY="YOUR_HOLYSHEEP_API_KEY"
export HOLYSHEEP_BASE_URL="https://api.holysheep.ai/v1"HolySheep AIでは、レートが¥1=$1という圧倒的なコストパフォーマンスを実現しており、公式¥7.3=$1比85%節約可能です。また、WeChat PayやAlipayにも対応しており、個人開発者でも気軽に始められます。登録することで無料クレジットも付与されるため、まずは今すぐ登録してください。
基本的な条件分支の実装
import os
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
from langchain_holySheep import ChatHolySheep
HolySheep AIクライアントの初期化
client = ChatHolySheep(
model="deepseek-chat", # 成本重視のタスクにはDeepSeek V3.2
api_key=os.getenv("HOLYSHEEP_API_KEY"),
base_url="https://api.holysheep.ai/v1"
)
状态的定義
class CustomerServiceState(TypedDict):
user_input: str
intent: str
order_id: str | None
context: dict
response: str
loop_count: int
ノード定義
def nlu_node(state: CustomerServiceState) -> CustomerServiceState:
"""自然言語理解ノード - ユーザー意図を分類"""
prompt = f"""ユーザーの入力を分析し、以下のカテゴリーから適切な意図を1つ選択してください：
- order_status: 注文状況の確認
- cancel_order: 注文のキャンセル
- refund: 返金依頼
- product_search: 商品検索
- other: その他
ユーザー入力: {state['user_input']}
結果のみを1語で返答してください。"""
response = client.invoke([{"role": "user", "content": prompt}])
intent = response.content.strip().lower()
return {
**state,
"intent": intent,
"loop_count": state.get("loop_count", 0) + 1
}
def order_status_node(state: CustomerServiceState) -> CustomerServiceState:
"""注文状況確認ノード"""
prompt = f"""あなたはECサイトの注文状況確認担当です。
ユーザーからの入力: {state['user_input']}
丁寧な口調で注文状況を確認し、必要な場合は注文IDを聞く応答を生成してください。"""
response = client.invoke([{"role": "user", "content": prompt}])
return {**state, "response": response.content}
def cancel_order_node(state: CustomerServiceState) -> CustomerServiceState:
"""注文キャンセルノード"""
prompt = f"""あなたはECサイトの注文キャンセル担当です。
ユーザーからの入力: {state['user_input']}
キャンセルポリシーを確認しつつ、丁寧な口調で処理を進める応答を生成してください。"""
response = client.invoke([{"role": "user", "content": prompt}])
return {**state, "response": response.content}
def refund_node(state: CustomerServiceState) -> CustomerServiceState:
"""返金処理ノード"""
prompt = f"""あなたはECサイトの返金処理担当です。
ユーザーからの入力: {state['user_input']}
返金手続きの案内を丁寧な口調で生成してください。"""
response = client.invoke([{"role": "user", "content": prompt}])
return {**state, "response": response.content}
def product_search_node(state: CustomerServiceState) -> CustomerServiceState:
"""商品検索ノード"""
prompt = f"""あなたはECサイトの商品検索担当です。
ユーザーからの入力: {state['user_input']}
商品検索の案内を生成してください。"""
response = client.invoke([{"role": "user", "content": prompt}])
return {**state, "response": response.content}
def other_node(state: CustomerServiceState) -> CustomerServiceState:
"""汎用対応ノード"""
prompt = f"""あなたはECサイトのカスタマーサービス担当です。
ユーザーからの入力: {state['user_input']}
親しみやすい口調で応答してください。"""
response = client.invoke([{"role": "user", "content": prompt}])
return {**state, "response": response.content}
条件分支関数
def route_based_on_intent(state: CustomerServiceState) -> Literal["order_status", "cancel_order", "refund", "product_search", "other"]:
"""NLU結果を基に次のノードを動的に選択"""
intent = state.get("intent", "other")
intent_mapping = {
"order_status": "order_status",
"cancel_order": "cancel_order",
"refund": "refund",
"product_search": "product_search"
}
return intent_mapping.get(intent, "other")
グラフの構築
def build_customer_service_graph():
graph = StateGraph(CustomerServiceState)
# ノードの追加
graph.add_node("nlu", nlu_node)
graph.add_node("order_status", order_status_node)
graph.add_node("cancel_order", cancel_order_node)
graph.add_node("refund", refund_node)
graph.add_node("product_search", product_search_node)
graph.add_node("other", other_node)
# 始点と終点の設定
graph.set_entry_point("nlu")
graph.add_edge("order_status", END)
graph.add_edge("cancel_order", END)
graph.add_edge("refund", END)
graph.add_node("product_search", product_search_node)
graph.add_edge("other", END)
# 条件分支の追加
graph.add_conditional_edges(
"nlu",
route_based_on_intent,
{
"order_status": "order_status",
"cancel_order": "cancel_order",
"refund": "refund",
"product_search": "product_search",
"other": "other"
}
)
return graph.compile()
使用例
graph = build_customer_service_graph()
initial_state = {
"user_input": "注文したシャツの配送状況を教えて",
"intent": "",
"order_id": None,
"context": {},
"response": "",
"loop_count": 0
}
result = graph.invoke(initial_state)
print(f"検出された意図: {result['intent']}")
print(f"応答: {result['response']}")ループ制御フローの実装
実際のカスタマーサービスでは、1回の対話で解決しない場合があります。例えば、注文IDを聞く→ユーザーがIDを入力→再度注文確認処理を行うといったループ構造が必要です。LangGraphでは、終了条件を設定することでこのループを実装できます。
from typing import Annotated
import operator
class LoopingCustomerServiceState(TypedDict):
user_input: str
messages: list
intent: str
requires_order_id: bool
order_id: str | None
loop_count: int
max_loops: int
def check_clarification_needed(state: LoopingCustomerServiceState) -> str:
"""追加情報の要否を判断"""
intent = state.get("intent", "")
# 注文確認・キャンセル・返金には注文IDが必要
needs_order_id = intent in ["order_status", "cancel_order", "refund"]
if needs_order_id and not state.get("order_id"):
return "request_order_id"
elif state.get("requires_order_id") and not state.get("order_id"):
return "request_order_id"
else:
return "process_request"
def request_order_id_node(state: LoopingCustomerServiceState) -> LoopingCustomerServiceState:
"""注文IDの要求"""
messages = state.get("messages", [])
messages.append({
"role": "assistant",
"content": "恐れ入りますが、注文ID，请您告诉我您的订单编号。您的订单ID可在确认邮件中找到。"
})
return {
**state,
"messages": messages,
"requires_order_id": True,
"loop_count": state.get("loop_count", 0) + 1
}
def extract_order_id_from_input(user_input: str) -> str | None:
"""入力から注文IDを抽出"""
import re
# 注文IDのパターンマッチング（例：ORD-123456）
patterns = [r'ORD-\d+', r'ORDER-\d+', r'[A-Z]{3}-\d{6}']
for pattern in patterns:
match = re.search(pattern, user_input, re.IGNORECASE)
if match:
return match.group()
# 数字のみの場合も抽出
numbers = re.findall(r'\d{6,}', user_input)
if numbers:
return numbers[0]
return None
def process_intent_node(state: LoopingCustomerServiceState) -> LoopingCustomerServiceState:
"""意図に基づく処理実行"""
intent = state.get("intent", "")
order_id = state.get("order_id")
user_input = state.get("user_input", "")
messages = state.get("messages", [])
# HolySheep APIを活用した処理（DeepSeek V3.2使用でコスト削減）
prompt = f"""ECサイトのカスタマーサービス応答を生成してください。
意図: {intent}
注文ID: {order_id or '未提供'}
ユーザー入力: {user_input}
応答は简洁で丁寧に。"""
response = client.invoke([{"role": "user", "content": prompt}])
messages.append({"role": "assistant", "content": response.content})
return {
**state,
"messages": messages
}
def should_continue_loop(state: LoopingCustomerServiceState) -> str:
"""ループ継続の判定"""
max_loops = state.get("max_loops", 5)
loop_count = state.get("loop_count", 0)
# 最大ループ回数に達したら終了
if loop_count >= max_loops:
return "end"
# 注文IDが未取得で、要求状態なら継続
if state.get("requires_order_id") and not state.get("order_id"):
return "continue"
return "end"
def build_looping_graph():
graph = StateGraph(LoopingCustomerServiceState)
graph.add_node("nlu", nlu_node)
graph.add_node("check_clarification", check_clarification_needed)
graph.add_node("request_order_id", request_order_id_node)
graph.add_node("process_intent", process_intent_node)
graph.set_entry_point("nlu")
# 条件分支：NLU後に処理方法を決定
graph.add_conditional_edges(
"nlu",
check_clarification_needed,
{
"request_order_id": "request_order_id",
"process_request": "process_intent"
}
)
# 注文ID要求後のループ
graph.add_edge("request_order_id", "check_clarification")
# 処理完了後のループ判定
graph.add_conditional_edges(
"process_intent",
should_continue_loop,
{
"continue": "nlu",
"end": END
}
)
return graph.compile()
実際の対話シミュレーション
def run_conversation_loop(graph, initial_input: str):
state = {
"user_input": initial_input,
"messages": [],
"intent": "",
"requires_order_id": False,
"order_id": None,
"loop_count": 0,
"max_loops": 5
}
while True:
result = graph.invoke(state)
state = result
# 最後のAssistantメッセージを表示
if result["messages"]:
last_message = result["messages"][-1]
if last_message["role"] == "assistant":
print(f"AI: {last_message['content']}")
# 終了条件のチェック
if state.get("loop_count", 0) >= state.get("max_loops", 5):
break
if not result["messages"]:
break
# 実際のアプリではここでユーザー入力を待つ
# user_input = input("あなた: ")
# if user_input.lower() in ["exit", "quit"]:
# break
# state["user_input"] = user_input
#
# # 注文IDの抽出
# extracted_id = extract_order_id_from_input(user_input)
# if extracted_id:
# state["order_id"] = extracted_id
# state["requires_order_id"] = False
return result
実行例
result = run_conversation_loop(build_looping_graph(), "注文の配送状況を確認したい")企業RAGシステムへの応用
企業向けのRAG（Retrieval-Augmented Generation）システムでも、条件分支は重要な役割を果たします。以下は、部門ごとに異なる知識ベースを参照するシステムの例です。
from typing import Literal
from langgraph.graph import StateGraph, END
from langchain_holySheep import ChatHolySheep
import time
class EnterpriseRAGState(TypedDict):
query: str
department: str # hr, it, sales, legal, general
retrieved_docs: list
answer: str
routing_confidence: float
def route_to_department(state: EnterpriseRAGState) -> Literal["hr_knowledge", "it_knowledge", "sales_knowledge", "legal_knowledge", "general_knowledge"]:
"""クエリ内容から適切な部門の知識ベースをルーティング"""
query = state.get("query", "")
department_keywords = {
"hr_knowledge": ["採用", "給与", "人事", "福利厚生", "休假", "퇴직"],
"it_knowledge": ["システム", "アクセス", "VPN", "パスワード", "ソフトウェア"],
"sales_knowledge": ["売上", "顧客", "契約", "报价", "提案"],
"legal_knowledge": ["法務", "コンプライアンス", "規約", "法的", "契約 条項"]
}
# キーワードマッチングによる部門判定
scores = {}
for dept, keywords in department_keywords.items():
score = sum(1 for kw in keywords if kw in query)
scores[dept] = score
# 最もスコアの高い部門を選択
if max(scores.values()) > 0:
return max(scores, key=scores.get)
return "general_knowledge"
def retrieve_and_answer(state: EnterpriseRAGState, kb_name: str) -> EnterpriseRAGState:
"""知識ベースからの検索と回答生成"""
# HolySheep API呼び出し（レイテンシ <50ms 目标）
start_time = time.time()
prompt = f"""あなたは企業内の{kb_name}担当です。
ユーザーからの質問: {state['query']}
社内の規定に準拠した回答を生成してください。简潔かつ正確である必要はありません。
関連する規定や手順がございましたら、それを含めてください。"""
response = client.invoke([{"role": "user", "content": prompt}])
latency_ms = (time.time() - start_time) * 1000
return {
**state,
"answer": response.content,
"department": kb_name,
"retrieved_docs": [f"{kb_name}_kb"], # 実際の実装ではベクトルDBから取得
"routing_confidence": 0.85
}
def build_enterprise_rag_graph():
graph = StateGraph(EnterpriseRAGState)
# 部門別の処理ノード
departments = ["hr_knowledge", "it_knowledge", "sales_knowledge", "legal_knowledge", "general_knowledge"]
for dept in departments:
graph.add_node(
dept,
lambda s, d=dept: retrieve_and_answer(s, d)
)
graph.set_entry_point("route")
# 条件分支ノード
graph.add_node("route", route_to_department)
# 各部門からENDへのエッジ
for dept in departments:
graph.add_edge(dept, END)
# 条件分支で部門を選択
graph.add_conditional_edges(
"route",
lambda x: x, # route_to_departmentは直接ノード名を返す
{dept: dept for dept in departments}
)
return graph.compile()
企業RAGシステムの特徴
print("""
企業RAGシステムの特徴:
- 部門別の専門知識ベースへの自動ルーティング
- HolySheep AIの<50msレイテンシで快速响应
- DeepSeek V3.2($0.42/MTok)でコスト効率最大化
- 必要に応じてClaude Sonnet 4.5($15/MTok)への切り替え可能
""")個人開発者向けプロジェクトテンプレート
個人開発者がLangGraphを活用したAIサービスを構築する際のプロジェクトテンプレートも紹介します。以下のコードは、最小構成で条件分支とループを実装した例です。
"""
LangGraph 条件分支 プロジェクトテンプレート
HolySheep AI API 활용 - 個人開発者向け
"""
import os
from typing import Literal
from langgraph.graph import StateGraph, END
from langchain_holySheep import ChatHolySheep
from pydantic import BaseModel
HolySheep AI клиент
class HolySheepClient:
def __init__(self):
self.client = ChatHolySheep(
model="deepseek-chat",
api_key=os.getenv("HOLYSHEEP_API_KEY"),
base_url="https://api.holysheep.ai/v1"
)
def chat(self, system: str, user: str) -> str:
response = self.client.invoke([
{"role": "system", "content": system},
{"role": "user", "content": user}
])
return response.content
class AgentState(BaseModel):
messages: list = []
current_step: str = "start"
result: str = ""
loop_count: int = 0
class SimpleRouter:
"""简易条件路由 实现"""
def __init__(self):
self.ai = HolySheepClient()
self.routes = {
"greeting": self.handle_greeting,
"question": self.handle_question,
"command": self.handle_command,
"unknown": self.handle_unknown
}
def classify(self, text: str) -> str:
prompt = f"""以下入力を分類してください：
- greeting: 挨拶
- question: 質問
- command: 命令・依頼
- unknown: 不明
入力: {text}
分類: """
result = self.ai.chat("あなたはテキスト分類器です。", prompt)
return result.strip().lower()
def handle_greeting(self, text: str) -> str:
return self.ai.chat(
"あなたは亲しみやすいAIアシスタントです。",
text
)
def handle_question(self, text: str) -> str:
return self.ai.chat(
"あなたは正確な情報を提供するアシスタントです。",
text
)
def handle_command(self, text: str) -> str:
return self.ai.chat(
"あなたはユーザーの依頼を诚実に実行するアシスタントです。",
text
)
def handle_unknown(self, text: str) -> str:
return "申し訳ありません。もう一度お願いできますか？"
def route(self, text: str) -> str:
intent = self.classify(text)
handler = self.routes.get(intent, self.handle_unknown)
return handler(text)
使用例
if __name__ == "__main__":
router = SimpleRouter()
# テスト
test_inputs = [
"你好、おはようございます",
"今日の天気を教えて",
"レポートを作成して"
]
for