看这里。
看。[quote=“zhc, post:1, topic:54056, username:zhhc88”]
async def create_completion_stream(self, model: str, messages: list, refresh_token: str, use_search: bool = True, retry_count: int = 0):
try:
logger.info(messages)
ref_file_urls = self.extract_ref_file_urls(messages)
refs = await asyncio.gather(*(self.upload_file(url, refresh_token) for url in ref_file_urls))
await self.fake_request(refresh_token)
conv_id = await self.create_conversation(refresh_token)
token = self.acquire_token(refresh_token)
payload = {
"messages": self.messages_prepare(messages),
"refs": refs,
"use_search": use_search
}
# 异步POST请求
async with ClientSession() as session:
async with session.post(f"{self.BASE_URL}/api/chat/{conv_id}/completion/stream", json=payload, headers={'Authorization': f'Bearer {token}', **self.FAKE_HEADERS}, timeout=120) as response:
response.raise_for_status() # Ensure response status is 200
stream_start_time = datetime.now()
created = datetime.now().timestamp()
search_flag = False
silent_search = 'silent_search' in model
# 初始化数据发送
yield (f"data: {json.dumps({'id': conv_id, 'model': model, 'object': 'chat.completion.chunk', 'choices': [{'index': 0, 'delta': {'role': 'assistant', 'content': ''}, 'finish_reason': None}], 'created': created})}\n\n".encode('utf-8'))
async for data, end_of_http_chunk in response.content.iter_chunks():
if not data or data == "":
continue
text = data.decode('utf-8') # 解码接收到的数据块
logger.info("====")
logger.info(f"text is {text}")
try:
if text.startswith("data:"):
event_data = text.lstrip("data:").strip()
event_data = event_data.split("data:")[0]
logger.info("----")
logger.info(f"event_data is {event_data}")
event = json.loads(event_data)
content = ''
if event.get('event') == 'cmpl':
except_char_index = event['text'].find("�")
chunk = event['text'][:except_char_index if except_char_index != -
1 else len(event['text'])]
content = (
'\n' if search_flag else '') + chunk
if search_flag:
search_flag = False
elif event.get('event') in ['all_done', 'error']:
content = '\n[内容由于不合规被停止生成，我们换个话题吧]' if event.get(
'event') == 'error' else ''
elif not silent_search and event.get('event') == 'search_plus' and event.get('msg') and event['msg'].get('type') == 'get_res':
if not search_flag:
search_flag = True
content = f"检索 {event['msg']['title']}({event['msg']['url']}) ...\n"
if content:
yield (f"data: {json.dumps({'id': conv_id, 'model': model, 'object': 'chat.completion.chunk', 'choices': [{'index': 0, 'delta': {'content': content}, 'finish_reason': None}], 'created': created})}\n\n")
except Exception as e:
logger.error(
f"Error processing SSE data: {e}", exc_info=True)
yield (f"data: {{'error': 'Internal server error'}}\n\n".encode('utf-8'))
return
# async for item in self.create_trans_stream(model, conv_id, response):
# # 对 item 进行处理
# print(item)
# return await self.create_trans_stream(model, conv_id, response)
From #dev to 开发调优
1 个赞