asyncio.gatherでLLMパイプラインに「タダで」分類機能を追加する

# 逐次実行（素朴な実装）
kb_query = await generate_kb_query(messages)          # ~100ms
problem_ctx = await classify_problem(raw_query)       # ~100ms
chunks = await retrieve_from_kb(kb_query, top_k=5)    # ~500ms
# 合計: ~700ms（回答生成前のオーバーヘッド）

async def _chat_stream(request: ChatRequest) -> AsyncGenerator[str]:
    is_first_turn = not any(m.role == "assistant" for m in request.messages)

    if is_first_turn:
        # Haiku 2並列: クエリ生成 + 問題分類
        kb_query, problem_ctx = await asyncio.gather(
            generate_kb_query(request.messages),
            classify_problem(request.messages[0].content),
        )
        # KB検索はクエリ生成の結果に依存するため、ここで逐次実行
        chunks = await retrieve_from_kb(kb_query, top_k=request.top_k)
        # ※ 以降、chunksとproblem_ctxを使ってstreaming回答を生成（yield）

    else:
        # 2ターン目: クエリ生成は会話コンテキストが必要なため先に実行
        kb_query = await generate_kb_query(request.messages)

        # KB検索（I/Oバウンド）と分類（CPUバウンド）を並列化
        chunks, problem_ctx = await asyncio.gather(
            retrieve_from_kb(kb_query, top_k=request.top_k),
            classify_problem(kb_query),
        )

from pydantic import BaseModel, Field

class ProblemClassification(BaseModel):
    """問題ドメインとカテゴリの分類結果"""
    software: str | None = Field(
        None,
        description="対象ソフトウェア名（例: Exchange Online, SAP）"
    )
    category: str | None = Field(
        None,
        description=(
            "問題カテゴリ: 認証・ログイン / 権限・アクセス / "
            "インストール・更新 / パフォーマンス / データ・ファイル / "
            "設定・構成 / その他"
        ),
    )

async def classify_problem(query: str) -> ProblemClassification:
    """問題ドメインとカテゴリを分類する。
    asyncio.gatherでKB検索と並列実行されるため、レイテンシ増加はゼロ。
    """
    tool_schema = ProblemClassification.model_json_schema()

    try:
        response = await _client.messages.create(
            model=settings.bedrock_haiku_model_id,  # Haiku 4.5
            messages=[{"role": "user", "content": query}],
            system=(
                "サービスデスクの問い合わせを分析し、"
                "対象ソフトウェアと問題カテゴリを特定してください。"
                "不明な場合はnullを返してください。"
            ),
            max_tokens=100,
            temperature=0.0,
            tools=[{
                "name": "classify_problem",
                "description": "問い合わせの対象ソフトウェアと問題カテゴリを分類する",
                "input_schema": tool_schema,
            }],
            tool_choice={"type": "tool", "name": "classify_problem"},
        )
        for block in response.content:
            if block.type == "tool_use" and block.name == "classify_problem":
                return ProblemClassification(**block.input)
        return ProblemClassification(software=None, category=None)
    except Exception:
        logger.warning("classify_problem failed silently", exc_info=True)
        return ProblemClassification(software=None, category=None)

def _build_domain_context(
    ticket_sw_map: dict[int, str],
    problem_ctx: ProblemClassification,
) -> str:
    """ドメインコンテキストを構築してSonnetのプロンプトに注入する"""
    # KBチャンクから抽出したソフトウェア情報を優先
    sw_counts = Counter(ticket_sw_map.values())
    kb_software = [sw for sw, _ in sw_counts.most_common(2)]
    systems = kb_software or (
        [problem_ctx.software] if problem_ctx.software else []
    )

    return (
        f"\n\n## ドメインコンテキスト（システム自動分析）\n"
        f"- 対象システム: {'、'.join(systems) or '不明'}\n"
        f"- 問題カテゴリ: {problem_ctx.category or '不明'}"
    )

# KBチャンクから正規表現でソフトウェア名を抽出（実データに基づく）
ticket_sw_map = _extract_software_from_chunks(chunks)

# Haiku分類（クエリテキストのみから推論）
_ctx_sw = problem_ctx.software

# KB証拠を優先、なければHaiku分類にフォールバック
_kb_sw = [
    sw for sw, _ in Counter(ticket_sw_map.values()).most_common()
    if sw not in _GENERIC_SW
]
dominant_sw = _kb_sw[0] if _kb_sw else _ctx_sw

async def retrieve_from_kb(query: str, top_k: int = 5) -> list[dict]:
    def _blocking() -> dict:
        client = get_opensearch_client()
        return client.search(index=index_name, body=search_body)

    # 同期関数をスレッドプールで実行し、イベントループをブロックしない
    response = await asyncio.to_thread(_blocking)
    return _parse_response(response)

# デフォルト: 1つでも例外が出たらgather全体が例外を投げる
results = await asyncio.gather(task_a(), task_b())

# return_exceptions=True: 例外もresultsに含める
results = await asyncio.gather(task_a(), task_b(), return_exceptions=True)

// Python: await asyncio.gather(task_a(), task_b())
const [kbQuery, problemCtx] = await Promise.all([
  generateKbQuery(messages),
  classifyProblem(rawQuery),
])