递归模式实现指南¶

from typing import List, Optional
from pydantic import BaseModel, Field


class RecursiveNode(BaseModel):
    """A node that can contain child nodes of the same type."""

    name: str = Field(..., description="Name of the node")
    value: Optional[str] = Field(
        None, description="Optional value associated with the node"
    )
    children: List["RecursiveNode"] = Field(
        default_factory=list, description="List of child nodes"
    )


# Required for recursive Pydantic models
RecursiveNode.model_rebuild()

import instructor
from openai import OpenAI

client = instructor.from_openai(OpenAI())


def parse_hierarchy(text: str) -> RecursiveNode:
    """Parse text into a hierarchical structure."""
    return client.chat.completions.create(
        model="gpt-4",
        messages=[
            {
                "role": "system",
                "content": "You are an expert at parsing text into hierarchical structures.",
            },
            {
                "role": "user",
                "content": f"Parse this text into a hierarchical structure: {text}",
            },
        ],
        response_model=RecursiveNode,
    )


# Example usage
hierarchy = parse_hierarchy(
    """
Company: Acme Corp
- Department: Engineering
  - Team: Frontend
    - Project: Website Redesign
    - Project: Mobile App
  - Team: Backend
    - Project: API v2
    - Project: Database Migration
- Department: Marketing
  - Team: Digital
    - Project: Social Media Campaign
  - Team: Brand
    - Project: Logo Refresh
"""
)

from pydantic import model_validator


class RecursiveNodeWithDepth(RecursiveNode):
    @model_validator(mode='after')
    def validate_depth(self) -> "RecursiveNodeWithDepth":
        def check_depth(node: "RecursiveNodeWithDepth", current_depth: int = 0) -> int:
            if current_depth > 10:  # Maximum allowed depth
                raise ValueError("Maximum depth exceeded")
            return max(
                [check_depth(child, current_depth + 1) for child in node.children],
                default=current_depth,
            )

        check_depth(self)
        return self