Using ActionWeaver with Pydantic V2
This guide is designed to demonstrate use cases using Pydantic along with ActionWeaver.
[1]:
%load_ext autoreload
%autoreload 2
[2]:
import os
from typing import List
from uuid import UUID, uuid4
from actionweaver import action
from actionweaver.utils.tokens import TokenUsageTracker
from actionweaver.llms import wrap
from actionweaver.actions.factories.pydantic_model_to_action import action_from_model
from pydantic import BaseModel, Field, PrivateAttr, validate_call
from openai import OpenAI
from datetime import datetime
Prerequisite: Patch OpenAI client with ActionWeaver
[4]:
from openai import AzureOpenAI
MODEL="gpt-35-turbo-0613-16k"
client = AzureOpenAI(
azure_endpoint = os.getenv("AZURE_OPENAI_ENDPOINT"),
api_key=os.getenv("AZURE_OPENAI_KEY"),
api_version="2023-10-01-preview"
)
client = wrap(client)
Use Case: Structure parsing using Pydantic V2
[5]:
class User(BaseModel):
_uid: UUID = PrivateAttr(default_factory=uuid4)
_created_at: datetime = PrivateAttr(default_factory=datetime.now)
name: str = Field(..., description="Name of the user")
age: int = Field(..., description="Age of the user", gt=0)
class Users(BaseModel):
users: List[User]
create_users = action_from_model(Users, stop=True)
create_users.json_schema()
[5]:
{'$defs': {'User': {'properties': {'name': {'description': 'Name of the user',
'title': 'Name',
'type': 'string'},
'age': {'description': 'Age of the user',
'exclusiveMinimum': 0,
'title': 'Age',
'type': 'integer'}},
'required': ['name', 'age'],
'title': 'User',
'type': 'object'}},
'properties': {'users': {'items': {'$ref': '#/$defs/User'},
'title': 'Users',
'type': 'array'}},
'required': ['users'],
'title': 'Users',
'type': 'object'}
[7]:
users = create_users.invoke(client, messages=[{"role": "user", "content": "Niel and Tom are 31 years old, James is 25 years old"}], model=MODEL, stream=False, force=False)
[8]:
users
[8]:
Users(users=[User(name='Niel', age=31), User(name='Tom', age=31), User(name='James', age=25)])
Use Case: You can also use Pydantic models as function parameter.
The @validate_call decorator allows the arguments passed to a function to be parsed and validated using the function’s annotations before the function is called.
[ ]:
user_db = []
@action(name="InsertUsers", stop=True)
@validate_call
def insert_users(users: Users) -> str:
"""Inserts user data into the database and returns a success message."""
# Add the provided users to the user database
user_db.extend(users.users)
return f"Inserted {len(users.users)} users ."
[ ]:
messages = [
{"role": "user", "content": "Add following users: Niel and Tom are 31 years old, James is 25 years old"}
]
response = client.create(
model="gpt-3.5-turbo",
messages=messages,
actions = [insert_users],
stream=False,
token_usage_tracker = TokenUsageTracker(500),
)
response, user_db