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

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

```mermaid
flowchart TD
    subgraph Python["Python環境 (Google Colab)"]
        PG[pydantic-graph]
    end

    subgraph Process["カレー作りプロセス"]
        Order[注文受付]
        Check[材料確認]
        Cook[調理工程]
        Serve[提供]
    end

    subgraph Management["品質管理システム"]
        QC[品質チェック]
        Time[調理時間管理]
        Taste[味の確認]
    end

    subgraph Improvements["改善要素"]
        Store[店舗別カスタマイズ]
        Spice[辛さ調整]
        Status[進捗管理]
    end

    Python --> Process
    Process --> Management
    Management --> Improvements
    
    Order --> Check
    Check --> Cook
    Cook --> Serve
    
    Cook --> QC
    QC --> Time
    QC --> Taste
    
    Taste --> Spice
    Time --> Status
    Status --> Store

    style Python fill:#f9f,stroke:#333,stroke-width:2px
    style Process fill:#bbf,stroke:#333,stroke-width:2px
    style Management fill:#bfb,stroke:#333,stroke-width:2px
    style Improvements fill:#fbb,stroke:#333,stroke-width:2px
```

## 目次

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

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

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

```python
!pip install pydantic-graph

import nest_asyncio
nest_asyncio.apply()
```

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

## ワークフロー設計

```mermaid
flowchart TD
    classDef process fill:#4263eb,stroke:#364fc7,color:white
    classDef check fill:#40c057,stroke:#2b8a3e,color:white
    classDef spice fill:#f76707,stroke:#d9480f,color:white
    classDef cook fill:#e64980,stroke:#c2255c,color:white
    classDef mild fill:#15aabf,stroke:#0b7285,color:white
    classDef serve fill:#7950f2,stroke:#5f3dc4,color:white

    Start([開始]) --> ReceiveOrder[注文を受ける<br>ReceiveOrder]
    ReceiveOrder --> CheckIngredients[材料を確認する<br>CheckIngredients]
    CheckIngredients --> CheckSpiciness{辛さを確認<br>CheckSpiciness}
    CheckSpiciness -->|辛い| CookSpicy[辛いカレーを調理<br>CookSpicyCurry]
    CheckSpiciness -->|普通| CookMild[マイルドカレーを調理<br>CookMildCurry]
    CookSpicy --> ServeCurry[カレーを提供する<br>ServeCurry]
    CookMild --> ServeCurry
    ServeCurry --> End([終了])

    class ReceiveOrder process
    class CheckIngredients check
    class CheckSpiciness spice
    class CookSpicy cook
    class CookMild mild
    class ServeCurry serve

    subgraph ワークフロー説明
        direction TB
        Note1[各ノードはpydantic-graphの<br>BaseNodeを継承したクラス]
        Note2[辛さに応じて調理プロセスが分岐]
    end
```

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

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

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

```python
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)
```

## グラフを実行

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/20032eeb-27eb-8d46-0818-1308253e5595.png)

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

```python
# グラフの実行と進行状況の管理
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()
```

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

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

## 辛さ選択

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/d6ed4f05-e01c-1cc3-af45-fffaa0f9f0fb.png)

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

```python
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)
```

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

```python
# ワークフローの再構築
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: 料理の調理時間や温度調整

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

```python
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: 品質管理 - 味見と調整

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

```python
@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: 複数店舗ごとのオーダー処理

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

```python
@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 を利用して状態を保存し、ノード間でデータを引き継ぐことができます。

```python
@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ダイアグラムとして可視化できます。以下のコードで、カレー作りのフローを可視化できます。

```python
# グラフのMermaidコードを生成
order_graph.mermaid_code(start_node=ReceiveOrder)
```
```mermaid
---
title: curry_graph
---
stateDiagram-v2
  [*] --> ReceiveOrder
  ReceiveOrder --> CheckIngredients
  CheckIngredients --> CheckSpiciness
  CheckSpiciness --> CookSpicyCurry
  CheckSpiciness --> CookMildCurry
  CookSpicyCurry --> ServeCurry
  CookMildCurry --> ServeCurry
  ServeCurry --> [*]
```
## 参考情報

https://ai.pydantic.dev/graph/#graph-types

## まとめ

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/aa9800da-8f52-7b85-4ab3-d7c14c813530.png)

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

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

## その後の記事
https://qiita.com/Tadataka_Takahashi/items/cb26cf4229cb2f3ef721

## 実装例

```json
{
 "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
}
```

![FireShot Capture 001 - curry-ipynb.json - Colab - colab.research.google.com.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/ab415882-4758-a962-fbd5-16412d737236.png)