kennethreitz/instructor
1
0
Fork 0
mirror of https://github.com/kennethreitz/instructor.git synced 2026-06-05 22:50:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

Experimental Prompting DSL (#25)

Browse Source
This commit is contained in:
Jason Liu
2023-07-03 17:22:12 +08:00
committed by GitHub
parent 730ff5e1e9
commit 8b7d28009e
11 changed files with 498 additions and 178 deletions
Download Patch File Download Diff File Expand all files Collapse all files
openai_function_call/__init__.py
+2 -110
View File
@@ -1,111 +1,3 @@
# MIT License
#
# Copyright (c) 2023 Jason Liu
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
from .function_calls import OpenAISchema, openai_function
import json
from functools import wraps
from typing import Any, Callable
from pydantic import validate_arguments, BaseModel
def _remove_a_key(d, remove_key) -> None:
"""Remove a key from a dictionary recursively"""
if isinstance(d, dict):
for key in list(d.keys()):
if key == remove_key:
del d[key]
else:
_remove_a_key(d[key], remove_key)
class openai_function:
def __init__(self, func: Callable) -> None:
self.func = func
self.validate_func = validate_arguments(func)
parameters = self.validate_func.model.schema()
parameters["properties"] = {
k: v
for k, v in parameters["properties"].items()
if k not in ("v__duplicate_kwargs", "args", "kwargs")
}
parameters["required"] = sorted(
parameters["properties"]
) # bug workaround see lc
_remove_a_key(parameters, "title")
_remove_a_key(parameters, "additionalProperties")
self.openai_schema = {
"name": self.func.__name__,
"description": self.func.__doc__,
"parameters": parameters,
}
self.model = self.validate_func.model
def __call__(self, *args: Any, **kwargs: Any) -> Any:
@wraps(self.func)
def wrapper(*args, **kwargs):
return self.validate_func(*args, **kwargs)
return wrapper(*args, **kwargs)
def from_response(self, completion, throw_error=True):
"""Execute the function from the response of an openai chat completion"""
message = completion.choices[0].message
if throw_error:
assert "function_call" in message, "No function call detected"
assert (
message["function_call"]["name"] == self.openai_schema["name"]
), "Function name does not match"
function_call = message["function_call"]
arguments = json.loads(function_call["arguments"])
return self.validate_func(**arguments)
class OpenAISchema(BaseModel):
@classmethod
@property
def openai_schema(cls):
schema = cls.schema()
parameters = {
k: v for k, v in schema.items() if k not in ("title", "description")
}
parameters["required"] = sorted(parameters["properties"])
_remove_a_key(parameters, "title")
return {
"name": schema["title"],
"description": schema["description"],
"parameters": parameters,
}
@classmethod
def from_response(cls, completion, throw_error=True):
message = completion.choices[0].message
if throw_error:
assert "function_call" in message, "No function call detected"
assert (
message["function_call"]["name"] == cls.openai_schema["name"]
), "Function name does not match"
function_call = message["function_call"]
arguments = json.loads(function_call["arguments"])
return cls(**arguments)
__all__ = ["OpenAISchema", "openai_function"]
openai_function_call/dsl/__init__.py
+5
View File
@@ -0,0 +1,5 @@
from .completion import ChatCompletion
from .multitask import MultiTask
from .messages import *
__all__ = ["ChatCompletion", "MultiTask", "messages"]
openai_function_call/dsl/completion.py
+1 -68
View File
@@ -102,71 +102,4 @@ class ChatCompletion(BaseModel):
if self.function:
return self.function.from_response(await completion)
return await completion
def MultiTask(
subtask_class: Type[OpenAISchema],
name: Optional[str] = None,
description: Optional[str] = None,
):
"""
Dynamically create a MultiTask OpenAISchema that can be used to segment multiple
tasks given a base class. This creates class that can be used to create a toolkit
for a specific task, names and descriptions are automatically generated. However
they can be overridden.
:param subtask_class: The base class to use for the MultiTask
:param name: The name of the MultiTask
:param description: The description of the MultiTask
:return: new schema class called `Multi{subtask_class.name}`
"""
task_name = subtask_class.__name__ if name is None else name
name = f"Multi{task_name}"
list_tasks = (
List[subtask_class],
Field(
default_factory=list,
repr=False,
description=f"Correctly segmented list of `{task_name}` tasks",
),
)
new_cls = create_model(name, tasks=list_tasks, __base__=(OpenAISchema,))
new_cls.__doc__ = (
f"Correct segmentation of `{task_name}` tasks"
if description is None
else description
)
return new_cls
if __name__ == "__main__":
from pprint import pprint
class Search(OpenAISchema):
id: int
query: str
task = (
ChatCompletion(name="Acme Inc Email Segmentation", model="gpt3.5-turbo-0613")
| ExpertSystem(task="Segment emails into search queries")
| MultiTask(subtask_class=Search)
| TaggedMessage(
tag="email",
content="Can you find the video I sent last week and also the post about dogs",
)
| TipsMessage(
tips=[
"When unsure about the correct segmentation, try to think about the task as a whole",
"If acronyms are used expand them to their full form",
"Use multiple phrases to describe the same thing",
]
)
| ChainOfThought()
)
assert isinstance(task, ChatCompletion)
openai_function_call/dsl/messages.py
+81
View File
@@ -0,0 +1,81 @@
from enum import Enum, auto
from pydantic.dataclasses import dataclass
from pydantic import Field
from typing import Optional, List
class MessageRole(Enum):
USER = auto()
SYSTEM = auto()
ASSISTANT = auto()
@dataclass
class Message:
content: str = Field(default=None, repr=True)
role: MessageRole = Field(default=MessageRole.USER, repr=False)
name: Optional[str] = Field(default=None)
def dict(self):
assert self.content is not None, "Content must be set!"
obj = {
"role": self.role.name.lower(),
"content": self.content,
}
if self.name and self.role == MessageRole.USER:
obj["name"] = self.name
return obj
@dataclass
class SystemMessage(Message):
def __post_init__(self):
self.role = MessageRole.SYSTEM
@dataclass
class UserMessage(Message):
def __post_init__(self):
self.role = MessageRole.USER
@dataclass
class TaggedMessage(Message):
tag: str = Field(default="data", repr=True)
def __post_init__(self):
self.role = MessageRole.USER
self.content = f"<{self.tag}>{self.content}</{self.tag}>"
@dataclass
class AssistantMessage(Message):
def __post_init__(self):
self.role = MessageRole.ASSISTANT
@dataclass
class ExpertSystem(Message):
task: str = Field(default=None, repr=True)
def __post_init__(self):
self.role = MessageRole.SYSTEM
self.content = f"You are a world class, state of the art agent capable of correctly completing the task: `{self.task}`"
@dataclass
class TipsMessage(Message):
tips: List[str] = Field(default_factory=list)
header: str = "Here are some tips to help you complete the task"
def __post_init__(self):
self.role = MessageRole.USER
tips = "\n* ".join(self.tips)
self.content = f"{self.header}:\n\n* {tips}"
@dataclass
class ChainOfThought(Message):
def __post_init__(self):
self.role = MessageRole.ASSISTANT
self.content = "Lets think step by step to get the correct answer:"
openai_function_call/dsl/multitask.py
+44
View File
@@ -0,0 +1,44 @@
from pydantic import create_model, Field
from typing import Optional, List, Type
from ..function_calls import OpenAISchema
def MultiTask(
subtask_class: Type[OpenAISchema],
name: Optional[str] = None,
description: Optional[str] = None,
):
"""
Dynamically create a MultiTask OpenAISchema that can be used to segment multiple
tasks given a base class. This creates class that can be used to create a toolkit
for a specific task, names and descriptions are automatically generated. However
they can be overridden.
:param subtask_class: The base class to use for the MultiTask
:param name: The name of the MultiTask
:param description: The description of the MultiTask
:return: new schema class called `Multi{subtask_class.name}`
"""
task_name = subtask_class.__name__ if name is None else name
name = f"Multi{task_name}"
list_tasks = (
List[subtask_class],
Field(
default_factory=list,
repr=False,
description=f"Correctly segmented list of `{task_name}` tasks",
),
)
new_cls = create_model(name, tasks=list_tasks, __base__=(OpenAISchema,))
new_cls.__doc__ = (
f"Correct segmentation of `{task_name}` tasks"
if description is None
else description
)
return new_cls
openai_function_call/function_calls.py
+111
View File
@@ -0,0 +1,111 @@
# MIT License
#
# Copyright (c) 2023 Jason Liu
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
import json
from functools import wraps
from typing import Any, Callable
from pydantic import validate_arguments, BaseModel
def _remove_a_key(d, remove_key) -> None:
"""Remove a key from a dictionary recursively"""
if isinstance(d, dict):
for key in list(d.keys()):
if key == remove_key:
del d[key]
else:
_remove_a_key(d[key], remove_key)
class openai_function:
def __init__(self, func: Callable) -> None:
self.func = func
self.validate_func = validate_arguments(func)
parameters = self.validate_func.model.schema()
parameters["properties"] = {
k: v
for k, v in parameters["properties"].items()
if k not in ("v__duplicate_kwargs", "args", "kwargs")
}
parameters["required"] = sorted(
parameters["properties"]
) # bug workaround see lc
_remove_a_key(parameters, "title")
_remove_a_key(parameters, "additionalProperties")
self.openai_schema = {
"name": self.func.__name__,
"description": self.func.__doc__,
"parameters": parameters,
}
self.model = self.validate_func.model
def __call__(self, *args: Any, **kwargs: Any) -> Any:
@wraps(self.func)
def wrapper(*args, **kwargs):
return self.validate_func(*args, **kwargs)
return wrapper(*args, **kwargs)
def from_response(self, completion, throw_error=True):
"""Execute the function from the response of an openai chat completion"""
message = completion.choices[0].message
if throw_error:
assert "function_call" in message, "No function call detected"
assert (
message["function_call"]["name"] == self.openai_schema["name"]
), "Function name does not match"
function_call = message["function_call"]
arguments = json.loads(function_call["arguments"])
return self.validate_func(**arguments)
class OpenAISchema(BaseModel):
@classmethod
@property
def openai_schema(cls):
schema = cls.schema()
parameters = {
k: v for k, v in schema.items() if k not in ("title", "description")
}
parameters["required"] = sorted(parameters["properties"])
_remove_a_key(parameters, "title")
return {
"name": schema["title"],
"description": schema["description"],
"parameters": parameters,
}
@classmethod
def from_response(cls, completion, throw_error=True):
message = completion.choices[0].message
if throw_error:
assert "function_call" in message, "No function call detected"
assert (
message["function_call"]["name"] == cls.openai_schema["name"]
), "Function name does not match"
function_call = message["function_call"]
arguments = json.loads(function_call["arguments"])
return cls(**arguments)