Streaming with PARALLEL_TOOLS

[jleguina]

Hi all, just wondering how to use parallel tools with streaming?

My current setup involves the agent responding to the user by calling a ResponseTool, within which I relay the streamed response to the user. This means that every step of the agent effectively needs to be streamed: if the chosen tool is not ResponseTool I simply wait for the full output to then call the function; otherwise I stream directly to the front-end.

I understand perhaps this is not the correct way of terminating agent loops. Any advice is welcome!

Thanks!

MRE:

import instructor

from openai import OpenAI
from pydantic import BaseModel, Field
from typing import Iterable


class PrintTool(BaseModel):
    text: str = Field(description="text to print")

    def execute(self):
        print("PrintTool:")
        print(self.text)


class ResponseTool(BaseModel):
    text: str = Field(description="text to send to respond to the user with")

    def execute(self):
        print("ResponseTool:")
        print(self.text)


class MRKLSchema(BaseModel):
    thought: str
    tools: Iterable[PrintTool | ResponseTool]


client = instructor.patch(OpenAI(api_key="xxx"), mode=instructor.Mode.PARALLEL_TOOLS)

while True:
    response = client.chat.completions.create(
        messages=[
            {
                "role": "system",
                "content": "you are a helpful assitant",
            },
            {
                "role": "user",
                "content": "print the following message and return ot to me: 'Hello world'",
            },
        ],
        response_model=Iterable[MRKLSchema],
        model="gpt-4-turbo-preview",
        stream=True,
    )
    response = cast(Iterable[MRKLSchema], response)

    stop = False
    for chunk in response:
        if stop:
          break 

        if chunk.tools:
            for tool in chunk.tools:  # For simplicity making this a normal for
                if isinstance(tool, ResponseTool):
                    ResponseTool(text=tool.text).execute()
                    stop = True
                    break
                else:
                    PrintTool(text=tool.text).execute()

raises:

/usr/local/lib/python3.10/dist-packages/instructor/process_response.py in handle_response_model(response_model, mode, **kwargs)
    199         if mode == Mode.PARALLEL_TOOLS:
    200             assert (
--> 201                 new_kwargs.get("stream", False) is False
    202             ), "stream=True is not supported when using PARALLEL_TOOLS mode"
    203             new_kwargs["tools"] = handle_parallel_model(response_model)

AssertionError: stream=True is not supported when using PARALLEL_TOOLS mode

[jxnl]

this should be tools not parallel tools.

also response and print tool both are {"text": str} and isomorphic so theres no way you'll be able to resolve the two.

try

class Tool
type: Literal["print", "response"]
text: str

[jleguina]

Thanks for getting back.

In this example I’m calling tools on a for loop for simplicity, but I would be calling them in parallel. I’ve also simplified the tools but in general they won’t have the same inputs.

The response I would expect from the model would be

• Step 1: tools=[PrintTool(…), PrintTool(…), PrintTool(…)] (I would call these in parallel)
• Step 2: tools=[ResponseTool(…)]

[jxnl]

right what im saying is you are not using parallel tool as intented.
https://python.useinstructor.com/concepts/parallel

[jleguina]

Sorry I might be missing something, but the example in the docs raises the same issue as above when stream=True. I assume there's no way to do that then?

[jxnl]

you are overcomplicating this, here is a sample to get you started.

you made an object that is an iterable, only to make that an iterable.

import instructor

from typing import Literal
from openai import OpenAI
from pydantic import BaseModel, Field
from typing import Iterable



class ActionTool(BaseModel):
    chain_of_thought: str = Field(description="the chain of thought that this response is a part of")
    action_type: Literal["print", "response"]
    text: str = Field(description="text to send to respond to the user with")

    def execute(self):
        print(f"Action type is '{self.action_type}'")
        if self.action_type == "print":
            print(self.text)
        elif self.action_type == "response":
            return self.text


client = instructor.patch(OpenAI(), mode=instructor.Mode.TOOLS)

response = client.chat.completions.create(
    messages=[
        {
            "role": "system",
            "content": "you are a helpful assitant",
        },
        {
            "role": "user",
            "content": "print the following message and return to to me: 'Hello world'",
        },
    ],
    response_model=Iterable[ActionTool],
    model="gpt-4-turbo-preview",
    stream=True,
)

for action in response:
    action.execute()

[jleguina]

Thanks a lot for that!

There's two problems with the above:

1. In general, each tool will take a different Pydantic model as an input (my bad, the MRE is leading to a lot of confusion). So I imagine ActionTool should become something like:

    class ActionTool(BaseModel):
        chain_of_thought: str = Field(description="the chain of thought that this response is a part of")
        action_type: Literal["foo", "bar"]
        - text: str = Field(description="text to send to respond to the user with")
        + action_input: FooModel | BarModel

2. I want multiple calls to those tools for each agent step, hence the inner Iterable and the need for PARALLEL_TOOLS. So then my intuition (based on this) was to do:

    class ActionTool(BaseModel):
        chain_of_thought: str = Field(description="the chain of thought that this response is a part of")
        action_type: Literal["foo", "bar"]
        - action_input: FooModel | BarModel
        + action_input: Iterable[FooModel | BarModel]

So essentially, the outer Iterable[ActionTool] is for the streaming, while the inner Iterable[FooModel | BarModel] is meant to indicate the need for parallel tools.