Conditional flow with tool calls

[r-leyshon]

Hi there,

I'm interested in this package for a more complex chatbot that I've been developing. I need to preferably stream openAI responses with tools. Getting this to work with the shiny chat component has been a little tricky, but then I came across a reference to chatlas in the docstring for the shiny Chat class. Working through your docs and examples, I can get so far with it.

This approach allows for streaming responses with tool calls and shiny ui
widgets. Though the tool calls need to be self-contained. Whatever the tool
returns is fed into the model for a subsequent response, rather than being
available for additional conditional flow.

I would like to instead receive the function name and parameter values and
call get_current_temperature(), passing the json response back to model.
This would allow me to display a notification without relying on a side
effect.

from chatlas import ChatOpenAI, Turn
from shiny.express import ui
import dotenv
import requests

openai_key = dotenv.dotenv_values()["OPENAI_KEY"]
messages = [Turn(role="system", contents="You are a helpful but terse assistant.")]
messages.append(Turn(role="assistant", contents = "Hi! How can I help you today?"))


# func contains side effect which I would prefer to handle outside
def get_current_temperature(latitude: float, longitude: float):
    """
    Get the current weather given a latitude and longitude.

    Parameters
    ----------
    latitude
        The latitude of the location.
    longitude
        The longitude of the location.
    """
    lat_lng = f"latitude={latitude}&longitude={longitude}"
    url = f"https://api.open-meteo.com/v1/forecast?{lat_lng}&current=temperature_2m,wind_speed_10m&hourly=temperature_2m,relative_humidity_2m,wind_speed_10m"
    response = requests.get(url)
    json = response.json()
    weather_now = json['current']
    ui.notification_show(f"Queried weather API... {weather_now}")
    return weather_now


chat_model = ChatOpenAI(
    api_key=openai_key,
    turns=messages,
    )
chat_model.register_tool(get_current_temperature)
chat = ui.Chat(
    id="ui_chat",
    messages=["Hi! How can I help you today?"],
)
chat.ui()


@chat.on_user_submit
async def handle_user_input():
    response = chat_model.stream(chat.user_input())
    await chat.append_message_stream(response)

[gadenbuie]

Hi, thanks for your question @r-leyshon. I'm having a hard time understanding the question though. Could you outline, maybe in pseudo-code, what you're wanting to do?

I think in particular I'm having a hard time seeing what you mean with "the tool call response ... being available for additional conditional flow".

[r-leyshon]

@gadenbuie thanks for the quick look at that, I'll try to show what I intend to do with the streamed model response below:

response_stream = chat_model.stream(chat.user_input())
async for part in response_stream:
    if part.tool_call:
        ui.notification_show("A tool call is required")
        tool_name = # collect func name
        tool_args = # collect func args
        if tool_name == "foo":
            foo(tool_args)
    else:
        await chat.append_message_stream(response_stream)

I've struggled to inspect the streamed parts and do things with them, at least when trying to await certain events for async appending to the message stream. My fallback is to switch off streaming but I'd rather not do that.

[gadenbuie]

Thanks for the clarification. Right now, I don't think it's possible to do what you're trying to do, at least not with the streaming response approach. Given that, I think it's very reasonable to use something like a notification to indicate that a tool was called. We'll definitely keep your use case in mind as we develop chatlas and the Shiny chat component.

[r-leyshon]

Gotcha, no worries I can fall back to removing streaming. Incase I wasn't clear, I need to handle the response for reasons other than showing a notification to the user. The notification would be a placeholder for the conditional logic that I'm using to pull the model response apart for other things that would be too onerous to share in the little example above.

I wanted to say that the chatlas approach seems very friendly and in the future, if I can keep each tool as a functional unit, I shall prefer to use chatlas over handling the response myself.

[gadenbuie]

The notification would be a placeholder for the conditional logic that I'm using to pull the model response apart for other things that would be too onerous to share in the little example above.

Could you describe the kinds of things you're wanting to do with the model response? That would help us understand the use case.

I'm going to reopen this issue so that we make sure we keep it in mind moving forward.

[r-leyshon]

Sure thing, the code I've fallen back to lives here: https://github.com/ministryofjustice/github-chat ,

Specifically this bit:

...
            elif (tool_call := resp.message.tool_calls):
                function_name = tool_call[0].function.name
                arguments = tool_call[0].function.arguments
                sanitised_func_nm = sanitise_string(function_name)
                sanitised_args = [
                    sanitise_string(arg) for arg in
                    json.loads(arguments)["keywords"]
                    ]
                if sanitised_func_nm == "ExtractKeywordEntities":
                    # Pydantic will raise if keywords violate schema rules
                    extracted_terms = ExtractKeywordEntities(
                        keywords=sanitised_args
                        )
                    ui.notification_show(
                        ("Searching database for keywords:"
                        f" {', '.join(extracted_terms.keywords)}")
                        )
                    summarise_this = chroma_pipeline.execute_pipeline(
                        keywords=extracted_terms.keywords,
                        n_results=input.selected_n(),
                        distance_threshold=input.dist_thresh(),
                        sanitised_prompt=sanitised_prompt
                    )
                    if (n_removed := chroma_pipeline.total_removed) > 0:
                        ui.notification_show(
                            f"{n_removed} results were removed."
                            )
                    if len(chroma_pipeline.results) == 0:
                        ui.notification_show(
                            "No results shown, increase distance threshold"
                            )
                    stream.append(summarise_this)
                    response = openai_client.chat.completions.create(
                        **completions_params
                        )
                    meta_resp = {
                        "role": "assistant",
                        "content": response.choices[0].message.content
                        }
                    await chat.append_message(response)
                    await chat.append_message(
                        {
                            "role": "asisstant",
                            "content": chroma_pipeline.chat_ui_results
                            })
                    stream.append(meta_resp)
...

[cpsievert]

@r-leyshon would it help you solve your use case if the chat_model had something like a .on_tool_request()/.on_tool_response() method which takes a callback that gets passed the tool request/response information?

[r-leyshon]

@r-leyshon would it help you solve your use case if the chat_model had something like a .on_tool_request()/.on_tool_response() method which takes a callback that gets passed the tool request/response information?

@cpsievert something like this?

async def handle_tool_request(tool_call):
    """Callback for handling tool requests."""
    ui.notification_show("A tool call is required")
    tool_name = tool_call.name
    tool_args = tool_call.arguments

    if tool_name == "foo":
        # Execute the required tool function
        result = foo(tool_args)
        ui.notification_show(f"Tool {tool_name} executed with args {tool_args}")
        return result

async def handle_tool_response(response_part):
    """Callback for handling regular responses."""
    await chat.append_message_stream(response_part)

# Instantiate the chat model and register callbacks
chat_model = ChatModel()
chat_model.on_tool_request(handle_tool_request)
chat_model.on_tool_response(handle_tool_response)

# Start streaming
response_stream = chat_model.stream(chat.user_input())
async for _ in response_stream:
    pass

[cpsievert]

Yea; although chatlas would still handle the tool execution on your behalf, and so in your case, I think it could be as simple as:

async def handle_tool_request(request: ContentToolRequest):
    name = request.name
    args = request.arguments
    ui.notification_show(f"Tool {name} executed with args {args}")

async def handle_tool_response(result: ContentToolResult):
    # ContentToolResult currently doesn't have a name attribute, but maybe it should?
    name = result.name
    res = result.value
    ui.notification_show(f"Tool {name} resulted in a value of: {res}")

# Instantiate the chat model and register callbacks
chat_model = ChatModel()
chat_model.on_tool_request(handle_tool_request)
chat_model.on_tool_response(handle_tool_response)

# Start streaming
response_stream = chat_model.stream(chat.user_input())
async for _ in response_stream:
    pass

Note that the model is given the tool result in the prompt, and most models tend to make it clear that they're using tool information in their response. so including the result in the middle of the message may lead to a weird result (this is why I changed the response handler to be a notification instead of inserted in the message stream)

[gadenbuie]

@cpsievert Could this approach be used to create an experience like the "Tool" demo here: https://www.assistant-ui.com/?

[cpsievert]

Could you be more specific? Are you referring to the "Tool UI" example?

[gadenbuie]

Are you referring to the "Tool UI" example?

Yeah exactly, there are examples there where the tool call results in a rich information display, e.g.

[cpsievert]

I don't know for sure, but seems it'd be possible with the proposal above