I am trying to use CrewAI and output_pydantic for my tasks. I have a guardrail that ensures the output does not have json fencing i.e `` json ```. However, sometimes the agent returns an invalid json with the fencing without going to the guardrail and causing the below error:
Exception: An error occurred while running the crew: Failed to convert text into a Pydantic model due to error: ‘NoneType’ object has no attribute ‘function_calling_llm’
This is because it cannot parse an invalid json:
Invalid JSON: expected value at line 1 column 1 [type=json_invalid, input_value='```json\n{\n "locations...i": "LOCATION Z",\n ', input_type=str] For further information visit https://errors.pydantic.dev/2.11/v/json_invalid
Why do my tasks most of the time go through the guardrail and rerun the task, but sometimes not?
Crew:
class LocationsPlaceholderOutput(BaseModel):
locations_mapping: Dict[str, str]
class PIIPlaceholderOutput(BaseModel):
pii_mapping: Dict[str, str]
class CompanyNamesPlaceholderOutput(BaseModel):
mapping: Dict[str, str]
# 2. Consolidated crew output Pydantic model
class PlaceholderOutput(BaseModel):
locations_mappings: Dict[str, str]
company_names_mappings: Dict[str, str]
pii_mappings: Dict[str, str]
@CrewBase
class PlaceholderAgents():
agents: List[BaseAgent]
tasks: List[Task]
current_directory = os.path.dirname(os.path.abspath(__file__))
agents_config = os.path.join(current_directory, 'config', 'placeholderAgents.yaml')
tasks_config = os.path.join(current_directory, 'config', 'placeholderTasks.yaml')
output_pydantic: Type[PlaceholderOutput] = PlaceholderOutput
@agent
def PlaceholderCreator(self) -> Agent:
return Agent(
config=self.agents_config['PlaceholderCreator'],
verbose=True,
llm=llm_gemini,
allow_delegation=False,
)
@task
def LocationsPlaceholdersTask(self) -> Task:
return Task(
config=self.tasks_config['LocationsPlaceholdersTask'],
agent=self.PlaceholderCreator(),
guardrail= clean_and_validate_json_output,
guardrail_max_retries=10,
max_retries=10,
output_pydantic=LocationsPlaceholderOutput
)
@task
def PIIPlaceholdersTask(self) -> Task:
return Task(
config=self.tasks_config['PIIPlaceholdersTask'],
agent=self.PlaceholderCreator(),
guardrail= ensure_not_empty,
guardrail_max_retries=10,
output_pydantic=PIIPlaceholderOutput
)
@task
def CompanyNamesPlaceholdersTask(self) -> Task:
return Task(
config=self.tasks_config['CompanyNamesPlaceholdersTask'],
output_pydantic=CompanyNamesPlaceholderOutput,
guardrail= ensure_not_empty,
guardrail_max_retries=10,
agent=self.PlaceholderCreator()
)
@task
def Combine(self) -> Task:
return Task(
config=self.tasks_config['CombineTask'],
output_pydantic=PlaceholderOutput,
guardrail= ensure_not_empty,
guardrail_max_retries=10,
agent=self.PlaceholderCreator(),
context=[self.PIIPlaceholdersTask(),
self.LocationsPlaceholdersTask(), self.CompanyNamesPlaceholdersTask()
]
)
@crew
def crew(self) -> Crew:
"""Creates the PlaceholderAgents crew"""
# To learn how to add knowledge sources to your crew, check out the documentation:
# https://docs.crewai.com/concepts/knowledge#what-is-knowledge
return Crew(
agents=self.agents, # Automatically created by the @agent decorator
tasks=self.tasks, # Automatically created by the @task decorator
process=Process.sequential,
verbose=True,
output_pydantic=self.output_pydantic
# process=Process.hierarchical, # In case you wanna use that instead https://docs.crewai.com/how-to/Hierarchical/
)
Guardrail:
def ensure_not_empty(result: TaskOutput) -> Tuple[bool, Any]:
raw_output = result.raw.strip()
try:
data = json.loads(result.raw)
except json.JSONDecodeError as e:
return (False, "Invalid JSON format. Do not use Json fencing")
match = re.search(r"```json\s*\n?(.*?)\n?```", raw_output, re.DOTALL)
if match:
return (False, f"Guardrail failed: Json fencing in the output: {raw_output}.")
try:
data = json.loads(raw_output)
if isinstance(data, dict) and not data:
return (False, "Guardrail failed: The JSON object is an empty dictionary '{}'.")
if isinstance(data, dict):
if not data.values():
return (False, "Guardrail failed: The JSON dictionary contains no values.")
for key, value in data.items():
if isinstance(value, dict) and not value:
return (False, f"Guardrail failed: The mapping for key '{key}' is an empty dictionary {{}}.")
if isinstance(value, list) and not value:
return (False, f"Guardrail failed: The list for key '{key}' is empty [].")
except Exception as e:
return(False, f"Error during check")
return(True, result.raw.strip())