Pydantic Graph

from dataclasses import dataclass, field
from pydantic_graph import BaseNode, End, Graph, GraphRunContext
 
@dataclass
class CounterState:
    count: int = 0
 
@dataclass
class Increment(BaseNode[CounterState]):
    """状態のカウンタを 1 増やして次のノードへ。"""
 
    async def run(self, ctx: GraphRunContext[CounterState]) -> 'CheckDone':
        ctx.state.count += 1
        return CheckDone()
 
@dataclass
class CheckDone(BaseNode[CounterState, None, int]):
    """3 まで来たら終了、未満ならまた Increment へ。"""
 
    async def run(self, ctx: GraphRunContext[CounterState]) -> Increment | End[int]:
        if ctx.state.count >= 3:
            return End(ctx.state.count)
        return Increment()
 
graph = Graph(nodes=[Increment, CheckDone])
 
async def main():
    result = await graph.run(Increment(), state=CounterState())
    print(f'最終カウント: {result.output}')
 
# import asyncio; asyncio.run(main())

from dataclasses import dataclass, field
from pydantic_ai import Agent
from pydantic_ai.models.google import GoogleModel
from pydantic_graph import BaseNode, End, Graph, GraphRunContext
 
model = GoogleModel('gemini-3-flash-preview')
writer = Agent(model, instructions='与えられたお題で短い日本語の見出しを 1 つ作ってください。')
critic = Agent(model, instructions='与えられた見出しを「OK」か「NG」で 1 単語だけ返してください。')
 
@dataclass
class WriteState:
    topic: str
    headline: str = ''
    review: str = ''
    iteration: int = 0
 
@dataclass
class Write(BaseNode[WriteState]):
    async def run(self, ctx: GraphRunContext[WriteState]) -> 'Review':
        r = await writer.run(ctx.state.topic)
        ctx.state.headline = r.output
        ctx.state.iteration += 1
        return Review()
 
@dataclass
class Review(BaseNode[WriteState, None, str]):
    async def run(self, ctx: GraphRunContext[WriteState]) -> Write | End[str]:
        r = await critic.run(ctx.state.headline)
        ctx.state.review = r.output.strip()
        if ctx.state.review.startswith('OK') or ctx.state.iteration >= 3:
            return End(ctx.state.headline)
        return Write()  # NG なら再生成
 
graph = Graph(nodes=[Write, Review])
 
async def main():
    state = WriteState(topic='生成 AI のセキュリティ')
    result = await graph.run(Write(), state=state)
    print(f'採用見出し: {result.output} (試行 {state.iteration} 回)')

from dataclasses import dataclass
from pydantic_graph import BaseNode, End, Graph, GraphRunContext
 
@dataclass
class GuessState:
    target: int = 7
    last_guess: int | None = None
    attempts: int = 0
 
@dataclass
class AskGuess(BaseNode[GuessState]):
    async def run(self, ctx: GraphRunContext[GuessState]) -> 'CheckGuess':
        ctx.state.attempts += 1
        # 実際は input() / Web フォーム入力等
        return CheckGuess()
 
@dataclass
class CheckGuess(BaseNode[GuessState, None, str]):
    async def run(self, ctx: GraphRunContext[GuessState]) -> AskGuess | End[str]:
        if ctx.state.last_guess == ctx.state.target:
            return End(f'{ctx.state.attempts} 回で正解!')
        return AskGuess()
 
graph = Graph(nodes=[AskGuess, CheckGuess])
 
async def play():
    state = GuessState()
    async with graph.iter(AskGuess(), state=state) as run:
        async for node in run:
            if isinstance(node, AskGuess):
                state.last_guess = int(input('数を入力: '))
            if isinstance(node, End):
                print(node.data)
                break

from pathlib import Path
from pydantic_graph.persistence.file import FileStatePersistence
 
persistence = FileStatePersistence(Path('graph_state.json'))
await graph.initialize(start, state=state, persistence=persistence)
 
async with graph.iter_from_persistence(persistence) as run:
    node = await run.next()
    if isinstance(node, End):
        print('完了:', node.data)