0

У меня есть функция:

async def event_generator():
    stream_gen = new_messages()
    while True:
        if await request.is_disconnected():
            break
        try:
            item = next(stream_gen)
            yield Item(
                user_uuid=user_uuid, session_id=session_id, message_id=item[0], token_id=item[0], message=item[1])
        except StopIteration:
            return

        await asyncio.sleep(STREAM_DELAY)

Она возвращает стриминговые ивенты в формате

data: user_uuid='we' token_id=51 session_id=23234234 message=' of'

data: user_uuid='we' token_id=52 session_id=23234234 message=' these'

data: user_uuid='we' token_id=53 session_id=23234234 message=' titles'

event: ping
data: 2023-04-21 14:12:09.880975

data: user_uuid='we' token_id=54 session_id=23234234 message=' or'

data: user_uuid='we' token_id=55 session_id=23234234 message=' provide'

Иногда ивентом есть событие типа event: ping data: 2023-04-21 14:12:09.880975

Я не хочу чтобы это событие отображалось. Т. е. должно быть так:

data: user_uuid='we' token_id=51 session_id=23234234 message=' of'

data: user_uuid='we' token_id=52 session_id=23234234 message=' these'

data: user_uuid='we' token_id=53 session_id=23234234 message=' titles'

data: user_uuid='we' token_id=54 session_id=23234234 message=' or'

data: user_uuid='we' token_id=55 session_id=23234234 message=' provide'

Вот часть кода для лучшего понимания

@unsync
async def message_stream(
        request: Request,
        system_info,
        mm_client,
        table_id,
        separator,
        message: str,
        user_uuid: str,
        session_id: int):

    ...

    def new_messages():
        str_response = ""
        messages.append({"role": "user", "content": message, "time": str(datetime.datetime.now(timezone))})

        status = False
        while not status:
            try:
                completion = openai.ChatCompletion.create(
                    model=model, messages=[{"role": i["role"], "content": i["content"]} for i in messages],
                    temperature=0.03, stream=True, request_timeout=5)
                for num, line in enumerate(completion):
                    token = line['choices'][0]['delta'].get('content', "")
                    yield num, token
                    str_response += token
                status = True

            except Exception as e:
                logger.warning(f"Session {session_id} with ChatGPT error {e} for the user {user_uuid}")
                time.sleep(1)

        ...

    async def event_generator():
        stream_gen = new_messages()
        while True:
            if await request.is_disconnected():
                break
            try:
                item = next(stream_gen)
                yield Item(
                    user_uuid=user_uuid, session_id=session_id, message_id=item[0], token_id=item[0], message=item[1])
            except StopIteration:
                return

            await asyncio.sleep(STREAM_DELAY)

    return EventSourceResponse(event_generator())


@app.get('/key_hire_stream')
async def key_hire(request: Request, message: str, user_uuid: str, session_id: int):
    return message_stream(
        request=request, system_info=key_hire_system_info, mm_client=key_hire_memcached_client,
        table_id=key_hire_table_id, separator="++++", message=message, user_uuid=user_uuid,
        session_id=session_id).result()


if __name__ == "__main__":
    uvicorn.run("main:app", host="127.0.0.1", port=8080, log_level="debug")

Как я могу удалить event: ping с возвращаемых событий?

Благодарю

2
  • Какую библиотеку вы используете? 21 апр в 14:55
  • @WTFisGoingOn я использую fastapi и sse_starlette. Добавил больше кода для лучшего понимания. Благодарю
    – ivan100096
    21 апр в 15:25

1 ответ 1

0

Одно из решений - установить огромный интервал для пинга.

event_source = EventSourceResponse(eg)
event_source.ping_interval = 600 # <---

Значение интервала указывается в секундах, в данном случае - 10 минут.

Ваш ответ

Нажимая на кнопку «Отправить ответ», вы соглашаетесь с нашими условиями использования и подтверждаете, что прочитали и поняли наши политику конфиденциальности и нормы поведения.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.