[OpenAI integration] Support IO capture when streaming function / tool call

[DanrForetellix]

Describe the bug

when using function calling and streaming, the output is not captured and tokens aren't counted

To reproduce

rom langfuse import Langfuse
from langfuse.decorators import observe
from langfuse.openai import openai # OpenAI integration
from openai import OpenAI

client = OpenAI()

tools = [
{
"type": "function",
"function": {
"name": "get_current_weather",
"description": "Get the current weather",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
}
},
"required": ["location"],
},
},
}
]
messages = [
{
"role": "system",
"content": "You are a helpful assistant.",
"role": "user",
"content": "What is the weather in boston",
}
]

@observe()
def get_openai_response(tools):
return client.chat.completions.create(
model="gpt-3.5-turbo-16k",
tools=tools,
tool_choice="auto",
messages=messages,
stream=True,
)

@observe()
def main():
num_chunk = 0
for chunk in get_openai_response(tools):
num_chunk += 1
print(f"num chunk = {num_chunk}")

main()

Additional information

No response

[marcklingen]

thanks for reporting

[hassiebp]

Thanks for reporting @DanrForetellix . This is a known limitation (tool call + streaming). Why exactly do you decide to stream the rather concise function call result?

[DanrForetellix]

@hassiebp @marcklingen
After digging into it a bit more, I actually figured the problem is somewhat more severe - it is not that function calling + streaming lacks reporting and counting of tokens, but actually that an exception is thrown - which happens in _extract_openai_response in openai.py, which tries to add objects of type ChoiceDeltaFunctionCall which is not supported.

Note also that a different issue when using the new tools parameters rather than the now deprecated function parameters. In this case an exception is not thrown, but I believe this is just a result of a typo in _extract_openai_response where sometimes the parameter is called tools_call and sometime tool_calls (note the plural s in different places). This in fact leads to the exception not being thrown and saves from the bug (but is a different sort of bug :) )

btw while badly documented, you can find quite a lot of posts in openai forums about people doing the streaming+functions solutions, for example
https://community.openai.com/t/help-for-function-calls-with-streaming/627170/5
https://community.openai.com/t/functions-calling-with-streaming/305742

[DanrForetellix]

The issue is that I am streaming the response, but I don't know from beforehand if it will be a function call or not; i.e. if it is not a function call I stream it to the user, otherwise I still use the stream API to build the function call iteratively with the same API. Indeed this is not well documented, but we are using a common solution which builds the function call response based on the received chunks (we use the function calling mainly to ensure the right format, and so the function parameters can include a lot of tokens - we might decide to stream the output to the user as well in the future). Hope that makes sense

…

On Tue, May 14, 2024 at 9:45 AM Hassieb Pakzad ***@***.***> wrote: Thanks for reporting @DanrForetellix <https://github.com/DanrForetellix> . This is a known limitation (tool call + streaming). Why exactly do you decide to *stream* the rather concise function call result? — Reply to this email directly, view it on GitHub <#1943 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/BHGH4KO6KVWX4YQN5CEBVODZCIIQ5AVCNFSM6AAAAABHCECATGVHI2DSMVQWIX3LMV43OSLTON2WKQ3PNVWWK3TUHMZDCMJQGI4DONJYGE> . You are receiving this because you were mentioned.Message ID: ***@***.***>

-- The information in this e-mail is confidential. The content may not be disclosed or used by anyone other than the addressee. Access to this e-mail by anyone else is unauthorised. If you are not the intended recipient, please notify Foretellix immediately and delete this e-mail. Foretellix cannot accept any responsibility for the accuracy or completeness of this e-mail as it has been sent over public networks. If you have any concerns over the content of this message or its Accuracy or Integrity, please contact Foretellix immediately. All outgoing e-mails from Foretellix are checked using regularly updated virus scanning software but you should take whatever measures you deem to be appropriate to ensure that this message and any attachments are virus free.  

[hassiebp]

Thanks for the explanation, @DanrForetellix! I'll convert this issue to a discussion, such that we can gauge interest by other users in supporting this use case which helps us prioritize building a solution for it.

[someshfengde]

Hi can this be resolved.

After some more debugging. I found it's indeed error from langfuse.openai

traceback

  + Exception Group Traceback (most recent call last):
  |   File "/Users/som/Downloads/code/cybergpt_backend/.venv/lib/python3.11/site-packages/uvicorn/protocols/http/h11_impl.py", line 408, in run_asgi
  |     result = await app(  # type: ignore[func-returns-value]
  |              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "/Users/som/Downloads/code/cybergpt_backend/.venv/lib/python3.11/site-packages/uvicorn/middleware/proxy_headers.py", line 84, in __call__
  |     return await self.app(scope, receive, send)
  |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "/Users/som/Downloads/code/cybergpt_backend/.venv/lib/python3.11/site-packages/fastapi/applications.py", line 1106, in __call__
  |     await super().__call__(scope, receive, send)
  |   File "/Users/som/Downloads/code/cybergpt_backend/.venv/lib/python3.11/site-packages/starlette/applications.py", line 122, in __call__
  |     await self.middleware_stack(scope, receive, send)
  |   File "/Users/som/Downloads/code/cybergpt_backend/.venv/lib/python3.11/site-packages/starlette/middleware/errors.py", line 184, in __call__
  |     raise exc
  |   File "/Users/som/Downloads/code/cybergpt_backend/.venv/lib/python3.11/site-packages/starlette/middleware/errors.py", line 162, in __call__
  |     await self.app(scope, receive, _send)
  |   File "/Users/som/Downloads/code/cybergpt_backend/.venv/lib/python3.11/site-packages/starlette/middleware/cors.py", line 91, in __call__
  |     await self.simple_response(scope, receive, send, request_headers=headers)
  |   File "/Users/som/Downloads/code/cybergpt_backend/.venv/lib/python3.11/site-packages/starlette/middleware/cors.py", line 146, in simple_response
  |     await self.app(scope, receive, send)
  |   File "/Users/som/Downloads/code/cybergpt_backend/.venv/lib/python3.11/site-packages/starlette/middleware/exceptions.py", line 79, in __call__
  |     raise exc
  |   File "/Users/som/Downloads/code/cybergpt_backend/.venv/lib/python3.11/site-packages/starlette/middleware/exceptions.py", line 68, in __call__
  |     await self.app(scope, receive, sender)
  |   File "/Users/som/Downloads/code/cybergpt_backend/.venv/lib/python3.11/site-packages/fastapi/middleware/asyncexitstack.py", line 20, in __call__
  |     raise e
  |   File "/Users/som/Downloads/code/cybergpt_backend/.venv/lib/python3.11/site-packages/fastapi/middleware/asyncexitstack.py", line 17, in __call__
  |     await self.app(scope, receive, send)
  |   File "/Users/som/Downloads/code/cybergpt_backend/.venv/lib/python3.11/site-packages/starlette/routing.py", line 718, in __call__
  |     await route.handle(scope, receive, send)
  |   File "/Users/som/Downloads/code/cybergpt_backend/.venv/lib/python3.11/site-packages/starlette/routing.py", line 276, in handle
  |     await self.app(scope, receive, send)
  |   File "/Users/som/Downloads/code/cybergpt_backend/.venv/lib/python3.11/site-packages/starlette/routing.py", line 69, in app
  |     await response(scope, receive, send)
  |   File "/Users/som/Downloads/code/cybergpt_backend/.venv/lib/python3.11/site-packages/starlette/responses.py", line 270, in __call__
  |     async with anyio.create_task_group() as task_group:
  |   File "/Users/som/Downloads/code/cybergpt_backend/.venv/lib/python3.11/site-packages/anyio/_backends/_asyncio.py", line 680, in __aexit__
  |     raise BaseExceptionGroup(
  | ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
  +-+---------------- 1 ----------------
    | Traceback (most recent call last):
    |   File "/Users/som/Downloads/code/cybergpt_backend/.venv/lib/python3.11/site-packages/starlette/responses.py", line 273, in wrap
    |     await func()
    |   File "/Users/som/Downloads/code/cybergpt_backend/.venv/lib/python3.11/site-packages/starlette/responses.py", line 262, in stream_response
    |     async for chunk in self.body_iterator:
    |   File "/Users/som/Downloads/code/cybergpt_backend/openai_function_calling.py", line 423, in run_conversation
    |     completion = openai.chat.completions.create(
    |                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    |   File "/Users/som/Downloads/code/cybergpt_backend/.venv/lib/python3.11/site-packages/langfuse/openai.py", line 151, in wrapper
    |     return func(open_ai_definitions, initialize, wrapped, args, kwargs)
    |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    |   File "/Users/som/Downloads/code/cybergpt_backend/.venv/lib/python3.11/site-packages/langfuse/openai.py", line 488, in _wrap
    |     generation, is_nested_trace = _get_langfuse_data_from_kwargs(
    |                                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    |   File "/Users/som/Downloads/code/cybergpt_backend/.venv/lib/python3.11/site-packages/langfuse/openai.py", line 264, in _get_langfuse_data_from_kwargs
    |     prompt = _extract_chat_prompt(kwargs)
    |              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    |   File "/Users/som/Downloads/code/cybergpt_backend/.venv/lib/python3.11/site-packages/langfuse/openai.py", line 181, in _extract_chat_prompt
    |     return _filter_image_data(kwargs.get("messages", []))
    |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    |   File "/Users/som/Downloads/code/cybergpt_backend/.venv/lib/python3.11/site-packages/langfuse/openai.py", line 642, in _filter_image_data
    |     if message.get("content", None) is not None:
    |        ^^^^^^^^^^^
    |   File "/Users/som/Downloads/code/cybergpt_backend/.venv/lib/python3.11/site-packages/pydantic/main.py", line 811, in __getattr__
    |     raise AttributeError(f'{type(self).__name__!r} object has no attribute {item!r}')
    | AttributeError: 'ChatCompletionMessage' object has no attribute 'get'
    +------------------------------------

the messages object was something similar to this

messages = [{'role': 'user', 'content': 'what is the most efficient way to store responses.'},
ChatCompletionMessage(content=None, role='assistant', function_call=None, tool_calls=[ChatCompletionMessageToolCall(id='call_wGM41oVUgboqYCpSsaYCXLpH', function=Function(arguments='{"query":"efficient way to store responses"}', name='get_general_cybersec_context'), type='function')]),
{'tool_call_id': 'call_wGM41oVUgboqYCpSsaYCXLpH', 'role': 'tool', 'name': 'get_general_cybersec_context', 'content': 'responding with therequested data notice that the server response repeats the functio...ave welldocumented and incident response p'}]

If you use same message_list in openai official client it works. but here we are getting it has no attribute get

[marcklingen]

are you on the latest version of the Langfuse Python SDK? Happy to investigate. Please open a bug a report for this: langfuse.com/issues

[someshfengde]

yes I've just installed it.