fix: [Coda] 修复流式请求中上下文管理器使用错误导致的AttributeError

(LogID: 202509222042560100911040167333F06)

Co-Authored-By: Coda <coda@bytedance.com>

Author: kasarolzzw

cozeloop/internal/httpclient/http_client.py

@@ -26,8 +26,9 @@ def __init__(self):
     def request(self, method: str, url: URL | str, **kwargs: Any) -> Response:
         return self.sync_client.request(method, url, **kwargs)
 
-    def stream(self, method: str, url: URL | str, **kwargs: Any) -> typing.Iterator[Response]:
-        yield self.sync_client.stream(method, url, **kwargs)
+    def stream(self, method: str, url: URL | str, **kwargs: Any):
+        """返回同步流上下文管理器"""
+        return self.sync_client.stream(method, url, **kwargs)
 
     async def arequest(self, method: str, url: URL | str, **kwargs: Any) -> Response:
         return await self.async_client.request(method, url, **kwargs)

cozeloop/internal/prompt/execute_stream_reader.py

@@ -128,7 +128,7 @@ def __exit__(self, exc_type, exc_val, exc_tb):
     async def __aenter__(self):
         """异步上下文管理器入口"""
         if not self._context_entered:
-            self._response = self._stream_context.__enter__()            # 检查HTTP状态码（同步版本逻辑）
+            self._response = await self._stream_context.__aenter__()            # 检查HTTP状态码（异步版本逻辑）
             if self._response.status_code != 200:
                 try:
                     # 先读取完整响应内容
@@ -139,10 +139,10 @@ async def __aenter__(self):
                     log_id = self._response.headers.get("x-tt-logid", "")
                     error_code = error_data.get('code', 0)
                     error_msg = error_data.get('msg', 'Unknown error')
-                    self._stream_context.__exit__(None, None, None)
+                    await self._stream_context.__aexit__(None, None, None)
                     raise RemoteServiceError(self._response.status_code, error_code, error_msg, log_id)
                 except Exception as e:
-                    self._stream_context.__exit__(None, None, None)
+                    await self._stream_context.__aexit__(None, None, None)
                     if isinstance(e, RemoteServiceError):
                         raise
                     from cozeloop.internal.consts.error import InternalError
@@ -158,7 +158,7 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
         """异步上下文管理器出口"""
         await self.aclose()
         if self._context_entered:
-            return self._stream_context.__exit__(exc_type, exc_val, exc_tb)
+            return await self._stream_context.__aexit__(exc_type, exc_val, exc_tb)
 
     def __iter__(self):
         """支持for循环直接读取"""
@@ -206,9 +206,9 @@ async def aclose(self) -> None:
         self._closed = True
         # 如果还没有进入上下文，直接关闭stream_context
         if not self._context_entered:
-            if hasattr(self._stream_context, '__exit__'):
+            if hasattr(self._stream_context, '__aexit__'):
                 try:
-                    self._stream_context.__exit__(None, None, None)
+                    await self._stream_context.__aexit__(None, None, None)
                 except Exception:
                     pass
             return
@@ -218,8 +218,8 @@ async def aclose(self) -> None:
             await super().aclose()
         else:
             # 如果response属性不存在，只关闭stream_context
-            if hasattr(self._stream_context, '__exit__'):
+            if hasattr(self._stream_context, '__aexit__'):
                 try:
-                    self._stream_context.__exit__(None, None, None)
+                    await self._stream_context.__aexit__(None, None, None)
                 except Exception:
                     pass

cozeloop/internal/stream/base_stream_reader.py

@@ -78,9 +78,8 @@ async def _aiter_events(self) -> AsyncIterator[ServerSentEvent]:
             ServerSentEvent: 解码后的SSE事件
         """
         try:
-            # 由于httpx.stream()返回的是同步流，即使在异步上下文中也需要使用同步迭代
-            # 将同步迭代包装成异步生成器
-            for sse in self._decoder.iter_bytes(self.response.iter_bytes()):
+            # 使用异步迭代器
+            async for sse in self._decoder.aiter_bytes(self.response.aiter_bytes()):
                 yield sse
         except Exception as e:
             logger.error(f"Error async iterating SSE events: {e}")

cozeloop/internal/stream/sse.py

@@ -4,7 +4,7 @@
 from __future__ import annotations
 
 import json
-from typing import Any, Iterator, Optional
+from typing import Any, Iterator, Optional, AsyncIterator
 
 
 class ServerSentEvent:
@@ -129,6 +129,44 @@ def _iter_chunks(self, iterator: Iterator[bytes]) -> Iterator[bytes]:
         if data:
             yield data
 
+    async def aiter_bytes(self, iterator: AsyncIterator[bytes]) -> AsyncIterator[ServerSentEvent]:
+        """
+        异步解码字节流为SSE事件
+        
+        Args:
+            iterator: 异步字节流迭代器
+            
+        Yields:
+            ServerSentEvent: 解码后的SSE事件
+        """
+        async for chunk in self._aiter_chunks(iterator):
+            # 先分割再解码，确保splitlines()只使用\r和\n
+            for raw_line in chunk.splitlines():
+                line = raw_line.decode("utf-8")
+                sse = self.decode(line)
+                if sse:
+                    yield sse
+
+    async def _aiter_chunks(self, iterator: AsyncIterator[bytes]) -> AsyncIterator[bytes]:
+        """
+        异步处理字节块，确保完整的SSE消息
+        
+        Args:
+            iterator: 异步字节流迭代器
+            
+        Yields:
+            bytes: 完整的SSE消息块
+        """
+        data = b""
+        async for chunk in iterator:
+            for line in chunk.splitlines(keepends=True):
+                data += line
+                if data.endswith((b"\r\r", b"\n\n", b"\r\n\r\n")):
+                    yield data
+                    data = b""
+        if data:
+            yield data
+
     def decode(self, line: str) -> Optional[ServerSentEvent]:
         """
         解码单行SSE数据
@@ -182,4 +220,4 @@ def decode(self, line: str) -> Optional[ServerSentEvent]:
                 pass
         # 其他字段被忽略
 
-        return None
+        return None