Blackbox stream

[phoen1x-dev]

I ask to return the data stream for the provider Blackbox. I changed part of your code below, I don't know how correct it turned out... <think> — this is for the deepseek-r1 model.

        async with session.post(cls.api_endpoint, json=data, proxy=proxy) as response:
            try:
                await raise_for_status(response)
                full_response = ""
                max_validation_attempts = 3
                validation_attempts = 0
                inside_think = False

                async for chunk in response.content.iter_any():
                    chunk_text = chunk.decode('utf-8', errors='replace')

                    if '<think>' in chunk_text:
                        inside_think = True
                        chunk_text = chunk_text.split('<think>', 1)[0]
                    if '</think>' in chunk_text and inside_think:
                        inside_think = False
                        chunk_text = chunk_text.split('</think>', 1)[1]

                    if inside_think:
                        continue

                    if chunk:
                        if "Generated by BLACKBOX.AI" in chunk_text and validation_attempts < max_validation_attempts:
                            validation_attempts += 1
                            if validation_attempts >= max_validation_attempts:
                                print("Max validation attempts reached. Unable to get a valid response.")
                                return
                            print(f"Attempting to refresh validation. Attempt {validation_attempts} "
                                  f"of {max_validation_attempts}")
                            conversation.validated_value = await cls.fetch_validated(force_refresh=True)
                            if conversation.validated_value:
                                data["validated"] = conversation.validated_value
                                async with session.post(cls.api_endpoint, json=data, proxy=proxy) as new_response:
                                    await raise_for_status(new_response)
                                    async for new_chunk in new_response.content.iter_any():
                                        new_chunk_text = new_chunk.decode('utf-8', errors='replace')
                                        yield new_chunk_text
                                        full_response += new_chunk_text
                        else:
                            yield chunk_text
                            full_response += chunk_text

                if not full_response or full_response.isspace():
                    return

                if model in cls.image_models:
                    image_url_match = re.search(r'!\[.*?\]\((.*?)\)', full_response)
                    if image_url_match:
                        image_url = image_url_match.group(1)
                        prompt = messages[-1]["content"]
                        yield ImageResponse(images=[image_url], alt=prompt)

                if full_response:
                    if max_tokens and len(full_response) >= max_tokens:
                        reason = "length"
                    else:
                        reason = "stop"

                    if return_conversation:
                        conversation.message_history.append({"role": "assistant", "content": full_response})
                        yield conversation

                    yield FinishReason(reason)

            except Exception as e:
                print(f"Error in response: {e}")
                raise

[TheFirstNoob]

It seems to me that a different approach is needed here than just catching the tag. Yes, for simple code where the response comes to the console, you can remove it, but for different bots, sites, etc., it is better to design a similar block.

I suppose it is better to make it a function.

Personally, in my bot, I implemented it like this:

def replace_think(match):
        think_content = match.group(1)
        formatted_think = '\n'.join([f"> {line}" for line in think_content.splitlines()])
        return f"**РАЗМЫШЛЕНИЕ:**\n{formatted_think}\n\n**ИТОГОВЫЙ ОТВЕТ:**"

    response = re.sub(r'<think>(.*?)</think>', replace_think, response, flags=re.DOTALL)

Yes, I catch the tag myself because at the moment g4f does not process this tag. But still, this approach seems better to me. If you have any ideas on how this can be implemented for g4f in general, then it is better to write here before the topic is closed.

[hlohaus]

Hey @phoen1x-dev, @TheFirstNoob, we should do it in the Reasoning Response Type so the UI can read it. We can still print those tags in the /chat/completions API.

HuggingChat's reasoning is better, plus it shows steps.

[phoen1x-dev]

I chose this option of code for the Telegram bot with data streaming:

        async with session.post(cls.api_endpoint, json=data, proxy=proxy) as response:
            try:
                await raise_for_status(response)
                full_response = ""

                async for chunk in response.content.iter_any():
                    chunk_text = chunk.decode('utf-8', errors='replace')

                    chunk_text = chunk_text.replace('<think>\n\n', '```РАЗМЫШЛЕНИЯ\n'
                                                    ).replace('<think>\n', '```РАЗМЫШЛЕНИЯ\n'
                                                              ).replace('</think>\n', '```')

                    if chunk:
                        if "Generated by BLACKBOX.AI" in chunk_text:
                            conversation.validated_value = await cls.fetch_validated(force_refresh=True)
                            if conversation.validated_value:
                                data["validated"] = conversation.validated_value
                                async with session.post(cls.api_endpoint, json=data, proxy=proxy) as new_response:
                                    await raise_for_status(new_response)
                                    async for new_chunk in new_response.content.iter_any():
                                        new_chunk_text = new_chunk.decode('utf-8', errors='replace')

                                        if new_chunk_text and not new_chunk_text.isspace():
                                            yield new_chunk_text
                                            full_response += new_chunk_text
                            else:
                                if chunk_text and not chunk_text.isspace():
                                    yield chunk_text
                                    full_response += chunk_text
                        else:
                            if chunk_text and not chunk_text.isspace():
                                yield chunk_text
                                full_response += chunk_text

                if not full_response or full_response.isspace():
                    return

                if model in cls.image_models:
                    image_url_match = re.search(r'!\[.*?\]\((.*?)\)', full_response)
                    if image_url_match:
                        image_url = image_url_match.group(1)
                        prompt = messages[-1]["content"]
                        yield ImageResponse(images=[image_url], alt=prompt)

                if full_response:
                    if max_tokens and len(full_response) >= max_tokens:
                        reason = "length"
                    else:
                        reason = "stop"

                    if return_conversation:
                        conversation.message_history.append({"role": "assistant", "content": full_response})
                        yield conversation

                    yield FinishReason(reason)

            except Exception as e:
                print(f"Error in response: {e}")
                raise