diff --git a/litellm/llms/oci/chat/cohere.py b/litellm/llms/oci/chat/cohere.py index 0c5c426478e..4b34c23f589 100644 --- a/litellm/llms/oci/chat/cohere.py +++ b/litellm/llms/oci/chat/cohere.py @@ -250,7 +250,13 @@ def handle_cohere_stream_chunk(dict_chunk: dict) -> ModelResponseStream: if typed_chunk.index is None: typed_chunk.index = 0 - text = typed_chunk.text or "" + # OCI Cohere's terminal SSE event re-sends the full assembled response in + # `text` alongside a populated `chatHistory`. Emitting that text would + # concatenate the whole response onto the already-streamed deltas. + # `chatHistory` is the correct discriminator: `finishReason` is a weaker + # signal that could in principle appear on a non-consolidated chunk. + is_terminal_consolidation = typed_chunk.chatHistory is not None + text = "" if is_terminal_consolidation else (typed_chunk.text or "") finish_reason = typed_chunk.finishReason if finish_reason == "COMPLETE": diff --git a/tests/llm_translation/test_oci_integration.py b/tests/llm_translation/test_oci_integration.py index 47d82ff9abe..c5658bbd131 100644 --- a/tests/llm_translation/test_oci_integration.py +++ b/tests/llm_translation/test_oci_integration.py @@ -212,6 +212,66 @@ def test_streaming(m: _M, oci_params): assert len(content) > 0 +@pytest.mark.parametrize( + "model", + ["cohere.command-latest", "cohere.command-r-plus-08-2024"], +) +def test_cohere_streaming_no_doubling(model, oci_params): + """Regression: OCI Cohere's terminal SSE event re-sends the full assembled + response in `text` alongside a populated `chatHistory`. Emitting that text + as another delta would concatenate the whole response onto the + already-streamed output (e.g. "How can I help?How can I help?"). + + Reported by @gotsysdba on PR #25177. Fix: drop terminal text when + `chatHistory` is present in `handle_cohere_stream_chunk`. + """ + import litellm + + streamed = "".join( + (c.choices[0].delta.content or "") + for c in litellm.completion( + model=f"oci/{model}", + messages=[{"role": "user", "content": "Hello!"}], + max_tokens=64, + stream=True, + **oci_params, + ) + if c.choices + ).strip() + + assert streamed, "expected non-empty streamed content" + + # Compare against a non-streamed call. With the doubling bug the streamed + # assembly is ~2x the real response; without it the two are the same order + # of magnitude (the model is non-deterministic, so allow generous slack). + non_streamed = ( + litellm.completion( + model=f"oci/{model}", + messages=[{"role": "user", "content": "Hello!"}], + max_tokens=64, + **oci_params, + ) + .choices[0] + .message.content + or "" + ).strip() + + assert len(streamed) < 2 * len(non_streamed) + 10, ( + f"streamed output appears doubled — " + f"streamed={len(streamed)} chars vs non_streamed={len(non_streamed)} chars\n" + f"streamed: {streamed!r}\n" + f"non_streamed: {non_streamed!r}" + ) + + # Stronger signal: the very start of the response should not appear twice. + head = streamed[:12] + assert streamed.count(head) == 1, ( + f"streamed output contains its own prefix {head!r} more than once — " + f"likely the terminal chunk re-emitted the full response.\n" + f"streamed: {streamed!r}" + ) + + @pytest.mark.parametrize("m", CHAT_MODELS) def test_multi_turn(m: _M, oci_params): import litellm diff --git a/tests/test_litellm/llms/oci/test_oci_coverage_boost.py b/tests/test_litellm/llms/oci/test_oci_coverage_boost.py index 89879426e7e..c2dc2d22251 100644 --- a/tests/test_litellm/llms/oci/test_oci_coverage_boost.py +++ b/tests/test_litellm/llms/oci/test_oci_coverage_boost.py @@ -585,21 +585,89 @@ def test_handle_cohere_stream_chunk_text(): def test_handle_cohere_stream_chunk_complete(): - chunk = {"apiFormat": "COHERE", "text": "", "finishReason": "COMPLETE"} + # Real OCI Cohere terminal events carry the full response in `text` plus a + # populated `chatHistory`; the parser must drop that text to avoid doubling. + chunk = { + "apiFormat": "COHERE", + "text": "How can I help you today?", + "finishReason": "COMPLETE", + "chatHistory": [ + {"role": "USER", "message": "Hello!"}, + {"role": "CHATBOT", "message": "How can I help you today?"}, + ], + } result = handle_cohere_stream_chunk(chunk) assert result.choices[0].finish_reason == "stop" + assert result.choices[0].delta.content == "" def test_handle_cohere_stream_chunk_max_tokens(): - chunk = {"apiFormat": "COHERE", "text": "", "finishReason": "MAX_TOKENS"} + chunk = { + "apiFormat": "COHERE", + "text": "truncated full response", + "finishReason": "MAX_TOKENS", + "chatHistory": [{"role": "CHATBOT", "message": "truncated full response"}], + } result = handle_cohere_stream_chunk(chunk) assert result.choices[0].finish_reason == "length" + assert result.choices[0].delta.content == "" def test_handle_cohere_stream_chunk_tool_call(): - chunk = {"apiFormat": "COHERE", "text": "", "finishReason": "TOOL_CALL"} + chunk = { + "apiFormat": "COHERE", + "text": "", + "finishReason": "TOOL_CALL", + "chatHistory": [{"role": "CHATBOT", "message": ""}], + } result = handle_cohere_stream_chunk(chunk) assert result.choices[0].finish_reason == "tool_calls" + assert result.choices[0].delta.content == "" + + +def test_handle_cohere_stream_chunk_terminal_drops_full_response_text(): + """Regression for double-output on cohere.command-* streaming. + + OCI's terminal SSE event re-sends the full assembled response in `text` + alongside a populated `chatHistory`. That text must be dropped — otherwise + it gets concatenated onto the already-streamed incremental deltas. + """ + chunk = { + "apiFormat": "COHERE", + "text": "How can I help you today?", + "finishReason": "COMPLETE", + "chatHistory": [ + {"role": "USER", "message": "Hello!"}, + {"role": "CHATBOT", "message": "How can I help you today?"}, + ], + } + result = handle_cohere_stream_chunk(chunk) + assert result.choices[0].delta.content == "" + + +def test_handle_cohere_stream_chunk_incremental_passes_text_through(): + """Non-terminal chunks (no chatHistory) must emit their incremental text.""" + chunk = { + "apiFormat": "COHERE", + "text": "How can I ", + "finishReason": None, + } + result = handle_cohere_stream_chunk(chunk) + assert result.choices[0].delta.content == "How can I " + assert result.choices[0].finish_reason is None + + +def test_handle_cohere_stream_chunk_finish_reason_without_chathistory_keeps_text(): + """`finishReason` alone (no `chatHistory`) must NOT trigger the drop — + `chatHistory` is the discriminator for the consolidated terminal event.""" + chunk = { + "apiFormat": "COHERE", + "text": "tail delta", + "finishReason": "COMPLETE", + } + result = handle_cohere_stream_chunk(chunk) + assert result.choices[0].delta.content == "tail delta" + assert result.choices[0].finish_reason == "stop" # ===========================================================================