Typed Workflow State

Give AgentFlow variables a Pydantic model or dataclass so typos are caught when the flow is built, not mid-run

Typed workflow state gives AgentFlow.variables a schema, so a misspelled variable is refused when the flow is built instead of dying halfway through a run.

Quick Start

1. Untyped baseline

   Without a model, variables stays the plain dict it has always been and flow.state is None.

   from praisonaiagents import AgentFlow
   
   flow = AgentFlow(steps=[], variables={"anything": 1})
   
   print(flow.state)      # None — no schema declared
   print(flow.variables)  # {'anything': 1}

2. Add a Pydantic model

   Declare a state_model and read fields with typed, dotted access.

   from pydantic import BaseModel
   from praisonaiagents import AgentFlow
   
   class ReviewState(BaseModel):
       draft: str = ""
       score: int = 0
   
   flow = AgentFlow(steps=[], state_model=ReviewState,
                    variables={"draft": "hi", "score": 7})
   
   print(flow.state.draft)  # "hi"
   print(flow.state.score)  # 7

3. Use a plain dataclass

   The same shape works as a @dataclass, so Pydantic is never required.

   import dataclasses
   from praisonaiagents import AgentFlow
   
   @dataclasses.dataclass
   class PlainState:
       draft: str = ""
       score: int = 0
   
   flow = AgentFlow(steps=[], state_model=PlainState, variables={"draft": "hi"})
   
   print(flow.state.draft)  # "hi"

How It Works

Validation runs at construction, so a typo costs nothing to find because the flow never starts.

Catching mistakes at build time

An unknown key is an error, not something to ignore, and the message names the declared fields so the fix is visible.

from pydantic import BaseModel
from praisonaiagents import AgentFlow
from praisonaiagents.workflows.state import WorkflowStateError

class ReviewState(BaseModel):
    draft: str = ""
    score: int = 0

# A misspelled variable is refused
AgentFlow(steps=[], state_model=ReviewState, variables={"reserach_results": 1})

Unknown workflow variable(s) ['reserach_results'] for state model 'ReviewState'. Declared fields: ['draft', 'score']. A misspelled variable would otherwise be written, never read, and never reported.

# A wrongly typed variable is refused
AgentFlow(steps=[], state_model=ReviewState, variables={"score": "not an int"})

Workflow variables do not fit state model 'ReviewState': ValidationError: ...

from praisonaiagents.workflows.state import validate_variables

# A model with no fields cannot describe the state
class Empty:
    pass

validate_variables(Empty, {"a": 1})

'Empty' declares no fields, so it cannot describe this flow's state. Use a Pydantic model or a dataclass.

Re-validating after a step writes

A step that writes into variables can re-check before the next step reads them.

from pydantic import BaseModel
from praisonaiagents import AgentFlow

class ReviewState(BaseModel):
    draft: str = ""
    score: int = 0

flow = AgentFlow(steps=[], state_model=ReviewState, variables={"score": 1})

flow.variables["draft"] = "later"
flow.validate_variables()   # raises if a step wrote an unknown or ill-typed key

When to use which model type

A dataclass works as well as a Pydantic model, so reach for the heavier dependency only when you need it.

API Reference

The public surface is one field, one property, one method, and one error type.

Best Practices

Related