Skip to content

anthropic.extractors

A class for extracting structured information using Anthropic Claude models.

AnthropicExtractor

Bases: BaseExtractor[AnthropicCall, AnthropicTool, Any, T], Generic[T]

A class for extracting structured information using Anthropic Claude models.

Example:

from typing import Literal, Type

from mirascope.anthropic import AnthropicExtractor
from pydantic import BaseModel


class TaskDetails(BaseModel):
    title: str
    priority: Literal["low", "normal", "high"]
    due_date: str


class TaskExtractor(AnthropicExtractor[TaskDetails]):
    extract_schema: Type[TaskDetails] = TaskDetails

    prompt_template = """
    Please extract the task details:
    {task}
    """

    task: str


task_description = "Submit quarterly report by next Friday. Task is high priority."
task = TaskExtractor(task=task_description).extract(retries=3)
assert isinstance(task, TaskDetails)
print(task)
#> title='Submit quarterly report' priority='high' due_date='next Friday'
Source code in mirascope/anthropic/extractors.py 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
class AnthropicExtractor(
    BaseExtractor[AnthropicCall, AnthropicTool, Any, T], Generic[T]
):
    '''A class for extracting structured information using Anthropic Claude models.

    Example:

    ```python
    from typing import Literal, Type

    from mirascope.anthropic import AnthropicExtractor
    from pydantic import BaseModel


    class TaskDetails(BaseModel):
        title: str
        priority: Literal["low", "normal", "high"]
        due_date: str


    class TaskExtractor(AnthropicExtractor[TaskDetails]):
        extract_schema: Type[TaskDetails] = TaskDetails

        prompt_template = """
        Please extract the task details:
        {task}
        """

        task: str


    task_description = "Submit quarterly report by next Friday. Task is high priority."
    task = TaskExtractor(task=task_description).extract(retries=3)
    assert isinstance(task, TaskDetails)
    print(task)
    #> title='Submit quarterly report' priority='high' due_date='next Friday'
    ```
    '''

    call_params: ClassVar[AnthropicCallParams] = AnthropicCallParams()

    def extract(self, retries: Union[int, Retrying] = 0, **kwargs: Any) -> T:
        """Extracts `extract_schema` from the Anthropic call response.

        The `extract_schema` is converted into an `AnthropicTool`, complete with a
        description of the tool, all of the fields, and their types. This allows us to
        take advantage of Anthropics's tool/function calling functionality to extract
        information from a prompt according to the context provided by the `BaseModel`
        schema.

        Args:
            retries: The maximum number of times to retry the query on validation error.
            **kwargs: Additional keyword arguments parameters to pass to the call. These
                will override any existing arguments in `call_params`.

        Returns:
            The `Schema` instance extracted from the completion.

        Raises:
            AttributeError: if there is no tool in the call creation.
            ValidationError: if the schema cannot be instantiated from the completion.
        """
        kwargs["system"] = EXTRACT_SYSTEM_MESSAGE
        return self._extract(AnthropicCall, AnthropicTool, retries, **kwargs)

    async def extract_async(
        self, retries: Union[int, AsyncRetrying] = 0, **kwargs: Any
    ) -> T:
        """Asynchronously extracts `extract_schema` from the Anthropic call response.

        The `extract_schema` is converted into an `AnthropicTool`, complete with a
        description of the tool, all of the fields, and their types. This allows us to
        take advantage of Anthropic's tool/function calling functionality to extract
        information from a prompt according to the context provided by the `BaseModel`
        schema.

        Args:
            retries: The maximum number of times to retry the query on validation error.
            **kwargs: Additional keyword arguments parameters to pass to the call. These
                will override any existing arguments in `call_params`.

        Returns:
            The `Schema` instance extracted from the completion.

        Raises:
            AttributeError: if there is no tool in the call creation.
            ValidationError: if the schema cannot be instantiated from the completion.
        """
        kwargs["system"] = EXTRACT_SYSTEM_MESSAGE
        return await self._extract_async(
            AnthropicCall, AnthropicTool, retries, **kwargs
        )

    def stream(
        self, retries: Union[int, Retrying] = 0, **kwargs: Any
    ) -> Generator[T, None, None]:
        """Streams partial instances of `extract_schema` as the schema is streamed.

        The `extract_schema` is converted into a `partial(AnthropicTool)`, which allows
        for any field (i.e.function argument) in the tool to be `None`. This allows us
        to stream partial results as we construct the tool from the streamed chunks.

        Args:
            retries: The maximum number of times to retry the query on validation error.
            **kwargs: Additional keyword argument parameters to pass to the call. These
                will override any existing arguments in `call_params`.

        Yields:
            The partial `extract_schema` instance from the current buffer.

        Raises:
            AttributeError: if there is no tool in the call creation.
            ValidationError: if the schema cannot be instantiated from the completion.
        """
        yield from self._stream(
            AnthropicCall, AnthropicTool, AnthropicToolStream, retries, **kwargs
        )

    async def stream_async(
        self, retries: Union[int, AsyncRetrying] = 0, **kwargs: Any
    ) -> AsyncGenerator[T, None]:
        """Asynchronously streams partial instances of `extract_schema` as streamed.

        The `extract_schema` is converted into a `partial(AnthropicTool)`, which allows
        for any field (i.e.function argument) in the tool to be `None`. This allows us
        to stream partial results as we construct the tool from the streamed chunks.

        Args:
            retries: The maximum number of times to retry the query on validation error.
            **kwargs: Additional keyword arguments parameters to pass to the call. These
                will override any existing arguments in `call_params`.

        Yields:
            The partial `extract_schema` instance from the current buffer.

        Raises:
            AttributeError: if there is no tool in the call creation.
            ValidationError: if the schema cannot be instantiated from the completion.
        """
        async for partial_tool in self._stream_async(
            AnthropicCall, AnthropicTool, AnthropicToolStream, retries, **kwargs
        ):
            yield partial_tool

extract(retries=0, **kwargs)

Extracts extract_schema from the Anthropic call response.

The extract_schema is converted into an AnthropicTool, complete with a description of the tool, all of the fields, and their types. This allows us to take advantage of Anthropics's tool/function calling functionality to extract information from a prompt according to the context provided by the BaseModel schema.

Parameters:

Name Type Description Default
retries Union[int, Retrying]

The maximum number of times to retry the query on validation error.

 0
**kwargs Any

Additional keyword arguments parameters to pass to the call. These will override any existing arguments in call_params.

 {}

Returns:

Type Description
T

The Schema instance extracted from the completion.

Raises:

Type Description
AttributeError

if there is no tool in the call creation.
ValidationError

if the schema cannot be instantiated from the completion.
Source code in mirascope/anthropic/extractors.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def extract(self, retries: Union[int, Retrying] = 0, **kwargs: Any) -> T:
    """Extracts `extract_schema` from the Anthropic call response.

    The `extract_schema` is converted into an `AnthropicTool`, complete with a
    description of the tool, all of the fields, and their types. This allows us to
    take advantage of Anthropics's tool/function calling functionality to extract
    information from a prompt according to the context provided by the `BaseModel`
    schema.

    Args:
        retries: The maximum number of times to retry the query on validation error.
        **kwargs: Additional keyword arguments parameters to pass to the call. These
            will override any existing arguments in `call_params`.

    Returns:
        The `Schema` instance extracted from the completion.

    Raises:
        AttributeError: if there is no tool in the call creation.
        ValidationError: if the schema cannot be instantiated from the completion.
    """
    kwargs["system"] = EXTRACT_SYSTEM_MESSAGE
    return self._extract(AnthropicCall, AnthropicTool, retries, **kwargs)

extract_async(retries=0, **kwargs) async

Asynchronously extracts extract_schema from the Anthropic call response.

The extract_schema is converted into an AnthropicTool, complete with a description of the tool, all of the fields, and their types. This allows us to take advantage of Anthropic's tool/function calling functionality to extract information from a prompt according to the context provided by the BaseModel schema.

Parameters:

Name Type Description Default
retries Union[int, AsyncRetrying]

The maximum number of times to retry the query on validation error.

 0
**kwargs Any

Additional keyword arguments parameters to pass to the call. These will override any existing arguments in call_params.

 {}

Returns:

Type Description
T

The Schema instance extracted from the completion.

Raises:

Type Description
AttributeError

if there is no tool in the call creation.
ValidationError

if the schema cannot be instantiated from the completion.
Source code in mirascope/anthropic/extractors.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
async def extract_async(
    self, retries: Union[int, AsyncRetrying] = 0, **kwargs: Any
) -> T:
    """Asynchronously extracts `extract_schema` from the Anthropic call response.

    The `extract_schema` is converted into an `AnthropicTool`, complete with a
    description of the tool, all of the fields, and their types. This allows us to
    take advantage of Anthropic's tool/function calling functionality to extract
    information from a prompt according to the context provided by the `BaseModel`
    schema.

    Args:
        retries: The maximum number of times to retry the query on validation error.
        **kwargs: Additional keyword arguments parameters to pass to the call. These
            will override any existing arguments in `call_params`.

    Returns:
        The `Schema` instance extracted from the completion.

    Raises:
        AttributeError: if there is no tool in the call creation.
        ValidationError: if the schema cannot be instantiated from the completion.
    """
    kwargs["system"] = EXTRACT_SYSTEM_MESSAGE
    return await self._extract_async(
        AnthropicCall, AnthropicTool, retries, **kwargs
    )

stream(retries=0, **kwargs)

Streams partial instances of extract_schema as the schema is streamed.

The extract_schema is converted into a partial(AnthropicTool), which allows for any field (i.e.function argument) in the tool to be None. This allows us to stream partial results as we construct the tool from the streamed chunks.

Parameters:

Name Type Description Default
retries Union[int, Retrying]

The maximum number of times to retry the query on validation error.

 0
**kwargs Any

Additional keyword argument parameters to pass to the call. These will override any existing arguments in call_params.

 {}

Yields:

Type Description
T

The partial extract_schema instance from the current buffer.

Raises:

Type Description
AttributeError

if there is no tool in the call creation.
ValidationError

if the schema cannot be instantiated from the completion.
Source code in mirascope/anthropic/extractors.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def stream(
    self, retries: Union[int, Retrying] = 0, **kwargs: Any
) -> Generator[T, None, None]:
    """Streams partial instances of `extract_schema` as the schema is streamed.

    The `extract_schema` is converted into a `partial(AnthropicTool)`, which allows
    for any field (i.e.function argument) in the tool to be `None`. This allows us
    to stream partial results as we construct the tool from the streamed chunks.

    Args:
        retries: The maximum number of times to retry the query on validation error.
        **kwargs: Additional keyword argument parameters to pass to the call. These
            will override any existing arguments in `call_params`.

    Yields:
        The partial `extract_schema` instance from the current buffer.

    Raises:
        AttributeError: if there is no tool in the call creation.
        ValidationError: if the schema cannot be instantiated from the completion.
    """
    yield from self._stream(
        AnthropicCall, AnthropicTool, AnthropicToolStream, retries, **kwargs
    )

stream_async(retries=0, **kwargs) async

Asynchronously streams partial instances of extract_schema as streamed.

The extract_schema is converted into a partial(AnthropicTool), which allows for any field (i.e.function argument) in the tool to be None. This allows us to stream partial results as we construct the tool from the streamed chunks.

Parameters:

Name Type Description Default
retries Union[int, AsyncRetrying]

The maximum number of times to retry the query on validation error.

 0
**kwargs Any

Additional keyword arguments parameters to pass to the call. These will override any existing arguments in call_params.

 {}

Yields:

Type Description
AsyncGenerator[T, None]

The partial extract_schema instance from the current buffer.

Raises:

Type Description
AttributeError

if there is no tool in the call creation.
ValidationError

if the schema cannot be instantiated from the completion.
Source code in mirascope/anthropic/extractors.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
async def stream_async(
    self, retries: Union[int, AsyncRetrying] = 0, **kwargs: Any
) -> AsyncGenerator[T, None]:
    """Asynchronously streams partial instances of `extract_schema` as streamed.

    The `extract_schema` is converted into a `partial(AnthropicTool)`, which allows
    for any field (i.e.function argument) in the tool to be `None`. This allows us
    to stream partial results as we construct the tool from the streamed chunks.

    Args:
        retries: The maximum number of times to retry the query on validation error.
        **kwargs: Additional keyword arguments parameters to pass to the call. These
            will override any existing arguments in `call_params`.

    Yields:
        The partial `extract_schema` instance from the current buffer.

    Raises:
        AttributeError: if there is no tool in the call creation.
        ValidationError: if the schema cannot be instantiated from the completion.
    """
    async for partial_tool in self._stream_async(
        AnthropicCall, AnthropicTool, AnthropicToolStream, retries, **kwargs
    ):
        yield partial_tool