mirror of
https://github.com/crewAIInc/crewAI.git
synced 2025-12-16 12:28:30 +00:00
Compare commits
5 Commits
devin/1762
...
devin/1740
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f80fe7d4c1 | ||
|
|
da0d37af03 | ||
|
|
f65c31bfd0 | ||
|
|
9322f06e7a | ||
|
|
326f406605 |
@@ -1,5 +1,7 @@
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
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,17 +257,57 @@ 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):
|
||||
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)
|
||||
|
||||
@@ -271,14 +315,14 @@ def generate_model_description(model: Type[BaseModel]) -> str:
|
||||
# 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:
|
||||
return f"Optional[{describe_field(non_none_args[0])}]"
|
||||
return f"Optional[{get_type_description(non_none_args[0])}]"
|
||||
else:
|
||||
return 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:
|
||||
return 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])
|
||||
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):
|
||||
return generate_model_description(field_type)
|
||||
@@ -287,8 +331,12 @@ def generate_model_description(model: Type[BaseModel]) -> str:
|
||||
else:
|
||||
return str(field_type)
|
||||
|
||||
fields = model.__annotations__
|
||||
field_descriptions = [
|
||||
f'"{name}": {describe_field(type_)}' for name, type_ in fields.items()
|
||||
]
|
||||
fields = model.model_fields
|
||||
field_descriptions = []
|
||||
for name, field in fields.items():
|
||||
field_desc = describe_field(field.annotation, field)
|
||||
if isinstance(field_desc, dict):
|
||||
field_descriptions.append(f'"{name}": {json.dumps(field_desc)}')
|
||||
else:
|
||||
field_descriptions.append(f'"{name}": {field_desc}')
|
||||
return "{\n " + ",\n ".join(field_descriptions) + "\n}"
|
||||
|
||||
@@ -4,7 +4,7 @@ from typing import Dict, List, Optional
|
||||
from unittest.mock import MagicMock, Mock, patch
|
||||
|
||||
import pytest
|
||||
from pydantic import BaseModel
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from crewai.llm import LLM
|
||||
from crewai.utilities.converter import (
|
||||
@@ -328,6 +328,51 @@ 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")
|
||||
|
||||
description = generate_model_description(ModelWithDescriptions)
|
||||
expected = '{\n "name": {"type": "str", "description": "The user\'s full name"},\n "age": {"type": "int", "description": "The user\'s age in years"}\n}'
|
||||
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
|
||||
|
||||
description = generate_model_description(MixedModel)
|
||||
expected = '{\n "name": {"type": "str", "description": "The user\'s name"},\n "age": int\n}'
|
||||
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")
|
||||
|
||||
Reference in New Issue
Block a user