Google ColabでPydantic-Graphを使って「カレー作り」を自動化する

Last updated at 2025-01-20Posted at 2025-01-18

はじめに

この記事では、Pythonの pydantic-graph ライブラリを使用して、カレー作りのプロセスを自動化する方法を紹介します。カレー作りは飲食店でよく行われる業務プロセスの一例であり、その流れをグラフで管理することで、業務の効率化や品質管理の向上を図ることができます。Google Colab上で実行し、実際の業務シナリオに即した改善案を加えながら進めます。

Pydantic-Graphのインストールとセットアップ
カレー作りのワークフロー設計
カレー作りのグラフを実行
追加機能: 辛さを選択する
改善案: 料理の品質管理と店舗ごとの調整
進行状況を確認する: ログと状態の追跡
Mermaidダイアグラムでグラフを可視化
まとめ

インストールとセットアップ

まず最初に、Google Colabに pydantic-graph ライブラリをインストールします。以下のコマンドを実行してください。

!pip install pydantic-graph

import nest_asyncio
nest_asyncio.apply()

これで、必要なライブラリのインストールと初期設定が完了しました。nest_asyncioは非同期処理の実行に必要で、環境変数の設定はAPIキーの管理のために使用します。これらの準備が整ったので、カレー作りのプロセスをグラフで管理していきます。

ワークフロー設計

カレー作りのプロセスを次のようなノードで設計します：

注文を受ける (ReceiveOrder)
材料を確認する (CheckIngredients)
カレーを調理する (CookCurry)
カレーを提供する (ServeCurry)

これらを順番に実行することで、カレー作りの流れを自動化できます。

from dataclasses import dataclass
from pydantic_graph import BaseNode, GraphRunContext, End

@dataclass
class ReceiveOrder(BaseNode[None]):
    order_details: str

    async def run(self, ctx: GraphRunContext[None]) -> 'CheckIngredients':
        print(f"注文を受けました: {self.order_details}")
        return CheckIngredients()

@dataclass
class CheckIngredients(BaseNode[None]):
    async def run(self, ctx: GraphRunContext[None]) -> 'CheckSpiciness':
        print("材料を確認しました")
        return CheckSpiciness(spice_level="普通")  # 辛さを指定する部分

@dataclass
class CheckSpiciness(BaseNode[None]):
    spice_level: str

    async def run(self, ctx: GraphRunContext[None]) -> Union['CookSpicyCurry', 'CookMildCurry']:
        print(f"辛さレベル: {self.spice_level}")
        if self.spice_level == "辛い":
            return CookSpicyCurry()
        else:
            return CookMildCurry()

@dataclass
class CookSpicyCurry(BaseNode[None]):
    async def run(self, ctx: GraphRunContext[None]) -> 'ServeCurry':
        print("辛いカレーを調理中です")
        return ServeCurry()  # ServeCurryに遷移

@dataclass
class CookMildCurry(BaseNode[None]):
    async def run(self, ctx: GraphRunContext[None]) -> 'ServeCurry':
        print("マイルドカレーを調理中です")
        return ServeCurry()  # ServeCurryに遷移

@dataclass
class ServeCurry(BaseNode[None]):
    async def run(self, ctx: GraphRunContext[None]) -> End:
        print("カレーが完成しました！")  # ここでカレーが完成したことを出力
        return End(None)

グラフを実行

次に、定義したノードを使って、カレー作りのグラフを実行します。pydantic-graph の Graph クラスを使用して、全体のワークフローを動かします。

# グラフの実行と進行状況の管理
from pydantic_graph import Graph

order_graph = Graph(nodes=[ReceiveOrder, CheckIngredients, CheckSpiciness, CookSpicyCurry, CookMildCurry, ServeCurry])

async def main():
    order_details = "チキンカレー、辛さ普通"
    state = None
    await order_graph.run(ReceiveOrder(order_details=order_details), state=state)

await main()

このコードを実行すると、以下のように出力されます：

注文を受けました: チキンカレー、辛さ普通
材料を確認しました
辛さレベル: 普通
マイルドカレーを調理中です
カレーが完成しました！

辛さ選択

次に、ユーザーがカレーの辛さを選べるようにします。注文時に「辛い」か「普通」かを選択し、それに応じて調理するプロセスを分岐させます。

from dataclasses import dataclass
from pydantic_graph import BaseNode, GraphRunContext, End
from typing import Union

@dataclass
class CheckSpiciness(BaseNode[None]):
    spice_level: str

    async def run(self, ctx: GraphRunContext[None]) -> Union['CookSpicyCurry', 'CookMildCurry']:
        print(f"辛さレベル: {self.spice_level}")
        if self.spice_level == "辛い":
            return CookSpicyCurry()
        else:
            return CookMildCurry()

@dataclass
class CookSpicyCurry(BaseNode[None]):
    async def run(self, ctx: GraphRunContext[None]) -> End:
        print("辛いカレーを調理中です")
        return End(None)

@dataclass
class CookMildCurry(BaseNode[None]):
    async def run(self, ctx: GraphRunContext[None]) -> End:
        print("マイルドカレーを調理中です")
        return End(None)

@dataclass
class ServeCurry(BaseNode[None]):
    async def run(self, ctx: GraphRunContext[None]) -> End:
        print("カレーが完成しました！")
        return End(None)

これにより、注文時に辛さを選択することができ、その選択に応じて調理プロセスが変わります。

# ワークフローの再構築
from pydantic_graph import Graph

spicy_order_graph = Graph(nodes=[ReceiveOrder, CheckIngredients, CheckSpiciness, CookSpicyCurry, CookMildCurry, ServeCurry])

async def main():
    order_details = "チキンカレー、辛さ普通"
    spice_level = "普通"  # または "辛い"
    state = None
    await spicy_order_graph.run(ReceiveOrder(order_details=order_details), state=state)

await main()

改善案: 料理の品質管理と店舗ごとの調整

実際の飲食店では、カレーの調理時間や品質管理が重要です。また、複数の店舗がある場合、各店舗の特徴に応じたオーダー処理が必要です。以下の改善案を反映させることで、業務の効率化や品質向上が期待できます。

改善案1: 料理の調理時間や温度調整

飲食店の厨房では、カレーの種類ごとに調理時間や温度が異なります。これを自動化することで、厨房のオペレーションが効率化されます。

from dataclasses import dataclass
from pydantic_graph import BaseNode, GraphRunContext, End
from types import SimpleNamespace

@dataclass
class CookSpicyCurry(BaseNode[None]):
    async def run(self, ctx: GraphRunContext[None]) -> End:
        print("辛いカレーを調理中です")
        # ステートの初期化
        if ctx.state is None:
            ctx.state = SimpleNamespace()
        # 辛いカレーは長時間調理する
        ctx.state.cooking_time = "60分"
        return End(None)

@dataclass
class CookMildCurry(BaseNode[None]):
    async def run(self, ctx: GraphRunContext[None]) -> End:
        print("マイルドカレーを調理中です")
        # ステートの初期化
        if ctx.state is None:
            ctx.state = SimpleNamespace()
        # マイルドカレーは短時間調理
        ctx.state.cooking_time = "30分"
        return End(None)

改善案2: 品質管理 - 味見と調整

カレーが完成した後に「味見」を行い、必要であれば辛さを調整するプロセスを追加します。これにより、一定の品質を保つことができます。

@dataclass
class TasteTest(BaseNode[None]):
    async def run(self, ctx: GraphRunContext[None]) -> Union['AdjustSpiciness', 'ServeCurry']:
        print("味見をしています")
        if ctx.state.spiciness_level == "辛すぎる":
            return AdjustSpiciness(spice_level="マイルド")
        else:
            return ServeCurry()

@dataclass
class AdjustSpiciness(BaseNode[None]):
    spice_level: str

    async def run(self, ctx: GraphRunContext[None]) -> 'ServeCurry':
        print(f"辛さを{self.spice_level}に調整しました")
        return ServeCurry()

改善案3: 複数店舗ごとのオーダー処理

店舗ごとの特色に応じて、オーダー処理をカスタマイズすることができます。例えば、「辛いカレー専門店」と「マイルドカレー専門店」で異なるプロセスを実行します。

@dataclass
class StoreOrderProcessing(BaseNode[None]):
    store_name: str

    async def run(self, ctx: GraphRunContext[None]) -> 'ServeCurry':
        # ステートの初期化
        if ctx.state is None:
            ctx.state = SimpleNamespace()
            
        # 店舗情報をステートに保存
        ctx.state.store_name = self.store_name
        print(f"{self.store_name} 店舗でオーダー処理中")

        # 店舗タイプに応じて処理を分岐
        if self.store_name == "辛口カレー専門店":
            return CookSpicyCurry()
        else:
            return CookMildCurry()

進行状況を確認する: ログと状態の追跡

実行中のデータの状態を追跡するために、ctx.state を利用して状態を保存し、ノード間でデータを引き継ぐことができます。

@dataclass
class MachineState:
    order_details: str
    ingredients_checked: bool = False
    curry_cooked: bool = False

@dataclass
class ReceiveOrder(BaseNode[MachineState]):
    order_details: str

    async def run(self, ctx: GraphRunContext[MachineState]) -> 'CheckIngredients':
        print(f"注文を受けました: {self.order_details}")
        ctx.state.order_details = self.order_details
        return CheckIngredients()

可視化: Mermaidダイアグラム

pydantic-graph では、作成したグラフの構造をMermaidダイアグラムとして可視化できます。以下のコードで、カレー作りのフローを可視化できます。

# グラフのMermaidコードを生成
order_graph.mermaid_code(start_node=ReceiveOrder)

参考情報

まとめ

この記事では、Google Colab上で pydantic-graph を使い、カレー作りのプロセスを自動化するワークフローを作成しました。実際にコードを実行して、注文からカレー提供までの一連の流れをシミュレーションできるようになっています。また、進行状況を追跡したり、Mermaid ダイアグラムを使ってグラフを可視化する方法も紹介しました。

業務フローの自動化に役立つ方法を学んでいただき、ぜひ自分のユースケースに合わせてカスタマイズしてみてください。

その後の記事

実装例

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# カレー作り自動化システム\n",
    "\n",
    "このノートブックでは、Pythonのpydantic-graphライブラリを使用して、カレー作りのプロセスを自動化する方法を実装します。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1. 環境設定\n",
    "\n",
    "最初に必要なライブラリをインストールし、環境を設定します。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!pip install pydantic-graph\n",
    "\n",
    "import nest_asyncio\n",
    "nest_asyncio.apply()\n",
    "\n"   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. 基本的なクラスの定義\n",
    "\n",
    "必要なクラスとノードを定義します。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dataclasses import dataclass\n",
    "from pydantic_graph import BaseNode, GraphRunContext, End\n",
    "from typing import Union\n",
    "\n",
    "@dataclass\n",
    "class ReceiveOrder(BaseNode[None]):\n",
    "    order_details: str\n",
    "\n",
    "    async def run(self, ctx: GraphRunContext[None]) -> 'CheckIngredients':\n",
    "        print(f\"注文を受けました: {self.order_details}\")\n",
    "        return CheckIngredients()\n",
    "\n",
    "@dataclass\n",
    "class CheckIngredients(BaseNode[None]):\n",
    "    async def run(self, ctx: GraphRunContext[None]) -> 'CheckSpiciness':\n",
    "        print(\"材料を確認しました\")\n",
    "        return CheckSpiciness(spice_level=\"普通\")\n",
    "\n",
    "@dataclass\n",
    "class CheckSpiciness(BaseNode[None]):\n",
    "    spice_level: str\n",
    "\n",
    "    async def run(self, ctx: GraphRunContext[None]) -> Union['CookSpicyCurry', 'CookMildCurry']:\n",
    "        print(f\"辛さレベル: {self.spice_level}\")\n",
    "        if self.spice_level == \"辛い\":\n",
    "            return CookSpicyCurry()\n",
    "        else:\n",
    "            return CookMildCurry()\n",
    "\n",
    "@dataclass\n",
    "class CookSpicyCurry(BaseNode[None]):\n",
    "    async def run(self, ctx: GraphRunContext[None]) -> 'ServeCurry':\n",
    "        print(\"辛いカレーを調理中です\")\n",
    "        ctx.state = {\"cooking_time\": \"60分\"}\n",
    "        return ServeCurry()\n",
    "\n",
    "@dataclass\n",
    "class CookMildCurry(BaseNode[None]):\n",
    "    async def run(self, ctx: GraphRunContext[None]) -> 'ServeCurry':\n",
    "        print(\"マイルドカレーを調理中です\")\n",
    "        ctx.state = {\"cooking_time\": \"30分\"}\n",
    "        return ServeCurry()\n",
    "\n",
    "@dataclass\n",
    "class ServeCurry(BaseNode[None]):\n",
    "    async def run(self, ctx: GraphRunContext[None]) -> End:\n",
    "        print(\"カレーが完成しました！\")\n",
    "        if hasattr(ctx, 'state') and ctx.state:\n",
    "            print(f\"調理時間: {ctx.state.get('cooking_time', '不明')}\")\n",
    "        return End(None)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3. グラフの作成と実行\n",
    "\n",
    "定義したノードを使ってグラフを作成し、実行します。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pydantic_graph import Graph\n",
    "\n",
    "# グラフの作成\n",
    "curry_graph = Graph(nodes=[\n",
    "    ReceiveOrder,\n",
    "    CheckIngredients,\n",
    "    CheckSpiciness,\n",
    "    CookSpicyCurry,\n",
    "    CookMildCurry,\n",
    "    ServeCurry\n",
    "])\n",
    "\n",
    "# 通常の注文を実行\n",
    "async def order_normal_curry():\n",
    "    print(\"=== 普通の辛さのカレーを注文 ===\")\n",
    "    await curry_graph.run(ReceiveOrder(order_details=\"チキンカレー、辛さ普通\"))\n",
    "\n",
    "# 辛いカレーの注文を実行\n",
    "async def order_spicy_curry():\n",
    "    print(\"\\n=== 辛いカレーを注文 ===\")\n",
    "    spicy_start = ReceiveOrder(order_details=\"チキンカレー、辛い\")\n",
    "    await curry_graph.run(spicy_start)\n",
    "\n",
    "# 実行\n",
    "await order_normal_curry()\n",
    "await order_spicy_curry()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 4. Mermaidダイアグラムの生成\n",
    "\n",
    "処理フローを視覚化します。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# グラフの可視化\n",
    "print(curry_graph.mermaid_code(start_node=ReceiveOrder))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 5. 注文処理のテスト\n",
    "\n",
    "異なる注文パターンをテストします。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "async def test_orders():\n",
    "    print(\"=== カレー注文のテスト ===\")\n",
    "    \n",
    "    # テストケース1: 普通のカレー\n",
    "    print(\"\\nテストケース1: 普通のカレー\")\n",
    "    await curry_graph.run(ReceiveOrder(order_details=\"チキンカレー、普通\"))\n",
    "    \n",
    "    # テストケース2: 辛いカレー\n",
    "    print(\"\\nテストケース2: 辛いカレー\")\n",
    "    await curry_graph.run(ReceiveOrder(order_details=\"ビーフカレー、辛い\"))\n",
    "    \n",
    "    # テストケース3: 特別な指示付き\n",
    "    print(\"\\nテストケース3: 特別な指示付き\")\n",
    "    await curry_graph.run(ReceiveOrder(order_details=\"野菜カレー、普通、ナンとライス大盛り\"))\n",
    "\n",
    "await test_orders()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up