|
1 | 1 | from collections.abc import AsyncGenerator
|
2 | 2 | from inspect import iscoroutinefunction, signature
|
3 |
| -from types import FunctionType |
4 |
| -from typing import Generator, get_origin, Union, Optional, Any, get_type_hints |
| 3 | +from typing import Generator, get_origin, Union, Optional, get_type_hints, Protocol, TypeVar, Callable |
5 | 4 | from typing import get_args
|
6 | 5 |
|
7 | 6 | from docstring_parser import parse
|
8 | 7 | from openai.types.beta import FunctionTool
|
| 8 | +from openai.types.chat import ChatCompletionToolParam |
9 | 9 | from openai.types.shared import FunctionDefinition
|
10 | 10 | from pydantic import create_model
|
11 | 11 |
|
12 | 12 |
|
13 |
| -def openai_streaming_function(func: FunctionType) -> Any: |
| 13 | +class OpenAIStreamingFunction(Protocol): |
| 14 | + """ |
| 15 | + A Protocol that represents a function that can be used with OpenAI Streaming. |
| 16 | + """ |
| 17 | + |
| 18 | + openai_schema: ChatCompletionToolParam # The OpenAI Schema for the function. |
| 19 | + |
| 20 | + def __call__(self, *args, **kwargs): |
| 21 | + pass |
| 22 | + |
| 23 | + |
| 24 | +F = TypeVar('F', bound=Callable[..., any]) |
| 25 | + |
| 26 | + |
| 27 | +def openai_streaming_function(func: F) -> OpenAIStreamingFunction: |
14 | 28 | """
|
15 | 29 | Decorator that creates an OpenAI Schema for your function, while support using Generators for Streaming.
|
16 | 30 |
|
@@ -75,7 +89,7 @@ async def error_message(typ: str, description: AsyncGenerator[str, None]):
|
75 | 89 | try:
|
76 | 90 | parameters = model.model_json_schema()
|
77 | 91 | except Exception as e:
|
78 |
| - parameters = model.schema() # Fallback to the default schema |
| 92 | + parameters = model.schema() # Fallback to the default schema |
79 | 93 |
|
80 | 94 | # extract parameter documentations from the docstring
|
81 | 95 | for param in docstring.params:
|
|
0 commit comments