fix httpcore GeneratorExit: explicitly aclose aiter_lines before response closes
This commit is contained in:
parent
7b54c97333
commit
a41160d3c9
1 changed files with 59 additions and 51 deletions
|
|
@ -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)
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue