L3: Retry-based Structured Output

# Wrap together with the instructor client
instructor_client = instructor.from_openai(together_client)

from pydantic import BaseModel


class Greeting(BaseModel):
    hello: str

response = instructor_client.chat.completions.create(
    model="meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo",
    messages=[{"role": "user", "content": "sup"}],
    # Note: OpenAI uses response_format, instructor
    #       uses response_model!
    response_model=Greeting,
)

response

from pydantic import Field
from datetime import date
from typing import List


# Person only has a name, but we can easily extend it with other fields
class Person(BaseModel):
    name: str


class CalendarEvent(BaseModel):
    name: str

    # not supported by OpenAI. We want a format like 2025-01-30
    date: date
    participants: List[Person]

    address_number: str
    street_name: str
    city_name: str

    # Inline regex patterns not supported by OpenAI restrict state code
    # to be two capital letters (OpenAI does not support pattern fields)
    state_code: str = Field(pattern=r"[A-Z]{2}")

    # Zip code must be five digits
    zip_code: str = Field(pattern=r"\d{6}")

event_description = """
Rahul and Priya are going to a science fair on Friday, January 2024. 
The science fair is hosted at the auditorium of National Public School, Indiranagar at 12th Main Road, HAL 2nd Stage in Bangalore, Karnataka.
"""

def generate(
    response_model,
    user_prompt,
    system_prompt="You are a helpful assistant.",
    model="meta-llama/Meta-Llama-3.1-70B-Instruct-Turbo",
    # Can use 70b for a higher-quality model
    # model="meta-llama/Meta-Llama-3.1-70B-Instruct-Turbo",
    max_retries=3,
):
    event = instructor_client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt},
        ],
        response_model=response_model,
        max_retries=max_retries,
    )

    return event

system_prompt = """
Make a calendar event. Respond in JSON with
the event name, date, list of participants,
and the address.
"""

user_prompt = "Event description: " + event_description

event = generate(CalendarEvent, user_prompt, system_prompt=system_prompt)

event

event.model_dump()

from utils import UsageTracker

# Clear any completion response hooks -- prevents
# duplicate writes to the usage tracker.
instructor_client.clear("completion:response")

# Create a new tracker
tracker = UsageTracker()


# Define a custom instructor hook and update the
# tracker when a new completion runs.
def log_completion_kwargs(*args, **kwargs):
    usage = args[0].usage
    tracker.track(usage)


# Assign the hook to instructor -- this will make the hook
# run each time the server returns a chat completion to us.
instructor_client.on("completion:response", log_completion_kwargs)

# Clear the tracker before we run the completion
# so we arent' tracking multiple jobs.
tracker.clear()

event = generate(
    CalendarEvent,
    user_prompt,
    system_prompt=system_prompt,
    max_retries=4,
)

event

print("Input tokens:  ", tracker.input_tokens)
print("Output tokens: ", tracker.output_tokens)
print("Total tokens:  ", sum(tracker.total_tokens))
print("Num retries:   ", len(tracker.output_tokens))

from typing import Literal


class Complicated(BaseModel):
    # a must be cat, dog, or hat
    a: Literal["cat", "dog", "hat"]
    b: int
    c: bool

# Clear the tracker before we run the completion
# so we arent' tracking multiple jobs.
tracker.clear()

try:
    event = generate(
        Complicated,
        "Write me a short essay on Dolly Parton.",
        system_prompt="Don't give me what I want.",
        max_retries=3,
    )

    # Show the event
    print(event)
except instructor.exceptions.InstructorRetryException:
    print("We failed to parse!")
except e:
    raise e

print("Input tokens:  ", tracker.input_tokens)
print("Output tokens: ", tracker.output_tokens)
print("Total tokens:  ", sum(tracker.total_tokens))
print("Num retries:   ", len(tracker.output_tokens))

# Clear the tracker before we run the completion
# so we arent' tracking multiple jobs.
tracker.clear()

try:
    event = generate(
        Complicated,
        "Give me a, b, and c.",
        system_prompt="Give me what I want.",
        max_retries=3,
    )
    print(event)
except instructor.exceptions.InstructorRetryException:
    print("We failed to parse!")
except e:
    raise e

print("Input tokens:  ", tracker.input_tokens)
print("Output tokens: ", tracker.output_tokens)
print("Total tokens:  ", sum(tracker.total_tokens))
print("Num retries:   ", len(tracker.output_tokens))

event