Merge branch 'rc/new-plugin' into refactor/new-plugin-system

pkg/provider/runners/dashscopeapi.py

@@ -102,8 +102,14 @@ class DashScopeAPIRunner(runner.RequestRunner):
         plain_text = ''  # 用户输入的纯文本信息
         image_ids = []  # 用户输入的图片ID列表 （暂不支持）
 
-        plain_text, image_ids = await self._preprocess_user_message(query)
+        think_start = False
+        think_end = False
 
+        plain_text, image_ids = await self._preprocess_user_message(query)
+        has_thoughts = True  # 获取思考过程
+        remove_think = self.pipeline_config['output'].get('misc', '').get('remove-think')
+        if remove_think:
+            has_thoughts = False
         # 发送对话请求
         response = dashscope.Application.call(
             api_key=self.api_key,  # 智能体应用的API Key
@@ -112,43 +118,108 @@ class DashScopeAPIRunner(runner.RequestRunner):
             stream=True,  # 流式输出
             incremental_output=True,  # 增量输出，使用流式输出需要开启增量输出
             session_id=query.session.using_conversation.uuid,  # 会话ID用于，多轮对话
+            has_thoughts=has_thoughts,
             # rag_options={                                     # 主要用于文件交互，暂不支持
             #     "session_file_ids": ["FILE_ID1"],             # FILE_ID1 替换为实际的临时文件ID,逗号隔开多个
             # }
         )
+        idx_chunk = 0
+        try:
+            is_stream = await query.adapter.is_stream_output_supported()
 
-        for chunk in response:
-            if chunk.get('status_code') != 200:
-                raise DashscopeAPIError(
-                    f'Dashscope API 请求失败: status_code={chunk.get("status_code")} message={chunk.get("message")} request_id={chunk.get("request_id")} '
-                )
-            if not chunk:
-                continue
+        except AttributeError:
+            is_stream = False
+        if is_stream:
+            for chunk in response:
+                if chunk.get('status_code') != 200:
+                    raise DashscopeAPIError(
+                        f'Dashscope API 请求失败: status_code={chunk.get("status_code")} message={chunk.get("message")} request_id={chunk.get("request_id")} '
+                    )
+                if not chunk:
+                    continue
+                idx_chunk += 1
+                # 获取流式传输的output
+                stream_output = chunk.get('output', {})
+                stream_think = stream_output.get('thoughts', [])
+                if stream_think[0].get('thought'):
+                    if not think_start:
+                        think_start = True
+                        pending_content += f'<think>\n{stream_think[0].get("thought")}'
+                    else:
+                        # 继续输出 reasoning_content
+                        pending_content += stream_think[0].get('thought')
+                elif stream_think[0].get('thought') == '' and not think_end:
+                    think_end = True
+                    pending_content += '\n</think>\n'
+                if stream_output.get('text') is not None:
+                    pending_content += stream_output.get('text')
+                # 是否是流式最后一个chunk
+                is_final = False if stream_output.get('finish_reason', False) == 'null' else True
 
-            # 获取流式传输的output
-            stream_output = chunk.get('output', {})
-            if stream_output.get('text') is not None:
-                pending_content += stream_output.get('text')
+                # 获取模型传出的参考资料列表
+                references_dict_list = stream_output.get('doc_references', [])
 
-        # 保存当前会话的session_id用于下次对话的语境
-        query.session.using_conversation.uuid = stream_output.get('session_id')
+                # 从模型传出的参考资料信息中提取用于替换的字典
+                if references_dict_list is not None:
+                    for doc in references_dict_list:
+                        if doc.get('index_id') is not None:
+                            references_dict[doc.get('index_id')] = doc.get('doc_name')
 
-        # 获取模型传出的参考资料列表
-        references_dict_list = stream_output.get('doc_references', [])
+                    # 将参考资料替换到文本中
+                    pending_content = self._replace_references(pending_content, references_dict)
 
-        # 从模型传出的参考资料信息中提取用于替换的字典
-        if references_dict_list is not None:
-            for doc in references_dict_list:
-                if doc.get('index_id') is not None:
-                    references_dict[doc.get('index_id')] = doc.get('doc_name')
+                if idx_chunk % 8 == 0 or is_final:
+                    yield provider_message.MessageChunk(
+                        role='assistant',
+                        content=pending_content,
+                        is_final=is_final,
+                    )
+            # 保存当前会话的session_id用于下次对话的语境
+            query.session.using_conversation.uuid = stream_output.get('session_id')
+        else:
+            for chunk in response:
+                if chunk.get('status_code') != 200:
+                    raise DashscopeAPIError(
+                        f'Dashscope API 请求失败: status_code={chunk.get("status_code")} message={chunk.get("message")} request_id={chunk.get("request_id")} '
+                    )
+                if not chunk:
+                    continue
+                idx_chunk += 1
+                # 获取流式传输的output
+                stream_output = chunk.get('output', {})
+                stream_think = stream_output.get('thoughts', [])
+                if stream_think[0].get('thought'):
+                    if not think_start:
+                        think_start = True
+                        pending_content += f'<think>\n{stream_think[0].get("thought")}'
+                    else:
+                        # 继续输出 reasoning_content
+                        pending_content += stream_think[0].get('thought')
+                elif stream_think[0].get('thought') == '' and not think_end:
+                    think_end = True
+                    pending_content += '\n</think>\n'
+                if stream_output.get('text') is not None:
+                    pending_content += stream_output.get('text')
 
-            # 将参考资料替换到文本中
-            pending_content = self._replace_references(pending_content, references_dict)
+            # 保存当前会话的session_id用于下次对话的语境
+            query.session.using_conversation.uuid = stream_output.get('session_id')
 
-        yield provider_message.Message(
-            role='assistant',
-            content=pending_content,
-        )
+            # 获取模型传出的参考资料列表
+            references_dict_list = stream_output.get('doc_references', [])
+
+            # 从模型传出的参考资料信息中提取用于替换的字典
+            if references_dict_list is not None:
+                for doc in references_dict_list:
+                    if doc.get('index_id') is not None:
+                        references_dict[doc.get('index_id')] = doc.get('doc_name')
+
+                # 将参考资料替换到文本中
+                pending_content = self._replace_references(pending_content, references_dict)
+
+            yield provider_message.Message(
+                role='assistant',
+                content=pending_content,
+            )
 
     async def _workflow_messages(
         self, query: pipeline_query.Query
@@ -176,52 +247,108 @@ class DashScopeAPIRunner(runner.RequestRunner):
             incremental_output=True,  # 增量输出，使用流式输出需要开启增量输出
             session_id=query.session.using_conversation.uuid,  # 会话ID用于，多轮对话
             biz_params=biz_params,  # 工作流应用的自定义输入参数传递
+            flow_stream_mode='message_format',  # 消息模式，输出/结束节点的流式结果
             # rag_options={                                     # 主要用于文件交互，暂不支持
             #     "session_file_ids": ["FILE_ID1"],             # FILE_ID1 替换为实际的临时文件ID,逗号隔开多个
             # }
         )
 
         # 处理API返回的流式输出
-        for chunk in response:
-            if chunk.get('status_code') != 200:
-                raise DashscopeAPIError(
-                    f'Dashscope API 请求失败: status_code={chunk.get("status_code")} message={chunk.get("message")} request_id={chunk.get("request_id")} '
-                )
-            if not chunk:
-                continue
+        try:
+            is_stream = await query.adapter.is_stream_output_supported()
 
-            # 获取流式传输的output
-            stream_output = chunk.get('output', {})
-            if stream_output.get('text') is not None:
-                pending_content += stream_output.get('text')
+        except AttributeError:
+            is_stream = False
+        idx_chunk = 0
+        if is_stream:
+            for chunk in response:
+                if chunk.get('status_code') != 200:
+                    raise DashscopeAPIError(
+                        f'Dashscope API 请求失败: status_code={chunk.get("status_code")} message={chunk.get("message")} request_id={chunk.get("request_id")} '
+                    )
+                if not chunk:
+                    continue
+                idx_chunk += 1
+                # 获取流式传输的output
+                stream_output = chunk.get('output', {})
+                if stream_output.get('workflow_message') is not None:
+                    pending_content += stream_output.get('workflow_message').get('message').get('content')
+                # if stream_output.get('text') is not None:
+                #     pending_content += stream_output.get('text')
 
-        # 保存当前会话的session_id用于下次对话的语境
-        query.session.using_conversation.uuid = stream_output.get('session_id')
+                is_final = False if stream_output.get('finish_reason', False) == 'null' else True
 
-        # 获取模型传出的参考资料列表
-        references_dict_list = stream_output.get('doc_references', [])
+                # 获取模型传出的参考资料列表
+                references_dict_list = stream_output.get('doc_references', [])
 
-        # 从模型传出的参考资料信息中提取用于替换的字典
-        if references_dict_list is not None:
-            for doc in references_dict_list:
-                if doc.get('index_id') is not None:
-                    references_dict[doc.get('index_id')] = doc.get('doc_name')
+                # 从模型传出的参考资料信息中提取用于替换的字典
+                if references_dict_list is not None:
+                    for doc in references_dict_list:
+                        if doc.get('index_id') is not None:
+                            references_dict[doc.get('index_id')] = doc.get('doc_name')
 
-            # 将参考资料替换到文本中
-            pending_content = self._replace_references(pending_content, references_dict)
+                    # 将参考资料替换到文本中
+                    pending_content = self._replace_references(pending_content, references_dict)
+                if idx_chunk % 8 == 0 or is_final:
+                    yield provider_message.MessageChunk(
+                        role='assistant',
+                        content=pending_content,
+                        is_final=is_final,
+                    )
 
-        yield provider_message.Message(
-            role='assistant',
-            content=pending_content,
-        )
+            # 保存当前会话的session_id用于下次对话的语境
+            query.session.using_conversation.uuid = stream_output.get('session_id')
+
+        else:
+            for chunk in response:
+                if chunk.get('status_code') != 200:
+                    raise DashscopeAPIError(
+                        f'Dashscope API 请求失败: status_code={chunk.get("status_code")} message={chunk.get("message")} request_id={chunk.get("request_id")} '
+                    )
+                if not chunk:
+                    continue
+
+                # 获取流式传输的output
+                stream_output = chunk.get('output', {})
+                if stream_output.get('text') is not None:
+                    pending_content += stream_output.get('text')
+
+                is_final = False if stream_output.get('finish_reason', False) == 'null' else True
+
+            # 保存当前会话的session_id用于下次对话的语境
+            query.session.using_conversation.uuid = stream_output.get('session_id')
+
+            # 获取模型传出的参考资料列表
+            references_dict_list = stream_output.get('doc_references', [])
+
+            # 从模型传出的参考资料信息中提取用于替换的字典
+            if references_dict_list is not None:
+                for doc in references_dict_list:
+                    if doc.get('index_id') is not None:
+                        references_dict[doc.get('index_id')] = doc.get('doc_name')
+
+                # 将参考资料替换到文本中
+                pending_content = self._replace_references(pending_content, references_dict)
+
+            yield provider_message.Message(
+                role='assistant',
+                content=pending_content,
+            )
 
     async def run(self, query: pipeline_query.Query) -> typing.AsyncGenerator[provider_message.Message, None]:
         """运行"""
+        msg_seq = 0
         if self.app_type == 'agent':
             async for msg in self._agent_messages(query):
+                if isinstance(msg, provider_message.MessageChunk):
+                    msg_seq += 1
+                    msg.msg_sequence = msg_seq
                 yield msg
         elif self.app_type == 'workflow':
             async for msg in self._workflow_messages(query):
+                if isinstance(msg, provider_message.MessageChunk):
+                    msg_seq += 1
+                    msg.msg_sequence = msg_seq
                 yield msg
         else:
             raise DashscopeAPIError(f'不支持的 Dashscope 应用类型: {self.app_type}')