refactor: address code review feedback

- Split describe_field into smaller functions
- Add error handling and logging
- Add comprehensive docstrings
- Add pytest marks for test organization
- Add edge case tests
- Add type hints and constants
- Add caching for performance

Co-Authored-By: Joe Moura <joao@crewai.com>
This commit is contained in:
Devin AI
2025-02-21 11:45:12 +00:00
parent 326f406605
commit 9322f06e7a
2 changed files with 81 additions and 17 deletions

View File

@@ -1,5 +1,7 @@
import json
import re
import logging
from functools import lru_cache
from typing import Any, Optional, Type, Union, get_args, get_origin
from pydantic import BaseModel, ValidationError
@@ -8,6 +10,8 @@ from crewai.agents.agent_builder.utilities.base_output_converter import OutputCo
from crewai.utilities.printer import Printer
from crewai.utilities.pydantic_schema_parser import PydanticSchemaParser
logger = logging.getLogger(__name__)
class ConverterError(Exception):
"""Error raised when Converter fails to parse the input."""
@@ -253,44 +257,79 @@ def create_converter(
return converter
FIELD_TYPE_KEY = "type"
FIELD_DESC_KEY = "description"
def generate_model_description(model: Type[BaseModel]) -> str:
"""
Generate a string description of a Pydantic model's fields and their types.
This function takes a Pydantic model class and returns a string that describes
the model's fields and their respective types. The description includes handling
of complex types such as `Optional`, `List`, and `Dict`, as well as nested Pydantic
models.
@lru_cache(maxsize=100)
def generate_model_description(model: Type[BaseModel]) -> str:
models and field descriptions when available.
Args:
model: A Pydantic BaseModel class to generate description for
Returns:
str: A JSON-like string describing the model's fields, their types, and descriptions
"""
def describe_field(field_type, field_info=None):
def describe_field(field_type: Any, field_info: Optional[Any] = None) -> Union[str, dict]:
"""
Generate a description for a model field including its type and description.
Args:
field_type: The type annotation of the field
field_info: Optional field information containing description
Returns:
Union[str, dict]: Field description either as string (type only) or
dict with type and description
"""
try:
type_desc = get_type_description(field_type)
if field_info and field_info.description:
return {FIELD_TYPE_KEY: type_desc, FIELD_DESC_KEY: field_info.description}
return type_desc
except Exception as e:
logger.warning(f"Error processing field description: {e}")
return str(field_type)
def get_type_description(field_type: Any) -> str:
"""
Get the type description for a field type.
Args:
field_type: The type annotation to describe
Returns:
str: A string representation of the type
"""
origin = get_origin(field_type)
args = get_args(field_type)
type_desc = ""
if origin is Union or (origin is None and len(args) > 0):
# Handle both Union and the new '|' syntax
non_none_args = [arg for arg in args if arg is not type(None)]
if len(non_none_args) == 1:
type_desc = f"Optional[{describe_field(non_none_args[0])}]"
return f"Optional[{get_type_description(non_none_args[0])}]"
else:
type_desc = f"Optional[Union[{', '.join(describe_field(arg) for arg in non_none_args)}]]"
return f"Optional[Union[{', '.join(get_type_description(arg) for arg in non_none_args)}]]"
elif origin is list:
type_desc = f"List[{describe_field(args[0])}]"
return f"List[{get_type_description(args[0])}]"
elif origin is dict:
key_type = describe_field(args[0])
value_type = describe_field(args[1])
type_desc = f"Dict[{key_type}, {value_type}]"
key_type = get_type_description(args[0])
value_type = get_type_description(args[1])
return f"Dict[{key_type}, {value_type}]"
elif isinstance(field_type, type) and issubclass(field_type, BaseModel):
type_desc = generate_model_description(field_type)
return generate_model_description(field_type)
elif hasattr(field_type, "__name__"):
type_desc = field_type.__name__
return field_type.__name__
else:
type_desc = str(field_type)
if field_info and field_info.description:
return {"type": type_desc, "description": field_info.description}
return type_desc
return str(field_type)
fields = model.model_fields
field_descriptions = []

View File

@@ -328,7 +328,12 @@ def test_generate_model_description_dict_field():
assert description == expected_description
@pytest.mark.field_descriptions
def test_generate_model_description_with_field_descriptions():
"""
Verify that the model description generator correctly includes field descriptions
when they are provided via Field(..., description='...').
"""
class ModelWithDescriptions(BaseModel):
name: str = Field(..., description="The user's full name")
age: int = Field(..., description="The user's age in years")
@@ -338,7 +343,12 @@ def test_generate_model_description_with_field_descriptions():
assert description == expected
@pytest.mark.field_descriptions
def test_generate_model_description_mixed_fields():
"""
Verify that the model description generator correctly handles a mix of fields
with and without descriptions.
"""
class MixedModel(BaseModel):
name: str = Field(..., description="The user's name")
age: int # No description
@@ -348,6 +358,21 @@ def test_generate_model_description_mixed_fields():
assert description == expected
@pytest.mark.field_descriptions
def test_generate_model_description_with_empty_description():
"""
Verify that the model description generator correctly handles fields with empty
descriptions by treating them as fields without descriptions.
"""
class ModelWithEmptyDescription(BaseModel):
name: str = Field(..., description="")
age: int = Field(..., description=None)
description = generate_model_description(ModelWithEmptyDescription)
expected = '{\n "name": "str",\n "age": "int"\n}'
assert description == expected
@pytest.mark.vcr(filter_headers=["authorization"])
def test_convert_with_instructions():
llm = LLM(model="gpt-4o-mini")