Azure OpenAI content streaming with asynchronous filter still streams responses in large bursts #22246
-
Checked other resources
Commit to Help
Example Codefrom langchain_openai.chat_models.azure import AzureChatOpenAI
...
conversation_llm = AzureChatOpenAI(
streaming=True,
callback_manager=conversation_manager,
temperature=conversation_temperature,
deployment_name=conversation_model,
azure_endpoint=azure_api_base,
openai_api_version=azure_api_version,
openai_api_key=azure_api_key,
openai_api_type="azure"
)
...
doc_chain = load_qa_chain(
conversation_llm, chain_type="stuff", prompt=chat_prompt, callback_manager=default_manager
)
conversation_chain = ConversationalRetrievalChain(
retriever=vectorstore.as_retriever(search_type="similarity_score_threshold",
search_kwargs={"score_threshold": rag_score_threshold, "k": rag_top_k}),
combine_docs_chain=doc_chain,
question_generator=question_generator,
return_source_documents=True,
callback_manager=default_manager,
rephrase_question=False,
memory=memory,
max_tokens_limit=max_retrieval_tokens,
)
...
result = await qa_chain.ainvoke(
{
"question": question,
"chat_history": chat_history,
"xls_file_name": file_name,
"xls_survey_sheet": survey_sheet,
"xls_choices_sheet": choices_sheet,
"xls_settings_sheet": settings_sheet
}
)DescriptionI've enabled content streaming via the Azure OpenAI asynchronous filtering. (This had been restricted, but recently they opened it up.) However, streamed responses still come in in big bursts rather than continuously, with 2-10 seconds between bursts. The app UX is therefore quite poor. I have confirmed that the asynchronous filtering is enabled for my Azure OpenAI deployment, and it's working well against that deployment using the Azure OpenAI playground as the client. However, in my application, the tokens are streamed in these large bursts. My code uses a FastAPI back-end to stream responses to a React front-end, via a Websocket connection. This works perfectly well with a I should say that I'm still getting a stream of tokens with Has anybody else tried Thanks so much, Chris System InfoPackage Information
Packages not installed (Not Necessarily a Problem)The following packages were not found:
|
Beta Was this translation helpful? Give feedback.
Replies: 2 comments 6 replies
-
|
Yes, the @pytest.mark.scheduled
async def test_async_chat_openai_streaming() -> None:
"""Test that streaming correctly invokes on_llm_new_token callback."""
callback_handler = FakeCallbackHandler()
callback_manager = CallbackManager([callback_handler])
chat = _get_llm(
max_tokens=10,
streaming=True,
temperature=0,
callback_manager=callback_manager,
verbose=True,
)
message = HumanMessage(content="Hello")
response = await chat.agenerate([[message], [message]])
assert callback_handler.llm_streams > 0
assert isinstance(response, LLMResult)
assert len(response.generations) == 2
for generations in response.generations:
assert len(generations) == 1
for generation in generations:
assert isinstance(generation, ChatGeneration)
assert isinstance(generation.text, str)
assert generation.text == generation.message.contentThis test confirms that the asynchronous streaming correctly invokes the @pytest.mark.scheduled
def test_openai_streaming(llm: AzureOpenAI) -> None:
"""Test streaming tokens from AzureOpenAI."""
generator = llm.stream("I'm Pickle Rick")
assert isinstance(generator, Generator)
full_response = ""
for token in generator:
assert isinstance(token, str)
full_response += token
assert full_response
@pytest.mark.scheduled
async def test_openai_astream(llm: AzureOpenAI) -> None:
"""Test streaming tokens from AzureOpenAI."""
async for token in llm.astream("I'm Pickle Rick"):
assert isinstance(token, str)
@pytest.mark.scheduled
async def test_openai_abatch(llm: AzureOpenAI) -> None:
"""Test streaming tokens from AzureOpenAI."""
result = await llm.abatch(["I'm Pickle Rick", "I'm not Pickle Rick"])
for token in result:
assert isinstance(token, str)These tests collectively ensure that
|
Beta Was this translation helpful? Give feedback.
-
|
To achieve smooth, unbuffered streaming with Synchronous Streamingdef stream(
self,
messages: List[BaseMessage],
stop: Optional[List[str]] = None,
run_manager: Optional[CallbackManager] = None,
**kwargs: Any,
) -> Iterator[ChatGenerationChunk]:
self.endpoint_url = self.endpoint_url.replace("/chat/completions", "")
timeout = None if "timeout" not in kwargs else kwargs["timeout"]
import openai
params = {}
client_params = {
"api_key": self.endpoint_api_key.get_secret_value(),
"base_url": self.endpoint_url,
"timeout": timeout,
"default_headers": None,
"default_query": None,
"http_client": None,
}
client = openai.OpenAI(**client_params)
message_dicts = [
CustomOpenAIChatContentFormatter._convert_message_to_dict(m)
for m in messages
]
params = {"stream": True, "stop": stop, "model": None, **kwargs}
default_chunk_class = AIMessageChunk
for chunk in client.chat.completions.create(messages=message_dicts, **params):
if not isinstance(chunk, dict):
chunk = chunk.dict()
if len(chunk["choices"]) == 0:
continue
choice = chunk["choices"][0]
chunk = _convert_delta_to_message_chunk(
choice["delta"], default_chunk_class
)
generation_info = {}
if finish_reason := choice.get("finish_reason"):
generation_info["finish_reason"] = finish_reason
logprobs = choice.get("logprobs")
if logprobs:
generation_info["logprobs"] = logprobs
default_chunk_class = chunk.__class__
chunk = ChatGenerationChunk(
message=chunk, generation_info=generation_info or None
)
if run_manager:
run_manager.on_llm_new_token(chunk.text, chunk=chunk, logprobs=logprobs)
yield chunkAsynchronous Streamingasync def astream(
self,
messages: List[BaseMessage],
stop: Optional[List[str]] = None,
run_manager: Optional[CallbackManager] = None,
**kwargs: Any,
) -> AsyncIterator[ChatGenerationChunk]:
self.endpoint_url = self.endpoint_url.replace("/chat/completions", "")
timeout = None if "timeout" not in kwargs else kwargs["timeout"]
import openai
params = {}
client_params = {
"api_key": self.endpoint_api_key.get_secret_value(),
"base_url": self.endpoint_url,
"timeout": timeout,
"default_headers": None,
"default_query": None,
"http_client": None,
}
async_client = openai.AsyncOpenAI(**client_params)
message_dicts = [
CustomOpenAIChatContentFormatter._convert_message_to_dict(m)
for m in messages
]
params = {"stream": True, "stop": stop, "model": None, **kwargs}
default_chunk_class = AIMessageChunk
async for chunk in await async_client.chat.completions.create(
messages=message_dicts, **params
):
if not isinstance(chunk, dict):
chunk = chunk.dict()
if len(chunk["choices"]) == 0:
continue
choice = chunk["choices"][0]
chunk = _convert_delta_to_message_chunk(
choice["delta"], default_chunk_class
)
generation_info = {}
if finish_reason := choice.get("finish_reason"):
generation_info["finish_reason"] = finish_reason
logprobs = choice.get("logprobs")
if logprobs:
generation_info["logprobs"] = logprobs
default_chunk_class = chunk.__class__
chunk = ChatGenerationChunk(
message=chunk, generation_info=generation_info or None
)
if run_manager:
await run_manager.on_llm_new_token(
token=chunk.text, chunk=chunk, logprobs=logprobs
)
yield chunkThese methods use the OpenAI client to stream responses in real-time, ensuring that tokens are received smoothly and continuously, similar to the Azure OpenAI playground. The
|
Beta Was this translation helpful? Give feedback.
-
|
@dosu, I appreciate that there are new APIs for this now, but:
To be clear, are you saying both that (a) the In trying to imagine what could cause the playground's pattern of streaming behavior to depart from the streaming behavior I'm observing via LangChain, I got to thinking that it might be either (a) the API package being used on the Python side not calling an Azure endpoint that supports streaming, or (b) the Azure API version I'm using is buffering even when async filtering is enabled (i.e., doesn't support async filtering). For the latter possibility, I see that I'm passing in |
Beta Was this translation helpful? Give feedback.
-
|
Switching to the Here is a relevant test case that confirms the functionality of the @pytest.mark.scheduled
async def test_openai_astream(llm: AzureOpenAI) -> None:
"""Test streaming tokens from AzureOpenAI."""
async for token in llm.astream("I'm Pickle Rick"):
assert isinstance(token, str)Regarding the API version, ensure that your environment variables are correctly set: OPENAI_API_VERSION = os.environ.get("AZURE_OPENAI_API_VERSION", "")
OPENAI_API_BASE = os.environ.get("AZURE_OPENAI_API_BASE", "")
OPENAI_API_KEY = os.environ.get("AZURE_OPENAI_API_KEY", "")
DEPLOYMENT_NAME = os.environ.get(
"AZURE_OPENAI_DEPLOYMENT_NAME",
os.environ.get("AZURE_OPENAI_LLM_DEPLOYMENT_NAME", ""),
)Additionally, make sure to handle the following configurations to avoid common errors:
Here are the relevant test cases for these validations: def test_openai_streaming_best_of_error() -> None:
"""Test validation for streaming fails if best_of is not 1."""
with pytest.raises(ValueError):
_get_llm(best_of=2, streaming=True)
def test_openai_streaming_n_error() -> None:
"""Test validation for streaming fails if n is not 1."""
with pytest.raises(ValueError):
_get_llm(n=2, streaming=True)
def test_openai_streaming_multiple_prompts_error() -> None:
"""Test validation for streaming fails if multiple prompts are given."""
with pytest.raises(ValueError):
_get_llm(streaming=True).generate(["I'm Pickle Rick", "I'm Pickle Rick"])To ensure smooth, unbuffered streaming, you can refer to the LangChain documentation, which provides examples of both synchronous and asynchronous streaming: Sync streamfrom langchain_openai import OpenAI
llm = OpenAI(model="gpt-3.5-turbo-instruct", temperature=0, max_tokens=512)
for chunk in llm.stream("Write me a 1 verse song about sparkling water."):
print(chunk, end="|", flush=True)Async streamingfrom langchain_openai import OpenAI
llm = OpenAI(model="gpt-3.5-turbo-instruct", temperature=0, max_tokens=512)
async for chunk in llm.astream("Write me a 1 verse song about sparkling water."):
print(chunk, end="|", flush=True)Async event streamingfrom langchain_openai import OpenAI
llm = OpenAI(model="gpt-3.5-turbo-instruct", temperature=0, max_tokens=512)
idx = 0
async for event in llm.astream_events(
"Write me a 1 verse song about goldfish on the moon", version="v1"
):
idx += 1
if idx >= 5: # Truncate the output
print("...Truncated")
break
print(event)Make sure these environment variables and configurations are set correctly to ensure smooth operation [1][2][3].
|
Beta Was this translation helpful? Give feedback.
-
|
@dosu, that wasn't required. To anybody else who struggles with this: changing the |
Beta Was this translation helpful? Give feedback.
-
|
It looks like the issue you were experiencing with buffered streaming was due to the API version you were using. Changing the For anyone else facing similar issues, ensure you are using the latest API version that supports asynchronous filtering to achieve smooth, unbuffered streaming with
|
Beta Was this translation helpful? Give feedback.
-
|
Hi @chrislrobert, Like you, I have an Am I missing anything? |
Beta Was this translation helpful? Give feedback.
@dosu, that wasn't required.
To anybody else who struggles with this: changing the
openai_api_versionparameter from2023-05-15to2024-02-01resolved the issue for me. For whatever reason, the older API version wasn't supporting the newer async filter (unbuffered stream) functionality.