autogen_core.utils#

schema_to_pydantic_model(schema: Dict[str, Any], model_name: str = 'GeneratedModel') → Type[BaseModel][来源]#

将 JSON Schema 字典转换为完全类型化的 Pydantic 模型。

此函数处理模式转换和验证逻辑以生成 Pydantic 模型。

支持的 JSON Schema 功能

  • 基本类型：string、integer、number、boolean、object、array、null

  • 字符串格式:

    • email、uri、uuid、uuid1、uuid3、uuid4、uuid5

    • hostname、ipv4、ipv6、ipv4-network、ipv6-network

    • date、time、date-time、duration

    • byte、binary、password、path

  • 字符串约束:

    • minLength、maxLength、pattern

  • 数字约束:

    • minimum、maximum、exclusiveMinimum、exclusiveMaximum

  • 数组约束:

    • minItems、maxItems、items

  • 对象模式支持:

    • properties、required、title、description、default

  • 枚举:

    • 转换为 Python Literal 类型

  • 联合类型:

    • anyOf、oneOf 支持可选的 discriminator

  • 继承和组合:

    • allOf 将多个模式合并为一个模型

  • $ref 和 $defs 解析:

    • 支持对同级定义和自引用模式的引用

from autogen_core.utils import schema_to_pydantic_model

# Example 1: Simple user model
schema = {
    "title": "User",
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "email": {"type": "string", "format": "email"},
        "age": {"type": "integer", "minimum": 0},
    },
    "required": ["name", "email"],
}

UserModel = schema_to_pydantic_model(schema)
user = UserModel(name="Alice", email="alice@example.com", age=30)

from autogen_core.utils import schema_to_pydantic_model

# Example 2: Nested model
schema = {
    "title": "BlogPost",
    "type": "object",
    "properties": {
        "title": {"type": "string"},
        "tags": {"type": "array", "items": {"type": "string"}},
        "author": {
            "type": "object",
            "properties": {"name": {"type": "string"}, "email": {"type": "string", "format": "email"}},
            "required": ["name"],
        },
    },
    "required": ["title", "author"],
}

BlogPost = schema_to_pydantic_model(schema)

from autogen_core.utils import schema_to_pydantic_model

# Example 3: allOf merging with $refs
schema = {
    "title": "EmployeeWithDepartment",
    "allOf": [{"$ref": "#/$defs/Employee"}, {"$ref": "#/$defs/Department"}],
    "$defs": {
        "Employee": {
            "type": "object",
            "properties": {"id": {"type": "string"}, "name": {"type": "string"}},
            "required": ["id", "name"],
        },
        "Department": {
            "type": "object",
            "properties": {"department": {"type": "string"}},
            "required": ["department"],
        },
    },
}

Model = schema_to_pydantic_model(schema)

from autogen_core.utils import schema_to_pydantic_model

# Example 4: Self-referencing (recursive) model
schema = {
    "title": "Category",
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "subcategories": {"type": "array", "items": {"$ref": "#/$defs/Category"}},
    },
    "required": ["name"],
    "$defs": {
        "Category": {
            "type": "object",
            "properties": {
                "name": {"type": "string"},
                "subcategories": {"type": "array", "items": {"$ref": "#/$defs/Category"}},
            },
            "required": ["name"],
        }
    },
}

Category = schema_to_pydantic_model(schema)

# Example 5: Serializing and deserializing with Pydantic

from uuid import uuid4
from pydantic import BaseModel, EmailStr, Field
from typing import Optional, List, Dict, Any
from autogen_core.utils import schema_to_pydantic_model


class Address(BaseModel):
    street: str
    city: str
    zipcode: str


class User(BaseModel):
    id: str
    name: str
    email: EmailStr
    age: int = Field(..., ge=18)
    address: Address


class Employee(BaseModel):
    id: str
    name: str
    manager: Optional["Employee"] = None


class Department(BaseModel):
    name: str
    employees: List[Employee]


class ComplexModel(BaseModel):
    user: User
    extra_info: Optional[Dict[str, Any]] = None
    sub_items: List[Employee]


# Convert ComplexModel to JSON schema
complex_schema = ComplexModel.model_json_schema()

# Rebuild a new Pydantic model from JSON schema
ReconstructedModel = schema_to_pydantic_model(complex_schema, "ComplexModel")

# Instantiate reconstructed model
reconstructed = ReconstructedModel(
    user={
        "id": str(uuid4()),
        "name": "Alice",
        "email": "alice@example.com",
        "age": 30,
        "address": {"street": "123 Main St", "city": "Wonderland", "zipcode": "12345"},
    },
    sub_items=[{"id": str(uuid4()), "name": "Bob", "manager": {"id": str(uuid4()), "name": "Eve"}}],
)

print(reconstructed.model_dump())
参数:

  • schema (Dict[str, Any]) – 一个有效的 JSON Schema 字典。

  • model_name (str, 可选) – 根模型的名称。默认为“GeneratedModel”。

返回:

Type[BaseModel] – 一个动态生成的 Pydantic 模型类。

抛出:

  • ReferenceNotFoundError – 如果 $ref 键引用了缺失的条目。

  • FormatNotSupportedError – 如果 format 关键字未知或不受支持。

  • UnsupportedKeywordError – 如果模式包含不受支持的 type。

另请参见

extract_json_from_str(content: str) → List[Dict[str, Any]][来源]#

从字符串中提取 JSON 对象。支持反引号括起来的 JSON 对象