genai-util: refactor streams to a generic ABC wrapper. by eternalcuriouslearner · Pull Request #4500 · open-telemetry/opentelemetry-python-contrib

eternalcuriouslearner · 2026-04-29T01:48:19Z

Description

This change introduces shared stream wrapper base classes in util-genai and updates the OpenAI v2 chat streaming wrappers to use that common lifecycle shape.

The new SyncStreamWrapper and AsyncStreamWrapper centralize stream iteration, close/context-manager handling, passthrough behavior, double-finalization protection, and failure handling. OpenAI chat streaming now uses dedicated ChatStreamWrapper and AsyncChatStreamWrapper classes that mirror the Messages and Responses wrapper structure while continuing to propagate SDK/API exceptions and user exceptions unchanged.

This also moves chat choice buffering into a separate chat_buffers.py module to avoid import cycles, switches the new chat path to LLMInvocation/InferenceInvocation lifecycle methods, and adds tests for stream wrapper behavior and sync/async chat exception propagation.

Fixes # (issue)

Type of change

Please delete options that are not relevant.

Bug fix (non-breaking change which fixes an issue)
New feature (non-breaking change which adds functionality)
Breaking change (fix or feature that would cause existing functionality to not work as expected)
This change requires a documentation update

How Has This Been Tested?

Added unit coverage for sync/async stream wrapper lifecycle behavior.
Added VCR-backed sync and async chat tests for API exception propagation.
Added sync and async streaming tests for user exception propagation.
Ran OpenAI v2 chat test slices against oldest and latest dependency envs.
Ran OpenAI v2 lint successfully.

Does This PR Require a Core Repo Change?

Yes. - Link to PR:
No.

Checklist:

See contributing.md for styleguide, changelog guidelines, and more.

Followed the style guidelines of this project
Changelogs have been updated
Unit tests have been added
Documentation has been updated

Copilot

Pull request overview

This PR introduces shared synchronous/asynchronous stream wrapper base classes in opentelemetry-util-genai and refactors the OpenAI v2 chat streaming implementation to use the shared lifecycle shape (including moving chat choice buffering into a dedicated module and switching chat spans to the InferenceInvocation lifecycle).

Changes:

Added SyncStreamWrapper / AsyncStreamWrapper ABCs in opentelemetry-util-genai plus unit tests for stream lifecycle behavior.
Refactored OpenAI v2 chat streaming to use new ChatStreamWrapper / AsyncChatStreamWrapper and extracted buffering into chat_buffers.py.
Added/updated OpenAI v2 tests for API exception and user exception propagation; updated changelogs.

Reviewed changes

Copilot reviewed 11 out of 11 changed files in this pull request and generated 8 comments.

Show a summary per file

File	Description
util/opentelemetry-util-genai/src/opentelemetry/util/genai/_stream.py	Adds new generic sync/async stream wrapper base classes and lifecycle finalization logic.
util/opentelemetry-util-genai/tests/test_stream.py	Adds unit coverage for wrapper iteration, close/context-manager behavior, and error handling.
util/opentelemetry-util-genai/CHANGELOG.md	Documents the new shared stream wrapper helpers.
instrumentation-genai/opentelemetry-instrumentation-openai-v2/src/opentelemetry/instrumentation/openai_v2/utils.py	Switches chat invocation creation to `TelemetryHandler.start_inference()` / `InferenceInvocation`.
instrumentation-genai/opentelemetry-instrumentation-openai-v2/src/opentelemetry/instrumentation/openai_v2/patch.py	Updates chat create wrappers to use the new invocation lifecycle and new chat stream wrappers.
instrumentation-genai/opentelemetry-instrumentation-openai-v2/src/opentelemetry/instrumentation/openai_v2/chat_wrappers.py	New chat streaming wrappers built on the shared stream wrapper base classes.
instrumentation-genai/opentelemetry-instrumentation-openai-v2/src/opentelemetry/instrumentation/openai_v2/chat_buffers.py	Extracts choice/tool-call buffering helpers to avoid import cycles.
instrumentation-genai/opentelemetry-instrumentation-openai-v2/tests/test_choice_buffer.py	Updates tests to import buffers from the new module.
instrumentation-genai/opentelemetry-instrumentation-openai-v2/tests/test_chat_completions.py	Adds tests for API exception propagation and streaming user-exception propagation.
instrumentation-genai/opentelemetry-instrumentation-openai-v2/tests/test_async_chat_completions.py	Adds async variants of exception propagation tests; adjusts stream close behavior by mode.
instrumentation-genai/opentelemetry-instrumentation-openai-v2/CHANGELOG.md	Documents the chat stream wrapper refactor.

Copilot

Pull request overview

Copilot reviewed 12 out of 12 changed files in this pull request and generated 2 comments.

Copilot

Pull request overview

Copilot reviewed 13 out of 13 changed files in this pull request and generated 1 comment.

Copilot

Pull request overview

Copilot reviewed 13 out of 13 changed files in this pull request and generated 3 comments.

Copilot

Pull request overview

Copilot reviewed 13 out of 13 changed files in this pull request and generated no new comments.

lmolkova · 2026-05-05T20:24:42Z

 - When catching exceptions from the underlying library to record telemetry, always re-raise the
  original exception unmodified.
 - Do not raise new exceptions in instrumentation/telemetry code.
+- For GenAI streaming wrappers, prefer the shared `SyncStreamWrapper` and `AsyncStreamWrapper`


this belongs in ./instrumentation-genai/AGENTS.md, not in the root

Co-authored-by: Liudmila Molkova <neskazu@gmail.com>

aabmass · 2026-05-06T18:50:45Z

+
+
+class ToolCallBuffer:
+    def __init__(self, index, tool_call_id, function_name):


If this is new code, can we add types?

aabmass · 2026-05-06T18:52:36Z

+        self.arguments = []
+
+    def append_arguments(self, arguments):
+        if arguments is not None:


it doesn't look like arguments is ever None

aabmass · 2026-05-06T18:55:40Z

+        if getattr(chunk, "model", None):
+            self.response_model = chunk.model


Why do we need getattr()?

aabmass · 2026-05-06T18:57:25Z

+                    arguments_str = "".join(tool_call.arguments)
+                    if arguments_str:
+                        try:
+                            arguments = json.loads(arguments_str)
+                        except json.JSONDecodeError:
+                            arguments = arguments_str


Ik this code already existed, but shouldn't this join by ","?

aabmass · 2026-05-06T19:03:20Z

+    internally by the wrapper lifecycle and are not part of the public API.
+    """
+
+    def __init__(self, stream: Any):


Can you type stream? it looks like Iterable[ChunkT] might work (or some other built in protocol). If not, can you define a new Protocol that describes all the operations it must support

aabmass · 2026-05-06T19:08:45Z

+    @abstractmethod
+    def _stop_stream(self) -> None:
+        """Finalize the stream successfully."""


Is the subclass supposed to do the finalization of the wrapped stream? If these are hooks I'd recommend naming them _on_<event> instead.

aabmass · 2026-05-06T19:09:32Z

+    """
+
+    def __init__(self, stream: Any):
+        self.stream = stream


Can this be prefixed with underscore? It's possible this can conflict with the public API of the wrapped object

aabmass · 2026-05-06T19:09:50Z

+    are owned by this base class.
+    """
+
+    def __init__(self, stream: Any):


same comment about protocol

aabmass · 2026-05-06T19:10:35Z

If this is new code, please add type annotations. It looks like some of it is copied/refactored which I'm OK to ignore to not bloat the scope of this PR.

aabmass · 2026-05-06T19:14:29Z

+    def __getattr__(self, name: str) -> Any:
+        return getattr(self.stream, name)


Also FYI this doesn't work for magic method function calls, so e.g. len() or str() will break. That's why we usually use wrapt, but I'm ok with this approach for now.

wip: move the streams to a generic ABC wrapper.

cb0156f

eternalcuriouslearner requested a review from a team as a code owner April 29, 2026 01:48

github-project-automation Bot added this to Python PR digest Apr 29, 2026

github-actions Bot assigned DylanRussell, keith-decker and lmolkova Apr 29, 2026

eternalcuriouslearner added 3 commits April 28, 2026 21:49

polish: add changelogs.

11258d9

wip: precommit.

6f4e510

Merge branch 'main' into feat/move-stream-to-genai-util-as-abc

787ecbb

lmolkova requested a review from Copilot April 29, 2026 15:29

Copilot started reviewing on behalf of lmolkova April 29, 2026 15:29 View session

Copilot AI reviewed Apr 29, 2026

View reviewed changes

lmolkova reviewed Apr 30, 2026

View reviewed changes

Comment thread util/opentelemetry-util-genai/src/opentelemetry/util/genai/stream.py

Comment thread util/opentelemetry-util-genai/src/opentelemetry/util/genai/stream.py

Comment thread ...metry-instrumentation-openai-v2/src/opentelemetry/instrumentation/openai_v2/chat_wrappers.py Outdated

eternalcuriouslearner added 2 commits April 29, 2026 22:27

Merge branch 'main' into feat/move-stream-to-genai-util-as-abc

8a273c5

wip: fix the merge issues.

07b4a66

xrmx changed the title ~~feat: refactor streams to a generic ABC wrapper.~~ genai-util: refactor streams to a generic ABC wrapper. Apr 30, 2026

xrmx added the gen-ai Related to generative AI label Apr 30, 2026

lzchen and others added 4 commits April 30, 2026 14:39

Merge branch 'main' into feat/move-stream-to-genai-util-as-abc

c496ecf

Merge branch 'main' into feat/move-stream-to-genai-util-as-abc

250a7bf

polish: pr feedback.

04e067c

polish: fixing the precommit.

2157d28

eternalcuriouslearner requested a review from Copilot May 1, 2026 13:07

Copilot started reviewing on behalf of eternalcuriouslearner May 1, 2026 13:08 View session

Copilot AI reviewed May 1, 2026

View reviewed changes

Comment thread ...metry-instrumentation-openai-v2/src/opentelemetry/instrumentation/openai_v2/chat_wrappers.py

Comment thread ...metry-instrumentation-openai-v2/src/opentelemetry/instrumentation/openai_v2/chat_wrappers.py

eternalcuriouslearner added 5 commits May 1, 2026 16:27

Merge branch 'main' into feat/move-stream-to-genai-util-as-abc

fc3bd4d

wip: renamed to stream.py and changed finish_reasons population style.

8a8b9c7

wip: implementing copilot suggestions.

ae2663b

wip: fixing the tox.

a246ee1

Merge branch 'main' into feat/move-stream-to-genai-util-as-abc

26743fb

eternalcuriouslearner requested a review from Copilot May 3, 2026 01:48

Copilot started reviewing on behalf of eternalcuriouslearner May 3, 2026 01:49 View session

Copilot AI reviewed May 3, 2026

View reviewed changes

Comment thread util/opentelemetry-util-genai/src/opentelemetry/util/genai/stream.py Outdated

polish: copilot feedback.

5689654

eternalcuriouslearner requested a review from Copilot May 3, 2026 02:28

Copilot started reviewing on behalf of eternalcuriouslearner May 3, 2026 02:28 View session

Copilot AI reviewed May 3, 2026

View reviewed changes

Comment thread util/opentelemetry-util-genai/src/opentelemetry/util/genai/stream.py

Comment thread util/opentelemetry-util-genai/src/opentelemetry/util/genai/stream.py

Comment thread ...emetry-instrumentation-openai-v2/src/opentelemetry/instrumentation/openai_v2/chat_buffers.py

polish: copilot suggestions.

c43b094

eternalcuriouslearner requested a review from Copilot May 3, 2026 02:51

Copilot started reviewing on behalf of eternalcuriouslearner May 3, 2026 02:52 View session

Copilot AI reviewed May 3, 2026

View reviewed changes

lzchen reviewed May 4, 2026

View reviewed changes

Comment thread ...opentelemetry-instrumentation-openai-v2/src/opentelemetry/instrumentation/openai_v2/patch.py

eternalcuriouslearner added 2 commits May 4, 2026 22:04

Merge branch 'main' into feat/move-stream-to-genai-util-as-abc

c06a7e7

Merge branch 'main' into feat/move-stream-to-genai-util-as-abc

57e2957

lzchen approved these changes May 5, 2026

View reviewed changes

lmolkova approved these changes May 5, 2026

View reviewed changes

eternalcuriouslearner and others added 3 commits May 5, 2026 20:48

Update CLAUDE.md

1cf0924

Co-authored-by: Liudmila Molkova <neskazu@gmail.com>

Merge branch 'main' into feat/move-stream-to-genai-util-as-abc

46540b3

Merge branch 'main' into feat/move-stream-to-genai-util-as-abc

3b31fc1

aabmass reviewed May 6, 2026

View reviewed changes



		class ToolCallBuffer:
		def __init__(self, index, tool_call_id, function_name):

		if getattr(chunk, "model", None):
		self.response_model = chunk.model

		def __getattr__(self, name: str) -> Any:
		return getattr(self.stream, name)

Conversation

eternalcuriouslearner commented Apr 29, 2026

Description

Type of change

How Has This Been Tested?

Does This PR Require a Core Repo Change?

Checklist:

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

lmolkova May 5, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

8 participants

lmolkova May 5, 2026 •

edited

Loading