diff --git a/pkg/provider/modelmgr/modelmgr.py b/pkg/provider/modelmgr/modelmgr.py index 737d3476..33a65ff3 100644 --- a/pkg/provider/modelmgr/modelmgr.py +++ b/pkg/provider/modelmgr/modelmgr.py @@ -6,7 +6,7 @@ from . import entities, requester from ...core import app from . import token -from .requesters import chatcmpl, anthropicmsgs, moonshotchatcmpl, deepseekchatcmpl, ollamachat, giteeaichatcmpl, xaichatcmpl, zhipuaichatcmpl, lmstudiochatcmpl, siliconflowchatcmpl, dashscopecmpl, qwenchatcmpl +from .requesters import chatcmpl, anthropicmsgs, moonshotchatcmpl, deepseekchatcmpl, ollamachat, giteeaichatcmpl, xaichatcmpl, zhipuaichatcmpl, lmstudiochatcmpl, siliconflowchatcmpl FETCH_MODEL_LIST_URL = "https://api.qchatgpt.rockchin.top/api/v2/fetch/model_list" diff --git a/pkg/provider/modelmgr/requesters/dashscopecmpl.py b/pkg/provider/modelmgr/requesters/dashscopecmpl.py deleted file mode 100644 index d2f2b2d2..00000000 --- a/pkg/provider/modelmgr/requesters/dashscopecmpl.py +++ /dev/null @@ -1,167 +0,0 @@ -from __future__ import annotations - -import re -import asyncio -import typing -import dashscope - -from .. import entities, errors, requester -from ....core import entities as core_entities, app -from ... import entities as llm_entities -from ...tools import entities as tools_entities - -#阿里云百炼平台的自定义应用支持资料引用,此函数可以将引用标签替换为参考资料 -def replace_references(text, references_dict): - # 修正正则表达式,匹配 [index_id] 形式的字符串 - pattern = re.compile(r'\[(.*?)\]') - - def replacement(match): - ref_key = match.group(1) # 获取引用编号 - if ref_key in references_dict: - return f"(参考资料来自:{references_dict[ref_key]})" - else: - return match.group(0) # 如果没有对应的参考资料,保留原样 - - # 使用 re.sub() 进行替换 - return pattern.sub(replacement, text) - - -@requester.requester_class("dashscope-chat-applications") -class DashscopeChatApplication(requester.LLMAPIRequester): - """Dashscope ChatApplications API 请求器""" - - requester_cfg: dict - - def __init__(self, ap: app.Application): - self.requester_cfg = ap.provider_cfg.data['requester']['dashscope-chat-applications'] - self.ap = ap - - async def initialize(self): - dashscope.api_key = self.ap.provider_cfg.data['keys']['dashscope'][0] - - async def _req(self, args: dict): - - #print("args:", args) - - #局部变量 - chunk = None - pending_content = "" - output = { - "role": "assistant", - "content": "", - "tool_calls": [], - "tool_call_id": None # Dashscope暂时不支持工具调用 - } #由于Dashscope的content的键值是text,所以需要定义一个新格式的字典适配llm_entities.Message - - references_dict = {} # 用于存储引用编号和对应的参考资料 - - #调用API - response = dashscope.Application.call( - api_key=dashscope.api_key, - app_id=args["model"], - prompt=args["messages"], - stream=True, # 设置流式输出 - tools=args.get("tools", None), - incremental_output = True, - ) - - #处理API返回的流式输出 - for chunk in response: - #print(chunk) - if not chunk: - continue - - #获取流式传输的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", []) - - #从模型传出的参考资料信息中提取用于替换的字典 - 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 = replace_references(pending_content, references_dict) - - #将流式传输的内容整合到output中 - output["content"] = pending_content - - return output if chunk else None - - async def _make_msg( - self, - chat_completion: dict, - ) -> llm_entities.Message: - chatcmpl_message = chat_completion - - # 确保 role 字段存在且不为 None - if 'role' not in chatcmpl_message or chatcmpl_message['role'] is None: - chatcmpl_message['role'] = 'assistant' - - message = llm_entities.Message(**chatcmpl_message) - #print("message:", message) - return message - - async def _closure( - self, - query: core_entities.Query, - req_messages: list[dict], - use_model: entities.LLMModelInfo, - use_funcs: list[tools_entities.LLMFunction] = None, - ) -> llm_entities.Message: - - args = self.requester_cfg['args'].copy() - args["model"] = use_model.name if use_model.model_name is None else use_model.model_name - - # 设置此次请求中的messages - messages = req_messages.copy() - - # 检查vision - for msg in messages: - 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"] - - args["messages"] = messages - - # 发送请求 - resp = await self._req(args) - - # 处理请求结果 - message = await self._make_msg(resp) - - return message - - async def call( - self, - query: core_entities.Query, - model: entities.LLMModelInfo, - messages: typing.List[llm_entities.Message], - funcs: typing.List[tools_entities.LLMFunction] = None, - ) -> llm_entities.Message: - req_messages = [] # req_messages 仅用于类内,外部同步由 query.messages 进行 - for m in messages: - msg_dict = m.dict(exclude_none=True) - content = msg_dict.get("content") - if isinstance(content, list): - # 检查 content 列表中是否每个部分都是文本 - if all(isinstance(part, dict) and part.get("type") == "text" for part in content): - # 将所有文本部分合并为一个字符串 - msg_dict["content"] = "\n".join(part["text"] for part in content) - req_messages.append(msg_dict) - - try: - return await self._closure(query=query, req_messages=req_messages, use_model=model, use_funcs=funcs) - except asyncio.TimeoutError: - raise errors.RequesterError('请求超时') diff --git a/pkg/provider/runnermgr.py b/pkg/provider/runnermgr.py index f45abfa4..4cd38c04 100644 --- a/pkg/provider/runnermgr.py +++ b/pkg/provider/runnermgr.py @@ -5,6 +5,7 @@ from ..core import app from .runners import localagent from .runners import difysvapi +from .runners import dashscopeapi class RunnerManager: diff --git a/pkg/provider/runners/dashscopeapi.py b/pkg/provider/runners/dashscopeapi.py new file mode 100644 index 00000000..8a6aef9d --- /dev/null +++ b/pkg/provider/runners/dashscopeapi.py @@ -0,0 +1,236 @@ +from __future__ import annotations + +import typing +import json +import base64 +import re + +import dashscope + +from .. import runner +from ...core import entities as core_entities +from .. import entities as llm_entities +from ...utils import image + +class DashscopeAPIError(Exception): + """Dashscope API 请求失败""" + + def __init__(self, message: str): + self.message = message + super().__init__(self.message) + + +@runner.runner_class("dashscope-service-api") +class DashScopeAPIRunner(runner.RequestRunner): + "阿里云百炼DashsscopeAPI对话请求器" + + # 运行器内部使用的配置 + app_type: str # 应用类型 + app_id: str # 应用ID + api_key: str # API Key + references_quote: str # 引用资料提示(当展示回答来源功能开启时,这个变量会作为引用资料名前的提示,可在provider.json中配置) + biz_params: dict = {} # 工作流应用参数(仅在工作流应用中生效) + + async def initialize(self): + """初始化""" + valid_app_types = ["agent", "workflow"] + self.app_type = self.ap.provider_cfg.data["dashscope-service-api"]["app-type"] + #检查配置文件中使用的应用类型是否支持 + if (self.app_type not in valid_app_types): + raise DashscopeAPIError( + f"不支持的 Dashscope 应用类型: {self.app_type}" + ) + + #初始化Dashscope 参数配置 + self.app_id = self.ap.provider_cfg.data["dashscope-service-api"][self.app_type]["app-id"] + self.api_key = self.ap.provider_cfg.data["dashscope-service-api"][self.app_type]["api-key"] + self.references_quote = self.ap.provider_cfg.data["dashscope-service-api"][self.app_type]["references_quote"] + self.biz_params = self.ap.provider_cfg.data["dashscope-service-api"]["workflow"]["biz_params"] + + def _replace_references(self, text, references_dict): + """阿里云百炼平台的自定义应用支持资料引用,此函数可以将引用标签替换为参考资料""" + + # 匹配 [index_id] 形式的字符串 + pattern = re.compile(r'\[(.*?)\]') + + def replacement(match): + # 获取引用编号 + ref_key = match.group(1) + if ref_key in references_dict: + # 如果有对应的参考资料按照provider.json中的reference_quote返回提示,来自哪个参考资料文件 + return f"({self.references_quote} {references_dict[ref_key]})" + else: + # 如果没有对应的参考资料,保留原样 + return match.group(0) + + # 使用 re.sub() 进行替换 + return pattern.sub(replacement, text) + + async def _preprocess_user_message( + self, query: core_entities.Query + ) -> tuple[str, list[str]]: + """预处理用户消息,提取纯文本,阿里云提供的上传文件方法过于复杂,暂不支持上传文件(包括图片)""" + plain_text = "" + image_ids = [] + if isinstance(query.user_message.content, list): + for ce in query.user_message.content: + if ce.type == "text": + plain_text += ce.text + # 暂时不支持上传图片,保留代码以便后续扩展 + # elif ce.type == "image_base64": + # image_b64, image_format = await image.extract_b64_and_format(ce.image_base64) + # file_bytes = base64.b64decode(image_b64) + # file = ("img.png", file_bytes, f"image/{image_format}") + # file_upload_resp = await self.dify_client.upload_file( + # file, + # f"{query.session.launcher_type.value}_{query.session.launcher_id}", + # ) + # image_id = file_upload_resp["id"] + # image_ids.append(image_id) + elif isinstance(query.user_message.content, str): + plain_text = query.user_message.content + + return plain_text, image_ids + + + async def _agent_messages( + self, query: core_entities.Query + ) -> typing.AsyncGenerator[llm_entities.Message, None]: + """Dashscope 智能体对话请求""" + + #局部变量 + chunk = None # 流式传输的块 + pending_content = "" # 待处理的Agent输出内容 + references_dict = {} # 用于存储引用编号和对应的参考资料 + plain_text = "" # 用户输入的纯文本信息 + image_ids = [] # 用户输入的图片ID列表 (暂不支持) + + plain_text, image_ids = await self._preprocess_user_message(query) + + #发送对话请求 + response = dashscope.Application.call( + api_key=self.api_key, # 智能体应用的API Key + app_id=self.app_id, # 智能体应用的ID + prompt=plain_text, # 用户输入的文本信息 + stream=True, # 流式输出 + incremental_output=True, # 增量输出,使用流式输出需要开启增量输出 + session_id=query.session.using_conversation.uuid, # 会话ID用于,多轮对话 + # rag_options={ # 主要用于文件交互,暂不支持 + # "session_file_ids": ["FILE_ID1"], # FILE_ID1 替换为实际的临时文件ID,逗号隔开多个 + # } + ) + + 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") + + #保存当前会话的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 llm_entities.Message( + role="assistant", + content=pending_content, + ) + + + async def _workflow_messages( + self, query: core_entities.Query + ) -> typing.AsyncGenerator[llm_entities.Message, None]: + """Dashscope 工作流对话请求""" + + #局部变量 + chunk = None # 流式传输的块 + pending_content = "" # 待处理的Agent输出内容 + references_dict = {} # 用于存储引用编号和对应的参考资料 + plain_text = "" # 用户输入的纯文本信息 + image_ids = [] # 用户输入的图片ID列表 (暂不支持) + + plain_text, image_ids = await self._preprocess_user_message(query) + + #发送对话请求 + response = dashscope.Application.call( + api_key=self.api_key, # 智能体应用的API Key + app_id=self.app_id, # 智能体应用的ID + prompt=plain_text, # 用户输入的文本信息 + stream=True, # 流式输出 + incremental_output=True, # 增量输出,使用流式输出需要开启增量输出 + session_id=query.session.using_conversation.uuid, # 会话ID用于,多轮对话 + biz_params=self.biz_params # 工作流应用的自定义输入参数传递 + # 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 + + #获取流式传输的output + stream_output = chunk.get("output", {}) + 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") + + #获取模型传出的参考资料列表 + 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 llm_entities.Message( + role="assistant", + content=pending_content, + ) + + + + async def run( + self, query: core_entities.Query + ) -> typing.AsyncGenerator[llm_entities.Message, None]: + """运行""" + if self.ap.provider_cfg.data["dashscope-service-api"]["app-type"] == "agent": + async for msg in self._agent_messages(query): + yield msg + elif self.ap.provider_cfg.data["dashscope-service-api"]["app-type"] == "workflow": + async for msg in self._workflow_messages(query): + yield msg + else: + raise DashscopeAPIError( + f"不支持的 Dashscope 应用类型: {self.ap.provider_cfg.data['dashscope-service-api']['app-type']}" + ) + + diff --git a/templates/provider.json b/templates/provider.json index 9db3c020..13044f1e 100644 --- a/templates/provider.json +++ b/templates/provider.json @@ -25,12 +25,6 @@ ], "siliconflow": [ "xxxxxxx" - ], - "dashscope": [ - "sk-1234567890" - ], - "qwen": [ - "sk-1234567890", ] }, "requester": { @@ -46,16 +40,6 @@ }, "timeout": 120 }, - "dashscope-chat-applications": { - "args": {}, - "base-url": "https://dashscope.aliyuncs.com/api/v1", - "timeout": 120 - }, - "qwen-chat-completions": { - "base-url": "https://dashscope.aliyuncs.com/compatible-mode/v1", - "args": {}, - "timeout": 120 - }, "moonshot-chat-completions": { "base-url": "https://api.moonshot.cn/v1", "args": {}, @@ -119,5 +103,22 @@ "output-key": "summary", "timeout": 120 } + }, + "dashscope-service-api": { + "agent": { + "api-key": "sk-1234567890", + "app-id": "Your_app_id", + "references_quote": "参考资料来自:" + }, + "app-type": "agent", + "workflow": { + "api-key": "sk-1234567890", + "app-id": "Your_app_id", + "references_quote": "参考资料来自:", + "biz_params": { + "city": "北京", + "date": "2023-08-10" + } + } } } \ No newline at end of file