Dynamically generated widgets for queries

[ahuang11]

Screen.Recording.2024-03-04.at.5.35.11.PM.mov

import json
import datetime
from typing import Tuple, Dict, Type, Callable, Union
from typing import Literal

import param
import instructor
from openai import OpenAI
from pydantic import BaseModel, Field, create_model as _create_model
from pydantic.fields import FieldInfo
import panel as pn

DATE_TYPE = Union[datetime.datetime, datetime.date]
PARAM_TYPE_MAPPING: Dict[param.Parameter, Type] = {
    param.String: str,
    param.Integer: int,
    param.Number: float,
    param.Boolean: bool,
    param.Event: bool,
    param.Date: DATE_TYPE,
    param.DateRange: Tuple[DATE_TYPE],
    param.CalendarDate: DATE_TYPE,
    param.CalendarDateRange: Tuple[DATE_TYPE],
    param.Parameter: object,
    param.Color: str,
    param.Callable: Callable,
    param.List: list,
    param.ObjectSelector: object,
}

pn.extension()


class FieldWidgetName(BaseModel):

    label: str
    widget_name: Literal[pn.widgets.__all__]


class BestMatch(BaseModel):

    response: str = Field(description="Be a helpful chatbot assistant.")

    requires_widget: bool = Field(
        description="Whether the query requires a widget to be created."
    )

    field_widget: FieldWidgetName | None = Field(
        default=None,
        description=(
            "The most suitable widgets to use to collect "
            "user input in a form based on the query."
        ),
    )


def _create_model_from_widget(widget_cls: Type[pn.widgets.Widget]) -> Type[BaseModel]:
    param_fields = {}
    common_keys = pn.widgets.Widget.param.values().keys()
    for key in widget_cls.param.values().keys() - common_keys | {"name"}:
        type_ = PARAM_TYPE_MAPPING.get(type(widget_cls.param[key]), str)
        param_fields[key] = (
            type_,
            FieldInfo(
                description=getattr(widget_cls.param, key).doc,
                default=None,
                required=False,
            ),
        )
    doc = (
        "Hydrate this based on the initial query. Ensure the `name` is human readable."
    )
    return _create_model(widget_cls.__name__, __doc__=doc, **param_fields)


def _hydrate_widget(widget_cls: Type[pn.widgets.Widget], **kwargs) -> pn.widgets.Widget:
    return widget_cls(
        **{key: value for key, value in kwargs.items() if value is not None}
    )


def _format_message(content: str, role: str = "user"):
    return {"role": role, "content": str(content)}


def _generate_response(messages: list, response_model: Type[BaseModel]):
    return client.chat.completions.create(
        model="gpt-4", response_model=response_model, messages=messages
    )


def _create_widget(best_match):
    widget_cls = getattr(pn.widgets, best_match.widget_name)
    widget_label = best_match.label
    widget_model = _create_model_from_widget(widget_cls)
    messages.extend(
        [
            _format_message(
                f"Creating {json.dumps(widget_model.model_json_schema())} for {widget_label}",
                role="assistant",
            )
        ]
    )
    kwargs = _generate_response(messages, widget_model)
    widget = _hydrate_widget(widget_cls, **dict(kwargs))
    pn.bind(
        react_to_value,
        widget,
        name=widget.name,
        watch=True,
    )
    return widget


def react_to_value(value, name):
    content = f"Selected: {value} from {name=} widget."
    messages.append(_format_message(content))
    chat.send(value)
    chat.widgets = [pn.chat.ChatAreaInput()]


def respond(query: str, user: str, instance: pn.chat.ChatInterface):
    messages.append(_format_message(query))

    best_match = _generate_response(messages, BestMatch)
    yield best_match.response

    messages.append(_format_message(best_match.response, role="assistant"))
    if best_match.requires_widget:
        widget = _create_widget(best_match.field_widget)
        instance.widgets = [widget]


messages = []
client = instructor.patch(OpenAI())
chat = pn.chat.ChatInterface(
    callback=respond,
    auto_send_types=[],
    help_text="Help answer your questions using Panel widgets.",
    callback_exception="raise",
)
chat.show()