演習：対話型推薦エージェントを構築しよう

from typing import TypedDict, Optional, Annotated
from operator import add
from langchain_core.tools import tool

class RecommendationState(TypedDict):
    messages: Annotated[list, add]
    intent: Optional[str]
    user_id: Optional[str]
    preferences: Optional[dict]
    recommended_items: Optional[list]
    reasons: Optional[list]
    error: Optional[str]

@tool
def parse_preference(query: str) -> dict:
    """自然言語からユーザーの嗜好条件を抽出する"""
    from langchain_openai import ChatOpenAI
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    prompt = f"以下の発言から商品の嗜好条件をJSON形式で抽出: {query}"
    response = llm.invoke(prompt)
    return {"preferences": response.content}

@tool
def recommend_items(user_id: str, preferences: str, n_items: int = 5) -> list:
    """推薦モデルで候補を生成する"""
    # 推薦モデルの呼び出し（ここではダミー）
    items = [
        {"item_id": f"item_{i}", "name": f"商品{i}", "score": 0.9 - i*0.1}
        for i in range(n_items)
    ]
    return items

@tool
def generate_reason(item_name: str, user_preferences: str) -> str:
    """推薦理由を生成する"""
    return f"{item_name}はご希望の条件に合った人気アイテムです"

from langgraph.graph import StateGraph, END

def classify_intent(state: RecommendationState) -> dict:
    """意図を分類"""
    last_message = state['messages'][-1].content
    if any(w in last_message for w in ['おすすめ', '推薦', '欲しい']):
        return {'intent': 'recommend'}
    elif any(w in last_message for w in ['絞り込', 'もっと', '別の']):
        return {'intent': 'filter'}
    return {'intent': 'recommend'}

def generate_recommendations(state: RecommendationState) -> dict:
    """推薦を生成"""
    preferences = parse_preference.invoke(
        {"query": state['messages'][-1].content}
    )
    items = recommend_items.invoke({
        "user_id": state.get('user_id', 'anonymous'),
        "preferences": str(preferences),
    })
    reasons = [
        generate_reason.invoke({
            "item_name": item['name'],
            "user_preferences": str(preferences),
        })
        for item in items[:3]
    ]
    return {
        'recommended_items': items,
        'reasons': reasons,
        'preferences': preferences,
    }

def respond(state: RecommendationState) -> dict:
    """応答を生成"""
    items = state.get('recommended_items', [])
    reasons = state.get('reasons', [])
    response = "おすすめの商品をご紹介します:\n"
    for i, item in enumerate(items[:3]):
        reason = reasons[i] if i < len(reasons) else ""
        response += f"\n{i+1}. {item['name']}（スコア: {item['score']:.1f}）\n   理由: {reason}\n"
    return {'messages': [AIMessage(content=response)]}

# グラフ構築
graph = StateGraph(RecommendationState)
graph.add_node("classify_intent", classify_intent)
graph.add_node("generate_recommendations", generate_recommendations)
graph.add_node("respond", respond)
graph.set_entry_point("classify_intent")
graph.add_edge("classify_intent", "generate_recommendations")
graph.add_edge("generate_recommendations", "respond")
graph.add_edge("respond", END)
app = graph.compile()

=== テストシナリオ ===

シナリオ1: 初回推薦
ユーザー: 「3万円以内でカジュアルなワンピースが欲しい」
期待: カテゴリ=ワンピース、スタイル=カジュアル、予算=30000円以内の推薦

シナリオ2: 絞り込み
ユーザー: 「もう少し明るい色はある？」
期待: 前回結果から色=明るい系でフィルタリング

シナリオ3: 比較
ユーザー: 「1番目と3番目を比較して」
期待: 2商品の比較表と推奨

=== A/Bテスト計画書 ===

■ 仮説
対話型推薦エージェント（B群）は、人気ランキング（A群）に比べてCTRが10%以上高い

■ 指標
主指標: CTR（クリック率）
副指標: CVR、セッション滞在時間、推薦からの購入率

■ サンプルサイズ
ベースラインCTR: 3%、最小検出効果: 10%相対改善
必要サンプル: 各群約120,000インプレッション

■ 期間
2週間（曜日効果の平均化）

■ 意思決定基準
p < 0.05 かつ CTR改善率 > 5%: 全面展開
p < 0.05 かつ CTR改善率 <= 5%: 追加改善後に再テスト
p >= 0.05: テスト延長または仮説見直し