Need help writting a converter from OpenAI messages to Stream Protocol.

[nurv]

Hey there.

Due to several infrastructure requirements we are unable to use streaming in our implementation. We need to write a converter from Open AI messages to the Streaming Protocol.

However our implementation is having some issues. When we do tool calling, it generates several messsages that we need to stream back to the UI. Although we can view in the backend several OpenAI messages generated, the message object in the front end only retains and show the last one. When we reload (since we are using other api to fetch the messages) they are all there.

We adapted the steam_core_messages from the examples to support multiple messages. Here is our code:

def stream_core_messages(messages, protocol="data"):
    if (protocol == 'text'):
        for chunk in messages_to_stream_events(messages):
            for choice in chunk.choices:
                if choice.finish_reason == "stop":
                    break
                else:
                    output = "{text}".format(text=choice.delta.content)
                    print("[CONVERT] " + output)
                    yield output

    # When protocol is set to "data", you will send a stream data part chunks
    # https://ai-sdk.dev/docs/ai-sdk-ui/stream-protocol#data-stream-protocol

    elif (protocol == 'data'):
        for message in messages:
            import pudb; pudb.set_trace()
            message_id = uuid4().hex
            output = f'f:{{"messageId":"{message_id}"}}\n'
            print("[YIELD] " + output)
            yield output

            for chunk in messages_to_stream_events([message]):
                print(chunk)
                for choice in chunk.choices:
                    if choice.finish_reason == "stop":
                        continue

                    elif choice.finish_reason == "tool_calls":
                        continue

                    elif choice.delta.tool_calls:
                        for tool_call in choice.delta.tool_calls:
                            id = tool_call['id']
                            name = tool_call['function']['name']
                            arguments = tool_call['function']['arguments']

                            output = '9:{{"toolCallId":"{id}","toolName":"{name}","args":{args}}}\n'.format(
                                id=id,
                                name=name,
                                args=arguments)
                            print("[YIELD] " + output)
                            yield output

                    elif choice.delta.tool_response:
                            output = 'a:{{"toolCallId":"{id}","result":{result}}}\n'.format(
                                id=choice.delta.tool_call_id,
                                result=json.dumps(choice.delta.tool_response))
                            print("[YIELD] " + output)
                            yield output
                    else:
                        output = '0:{text}\n'.format(text=json.dumps(choice.delta.content))
                        print("[YIELD] " + output)
                        yield output

            output = 'd:{{"finishReason":"{reason}","usage":{{"promptTokens":{prompt},"completionTokens":{completion}}}}}\n'.format(
                reason="stop",
                prompt=0,
                completion=0
            )
            print("[YIELD] " + output)
            yield output

For the following messages:

[{'content': '',
  'role': 'assistant',
  'tool_calls': [{'function': {'arguments': '{"input": {"plan": {"audience": '
                                            '"IPA Drinkers", "research_goal": '
                                            '"Food Interests"}, "subplan": '
                                            '{"type": "standard", "maxStudy": '
                                            '1}}}',
                               'name': 'create_plan'},
                  'id': 'call_gdDSs1PfHvojyw6QAmiioXgs',
                  'type': 'function'},
                 {'function': {'arguments': '{"input": {"plan": {"audience": '
                                            '"IPA Drinkers", "research_goal": '
                                            '"Food Interests"}, "subplan": '
                                            '{"type": "standard", "maxStudy": '
                                            '1}}}',
                               'name': 'create_plan'},
                  'id': 'call_JEXnAs5QnNIFNMTx04Ei9Hlf',
                  'type': 'function'}]},
 {'content': '"Created the following:\n'
             '* Plan named `Plan 66` with reference '
             'plan://ec40538e-5bd7-4649-b50d-3aad1dd3c142\n'
             '* Subplan with reference '
             'subplan://8c6b98a7-9b75-45ee-b239-48f712a78d61',
  'name': 'create_plan',
  'role': 'tool',
  'tool_call_id': 'call_gdDSs1PfHvojyw6QAmiioXgs'},
 {'content': '"Created the following:\n'
             '* Plan named `Plan 67` with reference '
             'plan://9826d0f9-231a-4f4d-81f9-a35f2fdcaf16\n'
             '* Subplan with reference '
             'subplan://375b46cd-a653-4327-8ac5-320b5c8b00ca',
  'name': 'create_plan',
  'role': 'tool',
  'tool_call_id': 'call_JEXnAs5QnNIFNMTx04Ei9Hlf'},
 {'content': 'I have created two standard research studies targeting the food '
             'interests of IPA Drinkers:\n'
             '\n'
             '1. **Plan 66** with a subplan reference: '
             '[subplan://8c6b98a7-9b75-45ee-b239-48f712a78d61](#).\n'
             '2. **Plan 67** with a subplan reference: '
             '[subplan://375b46cd-a653-4327-8ac5-320b5c8b00ca](#).\n'
             '\n'
             'These studies will help you gather insights into the food '
             'preferences within this specific audience.',
  'role': 'assistant'}]

Its generating the following event stream:

[YIELD] f:{"messageId":"b0928904ac8b465f9e7fb64b13c8c2c9"}

[YIELD] 9:{"toolCallId":"call_gdDSs1PfHvojyw6QAmiioXgs","toolName":"create_plan","args":{"input": {"plan": {"audience": "IPA Drinkers", "research_goal": "Food Interests"}, "subplan": {"type": "standard", "maxStudy": 1}}}}

[YIELD] 9:{"toolCallId":"call_JEXnAs5QnNIFNMTx04Ei9Hlf","toolName":"create_plan","args":{"input": {"plan": {"audience": "IPA Drinkers", "research_goal": "Food Interests"}, "subplan": {"type": "standard", "maxStudy": 1}}}}

[YIELD] d:{"finishReason":"stop","usage":{"promptTokens":0,"completionTokens":0}}

[YIELD] f:{"messageId":"29b48a7b6608423aa95a80a2246c72e4"}

[YIELD] a:{"toolCallId":"call_gdDSs1PfHvojyw6QAmiioXgs","result":"\"Created the following:\n* Plan named `Plan 66` with reference plan://ec40538e-5bd7-4649-b50d-3aad1dd3c142\n* Subplan with reference subplan://8c6b98a7-9b75-45ee-b239-48f712a78d61"}

[YIELD] d:{"finishReason":"stop","usage":{"promptTokens":0,"completionTokens":0}}

[YIELD] f:{"messageId":"01c1aa58774b422484123596ce6d8ffb"}

[YIELD] a:{"toolCallId":"call_JEXnAs5QnNIFNMTx04Ei9Hlf","result":"\"Created the following:\n* Plan named `Plan 66` with reference plan://9826d0f9-231a-4f4d-81f9-a35f2fdcaf16\n* Subplan with reference subplan://375b46cd-a653-4327-8ac5-320b5c8b00ca"}

[YIELD] d:{"finishReason":"stop","usage":{"promptTokens":0,"completionTokens":0}}

[YIELD] f:{"messageId":"903bfce3584a47df8e458c33e6d96706"}

[YIELD] 0:"I have created two standard research studies targeting the food interests of IPA Drinkers:\n\n1. **Plan 66** with a subplan reference: [subplan://8c6b98a7-9b75-45ee-b239-48f712a78d61](#).\n2. **Plan 66** with a subplan reference: [subplan://375b46cd-a653-4327-8ac5-320b5c8b00ca](#).\n\nThese studies will help you gather insights into the food preferences within this specific audience."

[YIELD] d:{"finishReason":"stop","usage":{"promptTokens":0,"completionTokens":0}}

We can't figure out what could be causing this.

[ErRickow]

Hey there.

Due to several infrastructure requirements we are unable to use streaming in our implementation. We need to write a converter from Open AI messages to the Streaming Protocol.

However our implementation is having some issues. When we do tool calling, it generates several messsages that we need to stream back to the UI. Although we can view in the backend several OpenAI messages generated, the message object in the front end only retains and show the last one. When we reload (since we are using other api to fetch the messages) they are all there.

We adapted the steam_core_messages from the examples to support multiple messages. Here is our code:

def stream_core_messages(messages, protocol="data"):
    if (protocol == 'text'):
        for chunk in messages_to_stream_events(messages):
            for choice in chunk.choices:
                if choice.finish_reason == "stop":
                    break
                else:
                    output = "{text}".format(text=choice.delta.content)
                    print("[CONVERT] " + output)
                    yield output

    # When protocol is set to "data", you will send a stream data part chunks
    # https://ai-sdk.dev/docs/ai-sdk-ui/stream-protocol#data-stream-protocol

    elif (protocol == 'data'):
        for message in messages:
            import pudb; pudb.set_trace()
            message_id = uuid4().hex
            output = f'f:{{"messageId":"{message_id}"}}\n'
            print("[YIELD] " + output)
            yield output

            for chunk in messages_to_stream_events([message]):
                print(chunk)
                for choice in chunk.choices:
                    if choice.finish_reason == "stop":
                        continue

                    elif choice.finish_reason == "tool_calls":
                        continue

                    elif choice.delta.tool_calls:
                        for tool_call in choice.delta.tool_calls:
                            id = tool_call['id']
                            name = tool_call['function']['name']
                            arguments = tool_call['function']['arguments']

                            output = '9:{{"toolCallId":"{id}","toolName":"{name}","args":{args}}}\n'.format(
                                id=id,
                                name=name,
                                args=arguments)
                            print("[YIELD] " + output)
                            yield output

                    elif choice.delta.tool_response:
                            output = 'a:{{"toolCallId":"{id}","result":{result}}}\n'.format(
                                id=choice.delta.tool_call_id,
                                result=json.dumps(choice.delta.tool_response))
                            print("[YIELD] " + output)
                            yield output
                    else:
                        output = '0:{text}\n'.format(text=json.dumps(choice.delta.content))
                        print("[YIELD] " + output)
                        yield output

            output = 'd:{{"finishReason":"{reason}","usage":{{"promptTokens":{prompt},"completionTokens":{completion}}}}}\n'.format(
                reason="stop",
                prompt=0,
                completion=0
            )
            print("[YIELD] " + output)
            yield output

For the following messages:

[{'content': '',
  'role': 'assistant',
  'tool_calls': [{'function': {'arguments': '{"input": {"plan": {"audience": '
                                            '"IPA Drinkers", "research_goal": '
                                            '"Food Interests"}, "subplan": '
                                            '{"type": "standard", "maxStudy": '
                                            '1}}}',
                               'name': 'create_plan'},
                  'id': 'call_gdDSs1PfHvojyw6QAmiioXgs',
                  'type': 'function'},
                 {'function': {'arguments': '{"input": {"plan": {"audience": '
                                            '"IPA Drinkers", "research_goal": '
                                            '"Food Interests"}, "subplan": '
                                            '{"type": "standard", "maxStudy": '
                                            '1}}}',
                               'name': 'create_plan'},
                  'id': 'call_JEXnAs5QnNIFNMTx04Ei9Hlf',
                  'type': 'function'}]},
 {'content': '"Created the following:\n'
             '* Plan named `Plan 66` with reference '
             'plan://ec40538e-5bd7-4649-b50d-3aad1dd3c142\n'
             '* Subplan with reference '
             'subplan://8c6b98a7-9b75-45ee-b239-48f712a78d61',
  'name': 'create_plan',
  'role': 'tool',
  'tool_call_id': 'call_gdDSs1PfHvojyw6QAmiioXgs'},
 {'content': '"Created the following:\n'
             '* Plan named `Plan 67` with reference '
             'plan://9826d0f9-231a-4f4d-81f9-a35f2fdcaf16\n'
             '* Subplan with reference '
             'subplan://375b46cd-a653-4327-8ac5-320b5c8b00ca',
  'name': 'create_plan',
  'role': 'tool',
  'tool_call_id': 'call_JEXnAs5QnNIFNMTx04Ei9Hlf'},
 {'content': 'I have created two standard research studies targeting the food '
             'interests of IPA Drinkers:\n'
             '\n'
             '1. **Plan 66** with a subplan reference: '
             '[subplan://8c6b98a7-9b75-45ee-b239-48f712a78d61](#).\n'
             '2. **Plan 67** with a subplan reference: '
             '[subplan://375b46cd-a653-4327-8ac5-320b5c8b00ca](#).\n'
             '\n'
             'These studies will help you gather insights into the food '
             'preferences within this specific audience.',
  'role': 'assistant'}]

Its generating the following event stream:

[YIELD] f:{"messageId":"b0928904ac8b465f9e7fb64b13c8c2c9"}

[YIELD] 9:{"toolCallId":"call_gdDSs1PfHvojyw6QAmiioXgs","toolName":"create_plan","args":{"input": {"plan": {"audience": "IPA Drinkers", "research_goal": "Food Interests"}, "subplan": {"type": "standard", "maxStudy": 1}}}}

[YIELD] 9:{"toolCallId":"call_JEXnAs5QnNIFNMTx04Ei9Hlf","toolName":"create_plan","args":{"input": {"plan": {"audience": "IPA Drinkers", "research_goal": "Food Interests"}, "subplan": {"type": "standard", "maxStudy": 1}}}}

[YIELD] d:{"finishReason":"stop","usage":{"promptTokens":0,"completionTokens":0}}

[YIELD] f:{"messageId":"29b48a7b6608423aa95a80a2246c72e4"}

[YIELD] a:{"toolCallId":"call_gdDSs1PfHvojyw6QAmiioXgs","result":"\"Created the following:\n* Plan named `Plan 66` with reference plan://ec40538e-5bd7-4649-b50d-3aad1dd3c142\n* Subplan with reference subplan://8c6b98a7-9b75-45ee-b239-48f712a78d61"}

[YIELD] d:{"finishReason":"stop","usage":{"promptTokens":0,"completionTokens":0}}

[YIELD] f:{"messageId":"01c1aa58774b422484123596ce6d8ffb"}

[YIELD] a:{"toolCallId":"call_JEXnAs5QnNIFNMTx04Ei9Hlf","result":"\"Created the following:\n* Plan named `Plan 66` with reference plan://9826d0f9-231a-4f4d-81f9-a35f2fdcaf16\n* Subplan with reference subplan://375b46cd-a653-4327-8ac5-320b5c8b00ca"}

[YIELD] d:{"finishReason":"stop","usage":{"promptTokens":0,"completionTokens":0}}

[YIELD] f:{"messageId":"903bfce3584a47df8e458c33e6d96706"}

[YIELD] 0:"I have created two standard research studies targeting the food interests of IPA Drinkers:\n\n1. **Plan 66** with a subplan reference: [subplan://8c6b98a7-9b75-45ee-b239-48f712a78d61](#).\n2. **Plan 66** with a subplan reference: [subplan://375b46cd-a653-4327-8ac5-320b5c8b00ca](#).\n\nThese studies will help you gather insights into the food preferences within this specific audience."

[YIELD] d:{"finishReason":"stop","usage":{"promptTokens":0,"completionTokens":0}}

We can't figure out what could be causing this.

the output schema is not similar with openai api response json structure. maybe thats causing error