[1fanwang]

What does this PR do?

Applications streaming several OpenAI chat completions at once receive each choice separately, but Langfuse combines their text and tool calls into one output. The trace can show BA10 and invalid JSON arguments instead of the two replies.

Record each choice separately, ordered by its API choice index. Each keeps its own text and tool calls, matching how non-streaming responses already appear. Single-choice streams retain their existing output shape.

Fixes langfuse/langfuse#16998

Type of change

• Bug fix
• New feature
• Breaking change
• Refactor
• Documentation update
• Tooling, CI, or repo maintenance

Verification

Testing Done

Python 3.12.13, Langfuse 5.0.4, OpenAI SDK 2.29.0, OpenTelemetry SDK 1.45.0.

The reproducer drives real sync and async OpenAI clients against a local SSE server and decodes the OTLP payload Langfuse sends over HTTP to a local collector. It exercises interleaved choices with fragmented tool calls and a single-choice text control. No live OpenAI service or deployed Langfuse backend was used.

#	Scenario	Command	Result
1	Without the fix	uv run --frozen python repro_stream_choices_against_main.py	Both clients merge the two replies; single-choice output stays Hello
2	With the fix	uv run --frozen python repro_stream_choices.py	Both clients record separate A0 and B1 replies with valid arguments; single-choice output stays Hello

The unchanged package for the first run comes from 0658c20.

Without the fix:

uv run --frozen python repro_stream_choices_against_main.py; printf 'exit_code=%s\n' "$?"

{"client":"sync","n":2,"openai_http":200,"collector_http":200,"received_indexes":[[1,0],[1,0],[0,1],[]],"output":{"role":"assistant","tool_calls":[{"function":{"name":"lookup","arguments":"{\"value\":\"B{\"value\":\"A0\"}1\"}"},"id":"call-0","type":"function"}],"content":"BA10"},"usage":{"completion_tokens":5,"prompt_tokens":7,"total_tokens":12,"completion_tokens_details":null,"prompt_tokens_details":null},"matches_expected":false}
{"client":"async","n":2,"openai_http":200,"collector_http":200,"received_indexes":[[1,0],[1,0],[0,1],[]],"output":{"role":"assistant","tool_calls":[{"function":{"name":"lookup","arguments":"{\"value\":\"B{\"value\":\"A0\"}1\"}"},"id":"call-0","type":"function"}],"content":"BA10"},"usage":{"completion_tokens":5,"prompt_tokens":7,"total_tokens":12,"completion_tokens_details":null,"prompt_tokens_details":null},"matches_expected":false}
{"client":"sync","n":1,"openai_http":200,"collector_http":200,"received_indexes":[[0],[0],[]],"output":"Hello","usage":{"completion_tokens":2,"prompt_tokens":3,"total_tokens":5,"completion_tokens_details":null,"prompt_tokens_details":null},"matches_expected":true}
{"client":"async","n":1,"openai_http":200,"collector_http":200,"received_indexes":[[0],[0],[]],"output":"Hello","usage":{"completion_tokens":2,"prompt_tokens":3,"total_tokens":5,"completion_tokens_details":null,"prompt_tokens_details":null},"matches_expected":true}
{"mismatched_cases":2}
exit_code=1

With the fix:

uv run --frozen python repro_stream_choices.py; printf 'exit_code=%s\n' "$?"

{"client":"sync","n":2,"openai_http":200,"collector_http":200,"received_indexes":[[1,0],[1,0],[0,1],[]],"output":[{"role":"assistant","tool_calls":[{"function":{"name":"lookup","arguments":"{\"value\":\"A0\"}"},"id":"call-0","type":"function"}],"content":"A0"},{"role":"assistant","tool_calls":[{"function":{"name":"lookup","arguments":"{\"value\":\"B1\"}"},"id":"call-1","type":"function"}],"content":"B1"}],"usage":{"completion_tokens":5,"prompt_tokens":7,"total_tokens":12,"completion_tokens_details":null,"prompt_tokens_details":null},"matches_expected":true}
{"client":"async","n":2,"openai_http":200,"collector_http":200,"received_indexes":[[1,0],[1,0],[0,1],[]],"output":[{"role":"assistant","tool_calls":[{"function":{"name":"lookup","arguments":"{\"value\":\"A0\"}"},"id":"call-0","type":"function"}],"content":"A0"},{"role":"assistant","tool_calls":[{"function":{"name":"lookup","arguments":"{\"value\":\"B1\"}"},"id":"call-1","type":"function"}],"content":"B1"}],"usage":{"completion_tokens":5,"prompt_tokens":7,"total_tokens":12,"completion_tokens_details":null,"prompt_tokens_details":null},"matches_expected":true}
{"client":"sync","n":1,"openai_http":200,"collector_http":200,"received_indexes":[[0],[0],[]],"output":"Hello","usage":{"completion_tokens":2,"prompt_tokens":3,"total_tokens":5,"completion_tokens_details":null,"prompt_tokens_details":null},"matches_expected":true}
{"client":"async","n":1,"openai_http":200,"collector_http":200,"received_indexes":[[0],[0],[]],"output":"Hello","usage":{"completion_tokens":2,"prompt_tokens":3,"total_tokens":5,"completion_tokens_details":null,"prompt_tokens_details":null},"matches_expected":true}
{"mismatched_cases":0}
exit_code=0

Reproducer source: repro_stream_choices.py

import asyncio
import gzip
import json
import os
from http import HTTPStatus
from typing import Any

from openai.types.chat import ChatCompletionChunk
from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import (
    ExportTraceServiceRequest,
)
from opentelemetry.proto.trace.v1.trace_pb2 import Span
from opentelemetry.sdk.trace import TracerProvider
from pytest_httpserver import HTTPServer
from pytest_httpserver.httpserver import RequestMatcher
from werkzeug.wrappers import Request, Response

from langfuse import Langfuse
from langfuse._client.attributes import LangfuseOtelSpanAttributes
from langfuse._client.environment_variables import (
    LANGFUSE_BASE_URL,
    LANGFUSE_PUBLIC_KEY,
    LANGFUSE_SECRET_KEY,
)
from langfuse.openai import AsyncOpenAI, OpenAI

BODY_BY_CHOICES: dict[int, str] = {
    2: (
        'data: {"id":"chatcmpl-test","object":"chat.completion.chunk",'
        '"created":1700000000,"model":"gpt-4o-mini",'
        '"choices":[{"index":1,"delta":{"role":"assistant","content":"B"},'
        '"finish_reason":null},{"index":0,"delta":{"role":"assistant","content":"A"},'
        '"finish_reason":null}]}\n\n'
        'data: {"id":"chatcmpl-test","object":"chat.completion.chunk",'
        '"created":1700000000,"model":"gpt-4o-mini",'
        '"choices":[{"index":1,"delta":{"content":"1","tool_calls":[{"index":0,'
        '"id":"call-1","type":"function","function":{"name":"lookup",'
        '"arguments":"{\\"value\\":\\"B"}}]},"finish_reason":null},{"index":0,'
        '"delta":{"content":"0","tool_calls":[{"index":0,"id":"call-0",'
        '"type":"function","function":{"name":"lookup",'
        '"arguments":"{\\"value\\":\\"A"}}]},"finish_reason":null}]}\n\n'
        'data: {"id":"chatcmpl-test","object":"chat.completion.chunk",'
        '"created":1700000000,"model":"gpt-4o-mini",'
        '"choices":[{"index":0,"delta":{"tool_calls":[{"index":0,'
        '"function":{"arguments":"0\\"}"}}]},"finish_reason":"tool_calls"},'
        '{"index":1,"delta":{"tool_calls":[{"index":0,'
        '"function":{"arguments":"1\\"}"}}]},"finish_reason":"tool_calls"}]}\n\n'
        'data: {"id":"chatcmpl-test","object":"chat.completion.chunk",'
        '"created":1700000000,"model":"gpt-4o-mini","choices":[],'
        '"usage":{"prompt_tokens":7,"completion_tokens":5,"total_tokens":12}}\n\n'
        "data: [DONE]\n\n"
    ),
    1: (
        'data: {"id":"chatcmpl-one","object":"chat.completion.chunk",'
        '"created":1700000000,"model":"gpt-4o-mini",'
        '"choices":[{"index":0,"delta":{"role":"assistant","content":"Hel"},'
        '"finish_reason":null}]}\n\n'
        'data: {"id":"chatcmpl-one","object":"chat.completion.chunk",'
        '"created":1700000000,"model":"gpt-4o-mini",'
        '"choices":[{"index":0,"delta":{"content":"lo"},"finish_reason":"stop"}]}\n\n'
        'data: {"id":"chatcmpl-one","object":"chat.completion.chunk",'
        '"created":1700000000,"model":"gpt-4o-mini","choices":[],'
        '"usage":{"prompt_tokens":3,"completion_tokens":2,"total_tokens":5}}\n\n'
        "data: [DONE]\n\n"
    ),
}


async def main() -> int:
    mismatches: int = 0
    with HTTPServer(host="127.0.0.1", port=0) as server:
        os.environ[LANGFUSE_PUBLIC_KEY] = "test-public-key"
        os.environ[LANGFUSE_SECRET_KEY] = "test-secret-key"
        os.environ[LANGFUSE_BASE_URL] = server.url_for("").rstrip("/")
        tracer_provider: TracerProvider = TracerProvider()
        langfuse: Langfuse = Langfuse(
            tracer_provider=tracer_provider,
            tracing_enabled=True,
            timeout=5,
        )
        try:
            for n in (2, 1):
                for async_client in (False, True):
                    server.clear()
                    server.expect_request(
                        uri="/api/public/otel/v1/traces", method="POST"
                    ).respond_with_data(
                        response_data=b"", content_type="application/x-protobuf"
                    )
                    args: dict[str, Any] = {
                        "model": "gpt-4o-mini",
                        "messages": [{"role": "user", "content": "choose"}],
                        "n": n,
                        "stream": True,
                        "stream_options": {"include_usage": True},
                    }
                    server.expect_request(
                        uri="/v1/chat/completions", method="POST", json=args
                    ).respond_with_data(
                        response_data=BODY_BY_CHOICES[n],
                        content_type="text/event-stream",
                    )
                    chunks: list[ChatCompletionChunk]
                    openai_http: int
                    if async_client:
                        async with AsyncOpenAI(
                            api_key="test", base_url=server.url_for("/v1")
                        ) as async_provider:
                            async with await async_provider.chat.completions.create(
                                **args
                            ) as async_stream:
                                openai_http = async_stream.response.status_code
                                chunks = [chunk async for chunk in async_stream]
                    else:
                        with OpenAI(
                            api_key="test", base_url=server.url_for("/v1")
                        ) as provider:
                            with provider.chat.completions.create(**args) as stream:
                                openai_http = stream.response.status_code
                                chunks = list(stream)
                    assert openai_http == HTTPStatus.OK
                    langfuse.flush()
                    ingestions: list[tuple[Request, Response]] = list(
                        server.iter_matching_requests(
                            matcher=RequestMatcher(
                                uri="/api/public/otel/v1/traces", method="POST"
                            )
                        )
                    )
                    assert len(ingestions) == 1
                    request: Request
                    response: Response
                    request, response = ingestions[0]
                    assert response.status_code == HTTPStatus.OK
                    body: bytes = request.get_data()
                    if request.headers.get("Content-Encoding") == "gzip":
                        body = gzip.decompress(body)
                    payload: ExportTraceServiceRequest = (
                        ExportTraceServiceRequest.FromString(body)
                    )
                    spans: list[Span] = [
                        span
                        for resource in payload.resource_spans
                        for scope in resource.scope_spans
                        for span in scope.spans
                    ]
                    assert len(spans) == 1
                    attributes: dict[str, str] = {
                        item.key: item.value.string_value
                        for item in spans[0].attributes
                    }
                    raw_output: str = attributes[
                        LangfuseOtelSpanAttributes.OBSERVATION_OUTPUT
                    ]
                    output: Any = raw_output if n == 1 else json.loads(raw_output)
                    matches_expected: bool = output == "Hello"
                    if n == 2:
                        matches_expected = (
                            isinstance(output, list)
                            and [choice["content"] for choice in output] == ["A0", "B1"]
                            and [
                                (
                                    tool_call["id"],
                                    tool_call["function"]["arguments"],
                                )
                                for choice in output
                                for tool_call in choice["tool_calls"]
                            ]
                            == [
                                ("call-0", '{"value":"A0"}'),
                                ("call-1", '{"value":"B1"}'),
                            ]
                        )
                    mismatches += not matches_expected
                    print(
                        json.dumps(
                            {
                                "client": "async" if async_client else "sync",
                                "n": n,
                                "openai_http": openai_http,
                                "collector_http": response.status_code,
                                "received_indexes": [
                                    [choice.index for choice in chunk.choices]
                                    for chunk in chunks
                                ],
                                "output": output,
                                "usage": json.loads(
                                    attributes[
                                        LangfuseOtelSpanAttributes.OBSERVATION_USAGE_DETAILS
                                    ]
                                ),
                                "matches_expected": matches_expected,
                            },
                            separators=(",", ":"),
                        )
                    )
        finally:
            langfuse.shutdown()
            tracer_provider.shutdown()
    print(json.dumps({"mismatched_cases": mismatches}, separators=(",", ":")))
    return int(mismatches != 0)


raise SystemExit(asyncio.run(main()))

Reproducer source: repro_stream_choices_against_main.py

import importlib
import io
import runpy
import subprocess
import sys
import tarfile
import tempfile
from pathlib import Path
from types import ModuleType

SOURCE_SHA: str = "0658c20db3ee028a4c63f4bddd6d58c6a20b058e"
repo: Path = Path.cwd()
with tempfile.TemporaryDirectory(prefix="langfuse-main-") as directory:
    baseline: Path = Path(directory)
    snapshot: subprocess.CompletedProcess[bytes] = subprocess.run(
        args=["git", "archive", SOURCE_SHA, "langfuse"],
        check=True,
        stdout=subprocess.PIPE,
    )
    with tarfile.open(fileobj=io.BytesIO(snapshot.stdout), mode="r:") as archive:
        archive.extractall(path=baseline, filter="data")
    sys.path.insert(0, str(baseline))
    provider: ModuleType = importlib.import_module("langfuse.openai")
    assert provider.__file__ is not None
    assert Path(provider.__file__).is_relative_to(baseline)
    print(f"Testing upstream source: {SOURCE_SHA}", flush=True)
    runpy.run_path(
        path_name=str(repo / "repro_stream_choices.py"),
        run_name="__main__",
    )

Checklist

• I self-reviewed the diff using code_review.md.
• I added or updated tests for behavior changes.
• I updated docs, examples, or .env.template if needed.
• I did not hand-edit generated files; if generated files changed, I used the upstream regeneration path.
• I did not commit secrets or credentials.

[claude]

Claude Code Review

This pull request is from a fork — automated review is disabled. A repository maintainer can comment @claude review to run a one-time review.

[CLAassistant]

All committers have signed the CLA.