Fix Flow initialization with Pydantic models having required fields

[devin-ai-integration]

Fix Flow initialization with Pydantic models having required fields

Summary

Fixes issue #3744 where Flow initialization fails with ValidationError when using Pydantic BaseModel state classes that have required (non-optional) fields.

Root cause: The _create_initial_state() method was instantiating Pydantic models without any arguments (state_type()), causing validation to fail for required fields. The kwargs passed to Flow.__init__() were only applied afterwards via _initialize_state(), which was too late for Pydantic's validation.

Solution: Modified _create_initial_state() to accept and use kwargs during state instantiation, allowing required fields to be provided upfront and pass Pydantic validation.

Changes made:

1. Updated _create_initial_state(self, kwargs: dict[str, Any] | None = None) to accept kwargs parameter
2. Pass kwargs when instantiating all state types (BaseModel, FlowState, dict)
3. Updated Flow.__init__() to pass kwargs directly to _create_initial_state()
4. Removed redundant _initialize_state(kwargs) call from __init__ since kwargs are now handled during state creation
5. Added 8 comprehensive tests covering required fields, mixed fields, FlowState subclasses, and flow execution

Backwards compatibility:

• All existing tests pass (26 in test_flow.py, 8 in test_flow_default_override.py and test_flow_persistence.py)
• Flows without required fields work exactly as before (kwargs defaults to empty dict)
• The _initialize_state() method remains unchanged and continues to be used by kickoff_async() for runtime state updates

Review & Testing Checklist for Human

• [CRITICAL] Test with a real Flow using required Pydantic fields - Create a flow like the example in issue [FEATURE] Flow initialization with Pydantic state models having required fields #3744 and verify it can be initialized with MyFlow(name="John", age=30)
• Verify backwards compatibility - Ensure existing flows without required fields still work (e.g., flows with only optional fields or dict states)
• Check all state initialization patterns - Test flows using: plain dict, BaseModel with id field, FlowState subclass, and BaseModel without id field
• Verify runtime state updates - Test that flow.kickoff(inputs={...}) still properly updates state during execution

Test Plan

# Test 1: Required fields (should now work)
from pydantic import BaseModel
from crewai import Flow, start

class UserState(BaseModel):
    name: str
    age: int

class UserFlow(Flow[UserState]):
    @start()
    def greet(self):
        return f"Hello {self.state.name}, age {self.state.age}"

flow = UserFlow(name="Alice", age=30)  # Should work now
result = flow.kickoff()  # Should execute successfully

# Test 2: Optional fields (backwards compatibility)
class OptionalState(BaseModel):
    message: str = "default"

class OptionalFlow(Flow[OptionalState]):
    @start()
    def run(self):
        return self.state.message

flow = OptionalFlow()  # Should still work

Notes

• Session: https://app.devin.ai/sessions/0135146e75754b429b71968b48e56b78
• Requested by: João (joao@crewai.com)
• This is a breaking change fix - it fixes behavior that was previously broken, but doesn't introduce new breaking changes
• The core change is minimal (passing kwargs earlier in the initialization flow) but touches critical infrastructure code

[devin-ai-integration]

🤖 Devin AI Engineer

I'll be helping with this pull request! Here's what you should know:

✅ I will automatically:

• Address comments on this PR. Add '(aside)' to your comment to have me ignore it.
• Look at CI failures and help fix them

Note: I can only respond to comments from users who have write access to this repository.

⚙️ Control Options:

• Disable automatic comment and CI monitoring