add dingtalk file support, fix video/voice file to bailian (#1683)

* add dingtalk file support, fix video/voice file to bailian

* fix bailian files conversation

* 更新 bailianchatcmpl.py

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Update libs/dingtalk_api/api.py

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* chore: bump langbot-plugin version to 0.1.3b1

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Junyan Qin <rockchinq@gmail.com>

libs/dingtalk_api/api.py

@@ -110,6 +110,24 @@ class DingTalkClient:
             else:
                 raise Exception(f'Error: {response.status_code}, {response.text}')
 
+    async def get_file_url(self, download_code: str):
+        if not await self.check_access_token():
+            await self.get_access_token()
+        url = 'https://api.dingtalk.com/v1.0/robot/messageFiles/download'
+        params = {'downloadCode': download_code, 'robotCode': self.robot_code}
+        headers = {'x-acs-dingtalk-access-token': self.access_token}
+        async with httpx.AsyncClient() as client:
+            response = await client.post(url, headers=headers, json=params)
+            if response.status_code == 200:
+                result = response.json()
+                download_url = result.get('downloadUrl')
+                if download_url:
+                    return download_url
+                else:
+                    await self.logger.error(f'failed to get file: {response.json()}')
+            else:
+                raise Exception(f'Error: {response.status_code}, {response.text}')
+
     async def update_incoming_message(self, message):
         """异步更新 DingTalkClient 中的 incoming_message"""
         message_data = await self.get_message(message)
@@ -189,6 +207,17 @@ class DingTalkClient:
                 message_data['Audio'] = await self.get_audio_url(incoming_message.to_dict()['content']['downloadCode'])
 
                 message_data['Type'] = 'audio'
+            elif incoming_message.message_type == 'file':
+                down_list = incoming_message.get_down_list()
+                if len(down_list) >= 2:
+                    message_data['File'] = await self.get_file_url(down_list[0])
+                    message_data['Name'] = down_list[1]
+                else:
+                    if self.logger:
+                        await self.logger.error(f'get_down_list() returned fewer than 2 elements: {down_list}')
+                    message_data['File'] = None
+                    message_data['Name'] = None
+                message_data['Type'] = 'file'
 
             copy_message_data = message_data.copy()
             del copy_message_data['IncomingMessage']

libs/dingtalk_api/dingtalkevent.py

@@ -31,6 +31,15 @@ class DingTalkEvent(dict):
     def audio(self):
         return self.get('Audio', '')
 
+    @property
+    def file(self):
+        return self.get('File', '')
+
+    @property
+    def name(self):
+        return self.get('Name', '')
+
+
     @property
     def conversation(self):
         return self.get('conversation_type', '')

pkg/pipeline/preproc/preproc.py

@@ -92,6 +92,9 @@ class PreProcessor(stage.PipelineStage):
                 if selected_runner != 'local-agent' or llm_model.model_entity.abilities.__contains__('vision'):
                     if me.base64 is not None:
                         content_list.append(provider_message.ContentElement.from_image_base64(me.base64))
+            elif isinstance(me, platform_message.File):
+                # if me.url is not None:
+                content_list.append(provider_message.ContentElement.from_file_url(me.url, me.name))
             elif isinstance(me, platform_message.Quote) and qoute_msg:
                 for msg in me.origin:
                     if isinstance(msg, platform_message.Plain):

pkg/pipeline/process/handlers/command.py

@@ -80,7 +80,8 @@ class CommandHandler(handler.MessageHandler):
                     self.ap.logger.info(f'Command({query.query_id}) error: {self.cut_str(str(ret.error))}')
 
                     yield entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query)
-                elif ret.text is not None or ret.image_url is not None or ret.image_base64 is not None:
+                elif (ret.text is not None or ret.image_url is not None or ret.image_base64 is not None
+                      or ret.file_url is not None):
                     content: list[provider_message.ContentElement] = []
 
                     if ret.text is not None:
@@ -92,6 +93,9 @@ class CommandHandler(handler.MessageHandler):
                     if ret.image_base64 is not None:
                         content.append(provider_message.ContentElement.from_image_base64(ret.image_base64))
 
+                    if ret.file_url is not None:
+                        # 此时为 file 类型
+                        content.append(provider_message.ContentElement.from_file_url(ret.file_url, ret.file_name))
                     query.resp_messages.append(
                         provider_message.Message(
                             role='command',

pkg/platform/sources/dingtalk.py

@@ -41,6 +41,8 @@ class DingTalkMessageConverter(abstract_platform_adapter.AbstractMessageConverte
             yiri_msg_list.append(platform_message.Plain(text=text_content))
         if event.picture:
             yiri_msg_list.append(platform_message.Image(base64=event.picture))
+        if event.file:
+            yiri_msg_list.append(platform_message.File(url=event.file, name=event.name))
         if event.audio:
             yiri_msg_list.append(platform_message.Voice(base64=event.audio))
 

pkg/provider/modelmgr/requesters/bailianchatcmpl.py

@@ -1,9 +1,14 @@
 from __future__ import annotations
 
 import typing
+import dashscope
 import openai
 
 from . import modelscopechatcmpl
+from .. import requester
+import langbot_plugin.api.entities.builtin.resource.tool as resource_tool
+import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
+import langbot_plugin.api.entities.builtin.provider.message as provider_message
 
 
 class BailianChatCompletions(modelscopechatcmpl.ModelScopeChatCompletions):
@@ -15,3 +20,211 @@ class BailianChatCompletions(modelscopechatcmpl.ModelScopeChatCompletions):
         'base_url': 'https://dashscope.aliyuncs.com/compatible-mode/v1',
         'timeout': 120,
     }
+
+    async def _closure_stream(
+        self,
+        query: pipeline_query.Query,
+        req_messages: list[dict],
+        use_model: requester.RuntimeLLMModel,
+        use_funcs: list[resource_tool.LLMTool] = None,
+        extra_args: dict[str, typing.Any] = {},
+        remove_think: bool = False,
+    ) -> provider_message.Message | typing.AsyncGenerator[provider_message.MessageChunk, None]:
+        self.client.api_key = use_model.token_mgr.get_token()
+
+        args = {}
+        args['model'] = use_model.model_entity.name
+
+        if use_funcs:
+            tools = await self.ap.tool_mgr.generate_tools_for_openai(use_funcs)
+
+            if tools:
+                args['tools'] = tools
+
+        # 设置此次请求中的messages
+        messages = req_messages.copy()
+
+        is_use_dashscope_call = False   # 是否使用阿里原生库调用
+        is_enable_multi_model = True  # 是否支持多轮对话
+        use_time_num = 0 # 模型已调用次数，防止存在多文件时重复调用
+        use_time_ids = [] # 已调用的ID列表
+        message_id = 0  # 记录消息序号
+
+        for msg in messages:
+            # print(msg)
+            if 'content' in msg and isinstance(msg['content'], list):
+                for me in msg['content']:
+                    if me['type'] == 'image_base64':
+                        me['image_url'] = {'url': me['image_base64']}
+                        me['type'] = 'image_url'
+                        del me['image_base64']
+                    elif me['type'] == 'file_url' and '.' in me.get('file_name', ''):
+                        # 1. 视频文件推理
+                        # https://bailian.console.aliyun.com/?tab=doc#/doc/?type=model&url=2845871
+                        file_type = me.get('file_name').lower().split('.')[-1]
+                        if file_type in ['mp4', 'avi', 'mkv', 'mov', 'flv', 'wmv']:
+                            me['type'] = 'video_url'
+                            me['video_url'] = {'url': me['file_url']}
+                            del me['file_url']
+                            del me['file_name']
+                            use_time_num +=1
+                            use_time_ids.append(message_id)
+                            is_enable_multi_model = False
+                        # 2. 语音文件识别, 无法通过openai的audio字段传递，暂时不支持
+                        # https://bailian.console.aliyun.com/?tab=doc#/doc/?type=model&url=2979031
+                        elif file_type in ['aac', 'amr', 'aiff', 'flac', 'm4a',
+                                           'mp3', 'mpeg', 'ogg', 'opus', 'wav', 'webm', 'wma']:
+                            me['audio'] = me['file_url']
+                            me['type'] = 'audio'
+                            del me['file_url']
+                            del me['type']
+                            del me['file_name']
+                            is_use_dashscope_call = True
+                            use_time_num +=1
+                            use_time_ids.append(message_id)
+                            is_enable_multi_model = False
+            message_id += 1
+
+        # 使用列表推导式，保留不在 use_time_ids[:-1] 中的元素，仅保留最后一个多媒体消息
+        if not is_enable_multi_model and use_time_num > 1:
+            messages = [msg for idx, msg in enumerate(messages) if idx not in use_time_ids[:-1]]
+
+        if not is_enable_multi_model:
+            messages = [msg for msg in messages if 'resp_message_id' not in msg]
+
+        args['messages'] = messages
+        args['stream'] = True
+
+        # 流式处理状态
+        # tool_calls_map: dict[str, provider_message.ToolCall] = {}
+        chunk_idx = 0
+        thinking_started = False
+        thinking_ended = False
+        role = 'assistant'  # 默认角色
+
+        if is_use_dashscope_call:
+            response = dashscope.MultiModalConversation.call(
+                # 若没有配置环境变量，请用百炼API Key将下行替换为：api_key = "sk-xxx"
+                api_key=use_model.token_mgr.get_token(),
+                model=use_model.model_entity.name,
+                messages=messages,
+                result_format="message",
+                asr_options={
+                    # "language": "zh", # 可选，若已知音频的语种，可通过该参数指定待识别语种，以提升识别准确率
+                    "enable_lid": True,
+                    "enable_itn": False
+                },
+                stream=True
+            )
+            content_length_list = []
+            previous_length = 0  # 记录上一次的内容长度
+            for res in response:
+                chunk = res["output"]
+                # 解析 chunk 数据
+                if hasattr(chunk, 'choices') and chunk.choices:
+                    choice = chunk.choices[0]
+                    delta_content = choice["message"].content[0]["text"]
+                    finish_reason = choice["finish_reason"]
+                    content_length_list.append(len(delta_content))
+                else:
+                    delta_content = ""
+                    finish_reason = None
+
+                # 跳过空的第一个 chunk（只有 role 没有内容）
+                if chunk_idx == 0 and not delta_content:
+                    chunk_idx += 1
+                    continue
+
+                # 检查 content_length_list 是否有足够的数据
+                if len(content_length_list) >= 2:
+                    now_content = delta_content[previous_length: content_length_list[-1]]
+                    previous_length = content_length_list[-1]  # 更新上一次的长度
+                else:
+                    now_content = delta_content  # 第一次循环时直接使用 delta_content
+                    previous_length = len(delta_content)  # 更新上一次的长度
+
+                # 构建 MessageChunk - 只包含增量内容
+                chunk_data = {
+                    'role': role,
+                    'content': now_content if now_content else None,
+                    'is_final': bool(finish_reason) and finish_reason != "null",
+                }
+
+                # 移除 None 值
+                chunk_data = {k: v for k, v in chunk_data.items() if v is not None}
+                yield provider_message.MessageChunk(**chunk_data)
+                chunk_idx += 1
+        else:
+            async for chunk in self._req_stream(args, extra_body=extra_args):
+                # 解析 chunk 数据
+                if hasattr(chunk, 'choices') and chunk.choices:
+                    choice = chunk.choices[0]
+                    delta = choice.delta.model_dump() if hasattr(choice, 'delta') else {}
+                    finish_reason = getattr(choice, 'finish_reason', None)
+                else:
+                    delta = {}
+                    finish_reason = None
+
+                # 从第一个 chunk 获取 role，后续使用这个 role
+                if 'role' in delta and delta['role']:
+                    role = delta['role']
+
+                # 获取增量内容
+                delta_content = delta.get('content', '')
+                reasoning_content = delta.get('reasoning_content', '')
+
+                # 处理 reasoning_content
+                if reasoning_content:
+                    # accumulated_reasoning += reasoning_content
+                    # 如果设置了 remove_think，跳过 reasoning_content
+                    if remove_think:
+                        chunk_idx += 1
+                        continue
+
+                    # 第一次出现 reasoning_content，添加 <think> 开始标签
+                    if not thinking_started:
+                        thinking_started = True
+                        delta_content = '<think>\n' + reasoning_content
+                    else:
+                        # 继续输出 reasoning_content
+                        delta_content = reasoning_content
+                elif thinking_started and not thinking_ended and delta_content:
+                    # reasoning_content 结束，normal content 开始，添加 </think> 结束标签
+                    thinking_ended = True
+                    delta_content = '\n</think>\n' + delta_content
+
+                # 处理工具调用增量
+                if delta.get('tool_calls'):
+                    for tool_call in delta['tool_calls']:
+                        if tool_call['id'] != '':
+                            tool_id = tool_call['id']
+                        if tool_call['function']['name'] is not None:
+                            tool_name = tool_call['function']['name']
+
+                        if tool_call['type'] is None:
+                            tool_call['type'] = 'function'
+                        tool_call['id'] = tool_id
+                        tool_call['function']['name'] = tool_name
+                        tool_call['function']['arguments'] = (
+                            '' if tool_call['function']['arguments'] is None else tool_call['function']['arguments']
+                        )
+
+                # 跳过空的第一个 chunk（只有 role 没有内容）
+                if chunk_idx == 0 and not delta_content and not reasoning_content and not delta.get('tool_calls'):
+                    chunk_idx += 1
+                    continue
+
+                # 构建 MessageChunk - 只包含增量内容
+                chunk_data = {
+                    'role': role,
+                    'content': delta_content if delta_content else None,
+                    'tool_calls': delta.get('tool_calls'),
+                    'is_final': bool(finish_reason),
+                }
+
+                # 移除 None 值
+                chunk_data = {k: v for k, v in chunk_data.items() if v is not None}
+
+                yield provider_message.MessageChunk(**chunk_data)
+                chunk_idx += 1
+                # return

pyproject.toml

@@ -62,7 +62,7 @@ dependencies = [
     "langchain>=0.2.0",
     "chromadb>=0.4.24",
     "qdrant-client (>=1.15.1,<2.0.0)",
-    "langbot-plugin==0.1.2",
+    "langbot-plugin==0.1.3b1",
     "asyncpg>=0.30.0",
     "line-bot-sdk>=3.19.0"
 ]