fix:修复了因为迭代数据只推入resq_messages和resq_message_chain导致缓存到内存中的数据和写入log中的数据量庞大，以及有思考的think处理

feat:增加带有深度思考模型的think的去think操作
feat:dify中聊天机器人，chatflow对流式的支持

pkg/platform/sources/lark.py

@@ -359,7 +359,7 @@ class LarkAdapter(adapter.MessagePlatformAdapter):
         self.listeners = {}
         self.message_id_to_card_id = {}
         self.card_id_dict = {}
-        self.seq = 0
+        self.seq = 1
 
         @self.quart_app.route('/lark/callback', methods=['POST'])
         async def lark_callback():
@@ -456,7 +456,7 @@ class LarkAdapter(adapter.MessagePlatformAdapter):
                     card_data = {"schema": "2.0", "header": {"title": {"content": "bot", "tag": "plain_text"}},
                                  "body": {"elements": [{"tag": "markdown", "content": "[思考中.....]","element_id":"markdown_1"}]},
                                  "config": {"streaming_mode": True,
-                                            "streaming_config": {"print_strategy": "fast"}}}
+                                            "streaming_config": {"print_strategy": "delay"}}}  # delay / fast
 
                     request: CreateCardRequest = CreateCardRequest.builder() \
                         .request_body(
@@ -620,7 +620,7 @@ class LarkAdapter(adapter.MessagePlatformAdapter):
             'type': 'card_json',
             'data': {'card_id': self.card_id_dict[message_id], 'elements': {'content': text_message}},
         }
-
+        print(self.seq)
         request: ContentCardElementRequest = ContentCardElementRequest.builder() \
             .card_id(self.card_id_dict[message_id]) \
             .element_id("markdown_1") \
@@ -632,7 +632,7 @@ class LarkAdapter(adapter.MessagePlatformAdapter):
             .build()
 
         if is_final:
-            self.seq = 0
+            self.seq = 1
         # 发起请求
         response: ContentCardElementResponse = self.api_client.cardkit.v1.card_element.content(request)
 

pkg/provider/modelmgr/requesters/chatcmpl.py

@@ -8,7 +8,7 @@ import openai.types.chat.chat_completion as chat_completion
 import httpx
 
 from .. import errors, requester
-from ....core import entities as core_entities
+from ....core import entities as core_entities, app
 from ... import entities as llm_entities
 from ...tools import entities as tools_entities
 
@@ -25,7 +25,6 @@ class OpenAIChatCompletions(requester.LLMAPIRequester):
     }
 
 
-
     async def initialize(self):
         self.client = openai.AsyncClient(
             api_key='',
@@ -53,6 +52,7 @@ class OpenAIChatCompletions(requester.LLMAPIRequester):
 
     async def _make_msg(
         self,
+        pipeline_config: dict[str, typing.Any],
         chat_completion: chat_completion.ChatCompletion,
     ) -> llm_entities.Message:
         chatcmpl_message = chat_completion.choices[0].message.model_dump()
@@ -64,8 +64,12 @@ class OpenAIChatCompletions(requester.LLMAPIRequester):
         reasoning_content = chatcmpl_message['reasoning_content'] if 'reasoning_content' in chatcmpl_message else None
 
         # deepseek的reasoner模型
-        if reasoning_content is not None:
+        print(pipeline_config['trigger'].get('misc', '').get('remove_think'))
-            chatcmpl_message['content'] = '<think>\n' + reasoning_content + '\n</think>\n' + chatcmpl_message['content']
+        if pipeline_config['trigger'].get('misc', '').get('remove_think'):
+            pass
+        else:
+            if reasoning_content is not None :
+                chatcmpl_message['content'] = '<think>\n' + reasoning_content + '\n</think>\n' + chatcmpl_message['content']
 
         message = llm_entities.Message(**chatcmpl_message)
 
@@ -73,7 +77,7 @@ class OpenAIChatCompletions(requester.LLMAPIRequester):
     
     async def _make_msg_chunk(
         self,
-        index:int,
+        pipeline_config: dict[str, typing.Any],
         chat_completion: chat_completion.ChatCompletion,
     ) -> llm_entities.MessageChunk:
 
@@ -96,16 +100,22 @@ class OpenAIChatCompletions(requester.LLMAPIRequester):
         reasoning_content = delta['reasoning_content'] if 'reasoning_content' in delta else None
 
         # deepseek的reasoner模型
-        if reasoning_content is not None and index == 0:
+        if pipeline_config['trigger'].get('misc', '').get('remove_think'):
-            delta['content']  += f'<think>\n{reasoning_content}'
+            if reasoning_content is not None :
-        elif reasoning_content is None:
+                pass
-            if self.is_content:
-                delta['content'] = delta['content']
             else:
-                delta['content'] = f'\n<think>\n\n{delta["content"]}'
+                delta['content'] = delta['content']
-                self.is_content = True
         else:
-            delta['content'] += reasoning_content
+            if reasoning_content is not None:
+                delta['content']  += f'<think>\n{reasoning_content}'
+            elif reasoning_content is None:
+                if self.is_content:
+                    delta['content'] = delta['content']
+                else:
+                    delta['content'] = f'\n<think>\n\n{delta["content"]}'
+                    self.is_content = True
+            else:
+                delta['content'] += reasoning_content
 
 
         message = llm_entities.MessageChunk(**delta)
@@ -151,20 +161,41 @@ class OpenAIChatCompletions(requester.LLMAPIRequester):
             args["stream"] = True
             chunk_idx = 0
             self.is_content = False
+            tool_calls_map: dict[str, llm_entities.ToolCall] = {}
+            pipeline_config = query.pipeline_config
             async for chunk in self._req_stream(args, extra_body=extra_args):
                 # 处理流式消息
-                delta_message = await self._make_msg_chunk(chunk_idx,chunk)
+                delta_message = await self._make_msg_chunk(pipeline_config,chunk)
-                # print(delta_message)
                 if delta_message.content:
                     current_content += delta_message.content
                     delta_message.content = current_content
+                    print(current_content)
                     # delta_message.all_content = current_content
+                if delta_message.tool_calls:
+                    for tool_call in delta_message.tool_calls:
+                        if tool_call.id not in tool_calls_map:
+                            tool_calls_map[tool_call.id] = llm_entities.ToolCall(
+                                id=tool_call.id,
+                                type=tool_call.type,
+                                function=llm_entities.FunctionCall(
+                                    name=tool_call.function.name if tool_call.function else '',
+                                    arguments=''
+                                ),
+                            )
+                        if tool_call.function and tool_call.function.arguments:
+                            # 流式处理中，工具调用参数可能分多个chunk返回，需要追加而不是覆盖
+                            tool_calls_map[tool_call.id].function.arguments += tool_call.function.arguments
+
+
                 chunk_idx += 1
                 chunk_choices = getattr(chunk, 'choices', None)
                 if chunk_choices and getattr(chunk_choices[0], 'finish_reason', None):
                     delta_message.is_final = True
+                    delta_message.content = current_content
 
-                yield delta_message
+                if chunk_idx % 64 == 0 or delta_message.is_final:
+
+                    yield delta_message
                 # return
 
     
@@ -208,7 +239,8 @@ class OpenAIChatCompletions(requester.LLMAPIRequester):
 
         resp = await self._req(args, extra_body=extra_args)
         # 处理请求结果
-        message = await self._make_msg(resp)
+        pipeline_config = query.pipeline_config
+        message = await self._make_msg(pipeline_config,resp)
 
 
         return message

pkg/provider/runners/difysvapi.py

@@ -95,6 +95,11 @@ class DifyServiceAPIRunner(runner.RequestRunner):
         cov_id = query.session.using_conversation.uuid or ''
         query.variables['conversation_id'] = cov_id
 
+        try:
+            is_stream = query.adapter.is_stream
+        except AttributeError:
+            is_stream = False
+
         plain_text, image_ids = await self._preprocess_user_message(query)
 
         files = [
@@ -144,40 +149,54 @@ class DifyServiceAPIRunner(runner.RequestRunner):
 
             if mode == 'workflow':
                 if chunk['event'] == 'node_finished':
-                    if chunk['data']['node_type'] == 'answer':
+                    if not is_stream:
-                        yield llm_entities.Message(
+
-                            role='assistant',
+                        if chunk['data']['node_type'] == 'answer':
-                            content=self._try_convert_thinking(chunk['data']['outputs']['answer']),
+                            yield llm_entities.Message(
-                        )
+                                role='assistant',
+                                content=self._try_convert_thinking(chunk['data']['outputs']['answer']),
+                            )
+                    else:
+                        if chunk['data']['node_type'] == 'answer':
+                            yield llm_entities.MessageChunk(
+                                role='assistant',
+                                content=self._try_convert_thinking(chunk['data']['outputs']['answer']),
+                                is_final=True,
+                            )
                 elif chunk['event'] == 'message':
                     stream_output_pending_chunk += chunk['answer']
-                    if self.pipeline_config['ai']['dify-service-api'].get('enable-streaming', False):
+                    if is_stream:
                         # 消息数超过量就输出，从而达到streaming的效果
                         batch_pending_index += 1
                         if batch_pending_index >= batch_pending_max_size:
-                            yield llm_entities.Message(
+                            yield llm_entities.MessageChunk(
                                 role='assistant',
                                 content=self._try_convert_thinking(stream_output_pending_chunk),
                             )
                             batch_pending_index = 0
             elif mode == 'basic':
-                if chunk['event'] == 'message':
+                if chunk['event'] == 'message' or chunk['event'] == 'message_end':
-                    stream_output_pending_chunk += chunk['answer']
+                    if chunk['event'] == 'message_end':
-                    if self.pipeline_config['ai']['dify-service-api'].get('enable-streaming', False):
+                        is_final = True
-                        # 消息数超过量就输出，从而达到streaming的效果
+                        if is_stream and batch_pending_index % batch_pending_max_size == 0:
-                        batch_pending_index += 1
+                            # 消息数超过量就输出，从而达到streaming的效果
-                        if batch_pending_index >= batch_pending_max_size:
+                            batch_pending_index += 1
+                            # if batch_pending_index >= batch_pending_max_size:
+                            yield llm_entities.MessageChunk(
+                                role='assistant',
+                                content=self._try_convert_thinking(stream_output_pending_chunk),
+                                is_final=is_final,
+                            )
+                            # batch_pending_index = 0
+                        elif not is_stream:
                             yield llm_entities.Message(
                                 role='assistant',
                                 content=self._try_convert_thinking(stream_output_pending_chunk),
                             )
-                            batch_pending_index = 0
+                            stream_output_pending_chunk = ''
-                elif chunk['event'] == 'message_end':
+                    else:
-                    yield llm_entities.Message(
+                        stream_output_pending_chunk += chunk['answer']
-                        role='assistant',
+                        is_final = False
-                        content=self._try_convert_thinking(stream_output_pending_chunk),
-                    )
-                    stream_output_pending_chunk = ''
 
         if chunk is None:
             raise errors.DifyAPIError('Dify API 没有返回任何响应，请检查网络连接和API配置')
@@ -191,6 +210,13 @@ class DifyServiceAPIRunner(runner.RequestRunner):
         cov_id = query.session.using_conversation.uuid or ''
         query.variables['conversation_id'] = cov_id
 
+        try:
+            is_stream = query.adapter.is_stream
+        except AttributeError:
+            is_stream = False
+
+        batch_pending_index = 0
+
         plain_text, image_ids = await self._preprocess_user_message(query)
 
         files = [
@@ -285,6 +311,13 @@ class DifyServiceAPIRunner(runner.RequestRunner):
 
         query.variables['conversation_id'] = query.session.using_conversation.uuid
 
+        try:
+            is_stream = query.adapter.is_stream
+        except AttributeError:
+            is_stream = False
+
+        batch_pending_index = 0
+
         plain_text, image_ids = await self._preprocess_user_message(query)
 
         files = [

templates/metadata/pipeline/trigger.yaml

@@ -132,3 +132,10 @@ stages:
         type: boolean
         required: true
         default: true
+      - name: remove_think
+        label:
+          en_US: remove think
+          zh_Hans: 删除深度思考消息
+        type: boolean
+        required: true
+        default: true