返回提交历史
Modified
g4f/client/async_client.py
+95
-56
XFEstudio/gpt4free
feat(g4f/client/async_client.py): improve async generator handling and logging
29f4fcb4
代码差异
1 个文件
+95
-56
@@ -33,6 +33,12 @@ except NameError:
33
33
except StopAsyncIteration:
34
34
raise StopIteration
35
35
36
async def safe_aclose(generator):
37
try:
38
await generator.aclose()
39
except Exception as e:
40
logging.warning(f"Error while closing generator: {e}")
41
36
42
async def iter_response(
37
43
response: AsyncIterator[str],
38
44
stream: bool,
@@ -45,48 +51,56 @@ async def iter_response(
45
51
completion_id = ''.join(random.choices(string.ascii_letters + string.digits, k=28))
46
52
idx = 0
47
53
48
async for chunk in response:
49
if isinstance(chunk, FinishReason):
50
finish_reason = chunk.reason
51
break
52
elif isinstance(chunk, BaseConversation):
53
yield chunk
54
continue
54
try:
55
async for chunk in response:
56
if isinstance(chunk, FinishReason):
57
finish_reason = chunk.reason
58
break
59
elif isinstance(chunk, BaseConversation):
60
yield chunk
61
continue
55
62
56
content += str(chunk)
57
idx += 1
63
content += str(chunk)
64
idx += 1
58
65
59
if max_tokens is not None and idx >= max_tokens:
60
finish_reason = "length"
66
if max_tokens is not None and idx >= max_tokens:
67
finish_reason = "length"
61
68
62
first, content, chunk = find_stop(stop, content, chunk if stream else None)
69
first, content, chunk = find_stop(stop, content, chunk if stream else None)
63
70
64
if first != -1:
65
finish_reason = "stop"
71
if first != -1:
72
finish_reason = "stop"
66
73
67
if stream:
68
yield ChatCompletionChunk(chunk, None, completion_id, int(time.time()))
74
if stream:
75
yield ChatCompletionChunk(chunk, None, completion_id, int(time.time()))
69
76
70
if finish_reason is not None:
71
break
77
if finish_reason is not None:
78
break
72
79
73
finish_reason = "stop" if finish_reason is None else finish_reason
80
finish_reason = "stop" if finish_reason is None else finish_reason
74
81
75
if stream:
76
yield ChatCompletionChunk(None, finish_reason, completion_id, int(time.time()))
77
else:
78
if response_format is not None and "type" in response_format:
79
if response_format["type"] == "json_object":
80
content = filter_json(content)
81
yield ChatCompletion(content, finish_reason, completion_id, int(time.time()))
82
if stream:
83
yield ChatCompletionChunk(None, finish_reason, completion_id, int(time.time()))
84
else:
85
if response_format is not None and "type" in response_format:
86
if response_format["type"] == "json_object":
87
content = filter_json(content)
88
yield ChatCompletion(content, finish_reason, completion_id, int(time.time()))
89
finally:
90
if hasattr(response, 'aclose'):
91
await safe_aclose(response)
82
92
83
93
async def iter_append_model_and_provider(response: AsyncIterator) -> AsyncIterator:
84
94
last_provider = None
85
async for chunk in response:
86
last_provider = get_last_provider(True) if last_provider is None else last_provider
87
chunk.model = last_provider.get("model")
88
chunk.provider = last_provider.get("name")
89
yield chunk
95
try:
96
async for chunk in response:
97
last_provider = get_last_provider(True) if last_provider is None else last_provider
98
chunk.model = last_provider.get("model")
99
chunk.provider = last_provider.get("name")
100
yield chunk
101
finally:
102
if hasattr(response, 'aclose'):
103
await safe_aclose(response)
90
104
91
105
class AsyncClient(BaseClient):
92
106
def __init__(
@@ -158,8 +172,6 @@ class Completions:
158
172
response = iter_append_model_and_provider(response)
159
173
return response if stream else await anext(response)
160
174
161
162
163
175
class Chat:
164
176
completions: Completions
165
177
@@ -168,14 +180,18 @@ class Chat:
168
180
169
181
async def iter_image_response(response: AsyncIterator) -> Union[ImagesResponse, None]:
170
182
logging.info("Starting iter_image_response")
171
async for chunk in response:
172
logging.info(f"Processing chunk: {chunk}")
173
if isinstance(chunk, ImageProviderResponse):
174
logging.info("Found ImageProviderResponse")
175
return ImagesResponse([Image(image) for image in chunk.get_list()])
176
177
logging.warning("No ImageProviderResponse found in the response")
178
return None
183
try:
184
async for chunk in response:
185
logging.info(f"Processing chunk: {chunk}")
186
if isinstance(chunk, ImageProviderResponse):
187
logging.info("Found ImageProviderResponse")
188
return ImagesResponse([Image(image) for image in chunk.get_list()])
189
190
logging.warning("No ImageProviderResponse found in the response")
191
return None
192
finally:
193
if hasattr(response, 'aclose'):
194
await safe_aclose(response)
179
195
180
196
async def create_image(client: AsyncClient, provider: ProviderType, prompt: str, model: str = "", **kwargs) -> AsyncIterator:
181
197
logging.info(f"Creating image with provider: {provider}, model: {model}, prompt: {prompt}")
@@ -220,12 +236,25 @@ class Images:
220
236
if isinstance(provider, type) and issubclass(provider, AsyncGeneratorProvider):
221
237
logging.info("Using AsyncGeneratorProvider")
222
238
messages = [{"role": "user", "content": prompt}]
223
async for response in provider.create_async_generator(model, messages, **kwargs):
224
if isinstance(response, ImageResponse):
225
return self._process_image_response(response)
226
elif isinstance(response, str):
227
image_response = ImageResponse([response], prompt)
228
return self._process_image_response(image_response)
239
generator = None
240
try:
241
generator = provider.create_async_generator(model, messages, **kwargs)
242
async for response in generator:
243
logging.debug(f"Received response: {type(response)}")
244
if isinstance(response, ImageResponse):
245
return self._process_image_response(response)
246
elif isinstance(response, str):
247
image_response = ImageResponse([response], prompt)
248
return self._process_image_response(image_response)
249
except RuntimeError as e:
250
if "async generator ignored GeneratorExit" in str(e):
251
logging.warning("Generator ignored GeneratorExit, handling gracefully")
252
else:
253
raise
254
finally:
255
if generator and hasattr(generator, 'aclose'):
256
await safe_aclose(generator)
257
logging.info("AsyncGeneratorProvider processing completed")
229
258
elif hasattr(provider, 'create'):
230
259
logging.info("Using provider's create method")
231
260
async_create = asyncio.iscoroutinefunction(provider.create)
@@ -241,7 +270,7 @@ class Images:
241
270
return self._process_image_response(image_response)
242
271
elif hasattr(provider, 'create_completion'):
243
272
logging.info("Using provider's create_completion method")
244
response = await create_image(provider, prompt, model, **kwargs)
273
response = await create_image(self.client, provider, prompt, model, **kwargs)
245
274
async for chunk in response:
246
275
if isinstance(chunk, ImageProviderResponse):
247
276
logging.info("Found ImageProviderResponse")
@@ -277,12 +306,24 @@ class Images:
277
306
if isinstance(provider, type) and issubclass(provider, AsyncGeneratorProvider):
278
307
messages = [{"role": "user", "content": "create a variation of this image"}]
279
308
image_data = to_data_uri(image)
280
async for response in provider.create_async_generator(model, messages, image=image_data, **kwargs):
281
if isinstance(response, ImageResponse):
282
return self._process_image_response(response)
283
elif isinstance(response, str):
284
image_response = ImageResponse([response], "Image variation")
285
return self._process_image_response(image_response)
309
generator = None
310
try:
311
generator = provider.create_async_generator(model, messages, image=image_data, **kwargs)
312
async for response in generator:
313
if isinstance(response, ImageResponse):
314
return self._process_image_response(response)
315
elif isinstance(response, str):
316
image_response = ImageResponse([response], "Image variation")
317
return self._process_image_response(image_response)
318
except RuntimeError as e:
319
if "async generator ignored GeneratorExit" in str(e):
320
logging.warning("Generator ignored GeneratorExit in create_variation, handling gracefully")
321
else:
322
raise
323
finally:
324
if generator and hasattr(generator, 'aclose'):
325
await safe_aclose(generator)
326
logging.info("AsyncGeneratorProvider processing completed in create_variation")
286
327
elif hasattr(provider, 'create_variation'):
287
328
if asyncio.iscoroutinefunction(provider.create_variation):
288
329
response = await provider.create_variation(image, **kwargs)
@@ -296,5 +337,3 @@ class Images:
296
337
return self._process_image_response(image_response)
297
338
else:
298
339
raise ValueError(f"Provider {provider} does not support image variation")
299
300
raise NoImageResponseError("Failed to create image variation")