はじめに
この記事では、Pythonの pydantic-graph を活用して、カレー作りの工程に 「AIによる辛さ・トッピングの提案」 を組み込む方法を紹介します。ユーザがAIの提案をそのまま採用するか、あるいは独自カスタマイズするかをフラグで切り替え、最終的にカレーを完成させる一連のフローを グラフ構造 で管理します。
AIには pydantic-ai を用い、「推奨トッピング」「推奨辛さ」を提案させるノードと、「味見フィードバック」を行うノードを定義。ノード間の遷移を可視化しやすくし、柔軟にカスタマイズ可能なシステムを実装します。
実験的な取り組みについての注意事項
pydantic-aiのβ版状態について
本記事で使用している pydantic-ai は2025年1月現在、β版の段階にあります。そのため：
- APIの仕様が変更される可能性があります
- 一部の機能が実験的な実装である可能性があります
- 本番環境での使用には十分な検証が必要です
前作からの展開について
前回の記事では、カレー調理プロセスをグラフ構造で管理する方法に焦点を当て、pydantic-graphを用いた実装を詳しく解説しました。今回は、そのグラフ構造に「AIによる提案機能」を組み込み、カスタマイズ可能な調理システムへと発展させています。
目次
- なぜAIがトッピングや辛さを提案するのか
- システム構成概要
- pydantic-graphによるフロー定義
- pydantic-aiによるAI連携のポイント
- 実装例のコード
- まとめ
1. なぜAIがトッピングや辛さを提案するのか
メニューのバリエーション向上
飲食店では、お客様の好みに合わせてカレーを少しずつカスタマイズする需要があります。AIから独自のトッピングや辛さを提案させることで、標準メニューにはない新しい味の選択肢を素早く提示できます。
接客効率の向上
AIが自動でおすすめを出してくれるため、スタッフの負荷を軽減しつつ、ユーザへの提案力を高めることができます。
トレンド対応や在庫活用
トッピングに季節の食材や在庫の多い材料を勧めれば、フードロス軽減や期間限定メニューとの相性提案なども視野に入ります。
2. システム構成概要
今回の例では、以下のようなフローを 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 クラスに登録することで「開始ノード → 次のノード → … → 終了ノード」のプロセスをわかりやすく管理できます。
本記事の例では以下のようにノードを定義しています:
受注～トッピング提案 (AIノード) → ユーザ確認 → 調理 → 味見(AIノード) → 再調整 or 提供
という流れが可視化されます。
4. pydantic-aiによるAI連携のポイント
AIRecommendation モデル
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 での実装例:
!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())
6. まとめ
- ユーザが注文したカレーに対して、AIが「辛さ」「トッピング」を提案するノードを作ることで、調理フローにオリジナルなおすすめ要素を組み込める。
- 「AIの提案をそのまま使う」か「独自カスタマイズする」かをフラグで分岐し、ユーザの自由度を保ちながらオペレーションを自動化できる。
- 味見チェックや再調整のノードを加えることで、辛さや具材をさらなる品質向上へ導くことも可能。
- pydantic-graph のノード＋ pydantic-ai のエージェントを組み合わせると、作業フローにAI提案を自然に挟み込み、ログ可視化や再実行時の管理が容易になる。
実際の飲食店業務シナリオでは、在庫確認やコスト計算などのノードも追加すると、より大規模な「AIアシスト付きカレー調理システム」に発展させられます。試しに自分のユースケースを拡張しながら、AI×業務フローの可能性を探ってみてください。