Issue Station
Sign In
Tools hereToolbox

crewai - ✅(Solved) Fix [BUG] Checkpointing + guardrail retry incompatible with Pydantic outputs in Flow [4 pull requests]

Official PRs (…)
ON THIS PAGE

Recommended Tools

×6

Utilities matched from this issue’s tags and category — try them while you read without losing context.

GitHub issue graph ai analysis

Paste a GitHub issue URL. We fetch that issue, discover linked issues from bodies/comments/timeline, collect linked pull requests, and produce a structured English report.

The report is written in English Markdown for sharing and archival.

Helpful · Quick feedback

Loading…

When using CrewAI Flows with Pydantic state and Crew tasks using output_pydantic, two issues occur:

  1. Checkpointing fails when the Flow state contains Pydantic models.
  2. Guardrail retry breaks when a task uses output_pydantic, causing a TaskOutput type mismatch during retry.

These issues make it difficult to reliably use structured outputs in Flows with guardrails enabled.

Error Message

PydanticSerializationError: Unable to serialize unknown type: <class 'pydantic._internal._model_construction.ModelMetaclass'>

Root Cause

When using CrewAI Flows with Pydantic state and Crew tasks using output_pydantic, two issues occur:

  1. Checkpointing fails when the Flow state contains Pydantic models.
  2. Guardrail retry breaks when a task uses output_pydantic, causing a TaskOutput type mismatch during retry.

These issues make it difficult to reliably use structured outputs in Flows with guardrails enabled.

Fix Action

Fixed

PR fix notes

PR #5545: fix(flow,task): handle pydantic outputs in guardrail retries and checkpoint serialization

Description (problem / solution / changelog)

Summary

Fixes #5544 by addressing two related issues when using Pydantic outputs in flows/tasks:

  1. Guardrail retries could rebuild TaskOutput with a non-string raw value when the model returned a BaseModel.
  2. Flow checkpoint serialization could fail with PydanticSerializationError when initial_state was assigned as a model class on flow subclasses.

What Changed

  • Unified TaskOutput construction in lib/crewai/src/crewai/task.py so initial execution and guardrail retry paths use the same output-shaping logic.
  • Updated sync and async guardrail retry code paths to use that shared builder, preserving expected raw/pydantic semantics for structured outputs.
  • Ensured Flow.initial_state remains excluded from serialized flow entities and preserved inherited field metadata when flow subclasses override parent fields in FlowMeta (lib/crewai/src/crewai/flow/flow.py).
  • Added focused regression tests:
    • test_guardrail_retry_preserves_pydantic_output (lib/crewai/tests/test_task.py)
    • test_ainvoke_guardrail_retry_preserves_pydantic_output (lib/crewai/tests/task/test_async_task.py)
    • test_flow_checkpoint_handles_nested_pydantic_models (lib/crewai/tests/test_checkpoint.py)

Why

  • Guardrail retries should behave identically to first-pass execution for typed outputs.
  • Checkpoint serialization should snapshot runtime state, not class-valued configuration artifacts.

Validation

Passed targeted suites for the touched behavior:

  • uv run pytest lib/crewai/tests/test_checkpoint.py -q
  • uv run pytest lib/crewai/tests/test_task.py -q -k 'guardrail or output_pydantic'
  • uv run pytest lib/crewai/tests/task/test_async_task.py -q -k 'guardrail'

Also confirmed new regression tests pass individually after formatting.

Notes

  • Full repo CI-equivalent runs in this local environment show broad unrelated failures due to missing optional dependencies/providers (for example a2a, provider SDK extras, qdrant, oxylabs) and interactive tool prompts in some tool tests.
  • No unrelated code changes are included in this patch scope.

Changed files

  • lib/crewai/src/crewai/flow/flow.py (modified, +17/-2)
  • lib/crewai/src/crewai/task.py (modified, +45/-90)
  • lib/crewai/tests/task/test_async_task.py (modified, +45/-1)
  • lib/crewai/tests/test_checkpoint.py (modified, +35/-1)
  • lib/crewai/tests/test_task.py (modified, +49/-0)

PR #5546: fix: checkpointing with Pydantic state and guardrail retries with output_pydantic (#5544)

Description (problem / solution / changelog)

Summary

Fixes the two regressions reported in #5544.

Issue I — Checkpointing with Pydantic state explodes on Task.output_pydantic. When a Flow whose state holds Pydantic models drives a Crew whose Tasks use output_pydantic, the auto-checkpoint serializer raises:

PydanticSerializationError: Unable to serialize unknown type:
<class 'pydantic._internal._model_construction.ModelMetaclass'>

Root cause is twofold:

  1. Task.output_json, Task.output_pydantic, Task.response_model and Task.converter_cls are typed as type[BaseModel] | None (i.e. they hold class references). Pydantic's default JSON serializer cannot turn a class into JSON.
  2. EventNode.event was serialized with PlainSerializer(lambda v: v.model_dump()) — note the missing mode="json". So even after the Task fields were made JSON-safe, RuntimeState.model_dump(mode="json") would walk into EventRecord → EventNode.event → TaskCompletedEvent.task (typed Any) → raw Task, dump it in default mode, and re-introduce a raw type[BaseModel] into the output that json.dumps then choked on.

Fix:

  • Add _serialize_class_ref / _validate_class_ref to task.py and wrap the four class-reference fields with Annotated[..., BeforeValidator, PlainSerializer(..., when_used="json")] so they round-trip as module.qualname strings (mirrors the existing pattern used by BaseAgent._serialize_llm_ref). Resolution degrades gracefully to None if the dotted path can't be imported (e.g. checkpoint produced in a different process), since user code re-instantiates the Task with the right class anyway.
  • Replace EventNode.event's lambda PlainSerializer with a @field_serializer that propagates info.mode == "json" into the nested event dump. Without this, JSON-mode dumping is silently dropped at the event boundary and any non-JSON-native nested value re-enters the output.

Issue II — Guardrail retry blows up when the task uses output_pydantic. When output_pydantic is set, agent.execute_task may return a BaseModel instance rather than a string. The guardrail retry path in _invoke_guardrail_function and _ainvoke_guardrail_function was constructing TaskOutput(raw=result, ...) directly, but TaskOutput.raw is typed str, so Pydantic raised ValidationError and aborted the retry — even though the LLM's structured response was valid.

Fix: extracted Task._normalize_agent_result which detects a BaseModel result, JSON-encodes it for raw, and assigns it to pydantic / json_dict based on the task configuration. Both retry sites now go through it.

Tests:

  • tests/test_checkpoint.py::TestPydanticTypeFieldSerialization — covers the dotted-path serialize, round-trip restore, graceful unknown-path restore, and an end-to-end regression that builds a RuntimeState containing a TaskCompletedEvent whose payload carries a Task with output_pydantic and asserts the full state.model_dump(mode="json") + json.dumps succeeds.
  • tests/test_task_guardrails.py::test_guardrail_retry_with_pydantic_agent_result — agent returns a BaseModel on first call, guardrail rejects it, agent returns a different BaseModel on retry, and the final TaskOutput has raw as JSON string and pydantic as the model.

Review & Testing Checklist for Human

  • Confirm the dotted-path serialization for Task.output_pydantic etc. is acceptable for checkpoint storage — the round-trip resolves the class via importlib.import_module, which works for module-level classes but returns None for closures/dynamically-defined classes (current tests cover this graceful degradation explicitly).
  • Run a real end-to-end Flow checkpoint with the user's reported scenario (Pydantic state + Crew with output_pydantic) and verify both the auto-save and the resume work.
  • Sanity-check that EventNode.event field-serializer change doesn't alter the on-disk shape for non-JSON dumps (only the JSON-mode branch was changed; default model_dump() still calls value.model_dump()).
  • Verify guardrail retry with output_pydantic produces the expected TaskOutput.pydantic model (vs. only being JSON-string in raw).

Notes

  • The 8 SQLite-related failures in tests/test_checkpoint.py::TestSqliteProviderFork and tests/test_checkpoint_cli.py are pre-existing on main (the local SQLite build lacks jsonb()) and are unrelated to this change.

Link to Devin session: https://app.devin.ai/sessions/77bc8af71d604726b751ae6e4db252fa

<!-- CURSOR_SUMMARY -->

[!NOTE] Medium Risk Touches checkpoint/state serialization and guardrail retry logic; changes are localized but could affect persistence compatibility and retry behavior if serialization/resolution differs across environments.

Overview Fixes checkpoint JSON serialization when Task fields contain class references (e.g. output_pydantic/output_json/response_model/converter_cls). These fields now serialize to dotted module.qualname strings in JSON mode and attempt to resolve back on restore, while EventNode.event now uses a field_serializer that preserves JSON mode when dumping nested events.

Fixes guardrail retry paths to handle agents returning a Pydantic BaseModel by normalizing results into (raw_json_str, pydantic, json_dict) before building TaskOutput, preventing retry-time validation errors. Adds regression tests covering round-trip checkpointing with embedded tasks in events and guardrail retries with Pydantic model outputs.

<sup>Reviewed by Cursor Bugbot for commit 316cf8c648d8257548c6331d6f242a351e612366. Bugbot is set up for automated code reviews on this repo. Configure here.</sup>

<!-- /CURSOR_SUMMARY -->

Changed files

  • lib/crewai/src/crewai/state/event_record.py (modified, +23/-2)
  • lib/crewai/src/crewai/task.py (modified, +98/-8)
  • lib/crewai/tests/test_checkpoint.py (modified, +107/-0)
  • lib/crewai/tests/test_task_guardrails.py (modified, +56/-0)

PR #5557: fix: handle BaseModel result in guardrail retry loop

Description (problem / solution / changelog)

Summary

  • Fixes guardrail retry failing with ValidationError when a task uses output_pydantic and a guardrail returns (False, feedback)
  • The retry path was passing a Pydantic object directly to TaskOutput.raw (expects str), now mirrors the isinstance(result, BaseModel) check from the initial execution path
  • Applied to both sync and async retry loops

Closes #5544 (part 1 — guardrail retry issue)

Test plan

  • Task with output_pydantic + guardrail that fails once should retry and succeed
  • Existing guardrail tests continue to pass
<!-- CURSOR_SUMMARY -->

[!NOTE] Low Risk Low risk bugfix limited to guardrail retry paths; main risk is subtle behavioral change in what TaskOutput.raw contains when an agent returns a Pydantic model during retries.

Overview Fixes guardrail retry loops (sync and async) to correctly handle agent results that are Pydantic BaseModels.

During retries, TaskOutput.raw is now always a string (using model_dump_json() for BaseModel results), while TaskOutput.pydantic/json_dict are populated consistently based on output_pydantic/output_json, preventing validation errors when regenerating outputs.

<sup>Reviewed by Cursor Bugbot for commit e239416ded68bed36b1307f7cdfec5318dbb91ee. Bugbot is set up for automated code reviews on this repo. Configure here.</sup>

<!-- /CURSOR_SUMMARY -->

Changed files

  • lib/crewai/src/crewai/task.py (modified, +32/-4)

PR #5559: fix: serialize Task class-reference fields for checkpointing

Description (problem / solution / changelog)

Summary

  • Fixes PydanticSerializationError: Unable to serialize unknown type: ModelMetaclass during checkpoint writes
  • Task fields output_pydantic, output_json, response_model, and converter_cls store type[BaseModel] class references that Pydantic can't JSON-serialize
  • Adds PlainSerializer (→ model_json_schema()) and BeforeValidator (→ create_model_from_schema()) using existing pydantic_schema_utils infrastructure

Closes #5544 (part 2 — checkpointing issue)

Test plan

  • Flow with Pydantic model state + checkpointing enabled should serialize without error
  • Checkpoint restore correctly hydrates model class fields
  • Existing task tests continue to pass
<!-- CURSOR_SUMMARY -->

[!NOTE] Medium Risk Changes how Task serializes/deserializes several schema/model-related fields, which can affect checkpoint compatibility and any consumers expecting raw class references. Risk is moderate since it’s limited to JSON serialization behavior and uses existing schema-hydration utilities.

Overview Fixes checkpoint JSON serialization for Task fields that store class references (e.g., output_json, output_pydantic, response_model, converter_cls) by introducing explicit Pydantic PlainSerializer/BeforeValidator handling.

These fields now serialize to Pydantic JSON Schema via model_json_schema() and can be re-hydrated from stored schema dicts using create_model_from_schema, avoiding ModelMetaclass serialization errors during checkpoint writes.

<sup>Reviewed by Cursor Bugbot for commit a4919db7e1ca3521fa18813e1704fb927108dd59. Bugbot is set up for automated code reviews on this repo. Configure here.</sup>

<!-- /CURSOR_SUMMARY -->

Changed files

  • lib/crewai/src/crewai/task.py (modified, +45/-4)

Code Example

PydanticSerializationError: Unable to serialize unknown type:
<class 'pydantic._internal._model_construction.ModelMetaclass'>

---

ValidationError for TaskOutput
raw
Input should be a valid string
input_value=FamilyList(...)

---

generator:
  role: "Family generator"
  goal: "Generate a list of families with structured data"
  backstory: "Expert in generating synthetic family datasets"

---

generation_task:
  description: >
    Generate EXACTLY ONE FAMILY with {sizes} members.
    Each family must include: id, name, size, members
    Each member must include: id, name, role, age, sports
    Output must be valid JSON matching the expected structure.
  expected_output: >
    A FamilyList containing exactly one Family with {sizes} members.
  agent: generator

---

from crewai import Agent, Crew, Process, Task
from crewai.agents.agent_builder.base_agent import BaseAgent
from crewai.project import CrewBase, agent, crew, task

from sporty_family_flow.models.family import FamilyList
from sporty_family_flow.guardrails.family_guardrails import is_family_guardrail


@CrewBase
class FamilyCrew:
    """Family Crew"""

    agents: list[BaseAgent]
    tasks: list[Task]

    agents_config = "config/agents.yaml"
    tasks_config = "config/tasks.yaml"

    @agent
    def generator(self) -> Agent:
        return Agent(
            config=self.agents_config["generator"]
        )

    @task
    def generation_task(self) -> Task:
        return Task(
            config=self.tasks_config["generation_task"],
            output_pydantic=FamilyList,
            guardrails=[is_family_guardrail],
        )

    @crew
    def crew(self) -> Crew:
        return Crew(
            agents=self.agents,  # Automatically created by the @agent decorator
            tasks=self.tasks,  # Automatically created by the @task decorator
            process=Process.sequential,
        )

---

from sporty_family_flow.models.family import FamilyList

def is_family_guardrail(output):
    families = output.pydantic.families
    for family in families:
        if family.size <= 1:
            return (False, f"Family {family.name} has an invalid size: {family.size}. Size must be greater than 1. Add members to this family.")
    return (True, output)

---

from pydantic import BaseModel
from typing import List


class Member(BaseModel):
    member_id: int
    name: str
    role: str
    age: int
    sports: List[str]


class Family(BaseModel):
    family_id: int
    name: str
    size: int
    members: List[Member]


class FamilyList(BaseModel):
    families: List[Family]

---

from pydantic import BaseModel
from crewai.flow import Flow, listen, start
from crewai import CheckpointConfig
from crewai.state import JsonProvider

import sporty_family_flow.listeners
from sporty_family_flow.crews.family_crew.family_crew import FamilyCrew
from sporty_family_flow.models.family import Family, FamilyList


class FamilyState(BaseModel):
    families: list[Family] = []


class FamilyFlow(Flow[FamilyState]):

    @start()
    def get_input(self):
        return(input("Enter the sizes of your families: "))

    @listen(get_input)
    def create_families(self, sizes: str):
        result = (FamilyCrew().crew().kickoff(inputs={"sizes": sizes}))
        families = FamilyList.model_validate_json(result.raw).families
        self.state.families = families

    @listen(create_families)
    def print_families(self):
        for family in self.state.families:
            log = f"""
            ----- Family {family.family_id} ----- \n\n
            Family Name: {family.name}\n
            Size: {family.size}\n
            MEMBERS: \n
            """
            for member in family.members:
                log += f"""
                \t Member: {member.name} \n
                \t Age: {member.age} \n
                \t Role: {member.role} \n
                \t Sports: {', '.join(member.sports)}
                """
            print(log)


def kickoff():
    cp_config = CheckpointConfig(
        location="./.checkpoints",
        on_events=["method_execution_finished"],
        provider=JsonProvider(),
        max_checkpoints=5
    )
    family_flow = FamilyFlow(checkpoint=cp_config)
    family_flow.kickoff()


def plot():
    FamilyFlow().plot()

if __name__ == "__main__":
    kickoff()

---

Auto-checkpoint failed for event method_execution_finished
pydantic_core._pydantic_core.PydanticSerializationError: Unable to serialize unknown type: <class 'pydantic._internal._model_construction.ModelMetaclass'>

---

pydantic_core._pydantic_core.ValidationError: 1 validation error for TaskOutput
raw
  Input should be a valid string [type=string_type, input_value=FamilyList(families=[Fami...occer', 'Swimming'])])]), input_type=FamilyList]
    For further information visit https://errors.pydantic.dev/2.11/v/string_type
An error occurred while running the flow: Command '['uv', 'run', 'kickoff']' returned non-zero exit status 1.

---

expected_output = f"""
You MUST return valid JSON matching this schema:

{json.dumps(FamilyList.model_json_schema(), indent=2)}

Do not include explanations or markdown.
"""
RAW_BUFFERClick to expand / collapse

Description

When using CrewAI Flows with Pydantic state and Crew tasks using output_pydantic, two issues occur:

  1. Checkpointing fails when the Flow state contains Pydantic models.
  2. Guardrail retry breaks when a task uses output_pydantic, causing a TaskOutput type mismatch during retry.

These issues make it difficult to reliably use structured outputs in Flows with guardrails enabled.

Steps to Reproduce

I - Checkpointing issue

  1. Define a Flow state using a Pydantic model
  2. Enable checkpointing with CheckpointConfig
  3. Run a flow with multiple executions

II - Guardrail retry issue

  1. Define a task with output_pydantic
  2. Add a guardrail that may return (False, feedback)
  3. Trigger a guardrail failure
  4. Observe retry execution

Expected behavior

Expected Behavior

I - Checkpointing

Flow state containing Pydantic models should be serialized correctly via state.model_dump(mode="json").

II - Guardrail retry

When a guardrail fails:

  • Feedback should be injected into the next LLM call
  • Task should retry successfully using structured output

Actual behavior

I - Checkpointing failure

PydanticSerializationError: Unable to serialize unknown type:
<class 'pydantic._internal._model_construction.ModelMetaclass'>

Checkpointing fails when trying to serialize Flow state containing Pydantic models.

II - Guardrail retry failure

When a guardrail fails once, retry breaks with:

ValidationError for TaskOutput
raw
Input should be a valid string
input_value=FamilyList(...)

The retry mechanism passes a Pydantic object where a string is expected.

Screenshots/Code snippets

Agents (agents.yaml)

generator:
  role: "Family generator"
  goal: "Generate a list of families with structured data"
  backstory: "Expert in generating synthetic family datasets"

Tasks (tasks.yaml)

generation_task:
  description: >
    Generate EXACTLY ONE FAMILY with {sizes} members.
    Each family must include: id, name, size, members
    Each member must include: id, name, role, age, sports
    Output must be valid JSON matching the expected structure.
  expected_output: >
    A FamilyList containing exactly one Family with {sizes} members.
  agent: generator

Crew (family_crew.py)

from crewai import Agent, Crew, Process, Task
from crewai.agents.agent_builder.base_agent import BaseAgent
from crewai.project import CrewBase, agent, crew, task

from sporty_family_flow.models.family import FamilyList
from sporty_family_flow.guardrails.family_guardrails import is_family_guardrail


@CrewBase
class FamilyCrew:
    """Family Crew"""

    agents: list[BaseAgent]
    tasks: list[Task]

    agents_config = "config/agents.yaml"
    tasks_config = "config/tasks.yaml"

    @agent
    def generator(self) -> Agent:
        return Agent(
            config=self.agents_config["generator"]
        )

    @task
    def generation_task(self) -> Task:
        return Task(
            config=self.tasks_config["generation_task"],
            output_pydantic=FamilyList,
            guardrails=[is_family_guardrail],
        )

    @crew
    def crew(self) -> Crew:
        return Crew(
            agents=self.agents,  # Automatically created by the @agent decorator
            tasks=self.tasks,  # Automatically created by the @task decorator
            process=Process.sequential,
        )

Guardrail (family_guardrails.py)

from sporty_family_flow.models.family import FamilyList

def is_family_guardrail(output):
    families = output.pydantic.families
    for family in families:
        if family.size <= 1:
            return (False, f"Family {family.name} has an invalid size: {family.size}. Size must be greater than 1. Add members to this family.")
    return (True, output)

Pydantic Model (family.py)

from pydantic import BaseModel
from typing import List


class Member(BaseModel):
    member_id: int
    name: str
    role: str
    age: int
    sports: List[str]


class Family(BaseModel):
    family_id: int
    name: str
    size: int
    members: List[Member]


class FamilyList(BaseModel):
    families: List[Family]

Flow (main.py)

from pydantic import BaseModel
from crewai.flow import Flow, listen, start
from crewai import CheckpointConfig
from crewai.state import JsonProvider

import sporty_family_flow.listeners
from sporty_family_flow.crews.family_crew.family_crew import FamilyCrew
from sporty_family_flow.models.family import Family, FamilyList


class FamilyState(BaseModel):
    families: list[Family] = []


class FamilyFlow(Flow[FamilyState]):

    @start()
    def get_input(self):
        return(input("Enter the sizes of your families: "))

    @listen(get_input)
    def create_families(self, sizes: str):
        result = (FamilyCrew().crew().kickoff(inputs={"sizes": sizes}))
        families = FamilyList.model_validate_json(result.raw).families
        self.state.families = families

    @listen(create_families)
    def print_families(self):
        for family in self.state.families:
            log = f"""
            ----- Family {family.family_id} ----- \n\n
            Family Name: {family.name}\n
            Size: {family.size}\n
            MEMBERS: \n
            """
            for member in family.members:
                log += f"""
                \t Member: {member.name} \n
                \t Age: {member.age} \n
                \t Role: {member.role} \n
                \t Sports: {', '.join(member.sports)}
                """
            print(log)


def kickoff():
    cp_config = CheckpointConfig(
        location="./.checkpoints",
        on_events=["method_execution_finished"],
        provider=JsonProvider(),
        max_checkpoints=5
    )
    family_flow = FamilyFlow(checkpoint=cp_config)
    family_flow.kickoff()


def plot():
    FamilyFlow().plot()

if __name__ == "__main__":
    kickoff()

Operating System

Ubuntu 24.04

Python Version

3.12

crewAI Version

1.14.2

crewAI Tools Version

Virtual Environment

Venv

Evidence

General remarks:

  • Checkpointing fails during state.model_dump(mode="json")
  • Guardrail retry fails during TaskOutput reconstruction
  • Issue occurs only when:
    • output_pydantic is enabled
    • guardrail returns (False, feedback)

I - Checkpoiting warning

Auto-checkpoint failed for event method_execution_finished
pydantic_core._pydantic_core.PydanticSerializationError: Unable to serialize unknown type: <class 'pydantic._internal._model_construction.ModelMetaclass'>

II - Guardrail blocking error

pydantic_core._pydantic_core.ValidationError: 1 validation error for TaskOutput
raw
  Input should be a valid string [type=string_type, input_value=FamilyList(families=[Fami...occer', 'Swimming'])])]), input_type=FamilyList]
    For further information visit https://errors.pydantic.dev/2.11/v/string_type
An error occurred while running the flow: Command '['uv', 'run', 'kickoff']' returned non-zero exit status 1.

Possible Solution

I was able to partially work around the issue by avoiding strict reliance on CrewAI’s internal Pydantic → JSON pipeline and manually controlling schema + parsing.

1. Explicit schema injection in prompt

Instead of relying on output_pydantic, I inject the JSON schema manually:

expected_output = f"""
You MUST return valid JSON matching this schema:

{json.dumps(FamilyList.model_json_schema(), indent=2)}

Do not include explanations or markdown.
"""

This bypasses part of CrewAI’s internal formatting pipeline.

2. Manual parsing of output

Instead of relying on result.pydantic, I manually parse: families = FamilyList.model_validate_json(result.raw).families

Additional context

It seems that:

1. Checkpointing issue

Pydantic model types inside Flow state are not properly serialized, possibly due to internal handling of ModelMetaclass instead of instances.

2. Guardrail retry issue

During retry, CrewAI reconstructs TaskOutput and incorrectly passes a Pydantic object into the raw field, which expects a string, causing a validation failure.

This suggests inconsistent handling between:

  • raw output (string)
  • parsed output (Pydantic model)
  • retry pipeline

extent analysis

TL;DR

To fix the issues with checkpointing and guardrail retry in CrewAI Flows with Pydantic state, manually control schema and parsing by injecting the JSON schema and parsing output, and consider modifying the retry pipeline to handle Pydantic objects correctly.

Guidance

  1. Manual schema injection: Instead of relying on output_pydantic, inject the JSON schema manually into the prompt to bypass CrewAI's internal formatting pipeline.
  2. Manual parsing of output: Manually parse the output using FamilyList.model_validate_json(result.raw).families instead of relying on result.pydantic.
  3. Modify retry pipeline: Consider modifying the retry pipeline to handle Pydantic objects correctly, possibly by converting them to strings before passing them to the raw field.
  4. Verify serialization: Verify that Pydantic model types inside Flow state are properly serialized by checking the model_dump method.

Example

expected_output = f"""
You MUST return valid JSON matching this schema:

{json.dumps(FamilyList.model_json_schema(), indent=2)}

Do not include explanations or markdown.
"""

Notes

  • The provided solution is a partial workaround and may not fully resolve the issues.
  • The root cause of the problems seems to be inconsistent handling of Pydantic objects between different parts of the CrewAI pipeline.
  • Further investigation and modification of the CrewAI code may be necessary to fully resolve the issues.

Recommendation

Apply the workaround by manually controlling schema and parsing, and consider modifying the retry pipeline to handle Pydantic objects correctly, as this is the most straightforward way to resolve the issues with the current version of CrewAI.

Vote matrix · Quick signals

Works
Did the solution work? Tap to confirm.
Easy Fix
Was it a quick fix?
Time Saver
Did it save you time?
Blocking
Was it severely blocking?
Common Issue
Are others likely hitting this too?
Flaky / Intermittent
Is it intermittent?
Verified / Reproducible
Can you reproduce it reliably?
Loading…
#api#ssr#model save/load#optimization#mixed precision

Still need to ship something?

×6

Another batch ranked right after the header list — different links, same matching logic.

Back to top recommendations

TRENDING

crewai - ✅(Solved) Fix [BUG] Checkpointing + guardrail retry incompatible with Pydantic outputs in Flow [4 pull requests]