feat(bytedance): add Seed Speech STT, TTS and Live Interpretation - #634
feat(bytedance): add Seed Speech STT, TTS and Live Interpretation#634Nash0x7E2 wants to merge 1 commit into
Conversation
Wrap the ByteDance / BytePlus (Volcengine) Seed Speech WebSocket APIs as a new plugin: bytedance.STT (streaming ASR on bigmodel_async), bytedance.TTS (bidirectional streaming TTS), and bytedance.Realtime (AST 2.0 Live Interpretation speech-to-speech translator). Includes a pure-Python v3 framing codec and AST proto codec, unit + integration tests, examples, and docs. Co-authored-by: Cursor <cursoragent@cursor.com>
github-actions
Bot
added
dependencies
agents-core
plugins
config
docs
project-info
labels
📝 WalkthroughWalkthroughAdded a ByteDance plugin for Seed Speech streaming STT and TTS, plus AST 2.0 realtime interpretation. The change includes authentication helpers, WebSocket framing, protobuf encoding and decoding, package metadata, workspace integration, public exports, documentation, runnable examples, and unit and integration tests. Realtime support validates languages and modes, streams resampled PCM audio, emits subtitle and translated-audio events, and manages session cleanup. Merge Risk: 🟠 High · up to The new ByteDance streaming integrations still have risks that can cause authentication with the wrong account, dropped or stale audio across sessions, continued usage after interruption, and connections that leak or fail to recover. These high-impact runtime and correctness issues should be fixed before merging. ✨ Finishing Touches 💡 1⚔️ Resolve merge conflicts 💡
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 12
🧹 Nitpick comments (9)
plugins/bytedance/vision_agents/plugins/bytedance/_v3.py (1)
17-17: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winRemove
Anyand baredictfrom the protocol contract.Use
object | NoneforMessage.payloadand a parameterized mapping type forbuild_full_client_request. The existing consumers already narrow JSON payloads before use.As per coding guidelines: "Avoid using
Anytype" and "Use type annotations everywhere."Also applies to: 90-91, 114-114, 212-213
Source: Coding guidelines
plugins/bytedance/tests/test_protocol.py (2)
4-4: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDo not import private codec modules in external tests.
_astand_v3are private modules. Lines 133-136 also call private wire helpers. Re-export the supported protocol API from the package, then test that API with known wire fixtures.As per coding guidelines: "Never import from private modules (
_foo) outside of the package's own__init__.py; use the public re-export."Also applies to: 133-136
Source: Coding guidelines
10-10: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd return annotations to test methods.
Add
-> Noneto each test method in this file.As per coding guidelines: "Use type annotations everywhere."
Also applies to: 23-23, 30-30, 52-52, 75-75, 86-86, 99-99, 117-117, 132-132, 143-143
Source: Coding guidelines
plugins/bytedance/vision_agents/plugins/bytedance/stt.py (1)
202-225: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueAdd the missing return annotation.
close()andstart()have no return annotation, whiletts.pyusesasync def close(self) -> None. Keep the annotations consistent across the plugin.Source: Coding guidelines
plugins/bytedance/tests/test_stt.py (1)
80-86: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winCredential-clearing test setup is duplicated across both test files. Each file repeats the same four
monkeypatch.delenvcalls for the ByteDance credential variables. The root cause is a missing shared fixture for the plugin's tests.
plugins/bytedance/tests/test_stt.py#L80-L86: replace the inlinedelenvcalls with a shared fixture from aconftest.pyinplugins/bytedance/tests/.plugins/bytedance/tests/test_tts.py#L29-L35: use the same shared fixture.plugins/bytedance/vision_agents/plugins/bytedance/tts.py (2)
70-77: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winValidate
speech_ratein the constructor.The docstring documents the range
[-50, 100], but no code enforces it. An out-of-range value fails remotely, mid-session, with a provider error. RaiseValueErrorin__init__instead.Proposed fix
super().__init__(provider_name="bytedance") + if not -50 <= speech_rate <= 100: + raise ValueError( + f"speech_rate must be in [-50, 100], got {speech_rate}" + ) self._credentials = Credentials.resolve(api_key, app_key, access_key)As per coding guidelines: "Raise
ValueErrorwith a descriptive message for invalid args".Source: Coding guidelines
193-198: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winClose the socket in a
finallyblock.If
super().close()raises, the WebSocket stays open and_on_disconnectednever runs. Move the socket teardown intofinally.Proposed fix
async def close(self) -> None: - await super().close() - if self._ws is not None: - await self._ws.close() - self._ws = None - self._on_disconnected() + try: + await super().close() + finally: + if self._ws is not None: + await self._ws.close() + self._ws = None + self._on_disconnected()As per coding guidelines: "Clean up resources in
finallyblocks".Source: Coding guidelines
plugins/bytedance/vision_agents/plugins/bytedance/realtime.py (1)
54-67: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd annotations to all new function boundaries and task state.
plugins/bytedance/vision_agents/plugins/bytedance/realtime.py#L54-L67: add-> Noneto__init__.plugins/bytedance/vision_agents/plugins/bytedance/realtime.py#L109-L110: parameterize_processing_taskasasyncio.Task[None] | None.plugins/bytedance/vision_agents/plugins/bytedance/realtime.py#L136-L193: add return annotations and concrete types for lifecycle and input methods.plugins/bytedance/vision_agents/plugins/bytedance/realtime.py#L243-L243: add-> Nonetoclose.plugins/bytedance/example/bytedance_realtime_example.py#L23-L33: type callback keyword arguments and return values.plugins/bytedance/tests/test_realtime.py#L34-L95: add-> Noneto tests and types for fixture parameters.plugins/bytedance/example/bytedance_stt_tts_example.py#L28-L40: type callback keyword arguments and return values.Use
objector concrete protocols where framework callback payload types are unknown. Do not introduceAny.As per coding guidelines: “Use type annotations everywhere.”
Source: Coding guidelines
plugins/bytedance/tests/test_realtime.py (1)
12-13: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAvoid private protocol imports in tests
Refactor
plugins/bytedance/tests/test_realtime.pyto avoid importing_astand its private wire helpers. AssertRealtimebehavior through its public interface, or expose a supported protocol API if message construction is required.Source: Coding guidelines
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: e67cd6f3-d7a9-4571-a16f-638bd7e8de9d
⛔ Files ignored due to path filters (1)
uv.lockis excluded by!**/*.lock
📒 Files selected for processing (23)
CHANGELOG.mdREADME.mdagents-core/pyproject.tomlplugins/bytedance/README.mdplugins/bytedance/example/README.mdplugins/bytedance/example/bytedance_realtime_example.pyplugins/bytedance/example/bytedance_stt_tts_example.pyplugins/bytedance/example/pyproject.tomlplugins/bytedance/py.typedplugins/bytedance/pyproject.tomlplugins/bytedance/tests/test_protocol.pyplugins/bytedance/tests/test_realtime.pyplugins/bytedance/tests/test_stt.pyplugins/bytedance/tests/test_tts.pyplugins/bytedance/vision_agents/plugins/bytedance/__init__.pyplugins/bytedance/vision_agents/plugins/bytedance/_ast.pyplugins/bytedance/vision_agents/plugins/bytedance/_auth.pyplugins/bytedance/vision_agents/plugins/bytedance/_v3.pyplugins/bytedance/vision_agents/plugins/bytedance/ast.protoplugins/bytedance/vision_agents/plugins/bytedance/realtime.pyplugins/bytedance/vision_agents/plugins/bytedance/stt.pyplugins/bytedance/vision_agents/plugins/bytedance/tts.pypyproject.toml
Included review availability: Your plan includes up to 8 reviews per rolling hour; 7 remain after this review.
| import gzip | ||
| import json | ||
|
|
||
| import pytest | ||
| from dotenv import load_dotenv | ||
|
|
||
| from vision_agents.core.edge.types import Participant | ||
| from vision_agents.core.stt import Transcript | ||
| from vision_agents.core.turn_detection import TurnEnded | ||
| from vision_agents.plugins import bytedance | ||
| from vision_agents.plugins.bytedance import _v3 | ||
|
|
||
| load_dotenv() | ||
|
|
||
|
|
||
| def _server_result_frame(result: dict) -> _v3.Message: | ||
| body = gzip.compress(json.dumps({"result": result}).encode()) | ||
| header = bytes( | ||
| [ | ||
| (0b0001 << 4) | 0b0001, | ||
| (int(_v3.MsgType.FULL_SERVER_RESPONSE) << 4) | _v3.Flags.POS_SEQ, | ||
| (int(_v3.Serialization.JSON) << 4) | int(_v3.Compression.GZIP), | ||
| 0, | ||
| ] | ||
| ) | ||
| frame = ( | ||
| header | ||
| + (1).to_bytes(4, "big", signed=True) | ||
| + len(body).to_bytes(4, "big") | ||
| + body | ||
| ) | ||
| return _v3.parse_response(frame) |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Do not import the private _v3 module from tests, and do not hand-build the frame header.
Two issues in the same block:
from vision_agents.plugins.bytedance import _v3reaches into a private module from outside the package._server_result_framere-encodes the v3 header bit layout by hand. That duplicates the framing rules in_v3and will drift silently if the header layout changes.
Add a server-frame builder in _v3 (or a small test helper exported by the package) and call it from the test.
As per coding guidelines: "Never import from private modules (_foo) outside of the package's own __init__.py; use the public re-export".
🧰 Tools
🪛 ast-grep (0.45.1)
[info] 16-16: use jsonify instead of json.dumps for JSON output
Context: json.dumps({"result": result})
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
Source: Coding guidelines
| async def test_error_frame_does_not_emit_transcript(self, stt, participant): | ||
| stt._current_participant = participant | ||
| error_message = _v3.Message(type=_v3.MsgType.ERROR, code=45000000, payload={}) | ||
|
|
||
| stt._handle_message(error_message) | ||
|
|
||
| assert stt.output.peek() == [] |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check where _emit_error_event routes events in the base STT class.
rg -nP --type=py -C6 'def _emit_error_event' agents-coreRepository: GetStream/Vision-Agents
Length of output: 3541
🏁 Script executed:
#!/bin/bash
set -eu
printf '%s\n' '--- STT error-event implementation ---'
sed -n '150,215p' agents-core/vision_agents/core/stt/stt.py
printf '%s\n' '--- Bytedance test and implementation references ---'
rg -n -C8 --type=py '_handle_message|_emit_error_event|output\.peek|class .*STT' plugins/bytedance agents-core/vision_agents/core/stt
printf '%s\n' '--- Stream/output definitions ---'
rg -n -C6 --type=py 'class .*Output|def peek|_output|output\s*=' agents-core/vision_agents/core/stt plugins/bytedanceRepository: GetStream/Vision-Agents
Length of output: 34347
Make the test synchronous.
The body has no await, so change async def to def. _handle_message sends STTErrorEvent through stt.events, not stt.output; keep assert stt.output.peek() == [].
| @pytest.mark.integration | ||
| async def test_transcribe_mia_audio(self, mia_audio_16khz, participant): | ||
| stt = bytedance.STT() | ||
| try: | ||
| await stt.start() | ||
| await stt.process_audio(mia_audio_16khz, participant=participant) | ||
| items = await stt.output.collect(timeout=15.0) | ||
| finally: | ||
| await stt.close() | ||
|
|
||
| finals = [i for i in items if isinstance(i, Transcript) and i.final] | ||
| full = " ".join(t.text for t in finals) | ||
| assert "forgotten treasures" in full.lower() |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Locate the mia_audio_16khz fixture and any conftest visible to the bytedance tests.
rg -nP --type=py -C3 'mia_audio_16khz'
fd -t f 'conftest.py'Repository: GetStream/Vision-Agents
Length of output: 161
🏁 Script executed:
#!/bin/bash
set -u
printf '%s\n' '--- relevant test files ---'
git ls-files 'plugins/bytedance/tests/*' 'plugins/bytedance/*' | sed -n '1,120p'
printf '%s\n' '--- fixture and conftest references ---'
rg -n -C4 'mia_audio_16khz|conftest|def close|async def close|negative|sequence' plugins/bytedance plugins 2>/dev/null | sed -n '1,240p'
printf '%s\n' '--- package exports and STT implementation references ---'
rg -n -C4 'from .* import _v3|import _v3|class STT|process_audio|output\.collect|def start|def close|async def close' plugins/bytedance 2>/dev/null | sed -n '1,260p'Repository: GetStream/Vision-Agents
Length of output: 31040
🏁 Script executed:
#!/bin/bash
set -u
printf '%s\n' '--- all tracked conftest files and pytest configuration ---'
git ls-files '*conftest.py' '*pytest*' | sed -n '1,200p'
find . -type f \( -name 'conftest.py' -o -name 'pytest.ini' -o -name 'pyproject.toml' \) -not -path './.git/*' -print | sed -n '1,240p'
rg -n -C5 'pytest_plugins|mia_audio_16khz|asyncio_mode|fixture' --glob 'conftest.py' --glob 'pyproject.toml' --glob 'pytest.ini' . 2>/dev/null | sed -n '1,260p'
printf '%s\n' '--- output stream implementation ---'
rg -n -C6 'class .*Channel|async def collect|def collect|collect\(' --glob '*.py' . 2>/dev/null | sed -n '1,320p'
printf '%s\n' '--- exact STT lifecycle ---'
sed -n '90,225p' plugins/bytedance/vision_agents/plugins/bytedance/stt.pyRepository: GetStream/Vision-Agents
Length of output: 41702
Send an end-of-audio frame before collecting
STT.close() sends the last=True frame only after collect() returns. This test therefore relies on server-side VAD to emit a final transcript within 15 seconds. Send an end-of-audio frame while keeping the listener active, then collect and close the stream.
Source: Path instructions
| @pytest.mark.integration | ||
| async def test_convert_text_to_audio(self): | ||
| tts = bytedance.TTS() | ||
| try: | ||
| out = [chunk async for chunk in tts.send_iter("你好，世界。")] | ||
| finally: | ||
| await tts.close() | ||
|
|
||
| assert len(out) > 0 | ||
| assert isinstance(out[0].data, PcmData) | ||
| assert out[-1].final |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Inspect the send_iter contract and the yielded chunk type in the base TTS class.
rg -nP --type=py -C12 'def send_iter' agents-coreRepository: GetStream/Vision-Agents
Length of output: 1993
🏁 Script executed:
#!/bin/bash
set -e
printf '%s\n' '--- base TTS definitions ---'
sed -n '1,280p' agents-core/vision_agents/core/tts/tts.py
printf '%s\n' '--- TTS output definitions and implementations ---'
rg -nP --type=py -C8 'class TTSOutputChunk|TTSOutputChunk|_ensure_connection|async def send_iter' agents-core plugins/bytedance
printf '%s\n' '--- test file ---'
cat -n plugins/bytedance/tests/test_tts.pyRepository: GetStream/Vision-Agents
Length of output: 29473
🏁 Script executed:
#!/bin/bash
set -e
printf '%s\n' '--- ByteDance TTS implementation ---'
cat -n plugins/bytedance/vision_agents/plugins/bytedance/tts.py
printf '%s\n' '--- project timeout patterns ---'
rg -nP --type=py -C4 'asyncio\.timeout|wait_for|recv\(|websockets\.connect|open_timeout|close_timeout|ping_timeout' plugins agents-core
printf '%s\n' '--- ByteDance dependency and test configuration ---'
rg -nP -C3 'websockets|integration|asyncio_mode|pytest' plugins/bytedance pyproject.toml agents-core/pyproject.tomlRepository: GetStream/Vision-Agents
Length of output: 50380
🏁 Script executed:
#!/bin/bash
set -e
python3 - <<'PY'
import ast
from pathlib import Path
base = ast.parse(Path("agents-core/vision_agents/core/tts/tts.py").read_text())
plugin = ast.parse(Path("plugins/bytedance/vision_agents/plugins/bytedance/tts.py").read_text())
test = ast.parse(Path("plugins/bytedance/tests/test_tts.py").read_text())
def find_class_method(tree, class_name, method_name):
for node in ast.walk(tree):
if isinstance(node, ast.ClassDef) and node.name == class_name:
for child in node.body:
if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)) and child.name == method_name:
return child
raise LookupError((class_name, method_name))
send_iter = find_class_method(base, "TTS", "send_iter")
print("send_iter_return_annotation:", ast.unparse(send_iter.returns))
print("send_iter_yield_types:", sorted({
ast.unparse(value.func)
for value in (
node.value for node in ast.walk(send_iter)
if isinstance(node, ast.Yield) and isinstance(node.value, ast.Call)
)
if isinstance(value, ast.Call)
}))
print("send_iter_final_values:", sorted({
ast.unparse(keyword.value)
for node in ast.walk(send_iter)
if isinstance(node, ast.Call)
and isinstance(node.func, ast.Name)
and node.func.id == "TTSOutputChunk"
for keyword in node.keywords
if keyword.arg == "final"
}))
print("send_iter_data_values:", sorted({
ast.unparse(keyword.value)
for node in ast.walk(send_iter)
if isinstance(node, ast.Call)
and isinstance(node.func, ast.Name)
and node.func.id == "TTSOutputChunk"
for keyword in node.keywords
if keyword.arg == "data"
}))
ensure = find_class_method(plugin, "TTS", "_ensure_connection")
receive = find_class_method(plugin, "TTS", "_receive_audio")
print("ensure_connection_recv_calls:", [
ast.unparse(node)
for node in ast.walk(ensure)
if isinstance(node, ast.Call)
and isinstance(node.func, ast.Attribute)
and node.func.attr == "recv"
])
print("ensure_connection_timeout_calls:", [
ast.unparse(node)
for node in ast.walk(ensure)
if isinstance(node, ast.Call)
and (
(isinstance(node.func, ast.Attribute) and node.func.attr in {"wait_for", "timeout"})
or (isinstance(node.func, ast.Name) and node.func.id == "wait_for")
)
])
print("receive_audio_async_for_targets:", [
ast.unparse(node.target)
for node in ast.walk(receive)
if isinstance(node, ast.AsyncFor)
])
print("receive_audio_timeout_calls:", [
ast.unparse(node)
for node in ast.walk(receive)
if isinstance(node, ast.Call)
and (
(isinstance(node.func, ast.Attribute) and node.func.attr in {"wait_for", "timeout"})
or (isinstance(node.func, ast.Name) and node.func.id == "wait_for")
)
])
integration_test = find_class_method(test, "TestBytedanceTTS", "test_convert_text_to_audio")
print("integration_test_async_comprehensions:", [
ast.unparse(node)
for node in ast.walk(integration_test)
if isinstance(node, ast.ListComp) and any(isinstance(gen, ast.comprehension) and gen.is_async for gen in node.generators)
])
print("integration_test_timeout_calls:", [
ast.unparse(node)
for node in ast.walk(integration_test)
if isinstance(node, ast.Call)
and (
(isinstance(node.func, ast.Attribute) and node.func.attr in {"wait_for", "timeout"})
or (isinstance(node.func, ast.Name) and node.func.id == "wait_for")
)
])
PYRepository: GetStream/Vision-Agents
Length of output: 641
Bound the integration test and WebSocket receive path.
send_iter yields TTSOutputChunk objects, so the assertions are valid. The integration test and ByteDance receive operations have no timeout. Add explicit timeouts to prevent stalled runs.
| api_key = ( | ||
| api_key | ||
| or os.environ.get("BYTEDANCE_API_KEY") | ||
| or os.environ.get("BYTEPLUS_API_KEY") | ||
| ) | ||
| app_key = app_key or os.environ.get("BYTEDANCE_APP_KEY") | ||
| access_key = access_key or os.environ.get("BYTEDANCE_ACCESS_KEY") |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
Preserve explicit legacy credentials over environment API keys.
When a caller passes app_key and access_key without api_key, BYTEDANCE_API_KEY or BYTEPLUS_API_KEY still populates api_key. headers() then selects X-Api-Key and ignores the explicit legacy credentials. This can authenticate requests with the wrong account or cause authentication failures.
Resolve the explicit credential scheme first. Use environment values only when the caller did not provide a credential scheme. Add a test with an API-key environment variable and explicit legacy credentials.
Suggested fix
- api_key = (
- api_key
- or os.environ.get("BYTEDANCE_API_KEY")
- or os.environ.get("BYTEPLUS_API_KEY")
- )
- app_key = app_key or os.environ.get("BYTEDANCE_APP_KEY")
- access_key = access_key or os.environ.get("BYTEDANCE_ACCESS_KEY")
+ if api_key is not None:
+ pass
+ elif app_key is not None or access_key is not None:
+ app_key = app_key or os.environ.get("BYTEDANCE_APP_KEY")
+ access_key = access_key or os.environ.get("BYTEDANCE_ACCESS_KEY")
+ else:
+ api_key = (
+ os.environ.get("BYTEDANCE_API_KEY")
+ or os.environ.get("BYTEPLUS_API_KEY")
+ )
+ app_key = os.environ.get("BYTEDANCE_APP_KEY")
+ access_key = os.environ.get("BYTEDANCE_ACCESS_KEY")📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| api_key = ( | |
| api_key | |
| or os.environ.get("BYTEDANCE_API_KEY") | |
| or os.environ.get("BYTEPLUS_API_KEY") | |
| ) | |
| app_key = app_key or os.environ.get("BYTEDANCE_APP_KEY") | |
| access_key = access_key or os.environ.get("BYTEDANCE_ACCESS_KEY") | |
| if api_key is not None: | |
| pass | |
| elif app_key is not None or access_key is not None: | |
| app_key = app_key or os.environ.get("BYTEDANCE_APP_KEY") | |
| access_key = access_key or os.environ.get("BYTEDANCE_ACCESS_KEY") | |
| else: | |
| api_key = ( | |
| os.environ.get("BYTEDANCE_API_KEY") | |
| or os.environ.get("BYTEPLUS_API_KEY") | |
| ) | |
| app_key = os.environ.get("BYTEDANCE_APP_KEY") | |
| access_key = os.environ.get("BYTEDANCE_ACCESS_KEY") |
| async def _listen(self): | ||
| ws = self._ws | ||
| if ws is None: | ||
| return | ||
| try: | ||
| async for message in ws: | ||
| if isinstance(message, (bytes, bytearray)): | ||
| self._handle_message(_v3.parse_response(bytes(message))) | ||
| except websockets.ConnectionClosedOK: | ||
| pass | ||
| except websockets.ConnectionClosedError as e: | ||
| self._emit_error_event(e, context="listen") |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Handle non-connection errors in the listener task.
_v3.parse_response and _handle_message can raise on a malformed or unexpected frame (gzip/JSON decode, unknown message type). That exception ends _listen with no log record. It then surfaces in close() at Line 215, where only asyncio.CancelledError is caught, so close() raises before self._ws.close() runs and the socket leaks.
Log the failure in the listener and treat any listener exception as terminal in close().
Proposed fix
except websockets.ConnectionClosedOK:
pass
except websockets.ConnectionClosedError as e:
self._emit_error_event(e, context="listen")
+ except (ValueError, KeyError, OSError) as e:
+ logger.exception("ByteDance STT listener failed")
+ self._emit_error_event(e, context="listen")And in close():
self._listen_task.cancel()
try:
await self._listen_task
except asyncio.CancelledError:
pass
+ except (ValueError, KeyError, OSError):
+ logger.exception("ByteDance STT listener terminated with an error")
self._listen_task = NoneSource: Coding guidelines
| async def _ensure_connection(self) -> websockets.ClientConnection: | ||
| if self._ws is not None: | ||
| return self._ws | ||
|
|
||
| async with self._connect_lock: | ||
| if self._ws is not None: | ||
| return self._ws |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🌐 Web query:
websockets 15.0.1 ClientConnection state attribute State.OPEN check
💡 Result:
In the websockets library (version 15.0.1), the state attribute on a ClientConnection object returns an enum value representing the connection state [1][2]. While you can programmatically check if the connection is in the OPEN state by comparing the attribute against the State.OPEN enum [3], the official documentation explicitly discourages this approach [1][4]. The state attribute is provided primarily for completeness and debugging purposes [1][2]. Instead of manually checking connection state, the library recommends the EAFP (Easier to Ask for Forgiveness than Permission) principle: call methods like recv or send and handle the ConnectionClosed exception when the connection is no longer usable [1][4]. You can import the necessary components from the library as follows: from websockets.asyncio.client import ClientConnection from websockets.protocol import State # Discouraged: checking state directly if connection.state is State.OPEN: pass # Recommended: try the operation and handle the exception try: await connection.send("hello") except ConnectionClosed: # Handle the closed connection pass The State enum includes four members: CONNECTING, OPEN, CLOSING, and CLOSED [3][5]. Because state transitions occur asynchronously, a direct check may lead to race conditions where the state changes immediately after your check, which is why handling exceptions remains the robust practice [1][6][7].
Citations:
- 1: https://websockets.readthedocs.io/en/15.0.1/reference/asyncio/client.html
- 2: https://github.com/python-websockets/websockets/blob/15.0.1/src/websockets/asyncio/connection.py
- 3: https://github.com/python-websockets/websockets/blob/15.0.1/src/websockets/protocol.py
- 4: https://websockets.readthedocs.io/en/15.0.1/reference/asyncio/common.html
- 5: https://websockets.readthedocs.io/en/stable/topics/design.html
- 6: https://websockets.readthedocs.io/en/8.1/_modules/websockets/protocol.html
- 7: https://dokk.org/documentation/python-websockets/7.0/_modules/websockets/protocol/
🏁 Script executed:
#!/bin/bash
set -eu
file=$(fd -t f 'tts\.py$' | grep '^plugins/bytedance/vision_agents/plugins/bytedance/tts\.py$' || true)
printf '%s\n' "$file"
sed -n '1,180p' "$file"
printf '\n--- dependency declarations ---\n'
rg -n -i 'websockets|bytedance|tts' pyproject.toml poetry.lock requirements*.txt uv.lock setup.cfg setup.py 2>/dev/null || true
printf '\n--- related symbols ---\n'
rg -n '_ensure_connection|_ws|_on_disconnected|stream_audio|ConnectionClosed|ClientConnection' plugins/bytedance/vision_agents/plugins/bytedanceRepository: GetStream/Vision-Agents
Length of output: 23678
🏁 Script executed:
#!/bin/bash
set -eu
printf '%s\n' '--- tts.py lifecycle and receive code ---'
sed -n '145,215p' plugins/bytedance/vision_agents/plugins/bytedance/tts.py
printf '%s\n' '--- realtime.py connection handling ---'
sed -n '120,185p' plugins/bytedance/vision_agents/plugins/bytedance/realtime.py
sed -n '235,275p' plugins/bytedance/vision_agents/plugins/bytedance/realtime.py
printf '%s\n' '--- stt.py connection handling ---'
sed -n '85,165p' plugins/bytedance/vision_agents/plugins/bytedance/stt.py
sed -n '195,230p' plugins/bytedance/vision_agents/plugins/bytedance/stt.py
printf '%s\n' '--- tests and package metadata ---'
fd -t f . plugins/bytedance | sort | rg 'test|pyproject|tts'
rg -n -C 3 'websockets|dependencies|ConnectionClosed|_ensure_connection' plugins/bytedanceRepository: GetStream/Vision-Agents
Length of output: 22716
🏁 Script executed:
#!/bin/bash
set -eu
tmp=$(mktemp -d)
trap 'rm -rf "$tmp"' EXIT
curl -fsSL https://raw.githubusercontent.com/python-websockets/websockets/15.0.1/src/websockets/asyncio/connection.py -o "$tmp/connection.py"
curl -fsSL https://raw.githubusercontent.com/python-websockets/websockets/15.0.1/src/websockets/protocol.py -o "$tmp/protocol.py"
curl -fsSL https://raw.githubusercontent.com/python-websockets/websockets/15.0.1/src/websockets/__init__.py -o "$tmp/init.py"
printf '%s\n' '--- ClientConnection state and send behavior ---'
rg -n -C 5 'class ClientConnection|def state|state:|async def send|ConnectionClosed' "$tmp/connection.py" "$tmp/protocol.py"
printf '%s\n' '--- public exports ---'
rg -n 'protocol|State|ClientConnection|ConnectionClosed' "$tmp/init.py"
printf '%s\n' '--- standalone stale-cache control-flow probe ---'
python3 - <<'PY'
class ClosedSocket:
def __init__(self):
self.send_calls = 0
async def send(self, _message):
self.send_calls += 1
raise RuntimeError("ConnectionClosed")
async def ensure_connection(ws):
if ws is not None:
return ws
return "new connection"
import asyncio
ws = ClosedSocket()
result = asyncio.run(ensure_connection(ws))
try:
asyncio.run(result.send(b"request"))
except RuntimeError as error:
print(f"cached object returned: {result is ws}")
print(f"send failure: {error}")
print(f"send calls: {result.send_calls}")
PYRepository: GetStream/Vision-Agents
Length of output: 27600
Reset the cached connection after ConnectionClosed.
When the server closes an idle connection, _ws remains non-None. _ensure_connection then returns the closed socket, and subsequent stream_audio calls raise websockets.ConnectionClosed. Clear _ws and call _on_disconnected() when send or receive detects closure so the next call can reconnect. Do not rely only on state is State.OPEN; the websockets API recommends handling ConnectionClosed because state checks are racy.
| ws = await websockets.connect( | ||
| self._ws_url, | ||
| additional_headers=headers, | ||
| max_size=10 * 1024 * 1024, | ||
| ) | ||
| await ws.send( | ||
| _v3.build_event_message( | ||
| _v3.MsgType.FULL_CLIENT_REQUEST, | ||
| _v3.EventType.START_CONNECTION, | ||
| payload=b"{}", | ||
| ) | ||
| ) | ||
| raw = await ws.recv() | ||
| raw_bytes = raw if isinstance(raw, bytes) else raw.encode() | ||
| started = _v3.parse_response(raw_bytes) | ||
| if started.event != _v3.EventType.CONNECTION_STARTED: | ||
| await ws.close() | ||
| raise RuntimeError( | ||
| f"ByteDance TTS handshake failed: {started.event} {started.payload}" | ||
| ) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Both WebSocket clients open connections without timeouts and leak the socket on setup failure. The STT and TTS clients duplicate the same connection-setup pattern: websockets.connect (and, for TTS, the handshake recv) waits without a bound, and a failure after the connection is established leaves the socket open.
plugins/bytedance/vision_agents/plugins/bytedance/tts.py#L88-L107: wrapwebsockets.connectandws.recv()inasyncio.wait_for, and closewson every failure path, not only the event-mismatch branch.plugins/bytedance/vision_agents/plugins/bytedance/stt.py#L97-L117: wrapwebsockets.connectinasyncio.wait_for, and assignself._wsonly after the config frame is sent so a failure closes the socket and does not leave_connection_readyunset.
Consider extracting the shared connect-and-handshake logic into a helper in _v3 or _auth so both clients get the same timeout and cleanup behavior.
📍 Affects 2 files
plugins/bytedance/vision_agents/plugins/bytedance/tts.py#L88-L107(this comment)plugins/bytedance/vision_agents/plugins/bytedance/stt.py#L97-L117
| async for message in ws: | ||
| if self._stop_event.is_set() or generation != self._generation: | ||
| return | ||
| if not isinstance(message, (bytes, bytearray)): | ||
| continue | ||
|
|
||
| parsed = _v3.parse_response(bytes(message)) | ||
| if parsed.session_id is not None and parsed.session_id != session_id: | ||
| continue |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Frame demultiplexing drops audio when sessions overlap.
_receive_audio iterates the shared connection directly and discards any frame whose session_id does not match. If a second stream_audio call starts before the first generator finishes, whichever generator reads first consumes and drops the other session's audio frames. That audio is lost permanently.
Either serialize sessions with a lock, or read the socket in one background task and dispatch frames to per-session queues.
| async def stop_audio(self) -> None: | ||
| self._stop_event.set() | ||
| self._generation += 1 |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟠 Major | ⚡ Quick win
stop_audio does not cancel the session on the server.
stop_audio sets a local flag and bumps _generation. The server keeps synthesizing the remaining text and keeps sending TTS_RESPONSE frames. Two consequences:
- Provider usage continues to accrue after the interruption.
- The stale frames stay queued on the shared connection and are read by the next session's generator.
Track the active session_id and send FINISH_SESSION (or the cancel event, if _v3.EventType defines one) for it.
Proposed direction
async def stop_audio(self) -> None:
self._stop_event.set()
self._generation += 1
+ ws, session_id = self._ws, self._active_session_id
+ if ws is not None and session_id is not None:
+ try:
+ await ws.send(
+ _v3.build_event_message(
+ _v3.MsgType.FULL_CLIENT_REQUEST,
+ _v3.EventType.FINISH_SESSION,
+ payload=b"{}",
+ session_id=session_id,
+ )
+ )
+ except websockets.ConnectionClosed:
+ pass
+ self._active_session_id = None1 participant
Why
Adds first-class support for ByteDance / BytePlus (Volcengine) Seed Speech, so agents can use Seed ASR, Seed TTS, and — most notably — AST 2.0 Live Interpretation, a speech-to-speech translator that slots into the same Agent role as Gemini Live Translate. There is no official Python SDK for these services, so the plugin talks to the WebSocket protocols directly. Both the new-console
X-Api-Keyauth and the legacyX-Api-App-Key+X-Api-Access-Keyauth are supported, and every class takes an overridablews_urlso the same code targets either the mainland or a BytePlus regional host.Changes
bytedance.STT— streaming ASR on Seed ASR 2.0bigmodel_async, emitting replacement/final transcripts and turn events.bytedance.TTS— bidirectional streaming TTS (seed-tts-2.0) over a persistent session, yielding PCM audio.bytedance.Realtime— AST 2.0 Live Interpretation (s2s / s2t) with virtual-microphone input pacing by default._auth,_v3framing and_astproto codecs; unit tests for the wire formats plus integration tests, runnable examples, README, and CHANGELOG/root-README integration entries.vision-agents[bytedance]extra and new workspace member.