ceos/unsloth
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix httpcore GeneratorExit: explicitly aclose aiter_lines before response closes

Browse source
This commit is contained in:
Roland Tannous 2026-03-30 10:54:24 +00:00
commit a41160d3c9
1 changed files with 59 additions and 51 deletions
Download patch file Download diff file Expand all files Collapse all files

98
studio/backend/core/inference/external_provider.py
View file

@ -117,9 +117,13 @@ class ExternalProviderClient:
) )
return return
async for line in response.aiter_lines(): lines_iter = response.aiter_lines()
if line.strip(): try:
yield line async for line in lines_iter:
if line.strip():
yield line
finally:
await lines_iter.aclose()
except httpx.ConnectError as exc: except httpx.ConnectError as exc:
logger.error("Connection error to %s: %s", self.provider_type, exc) logger.error("Connection error to %s: %s", self.provider_type, exc)
@ -238,54 +242,58 @@ class ExternalProviderClient:
yield _error_sse_line(response.status_code, error_text, self.provider_type) yield _error_sse_line(response.status_code, error_text, self.provider_type)
return return
async for line in response.aiter_lines(): lines_iter = response.aiter_lines()
if not line or line.startswith("event:"): try:
continue async for line in lines_iter:
if not line.startswith("data:"): if not line or line.startswith("event:"):
continue continue
if not line.startswith("data:"):
continue
data_str = line[len("data:"):].strip() data_str = line[len("data:"):].strip()
if not data_str: if not data_str:
continue continue
try: try:
event = _json.loads(data_str) event = _json.loads(data_str)
except _json.JSONDecodeError: except _json.JSONDecodeError:
continue continue
event_type = event.get("type") event_type = event.get("type")
if event_type == "content_block_delta": if event_type == "content_block_delta":
delta = event.get("delta", {}) delta = event.get("delta", {})
if delta.get("type") == "text_delta": if delta.get("type") == "text_delta":
chunk = { chunk = {
"id": completion_id, "id": completion_id,
"object": "chat.completion.chunk", "object": "chat.completion.chunk",
"choices": [{ "choices": [{
"index": 0, "index": 0,
"delta": {"content": delta.get("text", "")}, "delta": {"content": delta.get("text", "")},
"finish_reason": None, "finish_reason": None,
}], }],
} }
yield f"data: {_json.dumps(chunk)}" yield f"data: {_json.dumps(chunk)}"
elif event_type == "message_delta": elif event_type == "message_delta":
stop_reason = event.get("delta", {}).get("stop_reason") stop_reason = event.get("delta", {}).get("stop_reason")
if stop_reason: if stop_reason:
chunk = { chunk = {
"id": completion_id, "id": completion_id,
"object": "chat.completion.chunk", "object": "chat.completion.chunk",
"choices": [{ "choices": [{
"index": 0, "index": 0,
"delta": {}, "delta": {},
"finish_reason": _finish_reason_map.get(stop_reason, "stop"), "finish_reason": _finish_reason_map.get(stop_reason, "stop"),
}], }],
} }
yield f"data: {_json.dumps(chunk)}" yield f"data: {_json.dumps(chunk)}"
elif event_type == "message_stop": elif event_type == "message_stop":
yield "data: [DONE]" yield "data: [DONE]"
return return
finally:
await lines_iter.aclose()
except httpx.ConnectError as exc: except httpx.ConnectError as exc:
logger.error("Connection error to %s: %s", self.provider_type, exc) logger.error("Connection error to %s: %s", self.provider_type, exc)