AIが提案する「トッピング＆辛さ」のフローをpydantic-graphで実装する

## はじめに

この記事では、Pythonの pydantic-graph を活用して、カレー作りの工程に 「AIによる辛さ・トッピングの提案」 を組み込む方法を紹介します。ユーザがAIの提案をそのまま採用するか、あるいは独自カスタマイズするかをフラグで切り替え、最終的にカレーを完成させる一連のフローを グラフ構造 で管理します。

AIには pydantic-ai を用い、「推奨トッピング」「推奨辛さ」を提案させるノードと、「味見フィードバック」を行うノードを定義。ノード間の遷移を可視化しやすくし、柔軟にカスタマイズ可能なシステムを実装します。

![curry-flow.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/c646cb55-722b-7be1-cae8-9e7ab1b72ffc.png)

# 実験的な取り組みについての注意事項

## pydantic-aiのβ版状態について

本記事で使用している `pydantic-ai` は2025年1月現在、β版の段階にあります。そのため：

- APIの仕様が変更される可能性があります
- 一部の機能が実験的な実装である可能性があります
- 本番環境での使用には十分な検証が必要です


# 前作からの展開について

前回の記事では、カレー調理プロセスをグラフ構造で管理する方法に焦点を当て、pydantic-graphを用いた実装を詳しく解説しました。今回は、そのグラフ構造に「AIによる提案機能」を組み込み、カスタマイズ可能な調理システムへと発展させています。

https://qiita.com/Tadataka_Takahashi/items/d3b3e616c1b9ae7a5a16

## 目次
- なぜAIがトッピングや辛さを提案するのか
- システム構成概要
- pydantic-graphによるフロー定義
- pydantic-aiによるAI連携のポイント
- 実装例のコード
- まとめ

## 1. なぜAIがトッピングや辛さを提案するのか

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/20d355a5-460c-6684-cf86-42dd1494c266.png)

### メニューのバリエーション向上
飲食店では、お客様の好みに合わせてカレーを少しずつカスタマイズする需要があります。AIから独自のトッピングや辛さを提案させることで、標準メニューにはない新しい味の選択肢を素早く提示できます。

### 接客効率の向上
AIが自動でおすすめを出してくれるため、スタッフの負荷を軽減しつつ、ユーザへの提案力を高めることができます。

### トレンド対応や在庫活用
トッピングに季節の食材や在庫の多い材料を勧めれば、フードロス軽減や期間限定メニューとの相性提案なども視野に入ります。

## 2. システム構成概要

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/198ef0a5-7e9e-0c2e-4f08-137734537c89.png)

今回の例では、以下のようなフローを pydantic-graph のノードで実装します。

### 注文受付 (ReceiveOrder)
- ユーザが基本のカレーを注文する（例: "ビーフカレー"）。
- 「AIの提案を採用するか」の真偽値もここで持つ。

### AIによるトッピング・辛さの提案 (SuggestToppingsAndSpiceWithAI)
- AIが「辛さレベル」「おすすめのトッピング」を考えてくれる。

### ユーザがAI提案を採用するか確認 (ConfirmOrOverride)
- ユーザが「そのままAI提案を使う」か「独自に上書きしたい」かを選ぶ。
- AI提案を採用しない場合は、OverrideChoiceノードで手動設定する。

### カレーを調理 (CookCurry)
- 最終的に決まった辛さ・トッピングをもとに調理を行う（ログ出力のみ）。

### AIによる味見評価 (CheckTasteWithAI)
- 「OK」か「追加でレシピを調整すべきか」を判断する。
- 必要なら AdjustRecipe ノードで再調整後、再度 CookCurry に戻りループさせる。

### カレー提供 (ServeCurry)
- 調整も含めて満足のいく状態になったらお客様に提供して終了。

## 3. pydantic-graphによるフロー定義

pydantic-graphは、各工程を BaseNode で定義し、Graph クラスに登録することで「開始ノード → 次のノード → … → 終了ノード」のプロセスをわかりやすく管理できます。

本記事の例では以下のようにノードを定義しています:

```mermaid
flowchart TD
    Start(ReceiveOrder) --> Suggest
    Suggest --> Confirm
    Confirm -->|AI提案を採用| Cook
    Confirm -->|独自カスタム| Override
    Override --> Cook
    Cook --> Taste
    Taste -->|adjust_recipe| Adjust
    Taste -->|ok| Serve
    Adjust --> Cook
    Serve --> End
```

受注～トッピング提案 (AIノード) → ユーザ確認 → 調理 → 味見(AIノード) → 再調整 or 提供
という流れが可視化されます。

## 4. pydantic-aiによるAI連携のポイント

### AIRecommendation モデル

```python
class AIRecommendation(BaseModel):
    recommended_spiciness: str
    recommended_toppings: List[str]
```

「推奨辛さ」や「トッピングリスト」などを構造化データとして保持し、ノード間でやり取りできます。

### taste_agent
味見チェック用AI。辛さとトッピングを渡して評価コメント (comment) と、アクション (action="ok"または"adjust_recipe") を返します。

### ユーザの選択フラグ
CurryState 内に user_follow_ai_recommendation: bool を持ち、ConfirmOrOverride ノードで分岐することで、AIの提案をそのまま使うかどうかを簡単に制御できます。

## 5. 実装例のコード

以下がすべてをまとめたサンプルです。1セルですべて定義して、Cell末尾の if __name__ == "__main__": ... で実行する構成にしています。

google colab での実装例:

```python
!pip install --quiet pydantic-graph pydantic-ai nest_asyncio

from google.colab import userdata
import os

# 環境変数から読み込み (Colab左の歯車「環境変数の設定」で入力してもOK)
os.environ["OPENAI_API_KEY"] = userdata.get('OPENAI_API_KEY')

import nest_asyncio
nest_asyncio.apply()

import asyncio
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Union, List

from pydantic import BaseModel
from pydantic_graph import BaseNode, Graph, GraphRunContext, End
from pydantic_ai import Agent
from pydantic_ai.format_as_xml import format_as_xml
from pydantic_ai.messages import ModelMessage

# ============== 状態モデル ==============
@dataclass
class CurryState:
    order_details: str
    user_follow_ai_recommendation: bool = True
    spiciness: str = "普通"
    toppings: List[str] = field(default_factory=list)
    ai_messages: list[ModelMessage] = field(default_factory=list)
    adjustments: str = ""
    final_comment: str = ""

# ============== AI出力モデル ==============
class AIRecommendation(BaseModel):
    recommended_spiciness: str
    recommended_toppings: List[str]

class TasteFeedback(BaseModel):
    comment: str
    action: str  # "ok" or "adjust_recipe"

# ============== AIエージェント定義 ==============
recommend_agent = Agent[None, AIRecommendation](
    "openai:gpt-4o",
    result_type=AIRecommendation,
    system_prompt=(
        "You are a creative culinary AI. The user has ordered a curry. "
        "Propose 1) a recommended spiciness level, 2) up to 3 toppings."
    ),
)

taste_agent = Agent[None, TasteFeedback](
    "openai:gpt-4o",
    result_type=TasteFeedback,
    system_prompt=(
        "You are an AI cooking assistant. Given spiciness and toppings, "
        "give a short comment and action='ok' or 'adjust_recipe'."
    ),
)

# ============== ノード定義 ==============
@dataclass
class ReceiveOrder(BaseNode[CurryState]):
    async def run(self, ctx: GraphRunContext[CurryState]) -> "SuggestToppingsAndSpiceWithAI":
        print(f"【注文受付】注文: {ctx.state.order_details}")
        print(f"  (AI提案を採用するか: {ctx.state.user_follow_ai_recommendation})")
        return SuggestToppingsAndSpiceWithAI()

@dataclass
class SuggestToppingsAndSpiceWithAI(BaseNode[CurryState, None, AIRecommendation]):
    async def run(self, ctx: GraphRunContext[CurryState]) -> "ConfirmOrOverride":
        prompt_data = {"order_details": ctx.state.order_details}
        prompt = format_as_xml(prompt_data)

        result = await recommend_agent.run(prompt, message_history=ctx.state.ai_messages)
        ctx.state.ai_messages += result.all_messages()

        rec = result.data
        print("\n【AIの提案】")
        print(f"  推奨辛さ: {rec.recommended_spiciness}")
        print(f"  推奨トッピング: {', '.join(rec.recommended_toppings)}")

        # AI提案をひとまず state に保存
        ctx.state.spiciness = rec.recommended_spiciness
        ctx.state.toppings = rec.recommended_toppings
        return ConfirmOrOverride()

@dataclass
class ConfirmOrOverride(BaseNode[CurryState]):
    async def run(self, ctx: GraphRunContext[CurryState]) -> Union["OverrideChoice", "CookCurry"]:
        if ctx.state.user_follow_ai_recommendation:
            print("\n【選択】ユーザはAIの提案を採用します")
            return CookCurry()
        else:
            print("\n【選択】ユーザは独自の辛さ / トッピングを選択します")
            return OverrideChoice()

@dataclass
class OverrideChoice(BaseNode[CurryState]):
    async def run(self, ctx: GraphRunContext[CurryState]) -> "CookCurry":
        print("【独自カスタマイズ】辛さ=激辛、トッピング=[揚げナス, ゆで卵]")
        ctx.state.spiciness = "激辛"
        ctx.state.toppings = ["揚げナス", "ゆで卵"]
        return CookCurry()

@dataclass
class CookCurry(BaseNode[CurryState]):
    async def run(self, ctx: GraphRunContext[CurryState]) -> "CheckTasteWithAI":
        print(f"\n【調理】{ctx.state.spiciness} カレーを調理しました")
        if ctx.state.toppings:
            print(f"  トッピング: {', '.join(ctx.state.toppings)}")
        else:
            print("  トッピングなし")
        return CheckTasteWithAI()

@dataclass
class CheckTasteWithAI(BaseNode[CurryState, None, TasteFeedback]):
    async def run(self, ctx: GraphRunContext[CurryState]) -> Union["AdjustRecipe", "ServeCurry"]:
        prompt_data = {
            "spiciness": ctx.state.spiciness,
            "toppings": ctx.state.toppings
        }
        prompt = format_as_xml(prompt_data)

        result = await taste_agent.run(prompt, message_history=ctx.state.ai_messages)
        ctx.state.ai_messages += result.all_messages()

        feedback = result.data
        print("\n【AI味見フィードバック】")
        print(f"  comment: {feedback.comment}")
        print(f"  action: {feedback.action}")

        if feedback.action == "adjust_recipe":
            return AdjustRecipe(feedback=feedback.comment)
        else:
            return ServeCurry(ai_comment=feedback.comment)

@dataclass
class AdjustRecipe(BaseNode[CurryState]):
    feedback: str
    async def run(self, ctx: GraphRunContext[CurryState]) -> "CookCurry":
        print("\n【レシピ再調整】")
        print(f"  AIの提案: {self.feedback}")
        # デモ: 適当に辛さやトッピングを少し上げる
        ctx.state.spiciness = "さらに激辛"
        ctx.state.adjustments += f"- {self.feedback}\n"
        return CookCurry()

@dataclass
class ServeCurry(BaseNode[CurryState]):
    ai_comment: str
    async def run(self, ctx: GraphRunContext[CurryState]) -> End[CurryState]:
        print("\n【カレー提供】お客様にカレーを提供しました！")
        print(f"  AIの最終コメント: {self.ai_comment}")
        ctx.state.final_comment = self.ai_comment
        return End(ctx.state)

# ============== グラフ定義 ==============
curry_graph = Graph(nodes=[
    ReceiveOrder,
    SuggestToppingsAndSpiceWithAI,
    ConfirmOrOverride,
    OverrideChoice,
    CookCurry,
    CheckTasteWithAI,
    AdjustRecipe,
    ServeCurry
])

# ============== 実行関数 ==============
async def run_curry_with_ai():
    init_state = CurryState(order_details="ビーフカレー（ややスパイス強めを希望）", user_follow_ai_recommendation=True)
    final_result, _ = await curry_graph.run(ReceiveOrder(), state=init_state)

    print("\n=== フロー完了: 最終状態 ===")
    print(f"・最終辛さ: {final_result.spiciness}")
    print(f"・最終トッピング: {final_result.toppings}")
    if final_result.adjustments:
        print(f"・再調整内容:\n{final_result.adjustments}")

    print("\n=== AIとのメッセージログ ===")
    for i, msg in enumerate(final_result.ai_messages):
        print(f"[Message {i}] {msg}")

if __name__ == "__main__":
    asyncio.run(run_curry_with_ai())
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/b9465e42-27af-e04d-93b6-f0a5c4f5f56d.png)


## 6. まとめ

![summary-diagram.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/0c9552e0-1df7-b98b-acc2-10ffc914093e.png)

- ユーザが注文したカレーに対して、AIが「辛さ」「トッピング」を提案するノードを作ることで、調理フローにオリジナルなおすすめ要素を組み込める。
- 「AIの提案をそのまま使う」か「独自カスタマイズする」かをフラグで分岐し、ユーザの自由度を保ちながらオペレーションを自動化できる。
- 味見チェックや再調整のノードを加えることで、辛さや具材をさらなる品質向上へ導くことも可能。
- pydantic-graph のノード＋ pydantic-ai のエージェントを組み合わせると、作業フローにAI提案を自然に挟み込み、ログ可視化や再実行時の管理が容易になる。

実際の飲食店業務シナリオでは、在庫確認やコスト計算などのノードも追加すると、より大規模な「AIアシスト付きカレー調理システム」に発展させられます。試しに自分のユースケースを拡張しながら、AI×業務フローの可能性を探ってみてください。