@@ -15,6 +15,7 @@ from ml_model.exceptions import ( RequestBlocked, ) from poller.models import Proxy +from tools.chats.domain import RawSSEChunk logger = logging.getLogger(__name__) @@ -70,7 +71,7 @@ class BytedanceVideoTaskResponse(TypedDict, total=False): RunChatResult: TypeAlias = tuple[str, int, int] -RunStreamChatResult: TypeAlias = Generator[str, None, BytedanceUsage] +RunStreamChatResult: TypeAlias = Generator[RawSSEChunk, None, BytedanceUsage] RunImageResult: TypeAlias = list[str] RunVideoResult: TypeAlias = tuple[str, int] BytedanceRunResult: TypeAlias = RunChatResult | RunImageResult | RunVideoResult @@ -286,8 +287,6 @@ class BytedanceModelArkAdapter: ) raise GenerationException usage: BytedanceUsage = {} - reasoning_started = False - content_started = False for line in resp.iter_lines(): if not line: continue @@ -305,16 +304,10 @@ class BytedanceModelArkAdapter: delta = choices[0].get('delta', {}) reasoning_chunk = delta.get('reasoning_content') or '' if include_reasoning and reasoning_chunk: - if not reasoning_started: - yield '**Рассуждение:**\n\n' - reasoning_started = True - yield reasoning_chunk + yield RawSSEChunk(event='think', data={'content': reasoning_chunk}) chunk = delta.get('content') or '' if chunk: - if include_reasoning and reasoning_started and not content_started: - yield '\n\n**Основная мысль:**\n\n' - content_started = True - yield chunk + yield RawSSEChunk(event='token', data={'content': chunk}) if ( fr := choices[0].get('finish_reason') ) and fr not in ( @@ -7,7 +7,9 @@ import httpx import tiktoken from backend import settings +from messages.services.message_service import MessageService from poller.models import Proxy +from tools.chats.domain import RawSSEChunk from .models import ModelResponse @@ -35,7 +37,7 @@ class OpenrouterAdapter: @classmethod def run_streaming_api( cls, version: str, messages: list, callback_data: dict, model_name: str - ) -> Iterator[str]: + ) -> Iterator[RawSSEChunk]: for proxy in Proxy.objects.all(): with httpx.Client( base_url=cls.BASE_URL, @@ -55,8 +57,6 @@ class OpenrouterAdapter: }, ) as resp: content = '' - # reasoning используем только для фоллбэк-подсчёта токенизатора - # в ответ не кладём, заполняет буфер истории сообщений reasoning = '' input_tokens = output_tokens = cost = 0 for line in resp.iter_lines(): @@ -70,11 +70,14 @@ class OpenrouterAdapter: try: data_obj = json.loads(data) - chunk = data_obj['choices'][0]['delta'].get('content') or '' - reasoning += data_obj['choices'][0]['delta'].get('reasoning') or '' - if chunk: - content += chunk - yield chunk + content_chunk = data_obj['choices'][0]['delta'].get('content') or '' + reasoning_chunk = data_obj['choices'][0]['delta'].get('reasoning') or '' + if content_chunk: + content += content_chunk + yield RawSSEChunk(event='token', data={'content': content_chunk}) + if reasoning_chunk: + reasoning += reasoning_chunk + yield RawSSEChunk(event='think', data={'content': reasoning_chunk}) if data_obj.get('usage'): input_tokens = data_obj['usage']['prompt_tokens'] output_tokens = data_obj['usage']['completion_tokens'] @@ -100,15 +103,24 @@ class OpenrouterAdapter: cls, version: str, messages: list, callback_data: dict, model_name: str ) -> ModelResponse: stream = cls.run_streaming_api(version, messages, callback_data, model_name) - content_parts: list[str] = [] + reasoning = '' + content = '' try: while True: chunk = next(stream) if chunk: - content_parts.append(chunk) + if chunk.event == 'think': + reasoning += chunk.data['content'] + else: + content += chunk.data['content'] except StopIteration as exc: input_tokens, output_tokens, cost = exc.value - return ModelResponse(''.join(content_parts), input_tokens, output_tokens, cost) + return ModelResponse( + MessageService.prepare_output_message(reasoning, content), + input_tokens, + output_tokens, + cost, + ) @classmethod def _fallback_tokenize(cls, model_name: str, messages: list, content: str) -> tuple[int, int]: @@ -37,6 +37,7 @@ from payments.exceptions.insufficient_balance import InsufficientBalance from payments.selectors.payment_plan_selector import PaymentPlanSelector from poller.models import Proxy +from tools.chats.domain import RawSSEChunk class Chatgpt(Chatgpt_4, StreamSimpleService, OpenAIStreamMixin): @@ -265,7 +266,7 @@ class Chatgpt(Chatgpt_4, StreamSimpleService, OpenAIStreamMixin): ) return self.save_results([response], process_time, generated_image, save) - def make_stream(self, input_message: Message, save: bool = True) -> Iterator[str]: + def make_stream(self, input_message: Message, save: bool = True) -> Iterator[RawSSEChunk]: ctx: dict[str, Any] = {} content_parts: list[str] = [] input_tokens = output_tokens = 0 @@ -284,7 +285,7 @@ class Chatgpt(Chatgpt_4, StreamSimpleService, OpenAIStreamMixin): chunk = next(stream) if chunk: content_parts.append(chunk) - yield chunk + yield RawSSEChunk(event='token', data={'content': chunk}) except StopIteration as exc: input_tokens, output_tokens = exc.value or (0, 0) break @@ -11,6 +11,7 @@ from messages.models import Message from ml_model.exceptions import ModelVersionNotAvailable, PaidPlanRequiredError from ml_model.services.chatgpt import Chatgpt from poller.models import Proxy +from tools.chats.domain import RawSSEChunk class Chatgpt_5_4(Chatgpt): @@ -94,7 +95,7 @@ class Chatgpt_5_4(Chatgpt): price += self.TOKENS_COST[model]['generated_image'] return price.quantize(Decimal('0.1'), rounding='ROUND_UP') - def make_stream(self, input_message: Message, save: bool = True) -> Iterator[str]: + def make_stream(self, input_message: Message, save: bool = True) -> Iterator[RawSSEChunk]: return (yield from super().make_stream(input_message, save)) def _build_payload( @@ -11,6 +11,7 @@ from PIL import Image from django.utils.translation import gettext from messages.models import Message +from messages.services.message_service import MessageService from ml_model.adapters.openrouter import OpenrouterAdapter from ml_model.exceptions import ( CorruptedFileError, @@ -25,6 +26,7 @@ from ml_model.services.serper_mixin import SerperMixin from payments.exceptions.insufficient_balance import InsufficientBalance from payments.selectors.payment_plan_selector import PaymentPlanSelector from poller.models import Proxy +from tools.chats.domain import RawSSEChunk from tools.chats.models import Chat from tools.copywrite.models import Copywrite from tools.public_api.models import APIStore @@ -132,7 +134,7 @@ class Claude(SerperMixin, StreamSimpleService): ) return self.save_results(result.content, process_time, save) - def make_stream(self, input_message: Message, save: bool = True) -> Iterator[str]: + def make_stream(self, input_message: Message, save: bool = True) -> Iterator[RawSSEChunk]: start_time = time.time() version_slug = input_message.info.get('version') if version_slug is None or version_slug not in self.TOKENS_COST: @@ -140,9 +142,10 @@ class Claude(SerperMixin, StreamSimpleService): model_slug = f'anthropic/{version_slug}' callback_data = self._build_callback_data(input_message) messages, embedding_tokens = self._prepare_messages(input_message, version_slug, callback_data) - content_parts: list[str] = [] input_tokens = output_tokens = 0 cost = 0 + reasoning = '' + content = '' result = '' try: @@ -151,13 +154,16 @@ class Claude(SerperMixin, StreamSimpleService): while True: chunk = next(stream) if chunk: - content_parts.append(chunk) + if chunk.event == 'think': + reasoning += chunk.data['content'] + else: + content += chunk.data['content'] yield chunk except StopIteration as exc: input_tokens, output_tokens, cost = exc.value finally: - if content_parts: - result = ''.join(content_parts) + if reasoning or content: + result = MessageService.prepare_output_message(reasoning, content) process_time = timedelta(seconds=(time.time() - start_time)) self.handle_invoice( input_message.content_object.model, @@ -12,6 +12,7 @@ import filetype from PIL import Image from messages.models import Message +from messages.services.message_service import MessageService from ml_model.adapters.bytedance_model_ark import BytedanceContentType, BytedanceModelArkAdapter from ml_model.services.FileService import FileProcessingService from ml_model.exceptions import GenerationException, ModelVersionNotAvailable @@ -19,6 +20,7 @@ from ml_model.services.base import SimpleService from payments.exceptions.insufficient_balance import InsufficientBalance from payments.selectors.payment_plan_selector import PaymentPlanSelector from ml_model.tasks import bytedance_model_ark_run, stream_bytedance_model_ark_run +from tools.chats.domain import RawSSEChunk from tools.chats.models import Chat from tools.copywrite.models import Copywrite from tools.public_api.models import APIStore @@ -284,12 +286,13 @@ class Dola_Seed(SimpleService): msgs = self.save_results(result[0], process_time, save) return msgs - def make_stream(self, input_message: Message, save: bool = True) -> Iterator[str]: + def make_stream(self, input_message: Message, save: bool = True) -> Iterator[RawSSEChunk]: version, callback_data, messages = self._prepare_data(input_message) model = self.VERSION_MAPPING[version] start_time = time.time() - content_parts: list[str] = [] input_tokens = output_tokens = 0 + reasoning = '' + content = '' result = '' try: @@ -303,15 +306,18 @@ class Dola_Seed(SimpleService): while True: chunk = next(stream) if chunk: - content_parts.append(chunk) + if chunk.event == 'think': + reasoning += chunk.data['content'] + else: + content += chunk.data['content'] yield chunk except StopIteration as exc: usage = exc.value or {} input_tokens = int(usage.get('prompt_tokens') or 0) output_tokens = int(usage.get('completion_tokens') or 0) finally: - if content_parts: - result = ''.join(content_parts) + if reasoning or content: + result = MessageService.prepare_output_message(reasoning, content) if not (input_tokens + output_tokens): input_text_parts = [] for message in messages: @@ -10,6 +10,7 @@ import filetype from PIL import Image from messages.models import Message +from messages.services.message_service import MessageService from ml_model.adapters.openrouter import OpenrouterAdapter from ml_model.exceptions import CorruptedFileError, FileExtensionNotSupported, ModelVersionNotAvailable from ml_model.services.EmbeddingService import EmbeddingService @@ -17,6 +18,7 @@ from ml_model.services.FileService import FileProcessingService from ml_model.services.base import StreamSimpleService from ml_model.tasks import openrouter_run from poller.models import Proxy +from tools.chats.domain import RawSSEChunk from tools.chats.models import Chat from tools.copywrite.models import Copywrite from tools.public_api.models import APIStore @@ -94,7 +96,7 @@ class Gemini_3_1(StreamSimpleService): ) return self.save_results(result[0], process_time, save) - def make_stream(self, input_message: Message, save: bool = True) -> Iterator[str]: + def make_stream(self, input_message: Message, save: bool = True) -> Iterator[RawSSEChunk]: start_time = time.time() version_slug = input_message.info.get('version') if version_slug is None or version_slug not in self.TOKENS_COST: @@ -105,8 +107,9 @@ class Gemini_3_1(StreamSimpleService): **input_message.info, } messages, embedding_tokens = self._prepare_messages(input_message) - content_parts: list[str] = [] input_tokens = output_tokens = 0 + reasoning = '' + content = '' result = '' try: @@ -115,13 +118,16 @@ class Gemini_3_1(StreamSimpleService): while True: chunk = next(stream) if chunk: - content_parts.append(chunk) + if chunk.event == 'think': + reasoning += chunk.data['content'] + else: + content += chunk.data['content'] yield chunk except StopIteration as exc: input_tokens, output_tokens, _ = exc.value finally: - if content_parts: - result = ''.join(content_parts) + if reasoning or content: + result = MessageService.prepare_output_message(reasoning, content) process_time = timedelta(seconds=(time.time() - start_time)) self.handle_invoice( input_message.content_object.model, @@ -5,12 +5,14 @@ from decimal import Decimal from typing import Any, Iterator from messages.models import Message +from messages.services.message_service import MessageService from ml_model.adapters.bytedance_model_ark import BytedanceContentType, BytedanceModelArkAdapter from ml_model.exceptions import GenerationException from ml_model.services.base import SimpleService from ml_model.tasks import bytedance_model_ark_run, stream_bytedance_model_ark_run from payments.exceptions.insufficient_balance import InsufficientBalance from payments.selectors.payment_plan_selector import PaymentPlanSelector +from tools.chats.domain import RawSSEChunk from tools.chats.models import Chat from tools.copywrite.models import Copywrite from tools.public_api.models import APIStore @@ -107,11 +109,12 @@ class Glm_4_7(SimpleService): msgs = self.save_results(result[0], process_time, save) return msgs - def make_stream(self, input_message: Message, save: bool = True) -> Iterator[str]: + def make_stream(self, input_message: Message, save: bool = True) -> Iterator[RawSSEChunk]: version, callback_data, messages = self._prepare_data(input_message) start_time = time.time() - content_parts: list[str] = [] input_tokens = output_tokens = 0 + reasoning = '' + content = '' result = '' try: @@ -125,15 +128,18 @@ class Glm_4_7(SimpleService): while True: chunk = next(stream) if chunk: - content_parts.append(chunk) + if chunk.event == 'think': + reasoning += chunk.data['content'] + else: + content += chunk.data['content'] yield chunk except StopIteration as exc: usage = exc.value or {} input_tokens = int(usage.get('prompt_tokens') or 0) output_tokens = int(usage.get('completion_tokens') or 0) finally: - if content_parts: - result = ''.join(content_parts) + if reasoning or content: + result = MessageService.prepare_output_message(reasoning, content) if not (input_tokens + output_tokens): input_tokens = BytedanceModelArkAdapter.tokenize( version, ''.join([m['content'] for m in messages]) @@ -11,6 +11,7 @@ from PIL import Image from django.utils.translation import gettext from messages.models import Message +from messages.services.message_service import MessageService from ml_model.adapters.openrouter import OpenrouterAdapter from ml_model.exceptions import CorruptedFileError, FileExtensionNotSupported, PaidPlanRequiredError from ml_model.services.base import StreamSimpleService @@ -20,6 +21,7 @@ from ml_model.services.serper_mixin import SerperMixin from payments.exceptions.insufficient_balance import InsufficientBalance from payments.selectors.payment_plan_selector import PaymentPlanSelector from poller.models import Proxy +from tools.chats.domain import RawSSEChunk from tools.chats.models import Chat from tools.copywrite.models import Copywrite from tools.public_api.models import APIStore @@ -81,14 +83,15 @@ class Grok(SerperMixin, StreamSimpleService): ) return self.save_results(result.content, process_time, save) - def make_stream(self, input_message: Message, save: bool = True) -> Iterator[str]: + def make_stream(self, input_message: Message, save: bool = True) -> Iterator[RawSSEChunk]: start_time = time.time() version = input_message.info.get('version') or 'grok-4.5' callback_data = {**input_message.info, 'tools': []} messages, embedding_tokens = self._prepare_messages(input_message, version, callback_data) - content_parts: list[str] = [] input_tokens = output_tokens = 0 cost = 0 + reasoning = '' + content = '' result = '' try: @@ -97,13 +100,16 @@ class Grok(SerperMixin, StreamSimpleService): while True: chunk = next(stream) if chunk: - content_parts.append(chunk) + if chunk.event == 'think': + reasoning += chunk.data['content'] + else: + content += chunk.data['content'] yield chunk except StopIteration as exc: input_tokens, output_tokens, cost = exc.value finally: - if content_parts: - result = ''.join(content_parts) + if reasoning or content: + result = MessageService.prepare_output_message(reasoning, content) process_time = timedelta(seconds=(time.time() - start_time)) self.handle_invoice( input_message.content_object.model, @@ -1,4 +1,3 @@ -import json import time from datetime import timedelta from decimal import Decimal @@ -6,10 +5,9 @@ from pathlib import Path from typing import Iterator import filetype -import httpx -from backend import settings from messages.models import Message +from messages.services.message_service import MessageService from ml_model.adapters.openrouter import OpenrouterAdapter from ml_model.exceptions import CorruptedFileError, FileExtensionNotSupported from ml_model.services.EmbeddingService import EmbeddingService @@ -18,6 +16,7 @@ from ml_model.exceptions import ModelVersionNotAvailable from ml_model.services.base import StreamSimpleService from ml_model.tasks import openrouter_run from poller.models import Proxy +from tools.chats.domain import RawSSEChunk from tools.chats.models import Chat from tools.copywrite.models import Copywrite from tools.public_api.models import APIStore @@ -92,32 +91,34 @@ class Qwen_3_7(StreamSimpleService): msgs = self.save_results(result[0], process_time) return msgs - def make_stream(self, input_message: Message, save: bool = True) -> Iterator[str]: + def make_stream(self, input_message: Message, save: bool = True) -> Iterator[RawSSEChunk]: start_time = time.time() version_slug, model_slug, callback_data, messages, embedding_tokens = self._prepare_data( input_message ) - content_parts: list[str] = [] + input_tokens = output_tokens = 0 cost = 0 + reasoning = '' + content = '' result = '' try: - stream = self._run_streaming_api(model_slug, messages, callback_data) + stream = OpenrouterAdapter.run_streaming_api(model_slug, messages, callback_data, 'Qwen') try: while True: chunk = next(stream) if chunk: - content_parts.append(chunk) + if chunk.event == 'think': + reasoning += chunk.data['content'] + else: + content += chunk.data['content'] yield chunk except StopIteration as exc: - cost = exc.value + input_tokens, output_tokens, cost = exc.value finally: - if content_parts: - result = ''.join(content_parts) + if reasoning or content: + result = MessageService.prepare_output_message(reasoning, content) if not cost: - input_tokens, output_tokens = OpenrouterAdapter.count_tokens_fallback( - 'Qwen', messages, result - ) cost = self._estimate_cost(version_slug, input_tokens, output_tokens) process_time = timedelta(seconds=(time.time() - start_time)) self.handle_invoice( @@ -240,46 +241,3 @@ class Qwen_3_7(StreamSimpleService): + output_tokens * price_map['output'] / 1_000_000 ) return float(price / self.COEFFICIENT) - - def _run_streaming_api( - self, version: str, messages: list, callback_data: dict - ) -> Iterator[str]: - for proxy in Proxy.objects.all(): - with httpx.Client( - base_url='https://openrouter.ai/api/v1', - headers={'Authorization': f'Bearer {settings.OPENROUTER_API_KEY}'}, - proxy=f'{proxy.protocol}://{proxy.address}', - timeout=600, - ) as client: - with client.stream( - 'POST', - 'chat/completions', - json={ - 'model': version, - 'stream': True, - 'messages': messages, - 'transforms': ['middle-out'], - **callback_data, - }, - ) as resp: - cost = 0 - for line in resp.iter_lines(): - line = line.strip() - if not line or not line.startswith('data: '): - continue - - data = line[6:] - if data == '[DONE]': - break - - try: - data_obj = json.loads(data) - chunk = data_obj['choices'][0]['delta'].get('content') or '' - if chunk: - yield chunk - if data_obj.get('usage'): - cost = data_obj['usage'].get('cost') or 0 - except json.JSONDecodeError: - continue - - return cost